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