Asynchronous file writer : queue, mutex, condition variable

Question

As we know, it is better to write to a disk in a separate thread because the bottleneck may be the disk, which is slower than the CPU (at the condition only one thread writes to the disk).

So I have implemented a "Consumer - Producer".

In fact, I realize I think it could be used for any generic task queue.

I would like to know if the implementation is OK (performance, undefined behaviour especially). And also I am not sure if the implementation of a limit size of the queue is well done. I don't know if m_condition.wait() on two separate threads is okay.

AsyncFileWriter.h

#pragma once

#include <thread>
#include <condition_variable>
#include <mutex>
#include <opencv2/core.hpp>
#include <queue>

/**
 * Write OpenCV asynchronously to a file.
 * Because writing to a disk is slow, this is to achieve running the computation and writing to a disk in parallel.
 *
 * This class stores one thread to do this job.
 * Creating a thread is not cheap, only one instance of this class is probably needed.
 */
class AsynFileWriter
{
public:
    /**
     * @param maxPendingOperations
     *      To prevent the thread to be too much delayed with the computation, you can set
     *      a limit of pending operations. If the limit is reached, adding another operation will be blocking until the
     *      count of pending operations is below the limit.
     */
    explicit AsynFileWriter(int maxPendingOperations = 10);

    /**
     * If this object is destroyed, the thread is joined.
     * That means it will be blocking until all pending operations are completed.
     */
    ~AsynFileWriter();

    /**
     * @{
     * Prevent copy and assignation.
     */
    AsynFileWriter(const AsynFileWriter&) = delete;
    AsynFileWriter& operator=(const AsynFileWriter&) = delete;
    AsynFileWriter&& operator=(AsynFileWriter&&) = delete;
    AsynFileWriter(AsynFileWriter&&) = delete;
    /**
     * @}
     */

    /**
     * Append a pending operation to the thread.
     * If the limit of pending operations is reached, this call is blocking until the count of pending operations
     * is below the limit.
     *
     * @note About memory management:
     *      We don't know when the file will be written. The caller function may have exited his scope.
     *      Consequently, the caller matrix maybe won't exist anymore. We have to ensure the matrix is still
     *      valid in the thread. To do so, we can just store a cv::Mat variable which uses a reference counter.
     *      HOWEVER, the caller can't cache the Matrix and change it because it could be change the original matrix
     *      we want to write. That's why this function ALWAYS makes a copy of the original matrix and keep its own copy.
     */
    void async_imwrite(const std::string& path, const cv::Mat& image);

private:
    void runInThread();


    struct Operation
    {
        std::string path;
        cv::Mat image;
    };

    // THE ORDER OF VARIABLES IS IMPORTANT: they are initialized in the order they appear in class
    // The thread is last, because if it is run immediately and m_condition and/or m_mutex is not initialized,
    // then it is undefined behavior and will probably crash

    int m_maxPendingOperations;
    std::queue<Operation> m_pendingOperations;
    bool m_noMoreOperations;
    std::mutex m_mutex; ///< For synchronization of m_pendingOperations
    std::condition_variable m_condition; ///< To tell the thread: Wake up!!! There is job to do!
    std::thread m_thread;
};

AsyncFileWriter.cpp

#include "AsynFileWriter.h"
#include <opencv2/imgcodecs.hpp>
#include <cassert>

AsynFileWriter::AsynFileWriter(int maxPendingOperations)
    : m_maxPendingOperations(maxPendingOperations),
      m_noMoreOperations(false),
      m_thread(&AsynFileWriter::runInThread, this)
{

}

AsynFileWriter::~AsynFileWriter()
{
    {
        std::unique_lock lock(m_mutex);
        m_noMoreOperations = true;

        m_condition.notify_one(); // Notify the thread "We want you to stop immediately"
    }

    m_thread.join();
}

void AsynFileWriter::async_imwrite(const std::string& path, const cv::Mat& image)
{
    std::unique_lock lock(m_mutex);
    m_pendingOperations.push(Operation{path, image.clone()});
    m_condition.notify_one(); // Notify the thread "We have job for you"

    if(m_pendingOperations.size() >= m_maxPendingOperations)
    {
        // Thread says: "I have too much work boss, please come later"
        m_condition.wait(lock, [this]() {
            return m_pendingOperations.size() < m_maxPendingOperations;
        });
    }

}

void AsynFileWriter::runInThread()
{
    bool shouldStop = false;
    Operation operation;

    while(!shouldStop)
    {
        {
            // Only lock once per file to write because locking is costly
            std::unique_lock lock(m_mutex);

            // Check if there is pending work
            if (!m_pendingOperations.empty())
            {
                // Get the next operation if there is one
                operation = m_pendingOperations.front();
                m_pendingOperations.pop();
            }
            else if(m_noMoreOperations)
            {
                // No more operations and all operations are completed, stop the thread
                shouldStop = true;
            }
            else
            {
                // If there is no pending work, sleep until work is ready OR we are asked to stop
                m_condition.wait(lock, [this]() {
                    return !m_pendingOperations.empty() || m_noMoreOperations;
                });

                // Get the information again
                shouldStop = m_noMoreOperations;
                if(!shouldStop)
                {
                    // If shouldStop == false, then necessarily there is pending work du to the wait() condition !empty()
                    assert(!m_pendingOperations.empty());

                    operation = m_pendingOperations.front();
                    m_pendingOperations.pop();
                }
            }
        }

        if(!shouldStop)
        {
            // Do the work: one line!!! (all for that...)
            cv::imwrite(operation.path, operation.image);
        }
    }
}

Answer 1

