From 96a58cc251354954241aa2c9008b25d98daaad92 Mon Sep 17 00:00:00 2001
From: Dan Goodliffe <dan@randomdan.homeip.net>
Date: Tue, 19 Jan 2021 20:35:29 +0000
Subject: Add basic work and worker thread pool

Well... that requires GCC 11 cos 10 doesn't implement semaphore.
---
 lib/work.h     | 15 +++++++++++++++
 lib/worker.cpp | 46 ++++++++++++++++++++++++++++++++++++++++++++++
 lib/worker.h   | 46 ++++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 107 insertions(+)
 create mode 100644 lib/work.h
 create mode 100644 lib/worker.cpp
 create mode 100644 lib/worker.h

(limited to 'lib')

diff --git a/lib/work.h b/lib/work.h
new file mode 100644
index 0000000..2374673
--- /dev/null
+++ b/lib/work.h
@@ -0,0 +1,15 @@
+#ifndef WORK_H
+#define WORK_H
+
+#include <special_members.hpp>
+
+class Work {
+public:
+	virtual ~Work() = default;
+	NO_COPY(Work);
+	NO_MOVE(Work);
+
+	virtual void doWork() = 0;
+};
+
+#endif
diff --git a/lib/worker.cpp b/lib/worker.cpp
new file mode 100644
index 0000000..3c12caa
--- /dev/null
+++ b/lib/worker.cpp
@@ -0,0 +1,46 @@
+#include "worker.h"
+#include "work.h"
+#include <algorithm>
+#include <iterator>
+#include <mutex>
+
+Worker::Worker() : todoLen {0}
+{
+	std::generate_n(std::back_inserter(threads), std::thread::hardware_concurrency(), [this]() {
+		return std::thread {&Worker::worker, this};
+	});
+}
+
+Worker::~Worker()
+{
+	todoLen.release(std::thread::hardware_concurrency());
+	std::for_each(threads.begin(), threads.end(), [](auto & th) {
+		th.join();
+	});
+}
+
+void
+Worker::addWork(WorkPtr j)
+{
+	std::lock_guard<std::mutex> lck {todoMutex};
+	todoLen.release();
+	todo.emplace_back(std::move(j));
+}
+
+void
+Worker::worker()
+{
+	auto job = [this]() {
+		todoLen.acquire();
+		std::lock_guard<std::mutex> lck {todoMutex};
+		if (todo.size()) {
+			WorkPtr x = std::move(todo.front());
+			todo.pop_front();
+			return x;
+		}
+		return WorkPtr {};
+	};
+	while (auto j = job()) {
+		j->doWork();
+	}
+}
diff --git a/lib/worker.h b/lib/worker.h
new file mode 100644
index 0000000..5a13138
--- /dev/null
+++ b/lib/worker.h
@@ -0,0 +1,46 @@
+#ifndef WORKER_H
+#define WORKER_H
+
+#include <deque>
+#include <memory>
+#include <mutex>
+#include <semaphore>
+#include <special_members.hpp>
+#include <thread>
+#include <utility>
+#include <vector>
+
+class Work;
+
+class Worker {
+public:
+	Worker();
+	~Worker();
+
+	NO_COPY(Worker);
+	NO_MOVE(Worker);
+
+	using WorkPtr = std::unique_ptr<Work>;
+
+	template<typename T, typename... Params>
+	void
+	addWork(Params &&... params)
+	{
+		addWork(std::make_unique<T>(std::forward<Params>(params)...));
+	}
+
+	void addWork(WorkPtr w);
+
+private:
+	void worker();
+
+	using Threads = std::vector<std::thread>;
+	using ToDo = std::deque<WorkPtr>;
+
+	Threads threads;
+	ToDo todo;
+	std::counting_semaphore<16> todoLen;
+	std::mutex todoMutex;
+};
+
+#endif
-- 
cgit v1.2.3