C++11 Blocking Queue learning exercise

Question

As part of my own C++11 learning exercise I have implemented this Blocking Queue using the new C++11 thread API. One of the aspects I would like to improve is efficiency i.e. in the take member function while returning the Job template type to be returned by reference instead of by value.

/*
 * blockingqueue.h
 *
 *  Created on: May 20, 2016
 *      Author: bravegag
 */

#ifndef BLOCKINGQUEUE_H_
#define BLOCKINGQUEUE_H_

#include <condition_variable>
#include <mutex>
#include <thread>
#include <iostream>
#include <queue>
#include <chrono>
#include <cassert>

using namespace std;

template<class Job>
class blocking_queue {
private:
    queue<Job> queue_;
    mutex mutex_;
    condition_variable not_empty_cond_;
    condition_variable not_full_cond_;
    int capacity_ = 0;

public:
    blocking_queue(int capacity) : capacity_(capacity) {
        // empty
    }

    void put(const Job& job) {
        unique_lock<mutex> lock(mutex_);
        while (queue_.size() >= capacity_) {
            cout << "queue full, waiting for jobs to be taken" << endl;
            not_full_cond_.wait(lock, [&]() { return (queue_.size() < capacity_); } );
        }
        queue_.push(job);
        not_empty_cond_.notify_one();
        assert(!queue_.empty());
    }

    Job take() {
        unique_lock<mutex> lock(mutex_);
        while (queue_.empty()) {
            cout << "queue empty, waiting for jobs to be put" << endl;
            not_empty_cond_.wait(lock, [&]() { return !queue_.empty(); });
        }
        Job job = queue_.front();
        queue_.pop();
        not_full_cond_.notify_one();
        assert(queue_.size() < capacity_);
        return job;
    }

    int size() const {
        return queue_.size();
    }

    virtual ~blocking_queue() {
        // empty
    }
};

#endif /* BLOCKINGQUEUE_H_ */

Answer 1

One of the aspects I would like to improve is efficiency i.e. in the take member function while returning the Job template type to be returned by reference instead of by value.

So what you would like is:

Job& take()   ?

the problem is that you would return a reference to local variable which is Undefined Behaviour.

Just Job take() should be fine, compiler will use NRVO optimization so that no copy will be done, copy constructor call would be elided. You could return by moving: return std::move(job);, but that can prevent elision.

One place where you will have a copy constructor called is here:

Job job = queue_.front();

To prevent it you can implement move semantics into your Job class. Then replace above with:

Job job = std::move(queue_.front());

if you will not add any of the user defined constructor/destructor/assignment operator to your Job class, then compiler will add move constructor/assignment operator on its own - depending whether it is possible.

Answer 2

Some smaller issues in addition to marcinj's Job job = std::move(queue_.front()); (which is a big thing):

• The whiles are misleading - the loops will either run once or not at all, as the predicate-wait will only return if the predicate returns true. Use if.

• It's more efficient to call notify_one without holding the lock. Note that this means the asserts will have to be moved upwards.

• Virtual destructors are useless unless you are planning to use blocking_queue as a polymorphic base class. If you are worried about idiots, mark the class final.

Answer 3

Efficiency

I've made a number of tries at this myself. Based on them, I've concluded a few things. First of all, that take should usually have a timeout available.

Applying that here, we end up with take looking something like this:

template <class duration>
bool take(Job &job, duration const &d) {
    unique_lock<mutex> lock(mutex_);
    while (queue_.empty()) {
        std::cout << "queue empty, waiting for jobs to be put\n";
        if (!not_empty_cond_.wait_for(lock, d, [&]() { return !queue_.empty(); }))
            return false;
    }
    job = queue_.front();
    queue_.pop();
    not_full_cond_.notify_one();
    assert(queue_.size() < capacity_);
    return true;
}

This also gives good efficiency (regardless of optimization level) since we copy the data item directly from the source to its final destination.

Separation of Concerns

Although I suppose it's only for testing/education, it may be worth pointing out that the queue has no business printing anything to standard output.

Race Condition

Since your size() doesn't obtain the mutex, it can have a data race if any other thread is writing to the queue when it's called.

Fixing that doesn't really help matters much though. It does eliminate the undefined behavior, but almost any code that attempts to use it is still subject to a race condition, since the size might change between the time it called size() and the time it attempts to make any use of the size it retrieved.

You probably want to just eliminate this member altogether.

Adding timeout to put

You may want to add a timeout to your put. This makes it more symmetrical with the take, and can help deal with situations where (for example) you want to quit processing based on some criteria determined by the consumer (then with a long enough timeout on put, when the producer threads timeout, they "know" they should exit, move on to other processing, etc.

Answer 4

The only effective way to use references is to pass a Job& to the take function, as Jerry Coffin pointed out.

I suggest to offer a bool try_take(Job&) variant (maybe with an overload accepting time outs) which returns immediately false if the queue is empty, and a Job take() that blocks indefinitely if you need real blocking behaviour.

The same behaviour should be mirrored by put (try_put)