Split it into a thread-safe queue and a writer thread

So I have implemented a "Consumer - Producer".

In fact, I realize I think it could be used for any generic task queue.

Indeed, but by combining the queue with the image writer, you have made it more complex and at the same time less flexible. Consider splitting the code up into a generic thread-safe queue that doesn't have to know anything about image writing, and an image writer thread that just reads from the queue and doesn't have to know anything about thread safety. Ideally you would write code like this:

struct Operation {
    std::string path;
    cv::Mat image;
};

ThreadSafeQueue<Operation> imagesToWrite;

std::thread writerThread([&]{
    while (...) {
        auto operation = imagesToWrite.pop();
        cv::imWrite(operation.path, operation.image);
    }
});

I didn't address how to check if the writer thread should stop, but apart from that, it is possible to make a ThreadSafeQueue so that it's really that simple to write the above code.

Allow images to be moved instead of copied

Your async_imwrite() function takes an image by const reference. However, you need to put this in the queue somehow, so the only thing you can do then is to make a copy of it. But this can be very expensive if you have large images, and perhaps the caller doesn't need the image any more after they called async_imwrite(), so it would be better if the image could just be moved.

Consider writing an overload of async_imwrite() (or of push() if you implement a generic thread-safe queue) that takes the argument by r-value reference:

void async_imwrite(const std::string& path, cv::Mat&& image) {
    ...
    m_pendingOperations.emplace(path, std::move(image));
    ...
}

This will use cv::Mat's move constructor to efficiently move the image without having to make a copy of it.

You could avoid the resulting code duplication by using perfect forwarding.

Don't block unnecessarily

In async_imwrite(), you unconditionally push an image to the queue, and then check if the queue is full, and if so you block. But this causes you to wait unnecessarily if you don't have anything else to push. Instead, you should wait before you push.

Also consider that m_condition.wait() will first check the condition you give it, and only start waiting if it is false, so you don't need the if-statement around it. Finally, you can unlock the mutex before notifying, this avoids the write thread from receiving a signal but having to immediately wait again for the mutex to be unlocked:

void AsynFileWriter::async_imwrite(const std::string& path, const cv::Mat& image) {
    {
        std::unique_lock lock(m_mutex);
        m_condition.wait(lock, [this]() {
            return m_pendingOperations.size() < m_maxPendingOperations;
        });
        m_pendingOperations.emplace(path, image);
    }

    m_condition.notify_one();
}

Multiple threads waiting on the same condition variable

I don't know if m_condition.wait() on two separate threads is ok.

That is perfectly fine. Condition variables can be waited on by multiple threads. There's a notify_all() that should have given you a hint. However, you must make sure that every thread that waits gets notified at some point. I see a call to wait() in async_imwrite(), but no corresponding notify_one() in runInThread(). Make sure you add one there as well, so async_imwrite() gets woken up once the queue is no longer full.

Just return when m_noMoreOperations == true

Your runInThread() is quite complex, mainly because the checks for when to stop. Note that if you see that m_noMoreOperations == true, you can just immediately return, the lock object will automatically be released.

Also again, you are checking more often than necessary; just let m_condition.wait() do most of the work:

void AsynFileWriter::runInThread() {
    while (true) {
        Operation operation;

        {
            std::unique_lock lock(m_mutex);

            m_condition.wait(lock, [this]() {
                return !m_pendingOperations.empty() || m_noMoreOperations;
            });

            if (m_noMoreOperations && m_pendingOperations.empty()) {
               return;
            }

            operation = std::move(m_pendingOperations.front());
            m_pendingOperations.pop();
        }

        m_condition.notify_one();
        cv::imwrite(operation.path, operation.image);
    }
}

Race condition when destroying the AsynFileWriter

When the destructor of AsynFileWriter is called, you set m_noMoreOperations. It looks like runInThread() should empty the queue before exitting, however while it's blocked in wait(), it waits until !m_pendingOperations.empty() || m_noMoreOperations. Consider that another thread at that time might enqueue an element and set m_noMoreOperations in quick succession. Once runInThread() wakes up, it then checks if m_noMoreOperations is set, and if so quits, but there could still be elements in the queue. Make sure you only quit if the queue is really empty.

Naming

AsynFileWriter looks like it should have been AsyncFileWriter.

You also have some inconsistencies in how you name things: async_imwrite() uses snake_case, but runInThread() used camelCase.

Doxygen comments

I see you documented some of the functions and parameters using something that looks like Doxygen. However, it is not done very consistently. Ideally, every function and member variable is documented. Function documentation should start with a one-line @brief summary, and every parameter should be documented using @param. You can configure Doxygen to warn about all these things, and make sure you run doxygen on your code once in a while. If you are using a build system, make it run doxygen automatically.

Error handling

Consider that an error might occur when cv::imwrite() is called, either because it couldn't convert the image data to the desired file format, or because there was an I/O error writing to the file. cv::imwrite() returns a bool to indicate success or failure, and the documentation suggests that it can also throw an cv::Exception as well.

If an error happens and it didn't throw an exception, then you will have silently not written the file you wanted. If an exception is throw, you are not catching it, so this will cause your whole program to be terminated. Is either behavior OK?

Make sure you handle errors appropriately. Consider that the user might want to be informed that something bad happened, you can do this by writing an error message to std::cerr. If the program is running as part of a larger script, you want to make sure it exits with a non-zero exit code (preferrably EXIT_FAILURE), so the script can detect that something went wrong. Finally, it might be that your program might want to be able to do something when a file could not be written. Consider storing the result of the asynchronous actions in some way, for example by returning a std::future when enqueueing an image.