Multi producer/consumer queue (without Boost) in C++11

Question

Not lock-free, but still only C++11 and no Boost. It also supports timeouts.

#pragma once

#include <cstddef>
#include <chrono>
#include <memory>
#include <queue>
#include <mutex>
#include <condition_variable>

template<typename T>
class blocking_queue {
public:
    explicit blocking_queue(std::size_t max_size) :_max_size(max_size), _q()
    {}
    blocking_queue(const blocking_queue&) = delete;
    blocking_queue& operator=(const blocking_queue&) = delete;
    ~blocking_queue() = default;

    std::size_t size()
    {
        std::lock_guard<std::mutex> lock(_mutex);
        return _q.size();
    }
    //Return false if failed to push due to full queue after the timeout have passed
    bool push(const T& item, const std::chrono::milliseconds& timeout)
    {
        std::unique_lock<std::mutex> ul(_mutex);
        if (_q.size() >= _max_size) {
            if (_item_popped_cond.wait_for(ul, timeout) == std::cv_status::timeout || _q.size() >= _max_size)
                return false;
        }
        _q.push(item);
        if (_q.size() <= 1)
            _item_pushed_cond.notify_all();
        return true;
    }
    //Return false if failed to pop due to empty queue after the timeout have passed
    bool pop(T& item, const std::chrono::milliseconds& timeout)
    {
        std::unique_lock<std::mutex> ul(_mutex);
        if (_q.empty()) {
            if (_item_pushed_cond.wait_for(ul, timeout) == std::cv_status::timeout || _q.empty())
                return false;
        }
        item = _q.front();
        _q.pop();
        if (_q.size() >= _max_size - 1)
            _item_popped_cond.notify_all();
        return true;
    }
private:
    std::size_t _max_size;
    std::queue<T> _q;
    std::mutex _mutex;
    std::condition_variable _item_pushed_cond;
    std::condition_variable _item_popped_cond;
};

Answer 1

Your size is not mutating. So should declare it const.

std::size_t size() const
     //            ^^^^^

Still in the size() I don't see any in point in lock on such short term. As soon as the function exits the value can be mutated and will thus be out of date (so if you get a value as another thread is mutating it you will be in no better state).

// There is no point in this function.
// It has no intrinsic value alone but can only be used within a set of commands.
// But since you lock per call, if you make multiple calls another thread can 
// mutate the object and make this value stale.
{
    std::lock_guard<std::mutex> lock(_mutex);
    return _q.size();
}

Not sure your timed push will work as expected:

if (_item_popped_cond.wait_for(ul, timeout) == std::cv_status::timeout || _q.size() >= _max_size)

Say you set a timeout of 10 seconds. Now another threads pops at 1 second and signals the condition variable. This can potentially release this thread (this does not guarantee your thread will require the lock before another third thread manages to call push). This means that your size would still be max (when you eventually reacquire the lock) and thus you exit with false after only one second.

You need to use the other version of wait_for() that has a test function.

if (!_item_popped_cond.wait_for(ul, timeout, [&q,&_max_size](){return _q.size() < _max_size;})) {
    return false;
}

Or put it inside a loop:

while(_q.size() >= _max_size) {
    if (_item_popped_cond.wait_for(ul, timeout) == std::cv_status::timeout) {
        return false;
    }
    // probably need to adjust timeout here.
}

Don't see the need to test before notifying after a push. Also why are you notifying all threads, why not just notify one (there is only one new item available).

if (_q.size() <= 1)                 // Why test. Always notify
    _item_pushed_cond.notify_all(); // notify_one();

All the same comments apply to pop.

I don't like the name of your identifier names. Technically nothing wrong with them. But there are so many situations where identifiers with an initial underscore is incorrect that I avoid them. And most C++ code follows the same rules. What are the rules about using an underscore in a C++ identifier?

Note: The std lib with the compiler usually uses underscores but they are specifically allowed to (as they are part of the implementation). This is where most beginners pick up this hobbit; But as you are writing code that is not part of the implementation the same rules do not appy and thus this is not a reason to follow that convention.

Expansion of my point on locking above

The only way to use the lock is:

std::size_t  count = que.size();   // locks getting size.
for(int loop = 0;loop < count; ++loop)
{
    que.pop();                     // locks while popping.
}

I see very little use for the size() method in this context. Between the call to the size() method and other methods your object could have been mutated thus making your results stale. Any function that queries state like this is not returning you anything useful because it can immediately by out of date as soon as the lock is released.

So what should you do?

Where you have operations that can be combined together into a larger piece of work. What you normally do is not provide the interface directly. You provide a way to lock the object that returns an interface that you can use (because while you hold the interface object you hold the lock).

class blocking_queue
{
      friend class blocking_queue_interface;
      // Private interface can not be used directly.
      std::size_t p_size() const;
      bool p_push(T& item, const std::chrono::milliseconds& timeout, std::lock_guard<std::mutex>& lock);
      bool p_pop(T& item, const std::chrono::milliseconds& timeout, std::lock_guard<std::mutex>& lock);

      ......
      public:
          // Public interface
          // Just gets lock and calls private interface.
          // Should only be used as a last resort.
          // Personally I would not provide this interface at all.
          std::size_t size() const
          {
              std::lock_guard<std::mutex> lock(_mutex);
              return p_size();
          }
          bool p_push(T& item, const std::chrono::milliseconds& timeout)
          {
              std::lock_guard<std::mutex> lock(_mutex);
              return p_push(item, timeout, lock);
          }
          bool p_pop(T& item, const std::chrono::milliseconds& timeout)
          {
              std::lock_guard<std::mutex> lock(_mutex);
              return p_pop(item, timeout, lock);
          }
 };

 // A locked interface.
 // You can acquire the lock and make multiple calls without
 // the data becoming stale between calls.
 class blocking_queue_interface
 {
     blocking_queue              queue;
     std::lock_guard<std::mutex> lock;

     blocking_queue_interface(blocking_queue& q): queue(q), lock(queue._mutex) {}
     ~blocking_queue_interface()                                               {}
     // delete remove all copy and assignment operators.

      // Calls the private version of these methods
      // as it has already aquired the lock
      std::size_t size() const                                     {return queue.p_size());
      bool push(T& item, const std::chrono::milliseconds& timeout) {return queue.p_push(item, timeout, lock);}
      bool pop(T& item, const std::chrono::milliseconds& timeout)  {return queue.p_pop(item, timeout, lock);}
 };

Answer 2

I'd say that size limitation you've put on is more like policy rather than container limitation. If you'd have ring-buffer as underline container that max_size will be justified.

I'd suggest you to split this into max_size policy applying adapter and unlimited multi-threaded queue. And while you have no resize you can mark _max_size as const.

Note as well that C++11 provide very nice way to construct elements right in containers. You may want to forward that feature (see emplace method of std::queue)

Also don't forget about movable semantic for constructor of your queue and for elements you are pushing in (bool push (T &&item /*, ...*/)).