C++20 sort of infinite “Consumer-Producer”

Question

I think I can summarize the idea to the Producer-Consumer problem, with some modifications. And I think I misused the term "Producer" (it depends from which point of view :))

• An infinite consumer/producer produces a result from a given input
• The result is computed from another thread
• Only one product can be producted at a time.

That's it!

I was wondering if the code is OK, ESPECIALLY about thread-safety, and also about copy optimizations, C++ errors and so on.

#pragma once

#include <mutex>
#include <thread>
#include <condition_variable>

/**
 * Thread that infinitely make a task consuming each time a resource
 * When there is no more resource to consume, the thread exit.
 * When the thread is working, it cannot be canceled and wait the end of current operation to
 * ask if there is a pending request and see that there is no more pending request and also can end.
 */
template<typename Input, typename Output>
class ThreadConsumer
{
public:
    /**
     * Ensure cleanup before destruction
     */
    virtual ~ThreadConsumer()
    { stop(); }

    /**
     * Notify the consumer to shutdown and that no new input will be done.
     * If a task is currently running, wait the running task to finish before returns.
     * Used to join if a task is running before exiting, or free some output generated data.
     */
    void stop()
    {
        std::unique_lock lock(m_mutex);

        while(!m_waiting) {
            m_condition.wait(lock);
        }

        if(m_done) { // if zero tasks were accomplished, do not join the empty constructed default thread.
            m_thread.join(); // should returns immediately. Required & cleanup
        }
    }

    /**
     * @return true if the worker is waiting for an input resource to be processed.
     */
    bool ready() const
    {
        std::lock_guard lock(m_mutex);
        return m_waiting;
    }

    /**
     * Give a resource to the Thread. There is no process queue, the thread calling this function will wait
     * until the worker take the input. If the worker is waiting (that is ready() returned true in the current thread),
     * for an incoming resource, returns immediately.
     */
    void give(Input&& resource)
    {
        std::unique_lock lock(m_mutex);

        while(!m_waiting) {
            m_condition.wait(lock);
        }

        if(m_done) {
            m_thread.join(); // should return immediately. Required & cleanup
        }

        m_waiting = false;
        m_done = false;

        std::thread thread([&] {
            m_output = start(std::move(resource));

            std::lock_guard<std::mutex> lock(m_mutex);
            m_done = true;
            m_waiting = true;

            m_condition.notify_one();
        });

        m_thread = std::move(thread);
    }

    /**
     * @return true if the worker has finished a task and can provide an output result.
     * Not synonym for ready(): the only difference is just after construction of the consumer: at this time,
     * ready() returns true and done() returns false. In others cases, the two functions returns the same value.
     */
    bool done() const
    {
        std::lock_guard lock(m_mutex);
        return m_done;
    }

    /**
     * @return the output of the latest task. Do not check if the object is the one default-constructed with this
     * object. After at least one task finished, the output is always the result of a preceding task (unless moved from
     * caller).
     */
    Output& output()
    { return m_output; }

    const Output& output() const
    { return m_output; }

protected:
    virtual Output start(Input &&input) = 0;

private:
    /**
     * Result of last computation. Default-constructed if the consumer has not be launched one time.
     */
    Output m_output;

    /**
     * Protect all this class private fields except m_output that should be accessed only after a task finished,
     * also without concurrency.
     */
    mutable std::mutex m_mutex;
    std::condition_variable m_condition;

    /**
     * Represents current operation thread (if any)
     */
    std::thread m_thread;

    bool m_waiting = true;
    bool m_done = false;
};

template class ThreadConsumer<int, int>; // To debug syntax errors
```

Answer 1

I am confused about the design. Normally one reuses the same thread instead of creating one for each minor task. Thread creation is not cheap on most platforms.

1. void give(Input&& resource) will be clunky to use as input is an r-value reference which is inconvenient. In many template functions you see && used a lot but there it is most often interpreted as a universal/forwarding reference which accepts any input. In your case it will be r-values only, i.e., int a = 5; producer.give(a); will not compile and you'll have to write producer.give(std::move(a));. You should read more on r-values and forwarding references.

For 99% of the cases in should be preferable to have void give(Input resource) instead of void give(Input&& resource).

2. Okey,

std::thread thread([&] {
        m_output = start(std::move(resource)); // this is a bug

        std::lock_guard<std::mutex> lock(m_mutex);
        m_done = true;
        m_waiting = true;

        m_condition.notify_one();
    });

The operation might occur after leaving the function and destruction of resource which will make resource to be a dangling reference resulting in UB.

To fix it you can write it like this:

std::thread thread([this](Input res) {
        m_output = start(std::move(res)); // this is a bug

        std::lock_guard<std::mutex> lock(m_mutex);
        m_done = true;
        m_waiting = true;

        m_condition.notify_one();
    }, std::move(resource));

3. This isn't too good:

 std::lock_guard<std::mutex> lock(m_mutex);
 m_done = true;
 m_waiting = true;

 m_condition.notify_one();

You have the mutex locked while notifying another thread so it might result in "hurry up and wait" as it tries to lock the mutex. One should unlock the mutex prior to notifying.

4. About stopping:

void stop()
{
    std::unique_lock lock(m_mutex);

    while(!m_waiting) {
        m_condition.wait(lock);
    }

    if(m_done) { // if zero tasks were accomplished, do not join the empty constructed default thread.
        m_thread.join(); // should returns immediately. Required & cleanup
    }
}

You have lots of unnecessary code here. Just write:

void stop()
{
    if(m_thread.joinable()) m_thread.join();
}

Also the stop, doesn't actually do what the name implies - for what it does should be named wait() or something. stop would have to set the general state to "I refuse to get any more input".

P.S. don't know why you wrote C++20. There isn't any C++20 here.

Edit. also

  virtual ~ThreadConsumer()
  { stop(); }

Is a bug in design. Whatever class that derives from ThreadConsumer will first destroy its members and only then will trigger ~ThreadConsumer and subsequently stop() - leading to possible UB as members were likely destroyed before procedure finished.

---

Overall, I don't see much use for this ThreadConsumer class. It can be hard to figure out useful abstractions for multithreading. For myself, I figured messaging concept to be both most flexible and efficient.

What's messaging? You have a transmitter and receiver classes which act according to their names. So the whole ThreadConsumer can be trivially implemented via these two as:

std::thread([](receiver<Input> recv, transmitter<Output> trans, Func foo)
{
      Input in;
      while(recv.Receive(in)) // Receive returns false when communication ended.
      {
           if(not trans.Send(foo(in))) // Send forwards data, and returns false when communication is terminated.
           {
               return;
           } 
      }
}, ....);

You only need to figure out how to implement the messaging classes. I made mine via an additional shared control block class that manages the internal logic of how data transmission is performed between transmitter and receiver. Normally, one just needs a safe-thread queue of data but sometimes it is preferable to limit the size of the queue or forward data in different order according to some priorities or whatever. Or perhaps apply some minor conversion in between the operations (so that input type differs from output type).