task and async_manual_reset_event

.gitignore

@@ -30,3 +30,10 @@
 *.exe
 *.out
 *.app
+
+/build/
+/Debug/
+/RelWithDebInfo/
+/Release/
+
+/.vscode/

CMakeLists.txt

@@ -0,0 +1,30 @@
+cmake_minimum_required(VERSION 3.16)
+project(coro CXX)
+
+option(CORO_BUILD_TESTS    "Build the tests, Default=ON." ON)
+option(CORO_CODE_COVERAGE  "Enable code coverage, tests must also be enabled, Default=OFF" OFF)
+
+message("${PROJECT_NAME} CORO_BUILD_TESTS      = ${CORO_BUILD_TESTS}")
+message("${PROJECT_NAME} CORO_CODE_COVERAGE    = ${CORO_CODE_COVERAGE}")
+
+set(LIBCORO_SOURCE_FILES
+    src/coro/async_manual_reset_event.hpp
+    src/coro/coro.hpp
+    src/coro/task.hpp
+)
+
+add_library(${PROJECT_NAME} STATIC ${LIBCORO_SOURCE_FILES})
+set_target_properties(${PROJECT_NAME} PROPERTIES LINKER_LANGUAGE CXX)
+target_compile_features(${PROJECT_NAME} PUBLIC cxx_std_20)
+target_include_directories(${PROJECT_NAME} PUBLIC src)
+target_compile_options(${PROJECT_NAME} PUBLIC -fcoroutines)
+
+if(CORO_BUILD_TESTS)
+    if(CORO_CODE_COVERAGE)
+        target_compile_options(${PROJECT_NAME} PRIVATE --coverage)
+        target_link_libraries(${PROJECT_NAME} PRIVATE gcov)
+    endif()
+
+    enable_testing()
+    add_subdirectory(test)
+endif()

README.md

@@ -1 +1,3 @@
-# libcoro
+# libcoro
+
+This library is meant for learning the new C++20 coroutines.

src/coro/async_manual_reset_event.hpp

@@ -0,0 +1,122 @@
+#pragma once
+
+#include <coroutine>
+#include <optional>
+#include <atomic>
+
+namespace coro
+{
+
+template<typename return_type>
+class async_manual_reset_event
+{
+public:
+    async_manual_reset_event() noexcept
+        : m_state(nullptr)
+    {
+    }
+
+    async_manual_reset_event(const async_manual_reset_event&) = delete;
+    async_manual_reset_event(async_manual_reset_event&&) = delete;
+    auto operator=(const async_manual_reset_event&) -> async_manual_reset_event& = delete;
+    auto operator=(async_manual_reset_event&&) -> async_manual_reset_event& = delete;
+
+    bool is_set() const noexcept
+    {
+        return m_state.load(std::memory_order_acquire) == this;
+    }
+
+    struct awaiter
+    {
+        awaiter(const async_manual_reset_event& event) noexcept
+            : m_event(event)
+        {
+
+        }
+
+        auto await_ready() const noexcept -> bool
+        {
+            return m_event.is_set();
+        }
+
+        auto await_suspend(std::coroutine_handle<> awaiting_coroutine) noexcept -> bool
+        {
+            const void* const set_state = &m_event;
+
+            m_awaiting_coroutine = awaiting_coroutine;
+
+            // This value will update if other threads write to it via acquire.
+            void* old_value = m_event.m_state.load(std::memory_order_acquire);
+            do
+            {
+                // Resume immediately if already in the set state.
+                if(old_value == set_state)
+                {
+                    return false;
+                }
+
+                m_next = static_cast<awaiter*>(old_value);
+            } while(!m_event.m_state.compare_exchange_weak(
+                old_value,
+                this,
+                std::memory_order_release,
+                std::memory_order_acquire));
+
+            return true;
+        }
+
+        auto await_resume() noexcept
+        {
+
+        }
+
+        auto return_value() const & -> const return_type&
+        {
+            return m_event.m_return_value;
+        }
+
+        const async_manual_reset_event& m_event;
+        std::coroutine_handle<> m_awaiting_coroutine;
+        awaiter* m_next{nullptr};
+    };
+
+    auto operator co_await() const noexcept -> awaiter
+    {
+        return awaiter(*this);
+    }
+
+    auto set(return_type return_value) noexcept -> void
+    {
+        void* old_value = m_state.exchange(this, std::memory_order_acq_rel);
+        if(old_value != this)
+        {
+            m_return_value = std::move(return_value);
+
+            auto* waiters = static_cast<awaiter*>(old_value);
+            while(waiters != nullptr)
+            {
+                auto* next = waiters->m_next;
+                waiters->m_awaiting_coroutine.resume();
+                waiters = next;
+            }
+        }
+    }
+
+    auto reset() noexcept -> void
+    {
+        void* old_value = this;
+        m_state.compare_exchange_strong(old_value, nullptr, std::memory_order_acquire);
+    }
+
+    auto return_value() const -> const return_type&
+    {
+        return m_return_value;
+    }
+
+private:
+    friend struct awaiter;
+    return_type m_return_value;
+    mutable std::atomic<void*> m_state;
+};
+
+} // namespace coro

