Replace per-operation thread spawning with shared thread pool

Author: sgerbino

include/boost/corosio/detail/thread_pool.hpp

@@ -0,0 +1,213 @@
+//
+// Copyright (c) 2026 Steve Gerbino
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+// Official repository: https://github.com/cppalliance/corosio
+//
+
+#ifndef BOOST_COROSIO_DETAIL_THREAD_POOL_HPP
+#define BOOST_COROSIO_DETAIL_THREAD_POOL_HPP
+
+#include <boost/corosio/detail/config.hpp>
+#include <boost/corosio/detail/intrusive.hpp>
+#include <boost/capy/ex/execution_context.hpp>
+
+#include <condition_variable>
+#include <mutex>
+#include <stdexcept>
+#include <thread>
+#include <vector>
+
+namespace boost::corosio::detail {
+
+/** Base class for thread pool work items.
+
+    Derive from this to create work that can be posted to a
+    @ref thread_pool. Uses static function pointer dispatch,
+    consistent with the IOCP `op` pattern.
+
+    @par Example
+    @code
+    struct my_work : pool_work_item
+    {
+        int* result;
+        static void execute( pool_work_item* w ) noexcept
+        {
+            auto* self = static_cast<my_work*>( w );
+            *self->result = 42;
+        }
+    };
+
+    my_work w;
+    w.func_ = &my_work::execute;
+    w.result = &r;
+    pool.post( &w );
+    @endcode
+*/
+struct pool_work_item : intrusive_queue<pool_work_item>::node
+{
+    /// Static dispatch function signature.
+    using func_type = void (*)(pool_work_item*) noexcept;
+
+    /// Completion handler invoked by the worker thread.
+    func_type func_ = nullptr;
+};
+
+/** Shared thread pool for dispatching blocking operations.
+
+    Provides a fixed pool of reusable worker threads for operations
+    that cannot be integrated with async I/O (e.g. blocking DNS
+    calls). Registered as an `execution_context::service` so it
+    is a singleton per io_context.
+
+    Threads are created eagerly in the constructor. The default
+    thread count is 1.
+
+    @par Thread Safety
+    All public member functions are thread-safe.
+
+    @par Shutdown
+    Sets a shutdown flag, notifies all threads, and joins them.
+    In-flight blocking calls complete naturally before the thread
+    exits.
+*/
+class thread_pool final
+    : public capy::execution_context::service
+{
+    std::mutex mutex_;
+    std::condition_variable cv_;
+    intrusive_queue<pool_work_item> work_queue_;
+    std::vector<std::thread> threads_;
+    bool shutdown_ = false;
+
+    void worker_loop();
+
+public:
+    using key_type = thread_pool;
+
+    /** Construct the thread pool service.
+
+        Eagerly creates all worker threads.
+
+        @par Exception Safety
+        Strong guarantee. If thread creation fails, all
+        already-created threads are shut down and joined
+        before the exception propagates.
+
+        @param ctx Reference to the owning execution_context.
+        @param num_threads Number of worker threads. Must be
+               at least 1.
+
+        @throws std::logic_error If `num_threads` is 0.
+    */
+    explicit thread_pool(
+        capy::execution_context& ctx,
+        unsigned num_threads = 1)
+    {
+        (void)ctx;
+        if (!num_threads)
+            throw std::logic_error(
+                "thread_pool requires at least 1 thread");
+        threads_.reserve(num_threads);
+        try
+        {
+            for (unsigned i = 0; i < num_threads; ++i)
+                threads_.emplace_back([this] { worker_loop(); });
+        }
+        catch (...)
+        {
+            shutdown();
+            throw;
+        }
+    }
+
+    ~thread_pool() override = default;
+
+    thread_pool(thread_pool const&) = delete;
+    thread_pool& operator=(thread_pool const&) = delete;
+
+    /** Enqueue a work item for execution on the thread pool.
+
+        Zero-allocation: the caller owns the work item's storage.
+
+        @param w The work item to execute. Must remain valid until
+                 its `func_` has been called.
+
+        @return `true` if the item was enqueued, `false` if the
+                pool has already shut down.
+    */
+    bool post(pool_work_item* w) noexcept;
+
+    /** Shut down the thread pool.
+
+        Signals all threads to exit after draining any
+        remaining queued work, then joins them.
+    */
+    void shutdown() override;
+};
+
+inline void
+thread_pool::worker_loop()
+{
+    for (;;)
+    {
+        pool_work_item* w;
+        {
+            std::unique_lock<std::mutex> lock(mutex_);
+            cv_.wait(lock, [this] {
+                return shutdown_ || !work_queue_.empty();
+            });
+
+            w = work_queue_.pop();
+            if (!w)
+            {
+                if (shutdown_)
+                    return;
+                continue;
+            }
+        }
+        w->func_(w);
+    }
+}
+
+inline bool
+thread_pool::post(pool_work_item* w) noexcept
+{
+    {
+        std::lock_guard<std::mutex> lock(mutex_);
+        if (shutdown_)
+            return false;
+        work_queue_.push(w);
+    }
+    cv_.notify_one();
+    return true;
+}
+
+inline void
+thread_pool::shutdown()
+{
+    {
+        std::lock_guard<std::mutex> lock(mutex_);
+        shutdown_ = true;
+    }
+    cv_.notify_all();
+
+    for (auto& t : threads_)
+    {
+        if (t.joinable())
+            t.join();
+    }
+    threads_.clear();
+
+    {
+        std::lock_guard<std::mutex> lock(mutex_);
+        while (work_queue_.pop())
+            ;
+    }
+}
+
+} // namespace boost::corosio::detail
+
+#endif // BOOST_COROSIO_DETAIL_THREAD_POOL_HPP

