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.
---
 Jamroot.jam          |  5 ++++-
 application/main.cpp |  3 +++
 iwyu.json            |  8 ++++++++
 lib/work.h           | 15 +++++++++++++++
 lib/worker.cpp       | 46 ++++++++++++++++++++++++++++++++++++++++++++++
 lib/worker.h         | 46 ++++++++++++++++++++++++++++++++++++++++++++++
 6 files changed, 122 insertions(+), 1 deletion(-)
 create mode 100644 lib/work.h
 create mode 100644 lib/worker.cpp
 create mode 100644 lib/worker.h

diff --git a/Jamroot.jam b/Jamroot.jam
index 38a759d..397a167 100644
--- a/Jamroot.jam
+++ b/Jamroot.jam
@@ -1,4 +1,4 @@
-using gcc ;
+using gcc : 11.0.0 ;
 using pkg-config ;
 import pkg-config ;
 import type : register ;
@@ -7,6 +7,7 @@ import generators : register-standard ;
 pkg-config.import sdl2 ;
 pkg-config.import glew ;
 lib stb : : : : <include>/usr/include/stb ;
+lib pthread ;
 
 project : requirements
 	<cxxstd>20
@@ -48,7 +49,9 @@ exe test :
 	:
 	<include>.
 	<include>utility
+	<include>lib
 	<library>sdl2
 	<library>glew
+	<library>pthread
 	<use>stb
 	;
diff --git a/application/main.cpp b/application/main.cpp
index 73f337b..ed3c1c7 100644
--- a/application/main.cpp
+++ b/application/main.cpp
@@ -13,6 +13,7 @@
 #include <memory>
 #include <numbers>
 #include <special_members.hpp>
+#include <worker.h>
 
 static const int DISPLAY_WIDTH = 800;
 static const int DISPLAY_HEIGHT = 600;
@@ -60,6 +61,8 @@ public:
 		Collection<Window> windows;
 		windows.create(DISPLAY_WIDTH, DISPLAY_HEIGHT, "OpenGL");
 
+		Worker w;
+
 		World world;
 		world.create<Monkey>();
 
diff --git a/iwyu.json b/iwyu.json
index ccf35c5..ee63d1c 100644
--- a/iwyu.json
+++ b/iwyu.json
@@ -38,5 +38,13 @@
             "<glm/gtx/transform.hpp>",
             "public"
         ]
+    },
+    {
+        "symbol": [
+            "std::__atomic_wait",
+            "private",
+            "<mutex>",
+            "public"
+        ]
     }
 ]
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