3
\$\begingroup\$

I've written concurrent queue based on std::queue.

#include <queue>
#include <shared_mutex>
#include <condition_variable>

template <typename T, typename S = typename std::queue<T>::container_type>
class concurrent_queue {
    // no std::shared_mutex yet
    using mutex_type = std::shared_timed_mutex;

public:
    explicit concurrent_queue(const S &sequence)
            : queue_(sequence) { }

    explicit concurrent_queue(S &&sequence = S())
            : queue_(std::move(sequence)) { }

    concurrent_queue(const concurrent_queue &other)
    {
        std::shared_lock<mutex_type> other_lock(other.mutex_);
        queue_ = other.queue_;
    }

    concurrent_queue(concurrent_queue &&other)
    {
        std::unique_lock<mutex_type> lock(mutex_, std::defer_lock);
        std::shared_lock<mutex_type> other_lock(other.mutex_, std::defer_lock);
        std::lock(lock, other_lock);
        queue_ = std::move(other.queue_);
    }

    concurrent_queue &operator=(const concurrent_queue &other)
    {
        if (this != &other) {
            std::unique_lock<mutex_type> lock(mutex_, std::defer_lock);
            std::shared_lock<mutex_type> other_lock(other.mutex_, std::defer_lock);
            std::lock(lock, other_lock);
            queue_ = other.queue_;
        }
        return *this;
    }

    concurrent_queue &operator=(concurrent_queue &&other)
    {
        if (this != &other) { // not needed?
            std::unique_lock<mutex_type> lock(mutex_, std::defer_lock);
            std::shared_lock<mutex_type> other_lock(other.mutex_, std::defer_lock);
            std::lock(lock, other_lock);
            queue_ = std::move(other.queue_);
        }
        return *this;
    }

    ~concurrent_queue() = default;

    T pop()
    {
        std::unique_lock<mutex_type> lock(mutex_);
        condition_variable_.wait(lock, [this] { return !queue_.empty(); });
        T item = queue_.front(); // use auto?
        queue_.pop();
        return item;
    }

    bool pop(T &item)
    {
        std::unique_lock<mutex_type> lock(mutex_);
        if (queue_.empty()) {
            return false;
        } else {
            item = queue_.front();
            queue_.pop();
            return true;
        }
    }

    void push(const T &item)
    {
        std::unique_lock<mutex_type> lock(mutex_);
        queue_.push(item);
        lock.unlock();
        condition_variable_.notify_one();
    }

    void push(T &&item)
    {
        std::unique_lock<mutex_type> lock(mutex_);
        queue_.push(std::move(item));
        lock.unlock();
        condition_variable_.notify_one();
    }

    template <typename ...Args>
    void emplace(Args &&... args)
    {
        std::unique_lock<mutex_type> lock(mutex_);
        queue_.emplace(std::forward<Args>(args)...);
        lock.unlock();
        condition_variable_.notify_one();
    }

private:
    mutable mutex_type mutex_;
    std::condition_variable_any condition_variable_;
    std::queue<T, S> queue_;
};

Things I'd like to know about this code:

  1. Possible deadlocks - it seems to be working fine in my program, but I'm only using one producer and one consumer. (see #3)
  2. Exception safety - I guess pop methods can throw, but I don't see how this could be avoided without hacks like external locking (lock-front-pop-unlock)
  3. Testing - I've done some really basic tests, things like multiple readers and writers. It has never crashed but I don't really know how to test it.
  4. Anything that comes to your mind.
\$\endgroup\$
1
\$\begingroup\$

The two pop() functions have very different semantics, this is probably more of a taste issue, but i'd normally like to see such a difference in a difference in function name. Yes the signature is different, but there isn't really anything that would tell a user that one is blocking and the other one isn't.

With the copy and move operations, it might help if you thought about the semantics, when does one usually copy an object, when is it moved. Does the expectation that a user has from a copy constructor or the move constructor match what you are doing. It sounds like with the move operator you have some different functionality in mind. If that is that case I'd normally put that in a separate function. Also think of how this queue should be used, is a move or a copy a valid operation.

|improve this answer|||||
\$\endgroup\$
0
\$\begingroup\$

Moving the queue in

 concurrent_queue(concurrent_queue &&other)

or in

 concurrent_queue &operator=(concurrent_queue &&other)

can be problematic if there are threads blocked. When they wake up, their instance of queue is not valid anymore. That may sound quite of a corner situation, but the problem is the client doing the move does not know if there are blocked threads ready to gain access to the -moved- queue as soon as the moving operation releases the lock.

You may try this on a test: run several producers and consumers and additionally a thread that exercises the move operations. All working on the same instance of concurrent_queue.

|improve this answer|||||
\$\endgroup\$
  • \$\begingroup\$ This situation shouldn't be different from queue just beeing empty. According to standard, moved-from objects are left in valid but unspecified state. I can't make it specified because std::queue doesn't have clear() method but for sure I can recreate it: other.queue_ = std::queue<T, S>(). I don't know if it's idiomatic and doesn't abuse the move operations but it is a way to "steal" data from queue. \$\endgroup\$ – Przeszczep Nov 20 '16 at 20:42
  • \$\begingroup\$ If it's a way to "steal" data it would probably be better to indicate that by using an explicit function to do exactly that. Overloading move like this could lead to a hard to debug situation where clients are waiting on a queue that has been moved, to somewhere else. queue.takeData(otherQueue) would be more explicit therefore tell readers of your code what's going on. \$\endgroup\$ – Harald Scheirich Feb 18 '17 at 21:07

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.