Allow async_resolve to use a configurable number of internal threads.

The number of threads used to emulate asynchronous address resolution is
now adjustable via the "resolver" / "threads" configuration option. If
this is set to a non-zero value, that number of threads will be created
when the first resolver object is constructed. Otherwise, it defaults to
a single thread that is created on the first call to async_resolve.

Author: chriskohlhoff

asio/include/Makefile.am

@@ -140,6 +140,7 @@ nobase_include_HEADERS = \
 	asio/detail/impl/reactive_descriptor_service.ipp \
 	asio/detail/impl/reactive_socket_service_base.ipp \
 	asio/detail/impl/resolver_service_base.ipp \
+	asio/detail/impl/resolver_thread_pool.ipp \
 	asio/detail/impl/scheduler.ipp \
 	asio/detail/impl/select_reactor.hpp \
 	asio/detail/impl/select_reactor.ipp \
@@ -254,6 +255,7 @@ nobase_include_HEADERS = \
 	asio/detail/resolve_op.hpp \
 	asio/detail/resolve_query_op.hpp \
 	asio/detail/resolver_service_base.hpp \
+	asio/detail/resolver_thread_pool.hpp \
 	asio/detail/resolver_service.hpp \
 	asio/detail/scheduler.hpp \
 	asio/detail/scheduler_operation.hpp \

asio/include/asio/detail/impl/resolver_service_base.ipp

@@ -17,71 +17,21 @@
 
 #include "asio/detail/config.hpp"
 #include "asio/config.hpp"
