TLA Line data Source code
1 : //
2 : // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
3 : // Copyright (c) 2026 Steve Gerbino
4 : // Copyright (c) 2026 Michael Vandeberg
5 : //
6 : // Distributed under the Boost Software License, Version 1.0. (See accompanying
7 : // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
8 : //
9 : // Official repository: https://github.com/cppalliance/corosio
10 : //
11 :
12 : #ifndef BOOST_COROSIO_IO_CONTEXT_HPP
13 : #define BOOST_COROSIO_IO_CONTEXT_HPP
14 :
15 : #include <boost/corosio/detail/config.hpp>
16 : #include <boost/corosio/detail/platform.hpp>
17 : #include <boost/corosio/detail/scheduler.hpp>
18 : #include <boost/capy/continuation.hpp>
19 : #include <boost/capy/ex/execution_context.hpp>
20 :
21 : #include <chrono>
22 : #include <coroutine>
23 : #include <cstddef>
24 : #include <limits>
25 : #include <thread>
26 :
27 : namespace boost::corosio {
28 :
29 : /** Locking-safety tier for an @ref io_context.
30 :
31 : Selects which internal locks the scheduler and reactor elide, trading
32 : thread-safety guarantees for reduced synchronization overhead. This is
33 : the analog of Boost.Asio's `SAFE` / `UNSAFE_IO` / `UNSAFE` concurrency
34 : hint constants. The tier is chosen explicitly, not derived from the
35 : `concurrency_hint`. (The reverse does apply: a lockless tier reduces the
36 : effective hint used for performance tuning to 1.)
37 :
38 : @see io_context_options::locking
39 : */
40 : enum class locking_mode
41 : {
42 : /** Full thread safety (default). All locks enabled; equivalent to
43 : Boost.Asio's `SAFE`/`DEFAULT`. Any thread may use the context. */
44 : safe,
45 :
46 : /** Disable only the per-descriptor I/O locks; keep scheduler locking.
47 : Equivalent to Boost.Asio's `UNSAFE_IO`. The context must be run
48 : and driven by a single thread, but resolver and POSIX file
49 : services remain available (they rely on scheduler locking, which
50 : stays on). */
51 : unsafe_io,
52 :
53 : /** Disable all locking (fully lockless). Equivalent to Boost.Asio's
54 : `UNSAFE`.
55 :
56 : @par Restrictions
57 : - Only one thread may call `run()` (or any run variant).
58 : - Posting work from another thread is undefined behavior.
59 : - DNS resolution returns `operation_not_supported`.
60 : - POSIX file I/O returns `operation_not_supported`.
61 : - Signal sets should not be shared across contexts. */
62 : unsafe
63 : };
64 :
65 : /** Runtime tuning options for @ref io_context.
66 :
67 : All fields have defaults that match the library's built-in
68 : values, so constructing a default `io_context_options` produces
69 : identical behavior to an unconfigured context.
70 :
71 : Options that apply only to a specific backend family are
72 : silently ignored when the active backend does not support them.
73 :
74 : @par Example
75 : @par !example configure
76 :
77 : @see io_context, native_io_context
78 : */
79 : struct io_context_options
80 : {
81 : /** Maximum events fetched per reactor poll call.
82 :
83 : Controls the buffer size passed to `epoll_wait()` or
84 : `kevent()`. Larger values reduce syscall frequency under
85 : high load; smaller values improve fairness between
86 : connections. Ignored on IOCP and select backends.
87 : */
88 : unsigned max_events_per_poll = 128;
89 :
90 : /** Starting inline completion budget per handler chain.
91 :
92 : After a posted handler executes, the reactor grants this
93 : many speculative inline completions before forcing a
94 : re-queue. Applies to reactor backends only.
95 :
96 : @note Constructing an `io_context` with `concurrency_hint > 1`
97 : and all three budget fields at their defaults overrides
98 : them to disable inline completion (post-everything mode),
99 : since multi-thread workloads benefit from cross-thread
100 : work-stealing. Setting any budget field to a non-default
101 : value disables the override.
102 : */
103 : unsigned inline_budget_initial = 2;
104 :
105 : /** Hard ceiling on adaptive inline budget ramp-up.
106 :
107 : The budget doubles each cycle it is fully consumed, up to
108 : this limit. Applies to reactor backends only.
109 : */
110 : unsigned inline_budget_max = 16;
111 :
112 : /** Inline budget when no other thread assists the reactor.
113 :
114 : When only one thread is running the event loop, this
115 : value caps the inline budget to preserve fairness.
116 : Applies to reactor backends only.
117 : */
118 : unsigned unassisted_budget = 4;
119 :
120 : /** Thread pool size for blocking I/O (file I/O, DNS resolution).
121 :
122 : Sets the number of worker threads in the shared thread pool
123 : used by POSIX file services and DNS resolution. Must be at
124 : least 1. Applies to POSIX backends only; ignored on IOCP
125 : where file I/O uses native overlapped I/O.
126 : */
127 : unsigned thread_pool_size = 1;
128 :
129 : /** Thread-safety tier. See @ref locking_mode for the tiers and their
130 : restrictions.
131 : */
132 : locking_mode locking = locking_mode::safe;
133 :
134 : /** Enable IORING_SETUP_SQPOLL on the io_uring backend.
135 :
136 : With SQPOLL, the kernel forks a thread that busy-polls the
137 : submission ring; submission becomes a userspace-only memory
138 : store, eliminating the io_uring_enter syscall on the submit
139 : path. Most useful for sustained traffic. Idle thread parks
140 : after `sq_thread_idle_ms` of no activity.
141 :
142 : Independent of `locking`. Default: off.
143 :
144 : Ignored on non-io_uring backends.
145 : */
146 : bool enable_sqpoll = false;
147 :
148 : /** SQ-poll idle timeout in milliseconds.
149 :
150 : After this many ms of no submissions, the kernel polling
151 : thread sleeps; next submit re-wakes it via SQ_WAKEUP. 0
152 : means use the kernel default (1ms). Recommended for bursty
153 : workloads: 100-1000ms (avoids park/unpark thrash).
154 :
155 : Ignored unless `enable_sqpoll` is true. Ignored on
156 : non-io_uring backends.
157 : */
158 : unsigned sq_thread_idle_ms = 0;
159 :
160 : /** Pin the SQ-poll kernel thread to this CPU.
161 :
162 : -1 means do not pin (kernel scheduler picks). Pinning off
163 : the dispatch core is recommended on latency-sensitive
164 : deployments to avoid cache contention.
165 :
166 : Ignored unless `enable_sqpoll` is true. Ignored on
167 : non-io_uring backends.
168 : */
169 : int sq_thread_cpu = -1;
170 : };
171 :
172 : namespace detail {
173 : class timer_service;
174 :
175 : /** Return the hint used for performance tuning: the lockless tiers are
176 : single-threaded, so their effective hint is 1 whatever the caller passed.
177 : */
178 : inline unsigned
179 HIT 44 : effective_concurrency_hint(
180 : io_context_options const& opts, unsigned hint) noexcept
181 : {
182 44 : return opts.locking == locking_mode::safe ? hint : 1u;
183 : }
184 : } // namespace detail
185 :
186 : /** An I/O context for running asynchronous operations.
187 :
188 : The io_context provides an execution environment for async
189 : operations. It maintains a queue of pending work items and
190 : processes them when `run()` is called.
191 :
192 : The default and unsigned constructors select the platform's
193 : native backend:
194 : - Windows: IOCP
195 : - Linux: epoll
196 : - BSD/macOS: kqueue
197 : - Other POSIX: select
198 :
199 : The template constructor accepts a backend tag value to
200 : choose a specific backend at compile time:
201 :
202 : @par Example
203 : @par !example construct
204 :
205 : @par Preconditions
206 : The context must outlive every operation posted or dispatched
207 : through its executor, and no thread may be executing a run
208 : variant when the context is destroyed. Posting to the context
209 : concurrently with, or after, its destruction is undefined
210 : behavior. The safe teardown pattern is to stop submitting new
211 : work, let every `run()` call return (each returns once no
212 : outstanding work remains), and join the threads that ran the
213 : loop before destroying the context. Work launched with
214 : `capy::run` / `capy::run_async` is work-tracked, so a normal
215 : `run()` completion already waits for it.
216 :
217 : @par Exception Safety
218 : A context that constructs is usable. The infrastructure its
219 : backend needs — the completion port, the ring, the reactor's
220 : wakeup channel — is created during construction, so a system that
221 : refuses it throws from the constructor rather than from the first
222 : operation, and the failed construction leaves nothing open.
223 :
224 : @par Thread Safety
225 : Distinct objects: Safe.@n
226 : Shared objects: Safe, unless the context was constructed with a
227 : lockless @ref io_context_options::locking tier (`unsafe_io` or
228 : `unsafe`), in which case a single thread must drive it.
229 :
230 : @see epoll_t, select_t, kqueue_t, iocp_t
231 : */
232 : class BOOST_COROSIO_DECL io_context : public capy::execution_context
233 : {
234 : /// Reject invalid options before the backend is constructed.
235 : void apply_options_pre_(io_context_options const& opts);
236 :
237 : /** Create the blocking-I/O thread pool, apply runtime tuning to the
238 : scheduler and finish bringing the backend up. The tail of every
239 : options constructor: the backend infrastructure whose setup reads
240 : these options is created here, so a failure to create it throws
241 : from the constructor. */
242 : void apply_options_post_(
243 : io_context_options const& opts,
244 : unsigned concurrency_hint);
245 :
246 : /** Create the blocking-I/O thread pool and apply only the decomposed
247 : threading configuration (locking tiers), then finish bringing the
248 : backend up. The tail of every plain constructor, which — unlike
249 : the options constructors — deliberately leaves the reactor budget
250 : at its defaults rather than engaging the multi-thread
251 : post-everything heuristic. */
252 : void apply_threading_(io_context_options const& opts);
253 :
254 : protected:
255 : detail::scheduler* sched_;
256 :
257 : public:
258 : /** The executor type for this context. */
259 : class executor_type;
260 :
261 : /** Construct with default concurrency and platform backend.
262 :
263 : Uses `std::thread::hardware_concurrency()` (floored to 1, in
264 : case it reports 0) as the concurrency hint, and the default
265 : @ref locking_mode::safe tier. Select a lockless tier via
266 : @ref io_context_options::locking.
267 :
268 : @throws std::system_error If the backend's infrastructure
269 : could not be created.
270 : */
271 : io_context();
272 :
273 : /** Construct with a concurrency hint and platform backend.
274 :
275 : @param concurrency_hint Hint for the number of threads
276 : that will call `run()`.
277 :
278 : @throws std::system_error If the backend's infrastructure
279 : could not be created.
280 : */
281 : explicit io_context(unsigned concurrency_hint);
282 :
283 : /** Construct with runtime tuning options and platform backend.
284 :
285 : @param opts Runtime options controlling scheduler and
286 : service behavior.
287 : @param concurrency_hint Hint for the number of threads
288 : that will call `run()`.
289 :
290 : @throws std::invalid_argument If `opts.thread_pool_size` is
291 : less than 1 (POSIX).
292 :
293 : @throws std::system_error If the backend's infrastructure
294 : could not be created.
295 : */
296 : explicit io_context(
297 : io_context_options const& opts,
298 : unsigned concurrency_hint = std::thread::hardware_concurrency());
299 :
300 : /** Construct with an explicit backend tag.
301 :
302 : @param backend The backend tag value selecting the I/O
303 : multiplexer (e.g. `corosio::epoll`).
304 : @param concurrency_hint Hint for the number of threads
305 : that will call `run()`.
306 :
307 : @throws std::system_error If the backend's infrastructure
308 : could not be created.
309 : */
310 : template<class Backend>
311 : requires requires { Backend::construct; }
312 1719 : explicit io_context(
313 : [[maybe_unused]] Backend backend,
314 : unsigned concurrency_hint = std::thread::hardware_concurrency())
315 : : capy::execution_context(this)
316 1719 : , sched_(nullptr)
317 : {
318 1719 : sched_ = &Backend::construct(*this, concurrency_hint);
319 : // Apply threading config only (locking tier). Unlike the options
320 : // ctor, the plain path leaves the reactor budget at its defaults.
321 1707 : apply_threading_(io_context_options{});
322 1719 : }
323 :
324 : /** Construct with an explicit backend tag and runtime options.
325 :
326 : @param backend The backend tag value selecting the I/O
327 : multiplexer (e.g. `corosio::epoll`).
328 : @param opts Runtime options controlling scheduler and
329 : service behavior.
330 : @param concurrency_hint Hint for the number of threads
331 : that will call `run()`.
332 :
333 : @throws std::invalid_argument If `opts.thread_pool_size` is
334 : less than 1 (POSIX).
335 :
336 : @throws std::system_error If the backend's infrastructure
337 : could not be created.
338 : */
339 : template<class Backend>
340 : requires requires { Backend::construct; }
341 27 : explicit io_context(
342 : [[maybe_unused]] Backend backend,
343 : io_context_options const& opts,
344 : unsigned concurrency_hint = std::thread::hardware_concurrency())
345 : : capy::execution_context(this)
346 27 : , sched_(nullptr)
347 : {
348 27 : apply_options_pre_(opts);
349 : // Effective hint (1 for lockless tiers); see effective_concurrency_hint.
350 : unsigned const eff =
351 27 : detail::effective_concurrency_hint(opts, concurrency_hint);
352 27 : sched_ = &Backend::construct(*this, eff);
353 27 : apply_options_post_(opts, eff);
354 27 : }
355 :
356 : ~io_context();
357 :
358 : io_context(io_context const&) = delete;
359 : io_context& operator=(io_context const&) = delete;
360 :
361 : /** Return an executor for this context.
362 :
363 : The returned executor can be used to dispatch coroutines
364 : and post work items to this context.
365 :
366 : @return An executor associated with this context.
367 : */
368 : executor_type get_executor() const noexcept;
369 :
370 : /** Signal the context to stop processing.
371 :
372 : This causes `run()` to return as soon as possible. Any pending
373 : work items remain queued.
374 : */
375 13 : void stop()
376 : {
377 13 : sched_->stop();
378 13 : }
379 :
380 : /** Return whether the context has been stopped.
381 :
382 : @return `true` if `stop()` has been called and `restart()`
383 : has not been called since.
384 : */
385 117 : bool stopped() const noexcept
386 : {
387 117 : return sched_->stopped();
388 : }
389 :
390 : /** Restart the context after being stopped.
391 :
392 : This function must be called before `run()` can be called
393 : again after `stop()` has been called.
394 : */
395 455 : void restart()
396 : {
397 455 : sched_->restart();
398 455 : }
399 :
400 : /** Process all pending work items.
401 :
402 : This function blocks until all pending work items have been
403 : executed or `stop()` is called. The context is stopped
404 : when there is no more outstanding work.
405 :
406 : @note The context must be restarted with `restart()` before
407 : calling this function again after it returns.
408 :
409 : @return The number of handlers executed.
410 : */
411 1777 : std::size_t run()
412 : {
413 1777 : return sched_->run();
414 : }
415 :
416 : /** Process at most one pending work item.
417 :
418 : This function blocks until one work item has been executed
419 : or `stop()` is called. The context is stopped when there
420 : is no more outstanding work.
421 :
422 : @note The context must be restarted with `restart()` before
423 : calling this function again after it returns.
424 :
425 : @return The number of handlers executed (0 or 1).
426 : */
427 112 : std::size_t run_one()
428 : {
429 112 : return sched_->run_one();
430 : }
431 :
432 : /** Process work items for the specified duration.
433 :
434 : This function blocks until work items have been executed for
435 : the specified duration, or `stop()` is called. The context
436 : is stopped when there is no more outstanding work.
437 :
438 : @note The context must be restarted with `restart()` before
439 : calling this function again after it returns.
440 :
441 : @param rel_time The duration for which to process work.
442 :
443 : @return The number of handlers executed.
444 : */
445 : template<class Rep, class Period>
446 15 : std::size_t run_for(std::chrono::duration<Rep, Period> const& rel_time)
447 : {
448 15 : return run_until(std::chrono::steady_clock::now() + rel_time);
449 : }
450 :
451 : /** Process work items until the specified time.
452 :
453 : This function blocks until the specified time is reached
454 : or `stop()` is called. The context is stopped when there
455 : is no more outstanding work.
456 :
457 : @note The context must be restarted with `restart()` before
458 : calling this function again after it returns.
459 :
460 : @param abs_time The time point until which to process work.
461 :
462 : @return The number of handlers executed.
463 : */
464 : template<class Clock, class Duration>
465 : std::size_t
466 16 : run_until(std::chrono::time_point<Clock, Duration> const& abs_time)
467 : {
468 16 : std::size_t n = 0;
469 43 : while (run_one_until(abs_time))
470 27 : if (n != (std::numeric_limits<std::size_t>::max)())
471 27 : ++n;
472 16 : return n;
473 : }
474 :
475 : /** Process at most one work item for the specified duration.
476 :
477 : This function blocks until one work item has been executed,
478 : the specified duration has elapsed, or `stop()` is called.
479 : The context is stopped when there is no more outstanding work.
480 :
481 : @note The context must be restarted with `restart()` before
482 : calling this function again after it returns.
483 :
484 : @param rel_time The duration for which the call may block.
485 :
486 : @return The number of handlers executed (0 or 1).
487 : */
488 : template<class Rep, class Period>
489 75 : std::size_t run_one_for(std::chrono::duration<Rep, Period> const& rel_time)
490 : {
491 75 : return run_one_until(std::chrono::steady_clock::now() + rel_time);
492 : }
493 :
494 : /** Process at most one work item until the specified time.
495 :
496 : This function blocks until one work item has been executed,
497 : the specified time is reached, or `stop()` is called.
498 : The context is stopped when there is no more outstanding work.
499 :
500 : @note The context must be restarted with `restart()` before
501 : calling this function again after it returns.
502 :
503 : @param abs_time The time point until which the call may block.
504 :
505 : @return The number of handlers executed (0 or 1).
506 : */
507 : template<class Clock, class Duration>
508 : std::size_t
509 126 : run_one_until(std::chrono::time_point<Clock, Duration> const& abs_time)
510 : {
511 126 : typename Clock::time_point now = Clock::now();
512 27 : for (;;)
513 : {
514 153 : auto rel_time = abs_time - now;
515 : using rel_type = decltype(rel_time);
516 153 : if (rel_time < rel_type::zero())
517 5 : rel_time = rel_type::zero();
518 148 : else if (rel_time > std::chrono::seconds(1))
519 38 : rel_time = std::chrono::seconds(1);
520 :
521 153 : std::size_t s = sched_->wait_one(
522 : static_cast<long>(
523 153 : std::chrono::duration_cast<std::chrono::microseconds>(
524 : rel_time)
525 153 : .count()));
526 :
527 153 : if (s || stopped())
528 126 : return s;
529 :
530 53 : now = Clock::now();
531 53 : if (now >= abs_time)
532 26 : return 0;
533 : }
534 : }
535 :
536 : /** Process all ready work items without blocking.
537 :
538 : This function executes all work items that are ready to run
539 : without blocking for more work. The context is stopped
540 : when there is no more outstanding work.
541 :
542 : @note The context must be restarted with `restart()` before
543 : calling this function again after it returns.
544 :
545 : @return The number of handlers executed.
546 : */
547 47 : std::size_t poll()
548 : {
549 47 : return sched_->poll();
550 : }
551 :
552 : /** Process at most one ready work item without blocking.
553 :
554 : This function executes at most one work item that is ready
555 : to run without blocking for more work. The context is
556 : stopped when there is no more outstanding work.
557 :
558 : @note The context must be restarted with `restart()` before
559 : calling this function again after it returns.
560 :
561 : @return The number of handlers executed (0 or 1).
562 : */
563 11 : std::size_t poll_one()
564 : {
565 11 : return sched_->poll_one();
566 : }
567 : };
568 :
569 : /** An executor for dispatching work to an I/O context.
570 :
571 : The executor provides the interface for posting work items and
572 : dispatching coroutines to the associated context. It satisfies
573 : the `capy::Executor` concept.
574 :
575 : Executors are lightweight handles that can be copied and compared
576 : for equality. Two executors compare equal if they refer to the
577 : same context.
578 :
579 : @par Thread Safety
580 : Distinct objects: Safe.@n
581 : Shared objects: Safe.
582 : */
583 : class io_context::executor_type
584 : {
585 : io_context* ctx_ = nullptr;
586 :
587 : public:
588 : /** Default constructor.
589 :
590 : Constructs an executor not associated with any context.
591 : */
592 2053 : executor_type() = default;
593 :
594 : /** Construct an executor from a context.
595 :
596 : @param ctx The context to associate with this executor.
597 : */
598 4284 : explicit executor_type(io_context& ctx) noexcept : ctx_(&ctx) {}
599 :
600 : /** Return a reference to the associated execution context.
601 :
602 : @return Reference to the context.
603 : */
604 28045 : io_context& context() const noexcept
605 : {
606 28045 : return *ctx_;
607 : }
608 :
609 : /** Check if the current thread is running this executor's context.
610 :
611 : @return `true` if `run()` is being called on this thread.
612 : */
613 9439 : bool running_in_this_thread() const noexcept
614 : {
615 9439 : return ctx_->sched_->running_in_this_thread();
616 : }
617 :
618 : /** Informs the executor that work is beginning.
619 :
620 : Must be paired with `on_work_finished()`.
621 : */
622 9640 : void on_work_started() const noexcept
623 : {
624 9640 : ctx_->sched_->work_started();
625 9640 : }
626 :
627 : /** Informs the executor that work has completed.
628 :
629 : @par Preconditions
630 : A preceding call to `on_work_started()` on an equal executor.
631 : */
632 9578 : void on_work_finished() const noexcept
633 : {
634 9578 : ctx_->sched_->work_finished();
635 9578 : }
636 :
637 : /** Dispatch a continuation.
638 :
639 : Returns a handle for symmetric transfer. If called from
640 : within `run()`, returns `c.h`. Otherwise posts `c` for
641 : later execution and returns `std::noop_coroutine()`.
642 :
643 : @param c The continuation to dispatch.
644 :
645 : @return A handle for symmetric transfer or `std::noop_coroutine()`.
646 :
647 : @par Preconditions
648 : The associated context must outlive this call. Dispatching
649 : concurrently with, or after, the context's destruction is
650 : undefined behavior.
651 : */
652 9434 : std::coroutine_handle<> dispatch(capy::continuation& c) const
653 : {
654 9434 : if (running_in_this_thread())
655 939 : return c.h;
656 8495 : post(c);
657 8495 : return std::noop_coroutine();
658 : }
659 :
660 : /** Post a continuation for deferred execution.
661 :
662 : Enqueues `c` directly on the scheduler's ready queue.
663 : No heap allocation occurs.
664 :
665 : @par Preconditions
666 : The associated context must outlive this call. Posting
667 : concurrently with, or after, the context's destruction is
668 : undefined behavior.
669 : */
670 25723 : void post(capy::continuation& c) const
671 : {
672 25723 : ctx_->sched_->post(c);
673 25723 : }
674 :
675 : /** Post a bare coroutine handle for deferred execution.
676 :
677 : Heap-allocates a scheduler_op to wrap the handle. A caller
678 : that already owns a `scheduler_op` can post it directly via
679 : the `post(scheduler_op*)` overload to avoid the allocation.
680 :
681 : @param h The coroutine handle to post.
682 :
683 : @par Preconditions
684 : The associated context must outlive this call. Posting
685 : concurrently with, or after, the context's destruction is
686 : undefined behavior.
687 : */
688 3756 : void post(std::coroutine_handle<> h) const
689 : {
690 3756 : ctx_->sched_->post(h);
691 3756 : }
692 :
693 : /** Compare two executors for equality.
694 :
695 : @return `true` if both executors refer to the same context.
696 : */
697 2 : bool operator==(executor_type const& other) const noexcept
698 : {
699 2 : return ctx_ == other.ctx_;
700 : }
701 :
702 : /** Compare two executors for inequality.
703 :
704 : @return `true` if the executors refer to different contexts.
705 : */
706 : bool operator!=(executor_type const& other) const noexcept
707 : {
708 : return ctx_ != other.ctx_;
709 : }
710 : };
711 :
712 : inline io_context::executor_type
713 4284 : io_context::get_executor() const noexcept
714 : {
715 4284 : return executor_type(const_cast<io_context&>(*this));
716 : }
717 :
718 : } // namespace boost::corosio
719 :
720 : #endif // BOOST_COROSIO_IO_CONTEXT_HPP
|
