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);
}
}
}
cv::imwrite()
. But as with all performance-related issues, you ought to benchmark both approaches on hardware that you consider representative. \$\endgroup\$