+#include "asio/detail/memory.hpp"
 #include "asio/detail/resolver_service_base.hpp"
 
 #include "asio/detail/push_options.hpp"
 
 namespace asio {
 namespace detail {
 
-class resolver_service_base::work_scheduler_runner
-{
-public:
-  work_scheduler_runner(scheduler_impl& work_scheduler)
-    : work_scheduler_(work_scheduler)
-  {
-  }
-
-  void operator()()
-  {
-    asio::error_code ec;
-    work_scheduler_.run(ec);
-  }
-
-private:
-  scheduler_impl& work_scheduler_;
-};
-
 resolver_service_base::resolver_service_base(execution_context& context)
-  : scheduler_(asio::use_service<scheduler_impl>(context)),
-    work_scheduler_(new scheduler_impl(context)),
-    work_thread_(),
-    scheduler_locking_(config(context).get("scheduler", "locking", true))
+  : thread_pool_(asio::use_service<resolver_thread_pool>(context))
 {
-  work_scheduler_->work_started();
 }
 
 resolver_service_base::~resolver_service_base()
 {
-  base_shutdown();
-}
-
-void resolver_service_base::base_shutdown()
-{
-  if (work_scheduler_.get())
-  {
-    work_scheduler_->work_finished();
-    work_scheduler_->stop();
-    work_thread_.join();
-    work_scheduler_.reset();
-  }
-}
-
-void resolver_service_base::base_notify_fork(
-    execution_context::fork_event fork_ev)
-{
-  if (work_thread_.joinable())
-  {
-    if (fork_ev == execution_context::fork_prepare)
-    {
-      work_scheduler_->stop();
-      work_thread_.join();
-    }
-  }
-  else if (fork_ev != execution_context::fork_prepare)
-  {
-    work_scheduler_->restart();
-  }
 }
 
 void resolver_service_base::construct(
@@ -93,7 +43,7 @@ void resolver_service_base::construct(
 void resolver_service_base::destroy(
     resolver_service_base::implementation_type& impl)
 {
-  ASIO_HANDLER_OPERATION((scheduler_.context(),
+  ASIO_HANDLER_OPERATION((thread_pool_.context(),
         "resolver", &impl, 0, "cancel"));
 
   impl.reset();
@@ -115,34 +65,12 @@ void resolver_service_base::move_assign(implementation_type& impl,
 void resolver_service_base::cancel(
     resolver_service_base::implementation_type& impl)
 {
-  ASIO_HANDLER_OPERATION((scheduler_.context(),
+  ASIO_HANDLER_OPERATION((thread_pool_.context(),
         "resolver", &impl, 0, "cancel"));
 
   impl.reset(static_cast<void*>(0), socket_ops::noop_deleter());
 }
 
-void resolver_service_base::start_resolve_op(resolve_op* op)
-{
-  if (scheduler_locking_)
-  {
-    start_work_thread();
-    scheduler_.work_started();
-    work_scheduler_->post_immediate_completion(op, false);
-  }
-  else
-  {
-    op->ec_ = asio::error::operation_not_supported;
-    scheduler_.post_immediate_completion(op, false);
-  }
-}
-
-void resolver_service_base::start_work_thread()
-{
-  asio::detail::mutex::scoped_lock lock(mutex_);
-  if (!work_thread_.joinable())
-    work_thread_ = thread(work_scheduler_runner(*work_scheduler_));
-}
-
 } // namespace detail
 } // namespace asio
 

asio/include/asio/detail/impl/resolver_thread_pool.ipp

@@ -0,0 +1,124 @@
+//
+// detail/impl/resolver_thread_pool.ipp
+// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+//
+// Copyright (c) 2003-2025 Christopher M. Kohlhoff (chris at kohlhoff dot com)
+//
+// 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)
+//
+
+#ifndef ASIO_DETAIL_IMPL_RESOLVER_THREAD_POOL_IPP
+#define ASIO_DETAIL_IMPL_RESOLVER_THREAD_POOL_IPP
+
+#if defined(_MSC_VER) && (_MSC_VER >= 1200)
+# pragma once
+#endif // defined(_MSC_VER) && (_MSC_VER >= 1200)
+
+#include "asio/detail/config.hpp"
+#include "asio/config.hpp"
+#include "asio/detail/memory.hpp"
+#include "asio/detail/resolver_thread_pool.hpp"
+#include "asio/error.hpp"
+
+#include "asio/detail/push_options.hpp"
+
+namespace asio {
+namespace detail {
+
+class resolver_thread_pool::work_scheduler_runner
+{
+public:
+  work_scheduler_runner(scheduler_impl& work_scheduler)
+    : work_scheduler_(work_scheduler)
+  {
+  }
+
+  void operator()()
+  {
+    asio::error_code ec;
+    work_scheduler_.run(ec);
+  }
+
+private:
+  scheduler_impl& work_scheduler_;
+};
+
+resolver_thread_pool::resolver_thread_pool(execution_context& context)
+  : execution_context_service_base<resolver_thread_pool>(context),
+    scheduler_(asio::use_service<scheduler_impl>(context)),
+    work_scheduler_(scheduler_impl::internal(), context),
+    work_threads_(execution_context::allocator<void>(context)),
+    num_work_threads_(config(context).get("resolver", "threads", 0U)),
+    scheduler_locking_(config(context).get("scheduler", "locking", true)),
+    shutdown_(false)
+{
+  work_scheduler_.work_started();
+  if (num_work_threads_ > 0)
+    start_work_threads();
+  else
+    num_work_threads_ = 1;
+}
+
+resolver_thread_pool::~resolver_thread_pool()
+{
+  shutdown();
+}
+
+void resolver_thread_pool::shutdown()
+{
+  if (!shutdown_)
+  {
+    work_scheduler_.work_finished();
+    work_scheduler_.stop();
+    work_threads_.join();
+    work_scheduler_.shutdown();
+    shutdown_ = true;
+  }
+}
+
+void resolver_thread_pool::notify_fork(execution_context::fork_event fork_ev)
+{
+  if (!work_threads_.empty())
+  {
+    if (fork_ev == execution_context::fork_prepare)
+    {
+      work_scheduler_.stop();
+      work_threads_.join();
+    }
+  }
+  else if (fork_ev != execution_context::fork_prepare)
+  {
+    work_scheduler_.restart();
+  }
+}
+
+void resolver_thread_pool::start_resolve_op(resolve_op* op)
+{
+  if (scheduler_locking_)
+  {
+    start_work_threads();
+    scheduler_.work_started();
+    work_scheduler_.post_immediate_completion(op, false);
+  }
+  else
+  {
+    op->ec_ = asio::error::operation_not_supported;
+    scheduler_.post_immediate_completion(op, false);
+  }
+}
+
+void resolver_thread_pool::start_work_threads()
+{
+  asio::detail::mutex::scoped_lock lock(mutex_);
+  if (work_threads_.empty())
+    for (unsigned int i = 0; i < num_work_threads_; ++i)
+      work_threads_.create_thread(work_scheduler_runner(work_scheduler_));
+}
+
+} // namespace detail
+} // namespace asio
+
+#include "asio/detail/pop_options.hpp"
+
+#endif // ASIO_DETAIL_IMPL_RESOLVER_THREAD_POOL_IPP

asio/include/asio/detail/impl/scheduler.ipp

@@ -126,6 +126,22 @@ scheduler::scheduler(asio::execution_context& ctx,
   ASIO_HANDLER_TRACKING_INIT;
 }
 
+scheduler::scheduler(scheduler::internal, asio::execution_context& ctx)
+  : asio::detail::execution_context_service_base<scheduler>(ctx),
+    one_thread_(false),
+    mutex_(true, 0),
+    task_(0),
+    get_task_(&scheduler::get_default_task),
+    task_interrupted_(true),
+    stopped_(false),
+    shutdown_(false),
+    outstanding_work_(0),
+    task_usec_(-1L),
+    wait_usec_(-1L)
+{
+  ASIO_HANDLER_TRACKING_INIT;
+}
+
 scheduler::~scheduler()
 {
 }

asio/include/asio/detail/impl/win_iocp_io_context.ipp

@@ -104,6 +104,32 @@ win_iocp_io_context::win_iocp_io_context(
   }
 }
 
+win_iocp_io_context::win_iocp_io_context(
+    win_iocp_io_context::internal, asio::execution_context& ctx)
+  : execution_context_service_base<win_iocp_io_context>(ctx),
+    iocp_(),
+    outstanding_work_(0),
+    stopped_(0),
+    stop_event_posted_(0),
+    shutdown_(0),
+    gqcs_timeout_(get_gqcs_timeout()),
+    dispatch_required_(0),
+    concurrency_hint_(-1)
+{
+  ASIO_HANDLER_TRACKING_INIT;
+
+  iocp_.handle = ::CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0,
+      static_cast<DWORD>(concurrency_hint_ >= 0
+        ? concurrency_hint_ : DWORD(~0)));
+  if (!iocp_.handle)
+  {
+    DWORD last_error = ::GetLastError();
+    asio::error_code ec(last_error,
+        asio::error::get_system_category());
+    asio::detail::throw_error(ec, "iocp");
+  }
+}
+
 win_iocp_io_context::~win_iocp_io_context()
 {
 }

asio/include/asio/detail/resolver_service.hpp

@@ -60,13 +60,6 @@ class resolver_service :
   // Destroy all user-defined handler objects owned by the service.
   void shutdown()
   {
-    this->base_shutdown();
-  }
-
-  // Perform any fork-related housekeeping.
-  void notify_fork(execution_context::fork_event fork_ev)
-  {
-    this->base_notify_fork(fork_ev);
   }
 
   // Resolve a query to a list of entries.
@@ -93,12 +86,12 @@ class resolver_service :
     typedef resolve_query_op<Protocol, Handler, IoExecutor> op;
     typename op::ptr p = { asio::detail::addressof(handler),
       op::ptr::allocate(handler), 0 };
-    p.p = new (p.v) op(impl, qry, scheduler_, handler, io_ex);
+    p.p = new (p.v) op(impl, qry, thread_pool_.scheduler(), handler, io_ex);
 
-    ASIO_HANDLER_CREATION((scheduler_.context(),
+    ASIO_HANDLER_CREATION((thread_pool_.context(),
           *p.p, "resolver", &impl, 0, "async_resolve"));
 
-    start_resolve_op(p.p);
+    thread_pool_.start_resolve_op(p.p);
     p.v = p.p = 0;
   }
 
@@ -126,12 +119,13 @@ class resolver_service :
     typedef resolve_endpoint_op<Protocol, Handler, IoExecutor> op;
     typename op::ptr p = { asio::detail::addressof(handler),
       op::ptr::allocate(handler), 0 };
-    p.p = new (p.v) op(impl, endpoint, scheduler_, handler, io_ex);
+    p.p = new (p.v) op(impl, endpoint,
+        thread_pool_.scheduler(), handler, io_ex);
 
-    ASIO_HANDLER_CREATION((scheduler_.context(),
+    ASIO_HANDLER_CREATION((thread_pool_.context(),
           *p.p, "resolver", &impl, 0, "async_resolve"));
 
-    start_resolve_op(p.p);
+    thread_pool_.start_resolve_op(p.p);
     p.v = p.p = 0;
   }
 };