Merge branch 'worker' into assimp

9 files changed, 254 insertions, 79 deletions

diff --git a/assetFactory/assetFactory.cpp b/assetFactory/assetFactory.cpp
index 46b4642..e27e575 100644
--- a/assetFactory/assetFactory.cpp
+++ b/assetFactory/assetFactory.cpp
@@ -99,21 +99,25 @@ AssetFactory::createTexutre() const
 {
 	if (!textureFragments.empty() && !texture) {
 		// * layout images
-		std::vector<TexturePacker::Image> imageSizes;
+		std::map<std::string_view, std::unique_ptr<const Image>> images;
 		std::transform(
-				textureFragments.begin(), textureFragments.end(), std::back_inserter(imageSizes), [](const auto & tf) {
-					return TexturePacker::Image {tf.second->image->width, tf.second->image->height};
+				textureFragments.begin(), textureFragments.end(), std::inserter(images, images.end()), [](auto && tf) {
+					return decltype(images)::value_type {tf.first, tf.second->image->get()};
 				});
+		std::vector<TexturePacker::Image> imageSizes;
+		std::transform(images.begin(), images.end(), std::back_inserter(imageSizes), [](const auto & i) {
+			return TexturePacker::Image {i.second->width, i.second->height};
+		});
 		const auto [layout, outSize] = TexturePacker {imageSizes}.pack();
 		// * create texture
 		texture = std::make_shared<TextureAtlas>(outSize.x, outSize.y, layout.size());
-		std::transform(textureFragments.begin(), textureFragments.end(),
+		std::transform(images.begin(), images.end(),
 				std::inserter(textureFragmentPositions, textureFragmentPositions.end()),
-				[position = layout.begin(), size = imageSizes.begin(), this](const auto & tf) mutable {
-					const auto m = texture->add(*position, *size, tf.second->image->data.data());
+				[position = layout.begin(), size = imageSizes.begin(), this](const auto & i) mutable {
+					const auto m = texture->add(*position, *size, i.second->data.data());
 					position++;
 					size++;
-					return decltype(textureFragmentPositions)::value_type {tf.first, m};
+					return decltype(textureFragmentPositions)::value_type {i.first, m};
 				});
 	}
 }
diff --git a/assetFactory/assimp.cpp b/assetFactory/assimp.cpp
index 878e7e7..bab052e 100644
--- a/assetFactory/assimp.cpp
+++ b/assetFactory/assimp.cpp
@@ -102,18 +102,14 @@ AssImp::postLoad()
 					return AssetFactory::Shapes::value_type {m->mName.C_Str(), std::make_shared<AssImpNode>(scene, m)};
 				});
 		const auto textures = AIRANGE(scene, Textures);
-		auto textureFutures = textures * [](const aiTexture * t) {
-			return std::async(std::launch::async, [t]() {
-				auto texture = std::make_shared<TextureFragment>();
-				texture->id = texture->path = t->mFilename.C_Str();
-				texture->image = std::make_unique<Image>(
-						std::span {reinterpret_cast<unsigned char *>(t->pcData), t->mWidth}, STBI_rgb_alpha);
-				return texture;
-			});
-		};
-		std::transform(textureFutures.begin(), textureFutures.end(),
-				std::inserter(mf->textureFragments, mf->textureFragments.end()), [](auto && textureFuture) {
-					auto texture = textureFuture.get();
+		std::transform(textures.begin(), textures.end(),
+				std::inserter(mf->textureFragments, mf->textureFragments.end()), [](auto && t) {
+					auto texture = std::make_shared<TextureFragment>();
+					texture->id = texture->path = t->mFilename.C_Str();
+					texture->image = Worker::addWork([t]() {
+						return std::make_unique<Image>(
+								std::span {reinterpret_cast<unsigned char *>(t->pcData), t->mWidth}, STBI_rgb_alpha);
+					});
 					return AssetFactory::TextureFragments::value_type {texture->id, texture};
 				});
 	}
diff --git a/assetFactory/textureFragment.cpp b/assetFactory/textureFragment.cpp
index d153688..0a4ec1d 100644
--- a/assetFactory/textureFragment.cpp
+++ b/assetFactory/textureFragment.cpp
@@ -11,5 +11,7 @@ TextureFragment::persist(Persistence::PersistenceStore & store)
 void
 TextureFragment::postLoad()
 {
-	image = std::make_unique<Image>(Resource::mapPath(path), STBI_rgb_alpha);
+	image = Worker::addWork([this]() {
+		return std::make_unique<Image>(Resource::mapPath(path), STBI_rgb_alpha);
+	});
 }
diff --git a/assetFactory/textureFragment.h b/assetFactory/textureFragment.h
index d03affd..75fe96e 100644
--- a/assetFactory/textureFragment.h
+++ b/assetFactory/textureFragment.h
@@ -3,12 +3,13 @@
 #include "gfx/image.h"
 #include "persistence.h"
 #include "stdTypeDefs.hpp"
+#include "worker.h"
 
 class TextureFragment : public Persistence::Persistable, public StdTypeDefs<TextureFragment> {
 public:
 	std::string id;
 	std::string path;
-	std::unique_ptr<Image> image;
+	Worker::WorkPtrT<std::unique_ptr<Image>> image;
 
 private:
 	friend Persistence::SelectionPtrBase<Ptr>;
diff --git a/lib/work.h b/lib/work.h
deleted file mode 100644
index c780e13..0000000
--- a/lib/work.h
+++ /dev/null
@@ -1,12 +0,0 @@
-#pragma once
-
-#include <special_members.hpp>
-
-class Work {
-public:
-	virtual ~Work() = default;
-	NO_COPY(Work);
-	NO_MOVE(Work);
-
-	virtual void doWork() = 0;
-};
diff --git a/lib/worker.cpp b/lib/worker.cpp
index fd255c7..45fb6df 100644
--- a/lib/worker.cpp
+++ b/lib/worker.cpp
@@ -1,27 +1,24 @@
 #include "worker.h"
-#if __cpp_lib_semaphore
-#	include "work.h"
-#	include <algorithm>
-#	include <iterator>
-#	include <mutex>
+#include <algorithm>
+#include <iterator>
+#include <mutex>
+
+Worker Worker::instance;
 
 Worker::Worker() : todoLen {0}
 {
 	std::generate_n(std::back_inserter(threads), std::thread::hardware_concurrency(), [this]() {
-		return std::thread {&Worker::worker, this};
+		return std::jthread {&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)
+Worker::addWorkPtr(WorkPtr j)
 {
 	std::lock_guard<std::mutex> lck {todoMutex};
 	todoLen.release();
@@ -45,4 +42,24 @@ Worker::worker()
 		j->doWork();
 	}
 }
-#endif
+
+void
+Worker::assist()
+{
+	auto job = [this]() {
+		using namespace std::chrono_literals;
+		if (todoLen.try_acquire_for(100us)) {
+			if (std::lock_guard<std::mutex> lck {todoMutex}; todo.size()) {
+				WorkPtr x = std::move(todo.front());
+				if (x) {
+					todo.pop_front();
+				}
+				return x;
+			}
+		}
+		return WorkPtr {};
+	};
+	if (auto j = job()) {
+		j->doWork();
+	}
+}
diff --git a/lib/worker.h b/lib/worker.h
index 7cd06f9..d9a5a6f 100644
--- a/lib/worker.h
+++ b/lib/worker.h
@@ -1,60 +1,122 @@
 #pragma once
 
+#include <deque>
+#include <functional>
+#include <future>
+#include <memory>
+#include <mutex>
+#include <semaphore>
+#include <special_members.hpp>
+#include <thread>
 #include <utility>
-class Work;
-
-#if __cpp_lib_semaphore
-#	include <deque>
-#	include <memory>
-#	include <mutex>
-#	include <semaphore>
-#	include <special_members.hpp>
-#	include <thread>
-#	include <vector>
+#include <vector>
 
 class Worker {
 public:
+	class WorkItem {
+	protected:
+		WorkItem(Worker * worker) : worker {worker} { }
+		virtual ~WorkItem() = default;
+		NO_MOVE(WorkItem);
+		NO_COPY(WorkItem);
+
+		void
+		assist() const
+		{
+			worker->assist();
+		}
+		Worker * worker;
+
+	public:
+		virtual void doWork() = 0;
+	};
+
+	template<typename T> class WorkItemT : public WorkItem {
+	public:
+		T
+		get()
+		{
+			using namespace std::chrono_literals;
+			while (future.wait_for(0s) == std::future_status::timeout) {
+				assist();
+			}
+			return future.get();
+		}
+
+	protected:
+		WorkItemT(Worker * worker) : WorkItem {worker} { }
+
+		std::promise<T> promise;
+		std::future<T> future {promise.get_future()};
+		friend Worker;
+	};
+
+	template<typename... Params>
+	static auto
+	addWork(Params &&... params)
+	{
+		return instance.addWorkImpl(std::forward<Params>(params)...);
+	}
+	template<typename T> using WorkPtrT = std::shared_ptr<WorkItemT<T>>;
+
+private:
+	template<typename T, typename... Params> class WorkItemTImpl : public WorkItemT<T> {
+	public:
+		WorkItemTImpl(Worker * worker, Params &&... params) :
+			WorkItemT<T> {worker}, params {std::forward<Params>(params)...}
+		{
+		}
+
+	private:
+		void
+		doWork() override
+		{
+			try {
+				if constexpr (std::is_void_v<T>) {
+					std::apply(std::invoke<Params &...>, params);
+					WorkItemT<T>::promise.set_value();
+				}
+				else {
+					WorkItemT<T>::promise.set_value(std::apply(std::invoke<Params &...>, params));
+				}
+			}
+			catch (...) {
+				WorkItemT<T>::promise.set_exception(std::current_exception());
+			}
+		}
+
+		std::tuple<Params...> params;
+	};
+
 	Worker();
 	~Worker();
 
 	NO_COPY(Worker);
 	NO_MOVE(Worker);
 
-	using WorkPtr = std::unique_ptr<Work>;
+	using WorkPtr = std::shared_ptr<WorkItem>;
 
-	template<typename T, typename... Params>
-	void
-	addWork(Params &&... params)
-		requires std::is_base_of_v<Work, T>
+	template<typename... Params>
+	auto
+	addWorkImpl(Params &&... params)
 	{
-		addWork(std::make_unique<T>(std::forward<Params>(params)...));
+		using T = decltype(std::invoke(std::forward<Params>(params)...));
+		auto work = std::make_shared<WorkItemTImpl<T, Params...>>(this, std::forward<Params>(params)...);
+		addWorkPtr(work);
+		return work;
 	}
 
-	void addWork(WorkPtr w);
-
-private:
+	void addWorkPtr(WorkPtr w);
 	void worker();
+	void assist();
 
-	using Threads = std::vector<std::thread>;
+	using Threads = std::vector<std::jthread>;
 	using ToDo = std::deque<WorkPtr>;
 
-	Threads threads;
 	ToDo todo;
 	std::counting_semaphore<16> todoLen;
 	std::mutex todoMutex;
-};
-
-#else
+	Threads threads;
 
-class Worker {
-public:
-	template<typename T, typename... Params>
-	void
-	addWork(Params &&... params)
-		requires std::is_base_of_v<Work, T>
-	{
-		T(std::forward<Params>(params)...).doWork();
-	}
+	static Worker instance;
 };
-
-#endif
diff --git a/test/Jamfile.jam b/test/Jamfile.jam
index 390880d..482b388 100644
--- a/test/Jamfile.jam
+++ b/test/Jamfile.jam
@@ -57,6 +57,7 @@ run test-glContextBhvr.cpp ;
 run test-assetFactory.cpp : -- : [ sequence.insertion-sort [ glob-tree $(res) : *.* ] fixtures/rgb.txt ] : <library>test ;
 run perf-assetFactory.cpp : : : <library>benchmark <library>test <dependency>test-assetFactory ;
 run perf-persistence.cpp : : : <library>benchmark <library>test <dependency>test-persistence ;
+run test-worker.cpp ;
 compile test-static-enumDetails.cpp ;
 compile test-static-stream_support.cpp ;
 explicit perf-assetFactory ;
diff --git a/test/test-worker.cpp b/test/test-worker.cpp
new file mode 100644
index 0000000..76b7138
--- /dev/null
+++ b/test/test-worker.cpp
@@ -0,0 +1,104 @@
+#define BOOST_TEST_MODULE test_worker
+
+#include "testHelpers.h"
+#include <boost/test/unit_test.hpp>
+#include <set>
+#include <stream_support.hpp>
+#include <worker.h>
+
+uint32_t
+workCounter()
+{
+	static std::atomic_uint32_t n;
+	usleep(1000);
+	return n++;
+}
+
+void
+workVoid()
+{
+	usleep(1000);
+}
+
+void
+workFail()
+{
+	usleep(1000);
+	throw std::runtime_error {"test"};
+}
+
+BOOST_AUTO_TEST_CASE(basic_slow_counter)
+{
+	std::vector<Worker::WorkPtrT<uint32_t>> ps;
+	for (int i {}; i < 30; ++i) {
+		ps.push_back(Worker::addWork(workCounter));
+	}
+	std::set<uint32_t> out;
+	std::transform(ps.begin(), ps.end(), std::inserter(out, out.end()), [](auto && p) {
+		return p->get();
+	});
+	BOOST_REQUIRE_EQUAL(out.size(), ps.size());
+	BOOST_CHECK_EQUAL(*out.begin(), 0);
+	BOOST_CHECK_EQUAL(*out.rbegin(), ps.size() - 1);
+}
+
+BOOST_AUTO_TEST_CASE(basic_error_handler)
+{
+	auto workitem = Worker::addWork(workFail);
+	BOOST_CHECK_THROW(workitem->get(), std::runtime_error);
+}
+
+BOOST_AUTO_TEST_CASE(basic_void_work)
+{
+	auto workitem = Worker::addWork(workVoid);
+	BOOST_CHECK_NO_THROW(workitem->get());
+}
+
+BOOST_AUTO_TEST_CASE(lambda_void)
+{
+	BOOST_CHECK_NO_THROW(Worker::addWork([]() {})->get());
+	BOOST_CHECK_NO_THROW(Worker::addWork([](int) {}, 0)->get());
+	BOOST_CHECK_NO_THROW(Worker::addWork([](int, int) {}, 0, 0)->get());
+}
+
+BOOST_AUTO_TEST_CASE(lambda_value)
+{
+	BOOST_CHECK_EQUAL(1, Worker::addWork([]() {
+		return 1;
+	})->get());
+	BOOST_CHECK_EQUAL(2,
+			Worker::addWork(
+					[](int i) {
+						return i;
+					},
+					2)
+					->get());
+	BOOST_CHECK_EQUAL(3,
+			Worker::addWork(
+					[](int i, int j) {
+						return i + j;
+					},
+					1, 2)
+					->get());
+}
+
+BOOST_AUTO_TEST_CASE(recursive, *boost::unit_test::timeout(5))
+{
+	auto recurse = []() {
+		std::vector<Worker::WorkPtrT<uint32_t>> ps;
+		for (int i {}; i < 30; ++i) {
+			ps.push_back(Worker::addWork(workCounter));
+		}
+		return std::accumulate(ps.begin(), ps.end(), 0U, [](auto && out, auto && p) {
+			return out += p->get();
+		});
+	};
+	std::vector<Worker::WorkPtrT<uint32_t>> ps;
+	for (int i {}; i < 30; ++i) {
+		ps.push_back(Worker::addWork(recurse));
+	}
+	std::set<uint32_t> out;
+	std::transform(ps.begin(), ps.end(), std::inserter(out, out.end()), [](auto && p) {
+		return p->get();
+	});
+}