Implement await_worker_queue and test custom awaitables

Author: wravery

include/graphqlservice/GraphQLService.h

@@ -148,18 +148,32 @@ enum class ResolverContext
 	NotifyUnsubscribe,
 };
 
-// Resume coroutine execution on a worker thread.
+// Resume coroutine execution on a new worker thread any time co_await is called. This emulates the
+// behavior of std::async when passing std::launch::async.
 struct await_worker_thread : coro::suspend_always
 {
-	void await_suspend(coro::coroutine_handle<> h) const
-	{
-		std::thread(
-			[](coro::coroutine_handle<>&& h) noexcept {
-				h.resume();
-			},
-			std::move(h))
-			.detach();
-	}
+	GRAPHQLSERVICE_EXPORT void await_suspend(coro::coroutine_handle<> h) const;
+};
+
+// Queue coroutine execution on a single dedicated worker thread any time co_await is called from
+// the thread which created it.
+struct await_worker_queue : coro::suspend_always
+{
+	GRAPHQLSERVICE_EXPORT await_worker_queue();
+	GRAPHQLSERVICE_EXPORT ~await_worker_queue();
+
+	GRAPHQLSERVICE_EXPORT bool await_ready() const;
+	GRAPHQLSERVICE_EXPORT void await_suspend(coro::coroutine_handle<> h);
+
+private:
+	void resumePending();
+
+	const std::thread::id _startId;
+	std::mutex _mutex {};
+	std::condition_variable _cv {};
+	std::list<coro::coroutine_handle<>> _pending {};
+	bool _shutdown = false;
+	std::thread _worker;
 };
 
 // Type-erased awaitable.

src/GraphQLService.cpp

@@ -164,6 +164,72 @@ response::Value schema_exception::getErrors()
 	return buildErrorValues(std::move(_structuredErrors));
 }
 
+void await_worker_thread::await_suspend(coro::coroutine_handle<> h) const
+{
+	std::thread(
+		[](coro::coroutine_handle<>&& h) {
+			h.resume();
+		},
+		std::move(h))
+		.detach();
+}
+
+await_worker_queue::await_worker_queue()
+	: _startId { std::this_thread::get_id() }
+	, _worker { [this]() {
+		resumePending();
+	} }
+{
+}
+
+await_worker_queue::~await_worker_queue()
+{
+	std::unique_lock lock { _mutex };
+
+	_shutdown = true;
+	lock.unlock();
+	_cv.notify_one();
+
+	_worker.join();
+}
+
+bool await_worker_queue::await_ready() const
+{
+	return std::this_thread::get_id() != _startId;
+}
+
+void await_worker_queue::await_suspend(coro::coroutine_handle<> h)
+{
+	std::unique_lock lock { _mutex };
+
+	_pending.push_back(std::move(h));
+	lock.unlock();
+	_cv.notify_one();
+}
+
+void await_worker_queue::resumePending()
+{
+	std::unique_lock lock { _mutex };
+
+	while (!_shutdown)
+	{
+		_cv.wait(lock, [this]() {
+			return _shutdown || !_pending.empty();
+		});
+
+		auto pending = std::move(_pending);
+
+		lock.unlock();
+
+		for (auto h : pending)
+		{
+			h.resume();
+		}
+
+		lock.lock();
+	}
+}
+
 FieldParams::FieldParams(SelectionSetParams&& selectionSetParams, Directives directives)
 	: SelectionSetParams(std::move(selectionSetParams))
 	, fieldDirectives(std::move(directives))

test/CMakeLists.txt

@@ -23,6 +23,14 @@ target_link_libraries(today_tests PRIVATE
 add_bigobj_flag(today_tests)
 gtest_add_tests(TARGET today_tests)
 
+add_executable(coroutine_tests CoroutineTests.cpp)
+target_link_libraries(coroutine_tests PRIVATE
+  todaygraphql
+  graphqljson
+  GTest::GTest
+  GTest::Main)
+gtest_add_tests(TARGET coroutine_tests)
+
 add_executable(client_tests ClientTests.cpp)
 target_link_libraries(client_tests PRIVATE
   todaygraphql