src/coro/coro.hpp

@@ -0,0 +1,2 @@
+#include "coro/async_manual_reset_event.hpp"
+#include "coro/task.hpp"

src/coro/task.hpp

@@ -0,0 +1,121 @@
+#pragma once
+
+#include <coroutine>
+#include <optional>
+
+namespace coro
+{
+
+template<
+    typename return_type,
+    typename initial_suspend_type = std::suspend_never,
+    typename final_suspend_type = std::suspend_never>
+class task
+{
+public:
+    struct promise_type
+    {
+        /// The return value for this promise.
+        return_type m_return_value;
+        /// If an exception was thrown it will be stored here and re-thrown on accessing the return value.
+        std::optional<std::exception_ptr> m_e;
+
+        using coro_handle = std::coroutine_handle<promise_type>;
+
+        auto get_return_object() -> task<return_type, initial_suspend_type, final_suspend_type>
+        {
+            return coro_handle::from_promise(*this);
+        }
+
+        auto initial_suspend()
+        {
+            return initial_suspend_type();
+        }
+
+        auto final_suspend()
+        {
+            return final_suspend_type();
+        }
+
+        auto return_void() -> void
+        {
+            // no-op
+        }
+
+        auto unhandled_exception() -> void
+        {
+            m_e = std::current_exception();
+        }
+
+        auto return_value(return_type value) -> void
+        {
+            m_return_value = std::move(value);
+        }
+
+        auto return_value() const & -> const return_type&
+        {
+            if(m_e.has_value())
+            {
+                std::rethrow_exception(m_e.value());
+            }
+
+            return m_return_value;
+        }
+
+        auto return_value() && -> return_type&&
+        {
+            if(m_e.has_value())
+            {
+                std::rethrow_exception(m_e.value());
+            }
+
+            return std::move(m_return_value);
+        }
+
+        auto yield_value(return_type value)
+        {
+            m_return_value = std::move(value);
+            return std::suspend_always();
+        }
+    };
+
+    using coro_handle = std::coroutine_handle<promise_type>;
+
+    task(coro_handle handle) : m_handle(std::move(handle))
+    {
+
+    }
+    task(const task&) = delete;
+    task(task&&) = delete;
+    auto operator=(const task&) -> task& = delete;
+    auto operator=(task&&) -> task& = delete;
+
+    auto is_done() const noexcept -> bool
+    {
+        return m_handle.done();
+    }
+
+    auto resume() -> bool
+    {
+        if(!m_handle.done())
+        {
+            m_handle.resume();
+        }
+        return !m_handle.done();
+    }
+
+    auto return_value() const & -> const return_type&
+    {
+        return m_handle.promise().return_value();
+    }
+
+    auto return_value() && -> return_type&&
+    {
+        return std::move(m_handle.promise()).return_value();
+    }
+
+private:
+    coro_handle m_handle;
+};
+
+} // namespace coro

test/CMakeLists.txt

