mirror of
https://github.com/YACReader/yacreader
synced 2025-05-28 03:10:27 -04:00
61cd245037
ConcurrentQueue is currently used only by two classes and a test, but modifying concurrent_queue.h requires recompiling 30 source files. None of the member functions is so lightweight as to make it worth inlining. An alternative to `@note ConcurrentQueue is unable to execute jobs if @p threadCount == 0.` is `assert(threadCount != 0);`. But this would force classes that contain a ConcurrentQueue data member to always start a thread, even if they detect at runtime that they are never going to enqueue a job. Add Job type alias to avoid repeating the type. Use default member initializers instead of the member initializer list to make it clear [to the reader of the header] that no data member is left uninitialized.
128 lines
2.4 KiB
C++
128 lines
2.4 KiB
C++
#include "concurrent_queue.h"
|
|
|
|
#include <cassert>
|
|
#include <cstddef>
|
|
#include <mutex>
|
|
#include <utility>
|
|
|
|
using namespace YACReader;
|
|
|
|
ConcurrentQueue::ConcurrentQueue(std::size_t threadCount)
|
|
{
|
|
threads.reserve(threadCount);
|
|
for (; threadCount != 0; --threadCount)
|
|
threads.emplace_back(&ConcurrentQueue::nextJob, this);
|
|
}
|
|
|
|
ConcurrentQueue::~ConcurrentQueue()
|
|
{
|
|
joinAll();
|
|
}
|
|
|
|
void ConcurrentQueue::enqueue(Job job)
|
|
{
|
|
{
|
|
std::lock_guard<std::mutex> lock(jobsLeftMutex);
|
|
++jobsLeft;
|
|
}
|
|
|
|
{
|
|
std::lock_guard<std::mutex> lock(queueMutex);
|
|
_queue.emplace(std::move(job));
|
|
}
|
|
|
|
jobAvailableVar.notify_one();
|
|
}
|
|
|
|
std::size_t ConcurrentQueue::cancelPending()
|
|
{
|
|
decltype(_queue) oldQueue;
|
|
{
|
|
const std::lock_guard<std::mutex> lock(queueMutex);
|
|
// The mutex locking time is lower with swap() compared to assigning a
|
|
// temporary (which destroys _queue's elements and deallocates memory).
|
|
_queue.swap(oldQueue);
|
|
}
|
|
|
|
const auto size = oldQueue.size();
|
|
if (size != 0)
|
|
finalizeJobs(size);
|
|
return size;
|
|
}
|
|
|
|
void ConcurrentQueue::waitAll()
|
|
{
|
|
std::unique_lock<std::mutex> lock(jobsLeftMutex);
|
|
if (jobsLeft > 0) {
|
|
_waitVar.wait(lock, [this] {
|
|
return jobsLeft == 0;
|
|
});
|
|
}
|
|
}
|
|
|
|
void ConcurrentQueue::nextJob()
|
|
{
|
|
while (true) {
|
|
Job job;
|
|
|
|
{
|
|
std::unique_lock<std::mutex> lock(queueMutex);
|
|
|
|
if (bailout) {
|
|
return;
|
|
}
|
|
|
|
jobAvailableVar.wait(lock, [this] {
|
|
return _queue.size() > 0 || bailout;
|
|
});
|
|
|
|
if (bailout) {
|
|
return;
|
|
}
|
|
|
|
job = std::move(_queue.front());
|
|
_queue.pop();
|
|
}
|
|
|
|
job();
|
|
finalizeJobs(1);
|
|
}
|
|
}
|
|
|
|
void ConcurrentQueue::finalizeJobs(std::size_t count)
|
|
{
|
|
assert(count > 0);
|
|
|
|
std::size_t remainingJobs;
|
|
{
|
|
std::lock_guard<std::mutex> lock(jobsLeftMutex);
|
|
assert(jobsLeft >= count);
|
|
jobsLeft -= count;
|
|
remainingJobs = jobsLeft;
|
|
}
|
|
|
|
if (remainingJobs == 0)
|
|
_waitVar.notify_all();
|
|
}
|
|
|
|
void ConcurrentQueue::joinAll()
|
|
{
|
|
{
|
|
std::lock_guard<std::mutex> lock(queueMutex);
|
|
|
|
if (bailout) {
|
|
return;
|
|
}
|
|
|
|
bailout = true;
|
|
}
|
|
|
|
jobAvailableVar.notify_all();
|
|
|
|
for (auto &x : threads) {
|
|
if (x.joinable()) {
|
|
x.join();
|
|
}
|
|
}
|
|
}
|