Nikilite/suyu
1
1
Fork 0
forked from suyu/suyu
Code Pull requests Activity

common/thread_worker: Use unique function

Browse source
This commit is contained in:
ReinUsesLisp 2021-04-01 01:05:45 -03:00
parent 2c8d337418
commit bf5b5c1bf4
2 changed files with 25 additions and 29 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
src/common/thread_worker.cpp
View file

@ -8,36 +8,30 @@
namespace Common {
ThreadWorker::ThreadWorker(std::size_t num_workers, const std::string& name) {
for (std::size_t i = 0; i < num_workers; ++i)
threads.emplace_back([this, thread_name{std::string{name}}] {
Common::SetCurrentThreadName(thread_name.c_str());
const auto lambda = [this, thread_name{std::string{name}}] {
Common::SetCurrentThreadName(thread_name.c_str());
// Wait for first request
while (!stop) {
UniqueFunction<void> task;
{
std::unique_lock lock{queue_mutex};
condition.wait(lock, [this] { return stop || !requests.empty(); });
}
while (true) {
std::function<void()> task;
{
std::unique_lock lock{queue_mutex};
condition.wait(lock, [this] { return stop || !requests.empty(); });
if (stop || requests.empty()) {
return;
}
task = std::move(requests.front());
requests.pop();
if (requests.empty()) {
wait_condition.notify_one();
}
if (requests.empty()) {
wait_condition.notify_all();
}
task();
condition.wait(lock, [this] { return stop || !requests.empty(); });
if (stop || requests.empty()) {
break;
}
task = std::move(requests.front());
requests.pop();
}
});
task();
}
wait_condition.notify_all();
};
for (size_t i = 0; i < num_workers; ++i) {
threads.emplace_back(lambda);
}
}
ThreadWorker::~ThreadWorker() {
@ -51,10 +45,10 @@ ThreadWorker::~ThreadWorker() {
}
}
void ThreadWorker::QueueWork(std::function<void()>&& work) {
void ThreadWorker::QueueWork(UniqueFunction<void> work) {
{
std::unique_lock lock{queue_mutex};
requests.emplace(work);
requests.emplace(std::move(work));
}
condition.notify_one();
}

6
src/common/thread_worker.h
View file

@ -11,18 +11,20 @@
#include <vector>
#include <queue>
#include "common/unique_function.h"
namespace Common {
class ThreadWorker final {
public:
explicit ThreadWorker(std::size_t num_workers, const std::string& name);
~ThreadWorker();
void QueueWork(std::function<void()>&& work);
void QueueWork(UniqueFunction<void> work);
void WaitForRequests();
private:
std::vector<std::thread> threads;
std::queue<std::function<void()>> requests;
std::queue<UniqueFunction<void>> requests;
std::mutex queue_mutex;
std::condition_variable condition;
std::condition_variable wait_condition;