util-threadpool: Implement global thread pool

This thread pool can take large or small tasks and as such alleviates the burden of having a thread per source. Particularly for large setups, this drastically reduces the number of threads running in the background waiting for work.

CMakeLists.txt

@@ -364,6 +364,8 @@ set(PROJECT_PRIVATE_SOURCE
 	"${PROJECT_SOURCE_DIR}/source/utility.hpp"
 	"${PROJECT_SOURCE_DIR}/source/utility.cpp"
 	"${PROJECT_SOURCE_DIR}/source/util-event.hpp"
+	"${PROJECT_SOURCE_DIR}/source/util-threadpool.cpp"
+	"${PROJECT_SOURCE_DIR}/source/util-threadpool.hpp"
 	
 	# Graphics
 	"${PROJECT_SOURCE_DIR}/source/gfx/gfx-source-texture.hpp"

source/plugin.cpp

@@ -35,10 +35,14 @@
 #include "sources/source-mirror.hpp"
 #include "sources/source-shader.hpp"
 
+static std::shared_ptr<util::threadpool> global_threadpool;
+
 MODULE_EXPORT bool obs_module_load(void)
 try {
 	LOG_INFO("Loading Version %s", STREAMFX_VERSION_STRING);
 
+	global_threadpool = std::make_shared<util::threadpool>();
+
 	// Initialize Source Tracker
 	obs::source_tracker::initialize();
 
@@ -90,6 +94,8 @@ try {
 
 	// Finalize Source Tracker
 	obs::source_tracker::finalize();
+
+	global_threadpool.reset();
 } catch (...) {
 	LOG_ERROR("Unexpected exception in function '%s'.", __FUNCTION_NAME__);
 }
@@ -105,3 +111,8 @@ BOOL WINAPI DllMain(HINSTANCE, DWORD, LPVOID)
 	return TRUE;
 }
 #endif
+
+std::shared_ptr<util::threadpool> get_global_threadpool()
+{
+	return global_threadpool;
+}

source/plugin.hpp

@@ -20,6 +20,7 @@
 #pragma once
 #include "strings.hpp"
 #include "version.hpp"
+#include "util-threadpool.hpp"
 
 // OBS
 #ifdef _MSC_VER
@@ -46,3 +47,6 @@
 #define __FUNCTION_NAME__ __func__
 #endif
 #endif
+
+// Threadpool
+std::shared_ptr<util::threadpool> get_global_threadpool();

source/util-threadpool.cpp

@@ -0,0 +1,71 @@
+/*
+ * Modern effects for a modern Streamer
+ * Copyright (C) 2020 Michael Fabian Dirks
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA
+*/
+
+#include "util-threadpool.hpp"
+
+// Most Tasks likely wait for IO, so we can use that time for other tasks.
+#define CONCURRENCY_MULTIPLIER 2
+
+util::threadpool::threadpool() : _workers(), _tasks(), _tasks_lock(), _tasks_cv(), _worker_stop(false)
+{
+	size_t concurrency = static_cast<size_t>(std::thread::hardware_concurrency()) * CONCURRENCY_MULTIPLIER;
+	for (size_t n = 0; n < concurrency; n++) {
+		_workers.emplace_back(std::bind(&util::threadpool::work, this));
+	}
+}
+
+util::threadpool::~threadpool()
+{
+	_worker_stop = true;
+	_tasks_cv.notify_all();
+	for (auto& thread : _workers) {
+		_tasks_cv.notify_all();
+		if (thread.joinable())
+			thread.join();
+	}
+}
+
+void util::threadpool::push(threadpool_function_t fn, std::shared_ptr<void> data)
+{
+	std::unique_lock<std::mutex> lock(_tasks_lock);
+	_tasks.emplace_back(fn, data);
+	_tasks_cv.notify_one();
+}
+
+void util::threadpool::work()
+{
+	std::pair<threadpool_function_t, std::shared_ptr<void>> work;
+
+	while (!_worker_stop) {
+		// Wait for more work, or immediately continue if there is still work to do.
+		{
+			std::unique_lock<std::mutex> lock(_tasks_lock);
+			if (_tasks.size() == 0)
+				_tasks_cv.wait(lock, [this]() { return _worker_stop || _tasks.size() > 0; });
+			if (_tasks.size() == 0)
+				continue;
+			work = _tasks.front();
+			_tasks.pop_front();
+		}
+
+		// Execute work.
+		if (work.first)
+			work.first(work.second);
+	}
+}

source/util-threadpool.hpp

@@ -0,0 +1,46 @@
+/*
+ * Modern effects for a modern Streamer
+ * Copyright (C) 2020 Michael Fabian Dirks
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA
+ */
+
+#pragma once
+#include <atomic>
+#include <functional>
+#include <list>
+#include <mutex>
+#include <thread>
+
+namespace util {
+	typedef std::function<void(std::shared_ptr<void>)> threadpool_function_t;
+
+	class threadpool {
+		std::list<std::thread>                                             _workers;
+		std::atomic_bool                                                   _worker_stop;
+		std::list<std::pair<threadpool_function_t, std::shared_ptr<void>>> _tasks;
+		std::mutex                                                         _tasks_lock;
+		std::condition_variable                                            _tasks_cv;
+
+		public:
+		threadpool();
+		~threadpool();
+
+		void push(threadpool_function_t fn, std::shared_ptr<void> data);
+
+		private:
+		void work();
+	};
+} // namespace util