Multithreading, shared queue as synchronization point

Question

I have two threads, one produces images and one processes them. For the synchronization, I created a class where you can set and get images, and it always waits until an image is available or a worker thread is not busy anymore.

Additionally, a call to SetFinish stops both threads, while a call to Clear clears the currently (not yet processed) image.

Do you see any problems with this code (mainly threading issues)?

Header:

#ifndef SHARED_QUEUE_H_
#define SHARED_QUEUE_H_
#include <memory>
#include <condition_variable>
struct ImageData;
class SharedQueue {
public:
 void SetFinish();
 bool GetImage(std::shared_ptr<const ImageData> &image);
 void SetImage(std::shared_ptr<const ImageData> image);
 void Clear();
private:
 std::condition_variable image_available_;
 std::condition_variable image_processed_;
 std::shared_ptr<const ImageData> next_image_;
 bool stop_{false};
 std::mutex mutex_;
};
#endif // SHARED_QUEUE_H_

Implementation:

#include "shared_queue.h"
void SharedQueue::SetFinish() {
 { // Store flag and wake up the thread
 std::lock_guard<std::mutex> lock(mutex_);
 stop_ = true;
 }
 image_available_.notify_one();
 image_processed_.notify_one();
}
bool SharedQueue::GetImage(std::shared_ptr<const ImageData> &image) {
 {
 std::unique_lock<std::mutex> lock(mutex_);
 image_available_.wait(lock, [this]{
 return (next_image_.get() != nullptr || stop_);
 });
 if (stop_)
 return false;
 image = next_image_;
 next_image_.reset();
 }
 image_processed_.notify_one();
 return true;
}
void SharedQueue::SetImage(std::shared_ptr<const ImageData> image) {
 { // Store image for processing and wake up the thread
 std::unique_lock<std::mutex> lock(mutex_);
 image_processed_.wait(lock, [this]{
 return (next_image_.get() == nullptr || stop_);
 });
 if (stop_)
 return;
 next_image_ = image;
 }
 image_available_.notify_one();
}
void SharedQueue::Clear() {
 {
 std::unique_lock<std::mutex> lock(mutex_);
 next_image_.reset();
 }
 image_processed_.notify_one();
}

Answer 1

Not a Queue!

First and foremost, your SharedQueue isn't a queue. You can only store one element in it at a time. That doesn't make it super useful - what if the producer wants to write two images?

queue.setImage(img1);
queue.setImage(img2); // blocks?

It's more of a guarantee-one-at-a-time container. A queue would be much more useful, so I'd consider actually implementing one. This is a pretty major design flaw.

---

Beyond that, I just have minor comments.

Move semantics

You have a lot of copies where you can do moves. For instance, in SetImage():

next_image_ = image;

should be:

next_image_ = std::move(image);

Moving is cheaper than copying (no need to incur reference counting).

Checking shared_ptr

You don't need to use .get(), you can directly check the shared_ptr:

image_processed_.wait(lock, [this]{
 return !next_image_ || stop_;
});

Clear()

You use a std::unique_lock<> to Clear() where a std::lock_guard<> is sufficient. You use the correct one in SetFinish().

• Mat: Trivia: Java's SynchronousQueue<E> is even more extreme: " A synchronous queue does not have any internal capacity, not even a capacity of one."
• Barry: @Mat That is a weird "container". I guess if OP was trying to produce that one, my main comment is moot.
• Jan Rüegg: @Barry thanks a lot for your helpful comments! Yes I actually had a real queue in the beginning, but I changed it. The reason is that the producer is much faster than the worker getting the images, so if I have a queue, I get more and more latency and in the end the application runs out of memory. This way, I make sure that the producer and the consumer are in "sync", processing images at the same speed, and can at the same time still exploit the speedup of multithreading.