IOCP support (draft)

.gitignore

@@ -36,5 +36,7 @@
 /RelWithDebInfo/
 /Release/
 /Testing/
+/out/
 
 /.vscode/
+/.vs/

CMakeLists.txt

@@ -45,7 +45,7 @@ if (LIBCORO_RUN_GITCONFIG)
     )
 endif()
 
-cmake_dependent_option(LIBCORO_FEATURE_NETWORKING "Include networking features, Default=ON." ON "NOT EMSCRIPTEN; NOT MSVC" OFF)
+cmake_dependent_option(LIBCORO_FEATURE_NETWORKING "Include networking features, Default=ON." ON "NOT EMSCRIPTEN" OFF)
 cmake_dependent_option(LIBCORO_FEATURE_TLS "Include TLS encryption features, Default=ON." ON "NOT EMSCRIPTEN; NOT MSVC" OFF)
 
 message("${PROJECT_NAME} LIBCORO_ENABLE_ASAN           = ${LIBCORO_ENABLE_ASAN}")
@@ -100,6 +100,7 @@ set(LIBCORO_SOURCE_FILES
     include/coro/generator.hpp
     include/coro/latch.hpp
     include/coro/mutex.hpp src/mutex.cpp
+    include/coro/platform.hpp
     include/coro/queue.hpp
     include/coro/ring_buffer.hpp
     include/coro/semaphore.hpp src/semaphore.cpp
@@ -116,6 +117,7 @@ if(LIBCORO_FEATURE_NETWORKING)
     list(APPEND LIBCORO_SOURCE_FILES
         include/coro/detail/poll_info.hpp
         include/coro/detail/timer_handle.hpp src/detail/timer_handle.cpp
+        include/coro/signal.hpp
 
         include/coro/fd.hpp
         include/coro/io_scheduler.hpp src/io_scheduler.cpp
@@ -126,11 +128,20 @@ if(LIBCORO_FEATURE_NETWORKING)
     if(LINUX)
         list(APPEND LIBCORO_SOURCE_FILES
             include/coro/detail/io_notifier_epoll.hpp src/detail/io_notifier_epoll.cpp
+            include/coro/detail/signal_unix.hpp src/detail/signal_unix.cpp
         )
     endif()
     if(MACOSX)
         list(APPEND LIBCORO_SOURCE_FILES
             include/coro/detail/io_notifier_kqueue.hpp src/detail/io_notifier_kqueue.cpp
+            include/coro/detail/signal_unix.hpp src/detail/signal_unix.cpp
+        )
+    endif()
+    if(WIN32)
+        list(APPEND LIBCORO_SOURCE_FILES
+            include/coro/detail/io_notifier_iocp.hpp src/detail/io_notifier_iocp.cpp
+            include/coro/detail/signal_win32.hpp src/detail/signal_win32.cpp
+            include/coro/detail/winsock_handle.hpp src/detail/winsock_handle.cpp
         )
     endif()
 
@@ -142,6 +153,8 @@ if(LIBCORO_FEATURE_NETWORKING)
         include/coro/net/ip_address.hpp src/net/ip_address.cpp
         include/coro/net/recv_status.hpp src/net/recv_status.cpp
         include/coro/net/send_status.hpp src/net/send_status.cpp
+        include/coro/net/write_status.hpp
+        include/coro/net/read_status.hpp
         include/coro/net/socket.hpp src/net/socket.cpp
         include/coro/net/tcp/client.hpp src/net/tcp/client.cpp
         include/coro/net/tcp/server.hpp src/net/tcp/server.cpp

examples/CMakeLists.txt

@@ -6,6 +6,7 @@ if(${CMAKE_CXX_COMPILER_ID} MATCHES "GNU")
 elseif(${CMAKE_CXX_COMPILER_ID} MATCHES "Clang")
     set(LIBCORO_EXAMPLE_OPTIONS -Wall -Wextra -pipe)
 elseif(MSVC)
+    add_compile_definitions(NOMINMAX)
     set(LIBCORO_EXAMPLE_OPTIONS /W4)
 else()
     message(FATAL_ERROR "Unsupported compiler.")

examples/coro_tcp_echo_server.cpp

@@ -10,19 +10,13 @@ auto main() -> int
 
             while (true)
             {
-                // Wait for data to be available to read.
-                co_await client.poll(coro::poll_op::read);
-                auto [rstatus, rspan] = client.recv(buf);
+                auto [rstatus, rspan] = co_await client.read(buf);
                 switch (rstatus)
                 {
-                    case coro::net::recv_status::ok:
-                        // Make sure the client socket can be written to.
-                        co_await client.poll(coro::poll_op::write);
-                        client.send(std::span<const char>{rspan});
+                    case coro::net::read_status::ok:
+                        co_await client.write(rspan);
                         break;
-                    case coro::net::recv_status::would_block:
-                        break;
-                    case coro::net::recv_status::closed:
+                    case coro::net::read_status::closed:
                     default:
                         co_return;
                 }
@@ -34,24 +28,14 @@ auto main() -> int
 
         while (true)
         {
-            // Wait for a new connection.
-            auto pstatus = co_await server.poll();
-            switch (pstatus)
+            auto client = co_await server.accept_client();
+            if (client && client->socket().is_valid())
             {
-                case coro::poll_status::event:
-                {
-                    auto client = server.accept();
-                    if (client.socket().is_valid())
-                    {
-                        scheduler->spawn(make_on_connection_task(std::move(client)));
-                    } // else report error or something if the socket was invalid or could not be accepted.
-                }
-                break;
-                case coro::poll_status::error:
-                case coro::poll_status::closed:
-                case coro::poll_status::timeout:
-                default:
-                    co_return;
+                scheduler->spawn(make_on_connection_task(std::move(*client)));
+            }
+            else
+            {
+                co_return;
             }
         }
 

include/coro/concepts/executor.hpp

@@ -3,9 +3,13 @@
 #include "coro/concepts/awaitable.hpp"
 #include "coro/fd.hpp"
 #include "coro/task.hpp"
+#include "coro/platform.hpp"
 
 #ifdef LIBCORO_FEATURE_NETWORKING
     #include "coro/poll.hpp"
+#if defined(CORO_PLATFORM_WINDOWS)
+    #include "coro/detail/poll_info.hpp"
+#endif
 #endif // #ifdef LIBCORO_FEATURE_NETWORKING
 
 #include <chrono>
@@ -30,11 +34,19 @@ concept executor = requires(executor_type e, std::coroutine_handle<> c)
 };
 
 #ifdef LIBCORO_FEATURE_NETWORKING
+#if defined(CORO_PLATFORM_UNIX)
 template<typename executor_type>
 concept io_executor = executor<executor_type> and requires(executor_type e, std::coroutine_handle<> c, fd_t fd, coro::poll_op op, std::chrono::milliseconds timeout)
 {
     { e.poll(fd, op, timeout) } -> std::same_as<coro::task<poll_status>>;
 };
+#elif defined(CORO_PLATFORM_WINDOWS)
+template<typename executor_type>
+concept io_executor = executor<executor_type> and requires(executor_type e, coro::detail::poll_info pi, std::chrono::milliseconds timeout)
+{
+    { e.poll(pi, timeout) } -> std::same_as<coro::task<poll_status>>;
+};
+#endif
 #endif // #ifdef LIBCORO_FEATURE_NETWORKING
 
 // clang-format on

include/coro/detail/io_notifier_epoll.hpp

@@ -13,6 +13,7 @@
 #include "coro/detail/poll_info.hpp"
 #include "coro/fd.hpp"
 #include "coro/poll.hpp"
+#include "coro/signal.hpp"
 
 namespace coro::detail
 {
@@ -42,6 +43,8 @@ class io_notifier_epoll
 
     auto watch(fd_t fd, coro::poll_op op, void* data, bool keep = false) -> bool;
 
+    auto watch(const signal& signal, void* data) -> bool;
+
     auto watch(detail::poll_info& pi) -> bool;
 
     auto unwatch(detail::poll_info& pi) -> bool;

include/coro/detail/io_notifier_iocp.hpp

@@ -0,0 +1,58 @@
+#pragma once
+#include "coro/detail/poll_info.hpp"
+#include "coro/fd.hpp"
+#include "coro/poll.hpp"
+#include "coro/signal.hpp"
+#include <mutex>
+
+namespace coro::detail
+{
+class timer_handle;
+
+class io_notifier_iocp
+{
+public:
+    enum class completion_key : unsigned long long
+    {
+        signal_set,
+        signal_unset,
+        socket,
+        timer
+    };
+
+public:
+    io_notifier_iocp();
+
+    io_notifier_iocp(const io_notifier_iocp&)                    = delete;
+    io_notifier_iocp(io_notifier_iocp&&)                         = delete;
+    auto operator=(const io_notifier_iocp&) -> io_notifier_iocp& = delete;
+    auto operator=(io_notifier_iocp&&) -> io_notifier_iocp&      = delete;
+
+    ~io_notifier_iocp();
+
+    auto watch_timer(detail::timer_handle& timer, std::chrono::nanoseconds duration) -> bool;
+
+    auto watch(coro::signal& signal, void* data) -> bool;
+
+    auto unwatch_timer(detail::timer_handle& timer) -> bool;
+
+    auto next_events(
+        std::vector<std::pair<detail::poll_info*, coro::poll_status>>& ready_events,
+        std::chrono::milliseconds                                      timeout) -> void;
+
+    // static auto event_to_poll_status(const event_t& event) -> poll_status;
+
+    auto iocp() const noexcept -> void* { return m_iocp; }
+
+private:
+    void* m_iocp{};
+
+    void set_signal_active(void* data, bool active);
+    void process_active_signals(std::vector<std::pair<detail::poll_info*, coro::poll_status>>& ready_events);
+
+    std::mutex         m_active_signals_mutex;
+    std::vector<void*> m_active_signals;
+
+    static constexpr std::size_t max_events = 16;
+};
+} // namespace coro::detail

include/coro/detail/iocp_overlapped.hpp

@@ -0,0 +1,97 @@
+// NOTE: This file includes <Windows.h>, which pulls in many global symbols.
+// Do not include this file from headers. Include only in implementation files (.cpp) or modules.
+
+#pragma once
+#include "coro/io_scheduler.hpp"
+#include <coro/detail/poll_info.hpp>
+
+// clang-format off
+#define WIN32_LEAN_AND_MEAN
+#define NOMINMAX
+#include <winsock2.h>
+#include <ws2ipdef.h>
+#include <MSWSock.h>
+#include "coro/detail/iocp_overlapped.hpp"
+// clang-format on
+
+namespace coro::detail
+{
+struct overlapped_io_operation
+{
+    OVERLAPPED ov{}; // Base Windows OVERLAPPED structure for async I/O
+    poll_info  pi;
+    DWORD      bytes_transferred{}; // Number of bytes read or written once the operation completes
+
+    SOCKET socket{};
+};
+
+template<typename status_enum, typename buffer_type, bool is_read, typename operation_fn>
+    requires std::is_invocable_r_v<int, operation_fn, SOCKET, overlapped_io_operation&, WSABUF&>
+auto perform_write_read_operation(
+    const std::shared_ptr<io_scheduler>& scheduler,
+    SOCKET                               socket,
+    operation_fn&&                       operation,
+    buffer_type                          buffer,
+    std::chrono::milliseconds            timeout) -> task<std::pair<status_enum, buffer_type>>
+{
+    overlapped_io_operation ov{};
+    WSABUF                  buf{};
+
+    ov.socket = socket;
+
+    buf.buf = const_cast<char*>(buffer.data());
+    buf.len = buffer.size();
+
+    auto get_result_buffer = [&]()
+    {
+        if constexpr (is_read)
+            return ov.bytes_transferred == 0 ? buffer_type{} : buffer_type{buffer.data(), ov.bytes_transferred};
+        else
+            return ov.bytes_transferred == 0
+                       ? buffer_type{}
+                       : buffer_type{buffer.data() + ov.bytes_transferred, buffer.size() - ov.bytes_transferred};
+    };
+
+    auto r = operation(socket, std::ref(ov), std::ref(buf));
+
+    // Operation has been completed synchronously, no need to wait for event.
+    if (r == 0)
+    {
+        co_return {ov.bytes_transferred == 0 ? status_enum::closed : status_enum::ok, get_result_buffer()};
+    }
+    if (WSAGetLastError() != WSA_IO_PENDING)
+    {
+        co_return {status_enum::error, buffer};
+    }
+
+    // We need loop in case the operation completes right away with the timeout.
+    // In this case we just co_await our poll_info once more to correct status.
+    while (true)
+    {
+        switch (co_await scheduler->poll(ov.pi, timeout))
+        {
+            case poll_status::event:
+                co_return {status_enum::ok, get_result_buffer()};
+            case poll_status::timeout:
+            {
+                if (const BOOL success = CancelIoEx(reinterpret_cast<HANDLE>(socket), &ov.ov); !success)
+                {
+                    if (const auto err = GetLastError(); err == ERROR_NOT_FOUND)
+                    {
+                        // Operation has been completed, we need to co_await once more
+                        timeout = {}; // No need in timeout
+                        continue;
+                    }
+                }
+                co_return {status_enum::timeout, get_result_buffer()};
+            }
+            case poll_status::closed:
+                co_return {status_enum::closed, buffer};
+            case poll_status::error:
+            default:
+                co_return {status_enum::error, buffer};
+        }
+    }
+}
+
+} // namespace coro::detail

include/coro/detail/poll_info.hpp

@@ -1,8 +1,8 @@
 #pragma once
 
 #include "coro/fd.hpp"
-#include "coro/poll.hpp"
 #include "coro/time.hpp"
+#include "coro/poll.hpp"
 
 #include <atomic>
 #include <coroutine>
@@ -31,7 +31,9 @@ struct poll_info
     poll_info()  = default;
     ~poll_info() = default;
 
+    #if defined(CORO_PLATFORM_UNIX)
     poll_info(fd_t fd, coro::poll_op op) : m_fd(fd), m_op(op) {}
+    #endif
 
     poll_info(const poll_info&)                    = delete;
     poll_info(poll_info&&)                         = delete;
@@ -55,11 +57,13 @@ struct poll_info
 
     auto operator co_await() noexcept -> poll_awaiter { return poll_awaiter{*this}; }
 
+#if defined(CORO_PLATFORM_UNIX)
     /// The file descriptor being polled on.  This is needed so that if the timeout occurs first then
     /// the event loop can immediately disable the event within epoll.
     fd_t m_fd{-1};
     /// The operation that is being waited for to be performed on the file descriptor.
     coro::poll_op m_op;
+#endif
     /// The timeout's position in the timeout map.  A poll() with no timeout or yield() this is empty.
     /// This is needed so that if the event occurs first then the event loop can immediately disable
     /// the timeout within epoll.

include/coro/detail/signal_unix.hpp

@@ -0,0 +1,24 @@
+#pragma once
+#include "coro/fd.hpp"
+
+#include <array>
+
+namespace coro::detail
+{
+class signal_unix
+{
+public:
+    signal_unix();
+    ~signal_unix();
+
+    void set();
+
+    void unset();
+
+    [[nodiscard]] auto read_fd() const noexcept -> fd_t { return m_pipe[0]; }
+    [[nodiscard]] auto write_fd() const noexcept -> fd_t { return m_pipe[1]; }
+
+private:
+    std::array<fd_t, 2> m_pipe{-1};
+};
+} // namespace coro::detail