diff --git a/doc/modules/ROOT/pages/4.guide/4c.io-context.adoc b/doc/modules/ROOT/pages/4.guide/4c.io-context.adoc index 8e9490613..e74dd8368 100644 --- a/doc/modules/ROOT/pages/4.guide/4c.io-context.adoc +++ b/doc/modules/ROOT/pages/4.guide/4c.io-context.adoc @@ -53,27 +53,33 @@ corosio::io_context ioc; ---- Creates an `io_context` with a concurrency hint of -`std::max(2u, std::thread::hardware_concurrency())` — the default constructor -never selects single-threaded mode. Single-threaded (lockless) mode is keyed -on a `concurrency_hint` of exactly `1`; pass `1` explicitly to opt into it. +`std::max(1u, std::thread::hardware_concurrency())` and the default, fully +thread-safe locking tier. Thread safety is governed by the `locking` option +(see xref:4.guide/4c2.configuration.adoc#single-threaded-mode[Locking Tiers]). === With Concurrency Hint [source,cpp] ---- -corosio::io_context ioc(1); // Single-threaded, no synchronization -corosio::io_context ioc(4); // Up to 4 threads, thread-safe +corosio::io_context ioc(1); // one thread expected to call run() +corosio::io_context ioc(4); // up to 4 threads expected to call run() ---- -The concurrency hint affects: +The concurrency hint is a performance tuning value indicating how many threads +are expected to call `run()`. It tunes: -* Internal synchronization strategy -* On Windows, the hint is passed to `CreateIoCompletionPort` as +* The reactor's inline-completion budget defaults: a multi-thread context favors + posting completions for cross-thread work-stealing, while a single-thread + context keeps the inline fast path. +* On Windows, it is passed to `CreateIoCompletionPort` as `NumberOfConcurrentThreads`, bounding how many completion threads may run concurrently. It is not a library-managed thread pool, and it is distinct from `thread_pool_size`. -Use `1` for single-threaded programs to avoid synchronization overhead. +The hint does not affect thread safety, which is governed by the `locking` +tier. To reduce synchronization overhead in a single-threaded program, select a +lockless `locking` tier (see +xref:4.guide/4c2.configuration.adoc#single-threaded-mode[Locking Tiers]). == Running the Event Loop @@ -244,8 +250,9 @@ int main() == Thread Safety -The `io_context` can be used from multiple threads when constructed with -a concurrency hint greater than 1: +The `io_context` is thread-safe by default, regardless of the concurrency hint: +multiple threads may call `run()` concurrently, and any thread may `post()` work +into it. [source,cpp] ---- @@ -260,7 +267,10 @@ for (auto& t : threads) ---- Multiple threads can call `run()` concurrently. The `io_context` distributes -work across threads. +work across threads. The one exception is a lockless `locking` tier: constructing +with `io_context_options::locking` set to `unsafe_io` or `unsafe` drops these +guarantees (see +xref:4.guide/4c2.configuration.adoc#single-threaded-mode[Locking Tiers]). WARNING: Individual I/O objects (sockets, timers) are not thread-safe. Don't access the same socket from multiple threads without synchronization. diff --git a/doc/modules/ROOT/pages/4.guide/4c2.configuration.adoc b/doc/modules/ROOT/pages/4.guide/4c2.configuration.adoc index c566ee5d7..35e8d575e 100644 --- a/doc/modules/ROOT/pages/4.guide/4c2.configuration.adoc +++ b/doc/modules/ROOT/pages/4.guide/4c2.configuration.adoc @@ -80,12 +80,13 @@ corosio::native_io_context ioc(opts); blocking file I/O and DNS resolution. Ignored on IOCP where file I/O uses native overlapped I/O. -| `single_threaded` -| false +| `locking` +| `locking_mode::safe` | all -| Disable all scheduler mutex and condition variable operations. - Eliminates synchronization overhead when only one thread calls - `run()`. See <> for restrictions. +| Locking-safety tier: `safe` (default, full thread safety), + `unsafe_io` (per-descriptor I/O locks off), or `unsafe` (all locks + off). Governs the thread-safety contract. See + <> for the tiers and their restrictions. |=== Options that do not apply to the active backend are silently ignored. @@ -144,31 +145,73 @@ and DNS resolution use a shared thread pool. * *No file I/O*: leave at 1 (the pool is created lazily). [#single-threaded-mode] -=== Single-Threaded Mode (`single_threaded`) +=== Locking Tiers (`locking`) -Disables all mutex and condition variable operations inside the -scheduler and per-socket descriptor states. This eliminates -15-25% of overhead on the post-and-dispatch hot path. +The `locking` option selects which internal locks the scheduler and +reactor elide, trading thread-safety guarantees for reduced +synchronization overhead. It governs the thread-safety contract, +independently of the `concurrency_hint` (see +<>). + +[cols="1,3"] +|=== +| Tier | Behavior + +| `locking_mode::safe` (default) +| Full thread safety. All locks enabled; any thread may use the + context. + +| `locking_mode::unsafe_io` +| Disables only the per-descriptor I/O locks; scheduler locking stays + on. A single thread must run and drive the context, but DNS + resolution and POSIX file I/O remain available (they rely on + scheduler locking). + +| `locking_mode::unsafe` +| Disables all locking. Eliminates 15-25% of overhead on the + post-and-dispatch hot path. Imposes the full restrictions below. +|=== [source,cpp] ---- corosio::io_context_options opts; -opts.single_threaded = true; +opts.locking = corosio::locking_mode::unsafe; corosio::io_context ioc(opts); ioc.run(); // only one thread may call this ---- -WARNING: Single-threaded mode imposes hard restrictions. +WARNING: The `unsafe` and `unsafe_io` tiers impose hard restrictions. Violating them is undefined behavior. * Only **one thread** may call `run()` (or any run/poll variant). * **Posting work from another thread** is undefined behavior. -* **DNS resolution** returns `operation_not_supported`. -* **POSIX file I/O** (`stream_file`, `random_access_file`) returns - `operation_not_supported` on `open()`. * **Signal sets** should not be shared across contexts. -* **Delay and timeout cancellation via `stop_token`** from another - thread is not permitted: the cancel path posts into the scheduler, - whose synchronization is disabled in single-threaded mode. - Cross-thread cancellation requires a multi-threaded-capable context. +* **Delay and timeout cancellation via `stop_token`** from another thread + is not permitted: the cancel path posts into the scheduler, whose locking + these tiers disable. Cross-thread cancellation requires the `safe` tier. + +The `unsafe` tier additionally makes: + +* **DNS resolution** return `operation_not_supported`. +* **POSIX file I/O** (`stream_file`, `random_access_file`) return + `operation_not_supported` on `open()`. + +These two services stay available under `unsafe_io` because they rely +on scheduler locking, which that tier keeps enabled. + +[NOTE] +==== +These tiers correspond to Boost.Asio's `SAFE` / `UNSAFE_IO` / `UNSAFE` +concurrency-hint constants, exposed here as a dedicated option separate +from the `concurrency_hint`. +==== + +[#concurrency-hint-and-locking] +==== Concurrency hint and locking tier + +The `concurrency_hint` is a performance tuning value indicating how many +threads are expected to call `run()`; it tunes the reactor inline-completion +budget defaults and, on IOCP, the completion-port concurrency. A lockless +tier is single-threaded, so its effective hint for that tuning is `1` +regardless of the value passed. diff --git a/include/boost/corosio/backend.hpp b/include/boost/corosio/backend.hpp index e2a633edd..091c2cf65 100644 --- a/include/boost/corosio/backend.hpp +++ b/include/boost/corosio/backend.hpp @@ -359,7 +359,9 @@ struct iocp_t @param ctx The execution context that owns the scheduler. @param concurrency_hint Hint for the number of threads that - will call `run()`. Pass `1` for single-threaded mode. + will call `run()`; a performance tuning knob. The + thread-safety contract is set separately by + @ref io_context_options::locking. @return Reference to the newly created scheduler. */ diff --git a/include/boost/corosio/detail/scheduler.hpp b/include/boost/corosio/detail/scheduler.hpp index 20be8e39f..bd8f0e1f2 100644 --- a/include/boost/corosio/detail/scheduler.hpp +++ b/include/boost/corosio/detail/scheduler.hpp @@ -97,13 +97,25 @@ struct BOOST_COROSIO_DECL scheduler */ virtual void register_signal_reader(int read_fd) { (void)read_fd; } - /// True if the scheduler is configured for single-threaded use. - /// Default false; overridden by backends that support the mode. - virtual bool is_single_threaded() const noexcept { return false; } - - /// Enable or disable single-threaded mode. Default no-op for - /// backends that don't support the mode. - virtual void configure_single_threaded(bool) noexcept {} + /// Decomposed threading configuration applied via @ref configure_threading. + struct threading_config + { + /// Scheduler mutex/condvar enabled. Off only in the `unsafe` tier. + bool scheduler_locking = true; + /// Per-descriptor (reactor) or ring (io_uring) I/O lock enabled. + /// Off in the `unsafe_io` and `unsafe` tiers. + bool reactor_io_locking = true; + /// A single run thread is guaranteed (a lockless tier): elide + /// inter-run-thread wakeups. + bool one_thread = false; + }; + + /// True in the fully-lockless (`unsafe`) tier. The resolver and POSIX + /// file services gate their `operation_not_supported` result on this. + virtual bool scheduler_locking_disabled() const noexcept { return false; } + + /// Apply @ref threading_config. Default no-op. + virtual void configure_threading(threading_config) noexcept {} }; } // namespace boost::corosio::detail diff --git a/include/boost/corosio/io_context.hpp b/include/boost/corosio/io_context.hpp index 44fc6f6c0..51f0b32d1 100644 --- a/include/boost/corosio/io_context.hpp +++ b/include/boost/corosio/io_context.hpp @@ -26,6 +26,42 @@ namespace boost::corosio { +/** Locking-safety tier for an @ref io_context. + + Selects which internal locks the scheduler and reactor elide, trading + thread-safety guarantees for reduced synchronization overhead. This is + the analog of Boost.Asio's `SAFE` / `UNSAFE_IO` / `UNSAFE` concurrency + hint constants. The tier is chosen explicitly, not derived from the + `concurrency_hint`. (The reverse does apply: a lockless tier reduces the + effective hint used for performance tuning to 1.) + + @see io_context_options::locking +*/ +enum class locking_mode +{ + /** Full thread safety (default). All locks enabled; equivalent to + Boost.Asio's `SAFE`/`DEFAULT`. Any thread may use the context. */ + safe, + + /** Disable only the per-descriptor I/O locks; keep scheduler locking. + Equivalent to Boost.Asio's `UNSAFE_IO`. The context must be run + and driven by a single thread, but resolver and POSIX file + services remain available (they rely on scheduler locking, which + stays on). */ + unsafe_io, + + /** Disable all locking (fully lockless). Equivalent to Boost.Asio's + `UNSAFE`. + + @par Restrictions + - Only one thread may call `run()` (or any run variant). + - Posting work from another thread is undefined behavior. + - DNS resolution returns `operation_not_supported`. + - POSIX file I/O returns `operation_not_supported`. + - Signal sets should not be shared across contexts. */ + unsafe +}; + /** Runtime tuning options for @ref io_context. All fields have defaults that match the library's built-in @@ -97,26 +133,10 @@ struct io_context_options */ unsigned thread_pool_size = 1; - /** Enable single-threaded mode (disable scheduler locking). - - When true, the scheduler skips all mutex lock/unlock and - condition variable operations on the hot path. This - eliminates synchronization overhead when only one thread - calls `run()`. - - @par Restrictions - - Only one thread may call `run()` (or any run variant). - - Posting work from another thread is undefined behavior. - - DNS resolution returns `operation_not_supported`. - - POSIX file I/O returns `operation_not_supported`. - - Signal sets should not be shared across contexts. - - @note Constructing an `io_context` with `concurrency_hint == 1` - automatically enables single-threaded mode regardless of - this field's value, matching asio's convention. To opt out, - pass `concurrency_hint > 1`. + /** Thread-safety tier. See @ref locking_mode for the tiers and their + restrictions. */ - bool single_threaded = false; + locking_mode locking = locking_mode::safe; /** Enable IORING_SETUP_SQPOLL on the io_uring backend. @@ -126,7 +146,7 @@ struct io_context_options path. Most useful for sustained traffic. Idle thread parks after `sq_thread_idle_ms` of no activity. - Independent of `single_threaded`. Default: off. + Independent of `locking`. Default: off. Ignored on non-io_uring backends. */ @@ -158,6 +178,16 @@ struct io_context_options namespace detail { class timer_service; + +/** Return the hint used for performance tuning: the lockless tiers are + single-threaded, so their effective hint is 1 whatever the caller passed. +*/ +inline unsigned +effective_concurrency_hint( + io_context_options const& opts, unsigned hint) noexcept +{ + return opts.locking == locking_mode::safe ? hint : 1u; +} } // namespace detail /** An I/O context for running asynchronous operations. @@ -196,8 +226,9 @@ class timer_service; @par Thread Safety Distinct objects: Safe.@n - Shared objects: Safe, if using a concurrency hint greater - than 1. + Shared objects: Safe, unless the context was constructed with a + lockless @ref io_context_options::locking tier (`unsafe_io` or + `unsafe`), in which case a single thread must drive it. @see epoll_t, select_t, kqueue_t, iocp_t */ @@ -211,8 +242,11 @@ class BOOST_COROSIO_DECL io_context : public capy::execution_context io_context_options const& opts, unsigned concurrency_hint); - /// Switch the scheduler to single-threaded (lockless) mode. - void configure_single_threaded_(); + /** Apply only the decomposed threading configuration (locking tiers). + Used by the plain constructors, which — unlike the options + constructors — deliberately leave the reactor budget at its defaults + rather than engaging the multi-thread post-everything heuristic. */ + void apply_threading_(io_context_options const& opts); protected: detail::scheduler* sched_; @@ -223,11 +257,10 @@ class BOOST_COROSIO_DECL io_context : public capy::execution_context /** Construct with default concurrency and platform backend. - Uses `std::thread::hardware_concurrency()` clamped to a minimum - of 2 as the concurrency hint, so the default constructor never - silently engages single-threaded mode (see - @ref io_context_options::single_threaded). Pass an explicit - `concurrency_hint == 1` to opt into single-threaded mode. + Uses `std::thread::hardware_concurrency()` (floored to 1, in + case it reports 0) as the concurrency hint, and the default + @ref locking_mode::safe tier. Select a lockless tier via + @ref io_context_options::locking. */ io_context(); @@ -266,8 +299,9 @@ class BOOST_COROSIO_DECL io_context : public capy::execution_context { (void)backend; sched_ = &Backend::construct(*this, concurrency_hint); - if (concurrency_hint == 1) - configure_single_threaded_(); + // Apply threading config only (locking tier). Unlike the options + // ctor, the plain path leaves the reactor budget at its defaults. + apply_threading_(io_context_options{}); } /** Construct with an explicit backend tag and runtime options. @@ -290,8 +324,11 @@ class BOOST_COROSIO_DECL io_context : public capy::execution_context { (void)backend; apply_options_pre_(opts); - sched_ = &Backend::construct(*this, concurrency_hint); - apply_options_post_(opts, concurrency_hint); + // Effective hint (1 for lockless tiers); see effective_concurrency_hint. + unsigned const eff = + detail::effective_concurrency_hint(opts, concurrency_hint); + sched_ = &Backend::construct(*this, eff); + apply_options_post_(opts, eff); } ~io_context(); diff --git a/include/boost/corosio/native/detail/epoll/epoll_scheduler.hpp b/include/boost/corosio/native/detail/epoll/epoll_scheduler.hpp index 7f1843478..d8489bbb8 100644 --- a/include/boost/corosio/native/detail/epoll/epoll_scheduler.hpp +++ b/include/boost/corosio/native/detail/epoll/epoll_scheduler.hpp @@ -265,7 +265,7 @@ epoll_scheduler::register_descriptor(int fd, reactor_descriptor_state* desc) con desc->registered_events = ev.events; desc->fd = fd; desc->scheduler_ = this; - desc->mutex.set_enabled(!single_threaded_); + desc->mutex.set_enabled(reactor_io_locking_); desc->ready_events_.store(0, std::memory_order_relaxed); conditionally_enabled_mutex::scoped_lock lock(desc->mutex); diff --git a/include/boost/corosio/native/detail/io_uring/io_uring_scheduler.hpp b/include/boost/corosio/native/detail/io_uring/io_uring_scheduler.hpp index f784b222a..e892abd59 100644 --- a/include/boost/corosio/native/detail/io_uring/io_uring_scheduler.hpp +++ b/include/boost/corosio/native/detail/io_uring/io_uring_scheduler.hpp @@ -255,13 +255,17 @@ class BOOST_COROSIO_DECL io_uring_scheduler final completed_ops_.push(op); } - /// Single-threaded mode toggle (matches reactor_scheduler API). - void configure_single_threaded(bool v) noexcept override + void configure_threading(threading_config cfg) noexcept override { - single_threaded_ = v; - dispatch_mutex_.set_enabled(!v); - ring_mutex_.set_enabled(!v); - cond_.set_enabled(!v); + scheduler_locking_disabled_ = !cfg.scheduler_locking; + // reactor_io_locking off also drives SINGLE_ISSUER/DEFER_TASKRUN and + // the eventfd wake elision (see lazy_init_ring_unlocked and + // interrupt_reactor). one_thread is unused: the leader-follower wake + // model is not gated on it. + reactor_io_locking_ = cfg.reactor_io_locking; + dispatch_mutex_.set_enabled(cfg.scheduler_locking); + ring_mutex_.set_enabled(cfg.reactor_io_locking); + cond_.set_enabled(cfg.scheduler_locking); } /** Configure SQPOLL parameters. @@ -289,8 +293,11 @@ class BOOST_COROSIO_DECL io_uring_scheduler final sq_thread_cpu_ = cpu; } - /// Return true if single-threaded (lockless) mode is active. - bool is_single_threaded() const noexcept override { return single_threaded_; } + /// Return true when scheduler locking is disabled (fully-lockless tier). + bool scheduler_locking_disabled() const noexcept override + { + return scheduler_locking_disabled_; + } private: // ring_ + wakeup_eventfd_ are mutable so lazy_init_ring() (called @@ -334,7 +341,8 @@ class BOOST_COROSIO_DECL io_uring_scheduler final // Leader-follower flag: true while a thread is blocked in // io_uring_submit_and_wait_timeout. Protected by dispatch_mutex_. mutable bool task_running_ = false; - bool single_threaded_ = false; + bool scheduler_locking_disabled_ = false; + bool reactor_io_locking_ = true; bool enable_sqpoll_ = false; unsigned sq_thread_idle_ms_ = 0; int sq_thread_cpu_ = -1; @@ -401,8 +409,8 @@ class BOOST_COROSIO_DECL io_uring_scheduler final static constexpr unsigned long drain_cqes_kick_ns = 1'000'000; // ring_inited_ goes true once on first run/poll/submit. The init is - // deferred from the constructor so configure_single_threaded(true) - // can take effect before io_uring_queue_init_params chooses flags. + // deferred from the constructor so configure_threading() can take + // effect before io_uring_queue_init_params chooses flags. mutable std::once_flag ring_init_once_; mutable bool ring_inited_ = false; @@ -460,7 +468,8 @@ inline void io_uring_scheduler::lazy_init_ring_unlocked() const { io_uring_params params{}; - if (single_threaded_) + // The unsafe_io and unsafe tiers guarantee a single ring submitter. + if (!reactor_io_locking_) { // SINGLE_ISSUER promises the kernel one submitter thread, // letting it skip internal SQ locking. DEFER_TASKRUN tells @@ -498,7 +507,7 @@ io_uring_scheduler::lazy_init_ring_unlocked() const // submission becomes a userspace-only memory store. Combines // with SINGLE_ISSUER (the kernel accepts that pair) but NOT // with DEFER_TASKRUN (kernel returns -EINVAL); the - // single_threaded_ branch above suppresses DEFER_TASKRUN + // reactor-I/O-lockless branch above suppresses DEFER_TASKRUN // when SQPOLL is also set. Idle timeout 0 means kernel // default (1ms); we only forward when explicitly set so // the kernel default is preserved. @@ -653,10 +662,12 @@ io_uring_scheduler::interrupt_reactor() const noexcept if (!ring_inited_) return; - // Single-thread: the user's coroutines run on the leader thread, - // so when interrupt_reactor is called from user code the leader - // is not in kernel wait — there is nothing to wake. - if (single_threaded_) + // Lockless tiers (reactor-I/O locking off): cross-thread post() is + // forbidden, so interrupt_reactor is only ever reached from the leader + // thread's own coroutines — it is not in kernel wait, nothing to wake. + // Under the safe tier (including concurrency_hint == 1) cross-thread + // post() is allowed, so the eventfd write below must always fire. + if (!reactor_io_locking_) return; // Multi-thread: write the eventfd unconditionally. CAS-coalescing diff --git a/include/boost/corosio/native/detail/iocp/win_scheduler.hpp b/include/boost/corosio/native/detail/iocp/win_scheduler.hpp index d6c9e38e5..d521e2363 100644 --- a/include/boost/corosio/native/detail/iocp/win_scheduler.hpp +++ b/include/boost/corosio/native/detail/iocp/win_scheduler.hpp @@ -84,21 +84,18 @@ class BOOST_COROSIO_DECL win_scheduler final void work_started() noexcept override; void work_finished() noexcept override; - /** Enable or disable single-threaded (lockless) mode. - - When enabled, the dispatch mutex becomes a no-op. - Cross-thread post() is undefined behavior. - */ - void configure_single_threaded(bool v) noexcept override + // IOCP has only the dispatch mutex; reactor_io_locking and one_thread do + // not apply (the completion port provides its own synchronization). + void configure_threading(threading_config cfg) noexcept override { - single_threaded_ = v; - dispatch_mutex_.set_enabled(!v); + scheduler_locking_disabled_ = !cfg.scheduler_locking; + dispatch_mutex_.set_enabled(cfg.scheduler_locking); } - /// Return true if single-threaded (lockless) mode is active. - bool is_single_threaded() const noexcept override + /// Return true when scheduler locking is disabled (fully-lockless tier). + bool scheduler_locking_disabled() const noexcept override { - return single_threaded_; + return scheduler_locking_disabled_; } /** Signal that an overlapped I/O operation is now pending. @@ -124,7 +121,7 @@ class BOOST_COROSIO_DECL win_scheduler final mutable long stopped_; long stop_event_posted_; mutable long dispatch_required_; - bool single_threaded_ = false; + bool scheduler_locking_disabled_ = false; BOOST_COROSIO_MSVC_WARNING_PUSH BOOST_COROSIO_MSVC_WARNING_DISABLE(4251) // std::/detail:: members, dll-interface diff --git a/include/boost/corosio/native/detail/kqueue/kqueue_scheduler.hpp b/include/boost/corosio/native/detail/kqueue/kqueue_scheduler.hpp index 446a1ce28..3f7e9789f 100644 --- a/include/boost/corosio/native/detail/kqueue/kqueue_scheduler.hpp +++ b/include/boost/corosio/native/detail/kqueue/kqueue_scheduler.hpp @@ -253,7 +253,7 @@ kqueue_scheduler::register_descriptor(int fd, reactor_descriptor_state* desc) co desc->registered_events = reactor_event_read | reactor_event_write; desc->fd = fd; desc->scheduler_ = this; - desc->mutex.set_enabled(!single_threaded_); + desc->mutex.set_enabled(reactor_io_locking_); desc->ready_events_.store(0, std::memory_order_relaxed); conditionally_enabled_mutex::scoped_lock lock(desc->mutex); diff --git a/include/boost/corosio/native/detail/posix/posix_random_access_file_service.hpp b/include/boost/corosio/native/detail/posix/posix_random_access_file_service.hpp index c95d55686..ecfae2964 100644 --- a/include/boost/corosio/native/detail/posix/posix_random_access_file_service.hpp +++ b/include/boost/corosio/native/detail/posix/posix_random_access_file_service.hpp @@ -81,7 +81,9 @@ class BOOST_COROSIO_DECL posix_random_access_file_service final std::filesystem::path const& path, file_base::flags mode) override { - if (sched_->is_single_threaded()) + // Unavailable in the unsafe tier: the file thread pool completes + // cross-thread, which the lockless scheduler cannot accept. + if (sched_->scheduler_locking_disabled()) return std::make_error_code(std::errc::operation_not_supported); return static_cast(impl).open_file( path, mode); diff --git a/include/boost/corosio/native/detail/posix/posix_resolver_service.hpp b/include/boost/corosio/native/detail/posix/posix_resolver_service.hpp index 55b8d1a48..ff3e4d268 100644 --- a/include/boost/corosio/native/detail/posix/posix_resolver_service.hpp +++ b/include/boost/corosio/native/detail/posix/posix_resolver_service.hpp @@ -68,10 +68,12 @@ class BOOST_COROSIO_DECL posix_resolver_service final return pool_; } - /** Return true if single-threaded mode is active. */ - bool single_threaded() const noexcept + /// True when the resolver thread pool is unavailable: the `unsafe` tier, + /// whose lockless scheduler cannot accept the pool's cross-thread + /// completions. + bool resolver_unavailable() const noexcept { - return sched_->is_single_threaded(); + return sched_->scheduler_locking_disabled(); } private: @@ -378,7 +380,7 @@ posix_resolver::resolve( std::error_code* ec, resolver_results* out) { - if (svc_.single_threaded()) + if (svc_.resolver_unavailable()) { *ec = std::make_error_code(std::errc::operation_not_supported); op_.cont.h = h; @@ -424,7 +426,7 @@ posix_resolver::reverse_resolve( std::error_code* ec, reverse_resolver_result* result_out) { - if (svc_.single_threaded()) + if (svc_.resolver_unavailable()) { *ec = std::make_error_code(std::errc::operation_not_supported); reverse_op_.cont.h = h; diff --git a/include/boost/corosio/native/detail/posix/posix_stream_file_service.hpp b/include/boost/corosio/native/detail/posix/posix_stream_file_service.hpp index bf6dfe4b7..629dbc8f9 100644 --- a/include/boost/corosio/native/detail/posix/posix_stream_file_service.hpp +++ b/include/boost/corosio/native/detail/posix/posix_stream_file_service.hpp @@ -82,7 +82,9 @@ class BOOST_COROSIO_DECL posix_stream_file_service final std::filesystem::path const& path, file_base::flags mode) override { - if (sched_->is_single_threaded()) + // Unavailable in the unsafe tier: the file thread pool completes + // cross-thread, which the lockless scheduler cannot accept. + if (sched_->scheduler_locking_disabled()) return std::make_error_code(std::errc::operation_not_supported); return static_cast(impl).open_file(path, mode); } diff --git a/include/boost/corosio/native/detail/reactor/reactor_scheduler.hpp b/include/boost/corosio/native/detail/reactor/reactor_scheduler.hpp index 6e7cabf92..755eb0424 100644 --- a/include/boost/corosio/native/detail/reactor/reactor_scheduler.hpp +++ b/include/boost/corosio/native/detail/reactor/reactor_scheduler.hpp @@ -258,28 +258,28 @@ class reactor_scheduler return inline_budget_initial_; } - /// Return true if single-threaded (lockless) mode is active. - bool is_single_threaded() const noexcept override + /// Return true when scheduler locking is disabled (fully-lockless tier). + bool scheduler_locking_disabled() const noexcept override { - return single_threaded_; + return scheduler_locking_disabled_; } - /** Enable or disable single-threaded (lockless) mode. - - When enabled, all scheduler mutex and condition variable - operations become no-ops. Cross-thread post() is - undefined behavior. - */ - void configure_single_threaded(bool v) noexcept override + void configure_threading(threading_config cfg) noexcept override { - single_threaded_ = v; - mutex_.set_enabled(!v); - cond_.set_enabled(!v); + scheduler_locking_disabled_ = !cfg.scheduler_locking; + // reactor_io_locking takes effect at descriptor registration (see the + // register_descriptor overrides), not here. + reactor_io_locking_ = cfg.reactor_io_locking; + one_thread_ = cfg.one_thread; + mutex_.set_enabled(cfg.scheduler_locking); + cond_.set_enabled(cfg.scheduler_locking); } protected: timer_service* timer_svc_ = nullptr; - bool single_threaded_ = false; + bool scheduler_locking_disabled_ = false; + bool reactor_io_locking_ = true; + bool one_thread_ = false; reactor_scheduler() = default; @@ -912,7 +912,9 @@ reactor_scheduler::do_one( task_interrupted_ = task_timeout_us == 0; task_running_.store(true, std::memory_order_release); - if (more_handlers) + // Wake a peer to take the pending handlers while this thread + // polls the reactor; skipped when one_thread_ (no peer exists). + if (more_handlers && !one_thread_) unlock_and_signal_one(lock); try @@ -937,11 +939,16 @@ reactor_scheduler::do_one( { bool more = !completed_ops_.empty(); - if (more) + if (more && !one_thread_) + { + // Wake a peer for the remaining work; unassisted if none + // was parked to take it. ctx->unassisted = !unlock_and_signal_one(lock); + } else { - ctx->unassisted = false; + // No peer to wake (one_thread_, or nothing more queued). + ctx->unassisted = more; lock.unlock(); } diff --git a/include/boost/corosio/native/detail/select/select_scheduler.hpp b/include/boost/corosio/native/detail/select/select_scheduler.hpp index 566929915..b0489c6b4 100644 --- a/include/boost/corosio/native/detail/select/select_scheduler.hpp +++ b/include/boost/corosio/native/detail/select/select_scheduler.hpp @@ -225,7 +225,7 @@ select_scheduler::register_descriptor( desc->registered_events = reactor_event_read | reactor_event_write; desc->fd = fd; desc->scheduler_ = this; - desc->mutex.set_enabled(!single_threaded_); + desc->mutex.set_enabled(reactor_io_locking_); desc->ready_events_.store(0, std::memory_order_relaxed); { diff --git a/perf/bench/corosio/accept_churn_bench.cpp b/perf/bench/corosio/accept_churn_bench.cpp index 13003b467..1a50304a5 100644 --- a/perf/bench/corosio/accept_churn_bench.cpp +++ b/perf/bench/corosio/accept_churn_bench.cpp @@ -132,7 +132,7 @@ bench_sequential_churn_lockless(bench::state& state) using acceptor_type = corosio::native_tcp_acceptor; corosio::io_context_options opts; - opts.single_threaded = true; + opts.locking = corosio::locking_mode::unsafe; corosio::native_io_context ioc(opts, 1); acceptor_type acc(ioc); acc.open(); @@ -397,7 +397,7 @@ bench_burst_churn_lockless(bench::state& state) state.counters["burst_size"] = burst_size; corosio::io_context_options opts; - opts.single_threaded = true; + opts.locking = corosio::locking_mode::unsafe; corosio::native_io_context ioc(opts, 1); acceptor_type acc(ioc); acc.open(); diff --git a/perf/bench/corosio/fan_out_bench.cpp b/perf/bench/corosio/fan_out_bench.cpp index 582ab7a71..b13443c8a 100644 --- a/perf/bench/corosio/fan_out_bench.cpp +++ b/perf/bench/corosio/fan_out_bench.cpp @@ -332,7 +332,7 @@ bench_fork_join_lockless(bench::state& state) state.counters["fan_out"] = fan_out; corosio::io_context_options opts; - opts.single_threaded = true; + opts.locking = corosio::locking_mode::unsafe; corosio::native_io_context ioc(opts, 1); std::vector clients; @@ -403,7 +403,7 @@ bench_nested_lockless(bench::state& state) state.counters["subs_per_group"] = subs_per_group; corosio::io_context_options opts; - opts.single_threaded = true; + opts.locking = corosio::locking_mode::unsafe; corosio::native_io_context ioc(opts, 1); std::vector clients; @@ -487,7 +487,7 @@ bench_concurrent_parents_lockless(bench::state& state) state.counters["fan_out"] = fan_out; corosio::io_context_options opts; - opts.single_threaded = true; + opts.locking = corosio::locking_mode::unsafe; corosio::native_io_context ioc(opts, 1); std::vector clients; diff --git a/perf/bench/corosio/http_server_bench.cpp b/perf/bench/corosio/http_server_bench.cpp index bca5ba3be..c135572a6 100644 --- a/perf/bench/corosio/http_server_bench.cpp +++ b/perf/bench/corosio/http_server_bench.cpp @@ -179,7 +179,7 @@ bench_single_connection_lockless(bench::state& state) using socket_type = corosio::native_tcp_socket; corosio::io_context_options opts; - opts.single_threaded = true; + opts.locking = corosio::locking_mode::unsafe; corosio::native_io_context ioc(opts, 1); auto [client, server] = corosio::test::make_socket_pair< socket_type, corosio::native_tcp_acceptor>(ioc); diff --git a/perf/bench/corosio/io_context_bench.cpp b/perf/bench/corosio/io_context_bench.cpp index c99d4dab5..83e615a72 100644 --- a/perf/bench/corosio/io_context_bench.cpp +++ b/perf/bench/corosio/io_context_bench.cpp @@ -263,7 +263,7 @@ void bench_single_threaded_lockless(bench::state& state) { corosio::io_context_options opts; - opts.single_threaded = true; + opts.locking = corosio::locking_mode::unsafe; corosio::native_io_context ioc(opts, 1); auto ex = ioc.get_executor(); @@ -295,7 +295,7 @@ void bench_interleaved_lockless(bench::state& state) { corosio::io_context_options opts; - opts.single_threaded = true; + opts.locking = corosio::locking_mode::unsafe; int handlers_per_iteration = 100; diff --git a/perf/bench/corosio/local_socket_latency_bench.cpp b/perf/bench/corosio/local_socket_latency_bench.cpp index 50cd2aac3..f26085604 100644 --- a/perf/bench/corosio/local_socket_latency_bench.cpp +++ b/perf/bench/corosio/local_socket_latency_bench.cpp @@ -167,7 +167,7 @@ bench_unix_pingpong_latency_lockless(bench::state& state) state.counters["message_size"] = static_cast(message_size); corosio::io_context_options opts; - opts.single_threaded = true; + opts.locking = corosio::locking_mode::unsafe; corosio::native_io_context ioc(opts, 1); socket_type client(ioc), server(ioc); if (auto ec = corosio::connect_pair(client, server)) @@ -201,7 +201,7 @@ bench_unix_concurrent_latency_lockless(bench::state& state) state.counters["num_pairs"] = num_pairs; corosio::io_context_options opts; - opts.single_threaded = true; + opts.locking = corosio::locking_mode::unsafe; corosio::native_io_context ioc(opts, 1); std::vector clients; diff --git a/perf/bench/corosio/local_socket_throughput_bench.cpp b/perf/bench/corosio/local_socket_throughput_bench.cpp index 6c9c0af64..e1efa3dc9 100644 --- a/perf/bench/corosio/local_socket_throughput_bench.cpp +++ b/perf/bench/corosio/local_socket_throughput_bench.cpp @@ -195,7 +195,7 @@ bench_unix_throughput_lockless(bench::state& state) state.counters["chunk_size"] = static_cast(chunk_size); corosio::io_context_options opts; - opts.single_threaded = true; + opts.locking = corosio::locking_mode::unsafe; corosio::native_io_context ioc(opts, 1); socket_type writer(ioc), reader(ioc); if (auto ec = corosio::connect_pair(writer, reader)) @@ -259,7 +259,7 @@ bench_unix_bidirectional_throughput_lockless(bench::state& state) state.counters["chunk_size"] = static_cast(chunk_size); corosio::io_context_options opts; - opts.single_threaded = true; + opts.locking = corosio::locking_mode::unsafe; corosio::native_io_context ioc(opts, 1); socket_type sock1(ioc), sock2(ioc); if (auto ec = corosio::connect_pair(sock1, sock2)) diff --git a/perf/bench/corosio/socket_latency_bench.cpp b/perf/bench/corosio/socket_latency_bench.cpp index 872736d0c..be9bccc69 100644 --- a/perf/bench/corosio/socket_latency_bench.cpp +++ b/perf/bench/corosio/socket_latency_bench.cpp @@ -167,7 +167,7 @@ bench_pingpong_latency_lockless(bench::state& state) state.counters["message_size"] = static_cast(message_size); corosio::io_context_options opts; - opts.single_threaded = true; + opts.locking = corosio::locking_mode::unsafe; corosio::native_io_context ioc(opts, 1); auto [client, server] = corosio::test::make_socket_pair< socket_type, corosio::native_tcp_acceptor>(ioc); @@ -203,7 +203,7 @@ bench_concurrent_latency_lockless(bench::state& state) state.counters["num_pairs"] = num_pairs; corosio::io_context_options opts; - opts.single_threaded = true; + opts.locking = corosio::locking_mode::unsafe; corosio::native_io_context ioc(opts, 1); std::vector clients; diff --git a/perf/bench/corosio/socket_throughput_bench.cpp b/perf/bench/corosio/socket_throughput_bench.cpp index c29f2247c..e135cc859 100644 --- a/perf/bench/corosio/socket_throughput_bench.cpp +++ b/perf/bench/corosio/socket_throughput_bench.cpp @@ -254,7 +254,7 @@ bench_throughput_lockless(bench::state& state) state.counters["chunk_size"] = static_cast(chunk_size); corosio::io_context_options opts; - opts.single_threaded = true; + opts.locking = corosio::locking_mode::unsafe; corosio::native_io_context ioc(opts, 1); auto [writer, reader] = corosio::test::make_socket_pair< socket_type, corosio::native_tcp_acceptor>(ioc); @@ -320,7 +320,7 @@ bench_bidirectional_throughput_lockless(bench::state& state) state.counters["chunk_size"] = static_cast(chunk_size); corosio::io_context_options opts; - opts.single_threaded = true; + opts.locking = corosio::locking_mode::unsafe; corosio::native_io_context ioc(opts, 1); auto [sock1, sock2] = corosio::test::make_socket_pair< socket_type, corosio::native_tcp_acceptor>(ioc); diff --git a/src/corosio/src/io_context.cpp b/src/corosio/src/io_context.cpp index 6f0258e71..eff780408 100644 --- a/src/corosio/src/io_context.cpp +++ b/src/corosio/src/io_context.cpp @@ -170,21 +170,30 @@ pre_create_services( (void)opts; } -// Apply runtime tuning to the scheduler after construction. -// -// Concurrency-hint heuristic for budget defaults: when the io_context is -// constructed with concurrency_hint > 1 AND the user has not customized -// the budget settings (i.e. they remain at the struct defaults), we -// disable the inline-completion fast path. Multi-thread workloads -// benefit from "always-post" because cross-thread work-stealing wins -// over chained dispatch on the originating thread. Single-thread (or -// any custom budget) keeps the user/library setting unchanged. +// Map the locking tier to the scheduler's threading facilities. one_thread is +// set only for the lockless tiers, where a single run thread is guaranteed. +detail::scheduler::threading_config +make_threading_config(io_context_options const& opts) +{ + detail::scheduler::threading_config cfg; + cfg.scheduler_locking = opts.locking != locking_mode::unsafe; + cfg.reactor_io_locking = opts.locking == locking_mode::safe; + cfg.one_thread = opts.locking != locking_mode::safe; + return cfg; +} + +// Apply runtime tuning after construction. `concurrency_hint` is the effective +// hint (normalized to 1 for lockless tiers). Budget heuristic: with default +// budgets and hint > 1, disable the inline-completion fast path so multi-thread +// runs post everything for cross-thread work-stealing. void apply_scheduler_options( detail::scheduler& sched, io_context_options const& opts, unsigned concurrency_hint) { + sched.configure_threading(make_threading_config(opts)); + #if BOOST_COROSIO_HAS_EPOLL || BOOST_COROSIO_HAS_KQUEUE || BOOST_COROSIO_HAS_SELECT // dynamic_cast — when io_uring is also linked, the runtime probe may // have selected io_uring_scheduler instead of a reactor_scheduler. @@ -192,7 +201,7 @@ apply_scheduler_options( dynamic_cast(&sched)) { // Detect "user kept the defaults" by comparing all three to the - // io_context_options-defined struct defaults. + // io_context-options-defined struct defaults. io_context_options defaults; bool budget_at_defaults = opts.inline_budget_initial == defaults.inline_budget_initial && @@ -216,8 +225,6 @@ apply_scheduler_options( init, max, ua); - if (opts.single_threaded) - reactor->configure_single_threaded(true); } #endif @@ -225,20 +232,12 @@ apply_scheduler_options( if (auto* uring_sched = dynamic_cast(&sched)) { - if (opts.single_threaded) - uring_sched->configure_single_threaded(true); if (opts.enable_sqpoll) uring_sched->configure_sqpoll( true, opts.sq_thread_idle_ms, opts.sq_thread_cpu); } #endif -#if BOOST_COROSIO_HAS_IOCP - auto& iocp_sched = static_cast(sched); - if (opts.single_threaded) - iocp_sched.configure_single_threaded(true); -#endif - (void)sched; (void)opts; (void)concurrency_hint; @@ -258,19 +257,10 @@ construct_default(capy::execution_context& ctx, unsigned concurrency_hint) #endif } -// Tie concurrency_hint == 1 to single_threaded (asio precedent). -io_context_options -normalize_options(io_context_options opts, unsigned concurrency_hint) -{ - if (concurrency_hint == 1) - opts.single_threaded = true; - return opts; -} - } // anonymous namespace io_context::io_context() - : io_context(std::max(2u, std::thread::hardware_concurrency())) + : io_context(std::max(1u, std::thread::hardware_concurrency())) { } @@ -278,8 +268,9 @@ io_context::io_context(unsigned concurrency_hint) : capy::execution_context(this) , sched_(&construct_default(*this, concurrency_hint)) { - if (concurrency_hint == 1) - configure_single_threaded_(); + // Threading config only; the plain path leaves the reactor budget at its + // defaults (no options-ctor budget heuristic). + apply_threading_(io_context_options{}); } io_context::io_context( @@ -288,10 +279,13 @@ io_context::io_context( : capy::execution_context(this) , sched_(nullptr) { - auto opts = normalize_options(opts_in, concurrency_hint); - pre_create_services(*this, opts); - sched_ = &construct_default(*this, concurrency_hint); - apply_scheduler_options(*sched_, opts, concurrency_hint); + pre_create_services(*this, opts_in); + // Computed before construct_default so IOCP's completion port is created + // with the effective concurrency. + unsigned const eff = + detail::effective_concurrency_hint(opts_in, concurrency_hint); + sched_ = &construct_default(*this, eff); + apply_scheduler_options(*sched_, opts_in, eff); } void @@ -305,18 +299,13 @@ io_context::apply_options_post_( io_context_options const& opts_in, unsigned concurrency_hint) { - auto opts = normalize_options(opts_in, concurrency_hint); - apply_scheduler_options(*sched_, opts, concurrency_hint); + apply_scheduler_options(*sched_, opts_in, concurrency_hint); } void -io_context::configure_single_threaded_() +io_context::apply_threading_(io_context_options const& opts_in) { - // Dispatched through the scheduler base's virtual override; avoids - // unsafe downcasts when the active backend is io_uring rather than - // reactor (on Linux both BOOST_COROSIO_HAS_EPOLL and the io_uring - // backend may be enabled simultaneously). - sched_->configure_single_threaded(true); + sched_->configure_threading(make_threading_config(opts_in)); } io_context::~io_context() diff --git a/test/unit/io_context.cpp b/test/unit/io_context.cpp index 38be71691..a12233aff 100644 --- a/test/unit/io_context.cpp +++ b/test/unit/io_context.cpp @@ -17,6 +17,8 @@ #include #include +#include +#include #include #include #include @@ -124,6 +126,66 @@ make_atomic_coro(std::atomic& counter) return c; } +// Shared state for the peer-utilization barrier (see testHintOneWakesPeers). +struct gate_state +{ + std::mutex mtx; + std::condition_variable cv; + int active = 0; // gates currently in the barrier + int peak = 0; // max concurrent gates observed + int release_at = 0; // release once this many overlap +}; + +// Coroutine that, when resumed, blocks until `release_at` gates are running +// concurrently (or a deadline elapses). Observing peak >= 2 requires two +// gates to run on distinct run threads at once, i.e. a parked peer must have +// been woken. Note we release at 2 rather than the gate count: one run thread +// is always occupied servicing the reactor poll, so full N-way overlap is not +// expected. +struct gate_coro +{ + struct promise_type + { + gate_state* st = nullptr; + + gate_coro get_return_object() + { + return {std::coroutine_handle::from_promise(*this)}; + } + + std::suspend_always initial_suspend() noexcept { return {}; } + std::suspend_never final_suspend() noexcept { return {}; } + + void return_void() + { + std::unique_lock lk(st->mtx); + int cur = ++st->active; + if (cur > st->peak) + st->peak = cur; + if (st->active >= st->release_at) + st->cv.notify_all(); + else + st->cv.wait_for(lk, std::chrono::seconds(2), + [&] { return st->active >= st->release_at; }); + --st->active; + } + + void unhandled_exception() { std::terminate(); } + }; + + std::coroutine_handle h; + + operator std::coroutine_handle<>() const { return h; } +}; + +inline gate_coro +make_gate(gate_state& st) +{ + auto c = []() -> gate_coro { co_return; }(); + c.h.promise().st = &st; + return c; +} + // Coroutine whose promise destructor increments a counter. // Both initial_suspend and final_suspend return suspend_always so the // frame is only freed by an explicit .destroy() call. @@ -299,9 +361,46 @@ struct io_context_test void testConstructionSingleThreaded() { - // concurrency_hint == 1 enables single-threaded mode automatically. + // Lockless mode is enabled solely by opts.locking; the concurrency + // hint is unrelated to the safety contract. + io_context_options opts; + opts.locking = locking_mode::unsafe; + io_context ioc(Backend, opts, 1); + BOOST_TEST(!ioc.stopped()); + + int counter = 0; + auto ex = ioc.get_executor(); + post_coro(ex, make_coro(counter)); + std::size_t n = ioc.run(); + BOOST_TEST(n == 1); + BOOST_TEST(counter == 1); + } + + // A lockless tier normalizes the effective concurrency hint to 1 for + // downstream tuning (reactor budget heuristic, IOCP concurrency), + // regardless of the hint the caller passed; the safe tier passes it + // through unchanged. (Independent of Backend; runs per backend harmlessly.) + void testEffectiveConcurrencyHint() + { + io_context_options opts; + + opts.locking = locking_mode::safe; + BOOST_TEST(detail::effective_concurrency_hint(opts, 8) == 8u); + BOOST_TEST(detail::effective_concurrency_hint(opts, 1) == 1u); + + opts.locking = locking_mode::unsafe_io; + BOOST_TEST(detail::effective_concurrency_hint(opts, 8) == 1u); + + opts.locking = locking_mode::unsafe; + BOOST_TEST(detail::effective_concurrency_hint(opts, 8) == 1u); + } + + void testConstructionUnsafeIo() + { + // The unsafe_io tier elides only per-descriptor I/O locks; the + // scheduler still runs work normally on its single thread. io_context_options opts; - opts.single_threaded = true; + opts.locking = locking_mode::unsafe_io; io_context ioc(Backend, opts, 1); BOOST_TEST(!ioc.stopped()); @@ -642,6 +741,81 @@ struct io_context_test BOOST_TEST(counter.load() == total_handlers); } + void testHintOneIsThreadSafe() + { + // Regression for #310: a plain concurrency_hint of 1 must NOT + // enable lockless mode. Cross-thread post() into a hint==1 + // context must remain thread-safe (TSan-clean); lockless mode is + // reachable only via io_context_options::locking. + io_context ioc(Backend, 1); + auto ex = ioc.get_executor(); + std::atomic counter{0}; + constexpr int num_threads = 4; + constexpr int handlers_per_thread = 100; + constexpr int total_handlers = num_threads * handlers_per_thread; + + // Post handlers from multiple threads concurrently. + std::vector posters; + posters.reserve(num_threads); + for (int t = 0; t < num_threads; ++t) + { + posters.emplace_back([&ex, &counter]() { + for (int i = 0; i < handlers_per_thread; ++i) + post_coro(ex, make_atomic_coro(counter)); + }); + } + for (auto& t : posters) + t.join(); + + // Run with multiple threads. + std::vector runners; + runners.reserve(num_threads); + for (int t = 0; t < num_threads; ++t) + runners.emplace_back([&ioc]() { ioc.run(); }); + for (auto& t : runners) + t.join(); + + BOOST_TEST(counter.load() == total_handlers); + } + + void testHintOneWakesPeers() + { + // Regression: at concurrency_hint == 1 in the default (safe) tier, + // multiple run() threads must still be utilized. Internally-generated + // work has to wake parked peers — the one_thread wake-elision must NOT + // be engaged by the hint (only by a lockless tier). Each gate holds + // until two gates overlap, so peak >= 2 is reachable only if a parked + // peer is woken; if the hint wrongly elides peer wakeups the gates run + // sequentially (each times out) and peak stays 1. + constexpr int num_gates = 4; + constexpr int num_threads = 4; + io_context ioc(Backend, 1); + + gate_state st; + st.release_at = 2; + + // A leader runs on one thread, lets the other run() threads park, + // then generates the gates from inside the loop (internal work). + auto leader = [](io_context::executor_type ex, + gate_state& state, int gates) -> capy::task<> { + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + for (int i = 0; i < gates; ++i) + ex.post(make_gate(state)); + co_return; + }; + capy::run_async(ioc.get_executor())( + leader(ioc.get_executor(), st, num_gates)); + + std::vector runners; + runners.reserve(num_threads); + for (int t = 0; t < num_threads; ++t) + runners.emplace_back([&ioc]() { ioc.run(); }); + for (auto& t : runners) + t.join(); + + BOOST_TEST(st.peak >= 2); + } + void testMultithreadedStress() { // Stress test: multiple iterations of post-then-run with multiple threads @@ -794,6 +968,8 @@ struct io_context_test testConstructionWithOptions(); testConstructionWithThreadPoolSize(); testConstructionSingleThreaded(); + testConstructionUnsafeIo(); + testEffectiveConcurrencyHint(); testGetExecutor(); testRun(); testRunOne(); @@ -809,6 +985,8 @@ struct io_context_test testRunOneForWithOutstandingWork(); testExecutorRunningInThisThread(); testMultithreaded(); + testHintOneIsThreadSafe(); + testHintOneWakesPeers(); testMultithreadedStress(); testMultithreadedNotifyAndWaitFor(); testWhenAllSetEvent(); diff --git a/test/unit/resolver.cpp b/test/unit/resolver.cpp index 39fac0baf..898e7eeff 100644 --- a/test/unit/resolver.cpp +++ b/test/unit/resolver.cpp @@ -330,7 +330,9 @@ struct resolver_test { // Single-threaded contexts disable the resolver thread pool and // surface operation_not_supported instead of dispatching work. - io_context ioc(1); + io_context_options opts; + opts.locking = locking_mode::unsafe; + io_context ioc(opts, 1); resolver r(ioc); bool completed = false; @@ -359,7 +361,9 @@ struct resolver_test void testReverseResolveSingleThreadedNotSupported() { - io_context ioc(1); + io_context_options opts; + opts.locking = locking_mode::unsafe; + io_context ioc(opts, 1); resolver r(ioc); bool completed = false; @@ -384,6 +388,35 @@ struct resolver_test #endif } + void testResolveUnsafeIoStillSupported() + { + // The unsafe_io tier disables only the per-descriptor I/O locks; + // scheduler locking stays on, so the resolver thread pool remains + // available (matching asio, whose resolver restriction keys on the + // scheduler lock, not UNSAFE_IO). Resolution must NOT short-circuit + // with operation_not_supported. + io_context_options opts; + opts.locking = locking_mode::unsafe_io; + io_context ioc(opts, 1); + resolver r(ioc); + + bool completed = false; + std::error_code result_ec; + + auto task = [](resolver& r_ref, + std::error_code& ec_out, bool& done) -> capy::task<> { + auto [ec, res] = co_await r_ref.resolve("localhost", "80"); + ec_out = ec; + done = true; + (void)res; + }; + capy::run_async(ioc.get_executor())(task(r, result_ec, completed)); + ioc.run(); + + BOOST_TEST(completed); + BOOST_TEST(result_ec != std::errc::operation_not_supported); + } + void testResolveInvalidFlagsCombination() { // numeric_service with a non-numeric service should produce EAI_NONAME @@ -1079,6 +1112,7 @@ struct resolver_test testResolveWithVariedFlags(); testResolveSingleThreadedNotSupported(); testReverseResolveSingleThreadedNotSupported(); + testResolveUnsafeIoStillSupported(); testReverseResolveDatagramFlag(); // Cancellation diff --git a/test/unit/stream_file.cpp b/test/unit/stream_file.cpp index e036bee06..f313174ca 100644 --- a/test/unit/stream_file.cpp +++ b/test/unit/stream_file.cpp @@ -535,7 +535,9 @@ struct stream_file_test // POSIX file I/O requires the shared thread pool; in single-threaded // mode the service short-circuits with operation_not_supported. temp_file tmp("sf_st_", "data"); - io_context ioc(Backend, 1); + io_context_options opts; + opts.locking = locking_mode::unsafe; + io_context ioc(Backend, opts, 1); stream_file f(ioc); bool caught = false; @@ -549,6 +551,37 @@ struct stream_file_test } BOOST_TEST(caught); } + + void testOpenUnsafeIoStillSupported() + { +#if BOOST_COROSIO_HAS_IO_URING + if constexpr (std::is_same_v< + std::remove_const_t, io_uring_t>) + return; +#endif + // The unsafe_io tier keeps scheduler locking on, so the shared file + // thread pool remains available: open() must succeed (matching asio, + // whose file restriction keys on the scheduler lock, not UNSAFE_IO). + temp_file tmp("sf_st_io_", "data"); + io_context_options opts; + opts.locking = locking_mode::unsafe_io; + io_context ioc(Backend, opts, 1); + stream_file f(ioc); + + bool opened = false; + std::error_code caught_ec; + try + { + f.open(tmp.path, file_base::read_only); + opened = true; + } + catch (std::system_error const& e) + { + caught_ec = e.code(); + } + BOOST_TEST(opened); + BOOST_TEST(caught_ec != std::errc::operation_not_supported); + } #endif void testReadEmptyBuffer() @@ -839,6 +872,7 @@ struct stream_file_test testWriteEmptyBuffer(); #if BOOST_COROSIO_POSIX testOpenSingleThreadedNotSupported(); + testOpenUnsafeIoStillSupported(); #endif testTruncate();