@@ -0,0 +1,36 @@
+cmake_minimum_required(VERSION 3.16)
+project(libcoro_test)
+
+set(LIBCORO_TEST_SOURCE_FILES
+    test_async_manual_reset_event.cpp
+    test_task.cpp
+)
+
+add_executable(${PROJECT_NAME} main.cpp ${LIBCORO_TEST_SOURCE_FILES})
+target_compile_features(${PROJECT_NAME} PUBLIC cxx_std_20)
+target_link_libraries(${PROJECT_NAME} PRIVATE coro)
+target_compile_options(${PROJECT_NAME} PUBLIC -fcoroutines)
+
+if(CORO_CODE_COVERAGE)
+    target_compile_options(${PROJECT_NAME} PRIVATE --coverage)
+    target_link_libraries(${PROJECT_NAME} PRIVATE gcov)
+endif()
+
+if(${CMAKE_CXX_COMPILER_ID} MATCHES "GNU")
+    target_compile_options(
+        ${PROJECT_NAME} PRIVATE
+        -Wno-unknown-pragmas
+    )
+endif()
+if(${CMAKE_CXX_COMPILER_ID} MATCHES "Clang")
+    target_compile_options(${PROJECT_NAME} PRIVATE
+        -Wall
+        -Wextra
+        -Weffc++
+        -Werror
+        -Wpedantic
+        -pedantic-errors
+    )
+endif()
+
+add_test(NAME corohttp_test COMMAND ${PROJECT_NAME})

test/main.cpp

@@ -0,0 +1,2 @@
+#define CATCH_CONFIG_MAIN
+#include "catch.hpp"

test/test_async_manual_reset_event.cpp

@@ -0,0 +1,53 @@
+#include "catch.hpp"
+
+#include <coro/coro.hpp>
+
+#include <chrono>
+#include <thread>
+
+template<typename return_type>
+auto mre_producer(coro::async_manual_reset_event<return_type>& event, return_type produced_value) -> void
+{
+    // simulate complicated background task
+    using namespace std::chrono_literals;
+    std::this_thread::sleep_for(100ms);
+    event.set(std::move(produced_value));
+}
+
+template<typename return_type>
+auto mre_consumer(
+    const coro::async_manual_reset_event<return_type>& event
+) -> coro::task<return_type, std::suspend_never, std::suspend_always>
+{
+    co_await event;
+    co_return event.return_value();
+}
+
+TEST_CASE("manual reset event one watcher")
+{
+    coro::async_manual_reset_event<uint64_t> event{};
+
+    auto value = mre_consumer(event);
+    mre_producer<uint64_t>(event, 42);
+    value.resume();
+
+    REQUIRE(value.return_value() == 42);
+}
+
+TEST_CASE("manual reset event multiple watcher")
+{
+    std::string expected_value = "Hello World!";
+    coro::async_manual_reset_event<std::string> event{};
+
+    auto value1 = mre_consumer(event);
+    auto value2 = mre_consumer(event);
+    auto value3 = mre_consumer(event);
+    mre_producer<std::string>(event, expected_value);
+    value1.resume();
+    value2.resume();
+    value3.resume();
+
+    REQUIRE(value1.return_value() == expected_value);
+    REQUIRE(value2.return_value() == expected_value);
+    REQUIRE(value3.return_value() == expected_value);
+}

test/test_task.cpp

@@ -0,0 +1,55 @@
+#include "catch.hpp"
+
+#include <coro/coro.hpp>
+
+#include <chrono>
+#include <thread>
+
+static auto hello() -> coro::task<std::string, std::suspend_always>
+{
+    co_return "Hello";
+}
+
+static auto world() -> coro::task<std::string, std::suspend_always>
+{
+    co_return "World";
+}
+
+static auto throws_exception() -> coro::task<std::string, std::suspend_always>
+{
+    co_await std::suspend_always();
+    throw std::runtime_error("I'll be reached");
+    co_return "I'll never be reached";
+}
+
+TEST_CASE("hello world task")
+{
+    auto h = hello();
+    auto w = world();
+
+    REQUIRE(h.return_value().empty());
+    REQUIRE(w.return_value().empty());
+
+    h.resume(); // task suspends immediately
+    w.resume();
+
+    auto w_value = std::move(w).return_value();
+
+    REQUIRE(h.return_value() == "Hello");
+    REQUIRE(w_value == "World");
+    REQUIRE(w.return_value().empty());
+}
+
+TEST_CASE("Exception thrown")
+{
+    auto task = throws_exception();
+
+    try
+    {
+        task.resume();
+    }
+    catch(const std::exception& e)
+    {
+        REQUIRE(e.what() == "I'll be reached");
+    }
+}