include/boost/corosio/native/detail/iocp/win_resolver.hpp

@@ -23,6 +23,7 @@
 #endif
 
 #include <boost/corosio/detail/scheduler.hpp>
+#include <boost/corosio/detail/thread_pool.hpp>
 #include <boost/corosio/endpoint.hpp>
 #include <boost/corosio/resolver.hpp>
 #include <boost/corosio/resolver_results.hpp>
@@ -42,12 +43,9 @@
 #include <WS2tcpip.h>
 
 #include <atomic>
-#include <condition_variable>
 #include <cstring>
 #include <memory>
 #include <string>
-#include <thread>
-#include <unordered_map>
 
 // MinGW may not have GetAddrInfoExCancel declared
 #if defined(__MINGW32__) || defined(__MINGW64__)
@@ -72,30 +70,28 @@ extern "C"
     Reverse Resolution (GetNameInfoW)
     ---------------------------------
     Unlike GetAddrInfoExW, GetNameInfoW has no async variant. Reverse
-    resolution spawns a detached worker thread that calls GetNameInfoW
-    and posts the result to the scheduler upon completion.
+    resolution dispatches the blocking call to the shared
+    resolver_thread_pool service.
 
     Class Hierarchy
     ---------------
     - win_resolver_service (execution_context::service)
         - Owns all win_resolver instances via shared_ptr
         - Coordinates with win_scheduler for work tracking
-        - Tracks active worker threads for safe shutdown
     - win_resolver (one per resolver object)
         - Contains embedded resolve_op and reverse_resolve_op
         - Inherits from enable_shared_from_this for thread safety
     - resolve_op (overlapped_op subclass)
         - OVERLAPPED base enables IOCP integration
         - Static completion() callback invoked by Windows
     - reverse_resolve_op (overlapped_op subclass)
-        - Used by worker thread for reverse resolution
+        - Used by pool thread for reverse resolution
 
-    Shutdown Synchronization
-    ------------------------
-    The service uses condition_variable_any and win_mutex to track active
-    worker threads. During shutdown(), the service waits for all threads
-    to complete before destroying resources. Worker threads always post
-    their completions so the scheduler can properly drain them via destroy().
+    Shutdown
+    --------
+    The resolver service cancels all resolvers and clears the impl map.
+    The thread pool service shuts down separately via execution_context
+    service ordering, joining all worker threads.
 
     Cancellation
     ------------
@@ -128,17 +124,14 @@ extern "C"
 
     Reverse Resolution (GetNameInfoW)
     ---------------------------------
-    Unlike GetAddrInfoExW, GetNameInfoW has no async variant. We use a worker
-    thread approach similar to POSIX:
-    1. reverse_resolve() spawns a detached worker thread
-    2. Worker calls GetNameInfoW() (blocking)
-    3. Worker converts wide results to UTF-8 via WideCharToMultiByte
-    4. Worker posts completion to scheduler
+    Unlike GetAddrInfoExW, GetNameInfoW has no async variant. The blocking
+    call is dispatched to the shared resolver_thread_pool:
+    1. reverse_resolve() posts work to the thread pool
+    2. Pool thread calls GetNameInfoW() (blocking)
+    3. Pool thread converts wide results to UTF-8 via WideCharToMultiByte
+    4. Pool thread posts completion to scheduler
     5. op_() resumes the coroutine with results
 
-    Thread tracking (thread_started/thread_finished) ensures safe shutdown
-    by waiting for all worker threads before destroying the service.
-
     String Conversion
     -----------------
     Windows APIs require wide strings. We use MultiByteToWideChar for
@@ -250,6 +243,16 @@ class win_resolver final
     friend struct resolve_op;
 
 public:
+    /// Embedded pool work item for thread pool dispatch.
+    struct pool_op : pool_work_item
+    {
+        /// Resolver that owns this work item.
+        win_resolver* resolver_ = nullptr;
+
+        /// Prevent impl destruction while work is in flight.
+        std::shared_ptr<win_resolver> ref_;
+    };
+
     explicit win_resolver(win_resolver_service& svc) noexcept;
 
     std::coroutine_handle<> resolve(
@@ -276,6 +279,12 @@ class win_resolver final
     resolve_op op_;
     reverse_resolve_op reverse_op_;
 
+    /// Pool work item for reverse resolution.
+    pool_op reverse_pool_op_;
+
+    /// Execute blocking `GetNameInfoW()` on a pool thread.
+    static void do_reverse_resolve_work(pool_work_item*) noexcept;
+
 private:
     win_resolver_service& svc_;
 };