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