Commit e88a4669 authored by gabi's avatar gabi

blocking queue to notify one and not all

parent be9dd21f
...@@ -69,65 +69,60 @@ public: ...@@ -69,65 +69,60 @@ public:
template<typename Duration_Rep, typename Duration_Period, typename TT> template<typename Duration_Rep, typename Duration_Period, typename TT>
bool push(TT&& item, const std::chrono::duration<Duration_Rep, Duration_Period>& timeout) bool push(TT&& item, const std::chrono::duration<Duration_Rep, Duration_Period>& timeout)
{ {
std::unique_lock<std::mutex> ul(_mutex);
if (_q.size() >= _max_size)
{ {
if (!_item_popped_cond.wait_until(ul, clock::now() + timeout,[this]() std::unique_lock<std::mutex> ul(_mutex);
{ if (is_full())
return this->_q.size() < this->_max_size; {
})) if (!_item_popped_cond.wait_until(ul, clock::now() + timeout, [this]() {
return false; return !this->is_full();
} }))
_q.push(std::forward<TT>(item)); return false;
if (_q.size() <= 1) }
{ _q.push(std::forward<TT>(item));
ul.unlock(); //So the notified thread will have better chance to accuire the lock immediatly..
_item_pushed_cond.notify_all();
} }
_item_pushed_cond.notify_one();
return true; return true;
} }
// Push copy of item into the back of the queue. // Push copy of item into the back of the queue.
// If the queue is full, block the calling thread until there is room. // If the queue is full, block the calling thread until there is room.
template<typename TT> template<typename TT>
void push(TT&& item) void push(TT&& item)
{ {
while (!push(std::forward<TT>(item), std::chrono::seconds(10))); while (!push(std::forward<TT>(item), std::chrono::seconds(60)));
} }
// Pop a copy of the front item in the queue into the given item ref. // Pop a copy of the front item in the queue into the given item ref.
// If the queue is empty, block the calling thread util there is item to pop or timeout have passed. // If the queue is empty, block the calling thread util there is item to pop or timeout have passed.
// Return: false on timeout , true on successful pop/ // Return: false on timeout , true on successful pop/
template<class Duration_Rep, class Duration_Period> template<class Duration_Rep, class Duration_Period>
bool pop(T& item, const std::chrono::duration<Duration_Rep, Duration_Period>& timeout) bool pop(T& item, const std::chrono::duration<Duration_Rep, Duration_Period>& timeout)
{ {
std::unique_lock<std::mutex> ul(_mutex);
if (_q.empty())
{
if (!_item_pushed_cond.wait_until(ul, clock::now() + timeout, [this]()
{
return !this->_q.empty();
}))
return false;
}
item = std::move(_q.front());
_q.pop();
if (_q.size() >= _max_size - 1)
{ {
ul.unlock(); //So the notified thread will have better chance to accuire the lock immediatly.. std::unique_lock<std::mutex> ul(_mutex);
_item_popped_cond.notify_all(); if (is_empty())
{
if (!_item_pushed_cond.wait_until(ul, clock::now() + timeout, [this]()
{
return !this->is_empty();
}))
return false;
}
item = std::move(_q.front());
_q.pop();
} }
_item_popped_cond.notify_one();
return true; return true;
} }
// Pop a copy of the front item in the queue into the given item ref. // Pop a copy of the front item in the queue into the given item ref.
// If the queue is empty, block the calling thread util there is item to pop. // If the queue is empty, block the calling thread util there is item to pop.
void pop(T& item) void pop(T& item)
{ {
while (!pop(item, std::chrono::hours(1))); while (!pop(item, std::chrono::hours(1)));
} }
// Clear the queue // Clear the queue
void clear() void clear()
{ {
{ {
...@@ -143,6 +138,16 @@ private: ...@@ -143,6 +138,16 @@ private:
std::mutex _mutex; std::mutex _mutex;
std::condition_variable _item_pushed_cond; std::condition_variable _item_pushed_cond;
std::condition_variable _item_popped_cond; std::condition_variable _item_popped_cond;
inline bool is_full()
{
return _q.size() >= _max_size;
}
inline bool is_empty()
{
return _q.size() == 0;
}
}; };
} }
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment