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 33 : effective_concurrency_hint(
187 : io_context_options const& opts, unsigned hint) noexcept
188 : {
189 33 : 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 : explicit io_context(
282 : io_context_options const& opts,
283 : unsigned concurrency_hint = std::thread::hardware_concurrency());
284 :
285 : /** Construct with an explicit backend tag.
286 :
287 : @param backend The backend tag value selecting the I/O
288 : multiplexer (e.g. `corosio::epoll`).
289 : @param concurrency_hint Hint for the number of threads
290 : that will call `run()`.
291 : */
292 : template<class Backend>
293 : requires requires { Backend::construct; }
294 1150 : explicit io_context(
295 : Backend backend,
296 : unsigned concurrency_hint = std::thread::hardware_concurrency())
297 : : capy::execution_context(this)
298 1150 : , sched_(nullptr)
299 : {
300 : (void)backend;
301 1150 : sched_ = &Backend::construct(*this, concurrency_hint);
302 : // Apply threading config only (locking tier). Unlike the options
303 : // ctor, the plain path leaves the reactor budget at its defaults.
304 1150 : apply_threading_(io_context_options{});
305 1150 : }
306 :
307 : /** Construct with an explicit backend tag and runtime options.
308 :
309 : @param backend The backend tag value selecting the I/O
310 : multiplexer (e.g. `corosio::epoll`).
311 : @param opts Runtime options controlling scheduler and
312 : service behavior.
313 : @param concurrency_hint Hint for the number of threads
314 : that will call `run()`.
315 : */
316 : template<class Backend>
317 : requires requires { Backend::construct; }
318 18 : explicit io_context(
319 : Backend backend,
320 : io_context_options const& opts,
321 : unsigned concurrency_hint = std::thread::hardware_concurrency())
322 : : capy::execution_context(this)
323 18 : , sched_(nullptr)
324 : {
325 : (void)backend;
326 18 : apply_options_pre_(opts);
327 : // Effective hint (1 for lockless tiers); see effective_concurrency_hint.
328 : unsigned const eff =
329 18 : detail::effective_concurrency_hint(opts, concurrency_hint);
330 18 : sched_ = &Backend::construct(*this, eff);
331 18 : apply_options_post_(opts, eff);
332 18 : }
333 :
334 : ~io_context();
335 :
336 : io_context(io_context const&) = delete;
337 : io_context& operator=(io_context const&) = delete;
338 :
339 : /** Return an executor for this context.
340 :
341 : The returned executor can be used to dispatch coroutines
342 : and post work items to this context.
343 :
344 : @return An executor associated with this context.
345 : */
346 : executor_type get_executor() const noexcept;
347 :
348 : /** Signal the context to stop processing.
349 :
350 : This causes `run()` to return as soon as possible. Any pending
351 : work items remain queued.
352 : */
353 4 : void stop()
354 : {
355 4 : sched_->stop();
356 4 : }
357 :
358 : /** Return whether the context has been stopped.
359 :
360 : @return `true` if `stop()` has been called and `restart()`
361 : has not been called since.
362 : */
363 68 : bool stopped() const noexcept
364 : {
365 68 : return sched_->stopped();
366 : }
367 :
368 : /** Restart the context after being stopped.
369 :
370 : This function must be called before `run()` can be called
371 : again after `stop()` has been called.
372 : */
373 181 : void restart()
374 : {
375 181 : sched_->restart();
376 181 : }
377 :
378 : /** Process all pending work items.
379 :
380 : This function blocks until all pending work items have been
381 : executed or `stop()` is called. The context is stopped
382 : when there is no more outstanding work.
383 :
384 : @note The context must be restarted with `restart()` before
385 : calling this function again after it returns.
386 :
387 : @return The number of handlers executed.
388 : */
389 902 : std::size_t run()
390 : {
391 902 : return sched_->run();
392 : }
393 :
394 : /** Process at most one pending work item.
395 :
396 : This function blocks until one work item has been executed
397 : or `stop()` is called. The context is stopped when there
398 : is no more outstanding work.
399 :
400 : @note The context must be restarted with `restart()` before
401 : calling this function again after it returns.
402 :
403 : @return The number of handlers executed (0 or 1).
404 : */
405 22 : std::size_t run_one()
406 : {
407 22 : return sched_->run_one();
408 : }
409 :
410 : /** Process work items for the specified duration.
411 :
412 : This function blocks until work items have been executed for
413 : the specified duration, or `stop()` is called. The context
414 : is stopped when there is no more outstanding work.
415 :
416 : @note The context must be restarted with `restart()` before
417 : calling this function again after it returns.
418 :
419 : @param rel_time The duration for which to process work.
420 :
421 : @return The number of handlers executed.
422 : */
423 : template<class Rep, class Period>
424 10 : std::size_t run_for(std::chrono::duration<Rep, Period> const& rel_time)
425 : {
426 10 : return run_until(std::chrono::steady_clock::now() + rel_time);
427 : }
428 :
429 : /** Process work items until the specified time.
430 :
431 : This function blocks until the specified time is reached
432 : or `stop()` is called. The context is stopped when there
433 : is no more outstanding work.
434 :
435 : @note The context must be restarted with `restart()` before
436 : calling this function again after it returns.
437 :
438 : @param abs_time The time point until which to process work.
439 :
440 : @return The number of handlers executed.
441 : */
442 : template<class Clock, class Duration>
443 : std::size_t
444 10 : run_until(std::chrono::time_point<Clock, Duration> const& abs_time)
445 : {
446 10 : std::size_t n = 0;
447 28 : while (run_one_until(abs_time))
448 18 : if (n != (std::numeric_limits<std::size_t>::max)())
449 18 : ++n;
450 10 : return n;
451 : }
452 :
453 : /** Process at most one work item for the specified duration.
454 :
455 : This function blocks until one work item has been executed,
456 : the specified duration has elapsed, or `stop()` is called.
457 : The context is stopped when there is no more outstanding work.
458 :
459 : @note The context must be restarted with `restart()` before
460 : calling this function again after it returns.
461 :
462 : @param rel_time The duration for which the call may block.
463 :
464 : @return The number of handlers executed (0 or 1).
465 : */
466 : template<class Rep, class Period>
467 6 : std::size_t run_one_for(std::chrono::duration<Rep, Period> const& rel_time)
468 : {
469 6 : return run_one_until(std::chrono::steady_clock::now() + rel_time);
470 : }
471 :
472 : /** Process at most one work item until the specified time.
473 :
474 : This function blocks until one work item has been executed,
475 : the specified time is reached, or `stop()` is called.
476 : The context is stopped when there is no more outstanding work.
477 :
478 : @note The context must be restarted with `restart()` before
479 : calling this function again after it returns.
480 :
481 : @param abs_time The time point until which the call may block.
482 :
483 : @return The number of handlers executed (0 or 1).
484 : */
485 : template<class Clock, class Duration>
486 : std::size_t
487 42 : run_one_until(std::chrono::time_point<Clock, Duration> const& abs_time)
488 : {
489 42 : typename Clock::time_point now = Clock::now();
490 8 : for (;;)
491 : {
492 50 : auto rel_time = abs_time - now;
493 : using rel_type = decltype(rel_time);
494 50 : if (rel_time < rel_type::zero())
495 4 : rel_time = rel_type::zero();
496 46 : else if (rel_time > std::chrono::seconds(1))
497 22 : rel_time = std::chrono::seconds(1);
498 :
499 50 : std::size_t s = sched_->wait_one(
500 : static_cast<long>(
501 50 : std::chrono::duration_cast<std::chrono::microseconds>(
502 : rel_time)
503 50 : .count()));
504 :
505 50 : if (s || stopped())
506 42 : return s;
507 :
508 12 : now = Clock::now();
509 12 : if (now >= abs_time)
510 4 : return 0;
511 : }
512 : }
513 :
514 : /** Process all ready work items without blocking.
515 :
516 : This function executes all work items that are ready to run
517 : without blocking for more work. The context is stopped
518 : when there is no more outstanding work.
519 :
520 : @note The context must be restarted with `restart()` before
521 : calling this function again after it returns.
522 :
523 : @return The number of handlers executed.
524 : */
525 14 : std::size_t poll()
526 : {
527 14 : return sched_->poll();
528 : }
529 :
530 : /** Process at most one ready work item without blocking.
531 :
532 : This function executes at most one work item that is ready
533 : to run without blocking for more work. The context is
534 : stopped when there is no more outstanding work.
535 :
536 : @note The context must be restarted with `restart()` before
537 : calling this function again after it returns.
538 :
539 : @return The number of handlers executed (0 or 1).
540 : */
541 8 : std::size_t poll_one()
542 : {
543 8 : return sched_->poll_one();
544 : }
545 : };
546 :
547 : /** An executor for dispatching work to an I/O context.
548 :
549 : The executor provides the interface for posting work items and
550 : dispatching coroutines to the associated context. It satisfies
551 : the `capy::Executor` concept.
552 :
553 : Executors are lightweight handles that can be copied and compared
554 : for equality. Two executors compare equal if they refer to the
555 : same context.
556 :
557 : @par Thread Safety
558 : Distinct objects: Safe.@n
559 : Shared objects: Safe.
560 : */
561 : class io_context::executor_type
562 : {
563 : io_context* ctx_ = nullptr;
564 :
565 : public:
566 : /** Default constructor.
567 :
568 : Constructs an executor not associated with any context.
569 : */
570 2040 : executor_type() = default;
571 :
572 : /** Construct an executor from a context.
573 :
574 : @param ctx The context to associate with this executor.
575 : */
576 3253 : explicit executor_type(io_context& ctx) noexcept : ctx_(&ctx) {}
577 :
578 : /** Return a reference to the associated execution context.
579 :
580 : @return Reference to the context.
581 : */
582 16551 : io_context& context() const noexcept
583 : {
584 16551 : return *ctx_;
585 : }
586 :
587 : /** Check if the current thread is running this executor's context.
588 :
589 : @return `true` if `run()` is being called on this thread.
590 : */
591 6131 : bool running_in_this_thread() const noexcept
592 : {
593 6131 : return ctx_->sched_->running_in_this_thread();
594 : }
595 :
596 : /** Informs the executor that work is beginning.
597 :
598 : Must be paired with `on_work_finished()`.
599 : */
600 6413 : void on_work_started() const noexcept
601 : {
602 6413 : ctx_->sched_->work_started();
603 6413 : }
604 :
605 : /** Informs the executor that work has completed.
606 :
607 : @par Preconditions
608 : A preceding call to `on_work_started()` on an equal executor.
609 : */
610 6367 : void on_work_finished() const noexcept
611 : {
612 6367 : ctx_->sched_->work_finished();
613 6367 : }
614 :
615 : /** Dispatch a continuation.
616 :
617 : Returns a handle for symmetric transfer. If called from
618 : within `run()`, returns `c.h`. Otherwise posts `c` for
619 : later execution and returns `std::noop_coroutine()`.
620 :
621 : @param c The continuation to dispatch.
622 :
623 : @return A handle for symmetric transfer or `std::noop_coroutine()`.
624 :
625 : @par Preconditions
626 : The associated context must outlive this call. Dispatching
627 : concurrently with, or after, the context's destruction is
628 : undefined behavior.
629 : */
630 6127 : std::coroutine_handle<> dispatch(capy::continuation& c) const
631 : {
632 6127 : if (running_in_this_thread())
633 673 : return c.h;
634 5454 : post(c);
635 5454 : return std::noop_coroutine();
636 : }
637 :
638 : /** Post a continuation for deferred execution.
639 :
640 : Enqueues `c` directly on the scheduler's ready queue.
641 : No heap allocation occurs.
642 :
643 : @par Preconditions
644 : The associated context must outlive this call. Posting
645 : concurrently with, or after, the context's destruction is
646 : undefined behavior.
647 : */
648 14215 : void post(capy::continuation& c) const
649 : {
650 14215 : ctx_->sched_->post(c);
651 14215 : }
652 :
653 : /** Post a bare coroutine handle for deferred execution.
654 :
655 : Heap-allocates a scheduler_op to wrap the handle. A caller
656 : that already owns a `scheduler_op` can post it directly via
657 : the `post(scheduler_op*)` overload to avoid the allocation.
658 :
659 : @param h The coroutine handle to post.
660 :
661 : @par Preconditions
662 : The associated context must outlive this call. Posting
663 : concurrently with, or after, the context's destruction is
664 : undefined behavior.
665 : */
666 3686 : void post(std::coroutine_handle<> h) const
667 : {
668 3686 : ctx_->sched_->post(h);
669 3686 : }
670 :
671 : /** Compare two executors for equality.
672 :
673 : @return `true` if both executors refer to the same context.
674 : */
675 2 : bool operator==(executor_type const& other) const noexcept
676 : {
677 2 : return ctx_ == other.ctx_;
678 : }
679 :
680 : /** Compare two executors for inequality.
681 :
682 : @return `true` if the executors refer to different contexts.
683 : */
684 : bool operator!=(executor_type const& other) const noexcept
685 : {
686 : return ctx_ != other.ctx_;
687 : }
688 : };
689 :
690 : inline io_context::executor_type
691 3253 : io_context::get_executor() const noexcept
692 : {
693 3253 : return executor_type(const_cast<io_context&>(*this));
694 : }
695 :
696 : } // namespace boost::corosio
697 :
698 : #endif // BOOST_COROSIO_IO_CONTEXT_HPP
|
