TLA Line data Source code
1 : //
2 : // Copyright (c) 2026 Steve Gerbino
3 : //
4 : // Distributed under the Boost Software License, Version 1.0. (See accompanying
5 : // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6 : //
7 : // Official repository: https://github.com/cppalliance/corosio
8 : //
9 :
10 : #ifndef BOOST_COROSIO_DETAIL_TIMEOUT_AWAITABLE_HPP
11 : #define BOOST_COROSIO_DETAIL_TIMEOUT_AWAITABLE_HPP
12 :
13 : #include <boost/corosio/io_context.hpp>
14 : #include <boost/corosio/detail/timeout_coro.hpp>
15 : #include <boost/corosio/detail/timer.hpp>
16 : #include <boost/corosio/detail/except.hpp>
17 : #include <boost/capy/cond.hpp>
18 : #include <boost/capy/error.hpp>
19 : #include <boost/capy/ex/frame_allocator.hpp>
20 : #include <boost/capy/ex/await_context.hpp>
21 : #include <boost/capy/ex/recycling_memory_resource.hpp>
22 : #include <boost/capy/results.hpp>
23 :
24 : #include <chrono>
25 : #include <coroutine>
26 : #include <new>
27 : #include <optional>
28 : #include <stdexcept>
29 : #include <stop_token>
30 : #include <type_traits>
31 : #include <utility>
32 :
33 : /* Races an inner ContextAwaitable against a timer via a shared
34 : stop_source. await_suspend arms the timer by launching a
35 : fire-and-forget timeout_coro, then starts the inner op with
36 : an interposed stop_token. Whichever completes first signals
37 : the stop_source, cancelling the other.
38 :
39 : Parent cancellation is forwarded through a stop_callback
40 : stored in a placement-new buffer (stop_callback is not
41 : movable, but the awaitable must be movable for
42 : transform_awaiter). The buffer is inert during moves
43 : (before await_suspend) and constructed in-place once the
44 : awaitable is pinned on the coroutine frame.
45 :
46 : The timeout_coro can outlive this awaitable — it owns its
47 : env and self-destroys via suspend_never. The timer lives in
48 : std::optional and is constructed lazily in await_suspend,
49 : once the awaiting coroutine's executor context is known. */
50 :
51 : namespace boost::corosio::detail {
52 :
53 : // Local stand-in for capy::detail's `results` trait: corosio must not
54 : // reach into capy::detail, but the result-mapping switch in
55 : // await_resume needs to distinguish `results` from other return types.
56 : template<typename T>
57 : struct is_results : std::false_type
58 : {};
59 :
60 : template<typename... Ts>
61 : struct is_results<capy::results<Ts...>> : std::true_type
62 : {};
63 :
64 : template<typename T>
65 : inline constexpr bool is_results_v = is_results<T>::value;
66 :
67 : /** Awaitable adapter that cancels an inner operation after a deadline.
68 :
69 : Races the inner awaitable against a timer. A shared stop_source
70 : ties them together: whichever completes first cancels the other.
71 : Parent cancellation is forwarded via stop_callback.
72 :
73 : The timer is constructed internally in `await_suspend` from the
74 : execution context in `await_context`.
75 :
76 : @tparam A The inner ContextAwaitable type (decayed).
77 : */
78 : template<typename A>
79 : struct timeout_awaitable
80 : {
81 : struct stop_forwarder
82 : {
83 : std::stop_source* src_;
84 HIT 1883 : void operator()() const noexcept
85 : {
86 1883 : src_->request_stop();
87 1883 : }
88 : };
89 :
90 : using time_point = std::chrono::steady_clock::time_point;
91 : using stop_cb_type = std::stop_callback<stop_forwarder>;
92 :
93 : // Declared before inner_: a task destroyed while lent reads the
94 : // environment it was given.
95 : std::stop_source stop_src_;
96 : std::stop_token parent_token_;
97 : capy::await_context inner_env_;
98 : A inner_;
99 : std::optional<timer> timer_;
100 : time_point deadline_;
101 : std::chrono::nanoseconds dur_{};
102 : alignas(stop_cb_type) unsigned char cb_buf_[sizeof(stop_cb_type)];
103 : bool has_deadline_ = true;
104 : bool cb_active_ = false;
105 :
106 : /// Construct without a timer, deadline given as an absolute time.
107 4 : timeout_awaitable(A&& inner, time_point deadline)
108 4 : : inner_(std::move(inner))
109 8 : , deadline_(deadline)
110 : {
111 4 : }
112 :
113 : /// Construct without a timer, deadline measured from suspension.
114 2059 : timeout_awaitable(A&& inner, std::chrono::nanoseconds dur)
115 2059 : : inner_(std::move(inner))
116 2059 : , dur_(dur)
117 4118 : , has_deadline_(false)
118 : {
119 2059 : }
120 :
121 4128 : ~timeout_awaitable()
122 : {
123 4128 : destroy_parent_cb();
124 4128 : }
125 :
126 : // Only moved before await_suspend, when cb_active_ is false
127 2065 : timeout_awaitable(timeout_awaitable&& o) noexcept(
128 : std::is_nothrow_move_constructible_v<A>)
129 2065 : : stop_src_(std::move(o.stop_src_))
130 2065 : , inner_(std::move(o.inner_))
131 2065 : , timer_(std::move(o.timer_))
132 2065 : , deadline_(o.deadline_)
133 2065 : , dur_(o.dur_)
134 2065 : , has_deadline_(o.has_deadline_)
135 : {
136 2065 : }
137 :
138 : timeout_awaitable(timeout_awaitable const&) = delete;
139 : timeout_awaitable& operator=(timeout_awaitable const&) = delete;
140 : timeout_awaitable& operator=(timeout_awaitable&&) = delete;
141 :
142 : // Forwarding here is load-bearing, not an optimization: awaitables
143 : // may perform setup in await_ready (type-erased stream wrappers
144 : // construct their cached inner op there), so the full awaiter
145 : // protocol must reach inner_ before await_suspend is driven. An
146 : // already-ready inner op also skips arming the timer entirely.
147 2061 : bool await_ready()
148 : {
149 2061 : return inner_.await_ready();
150 : }
151 :
152 2059 : auto await_suspend(std::coroutine_handle<> h, capy::await_context const* env)
153 : {
154 2059 : parent_token_ = env->stop_token;
155 :
156 : // The deadline timer is built here from the awaiting
157 : // coroutine's executor context, the first point at which it
158 : // is known. await_suspend is driven through a noexcept
159 : // wrapper, so a failure cannot be surfaced as a catchable
160 : // exception. An executor whose context is not an io_context
161 : // cannot supply a timer service; silently running the
162 : // operation with no deadline would be a worse failure than
163 : // aborting, so translate the service-lookup error into a
164 : // clear precondition diagnostic. This terminates by design
165 : // (a usage error) rather than dropping the requested timeout.
166 : // The detached timeout coroutine must own its executor by
167 : // value (see timeout_coro::set_env_owned); await_context carries
168 : // only a non-owning executor_ref. Recover the concrete
169 : // executor from the context rather than the executor_ref:
170 : // wrapped executors (a strand over the io_context) satisfy
171 : // the documented precondition but do not expose the io
172 : // executor as their target. The timer construction below
173 : // validates the context is an io_context, and the detached
174 : // coroutine shares only the thread-safe stop_source with
175 : // the caller, so resuming it on the raw io executor instead
176 : // of the caller's wrapper is safe.
177 : try
178 : {
179 2059 : timer_.emplace(env->executor.context());
180 : }
181 4 : catch (std::logic_error const&)
182 : {
183 2 : throw_logic_error("timeout requires an io_context-backed executor");
184 : }
185 : auto ex =
186 2057 : static_cast<io_context&>(env->executor.context()).get_executor();
187 :
188 2057 : if (has_deadline_)
189 4 : timer_->expires_at(deadline_);
190 : else
191 2053 : timer_->expires_after(dur_);
192 :
193 : // The detached coroutine can outlive this chain and its frame
194 : // allocator, so it allocates from the process-wide recycling
195 : // resource instead; the chain's allocator is put back for the
196 : // inner operation.
197 2057 : auto* const rmr = capy::get_recycling_memory_resource();
198 2057 : auto* const prev = capy::get_current_frame_allocator();
199 2057 : capy::set_current_frame_allocator(rmr);
200 : // Launch fire-and-forget timeout (starts suspended)
201 2057 : auto timeout = make_timeout(*timer_, stop_src_);
202 2057 : timeout.h_.promise().set_env_owned(ex, stop_src_.get_token(), rmr);
203 : // Runs synchronously until timer.wait() suspends
204 2057 : timeout.h_.resume();
205 2057 : capy::set_current_frame_allocator(prev);
206 : // timeout goes out of scope; destructor is a no-op,
207 : // the coroutine self-destroys via suspend_never
208 :
209 : // Forward parent cancellation
210 2057 : new (cb_buf_) stop_cb_type(env->stop_token, stop_forwarder{&stop_src_});
211 2057 : cb_active_ = true;
212 :
213 : // Start the inner op with our interposed stop_token
214 2057 : inner_env_ = {
215 2057 : env->executor, stop_src_.get_token(), env->frame_allocator,
216 2057 : env->owner};
217 4114 : return inner_.await_suspend(h, &inner_env_);
218 2057 : }
219 :
220 2055 : decltype(auto) await_resume()
221 : {
222 : // Read before request_stop: afterwards stop_requested()
223 : // can no longer distinguish who fired first. This must also
224 : // happen before inner_.await_resume() rather than after: when
225 : // the inner awaitable is itself a timeout_awaitable (nested
226 : // timeout()), our own request_stop() below is visible through
227 : // its parent_token_ (aliasing our stop_src_), and would
228 : // otherwise make its read of "parent" look like a
229 : // cancellation that never happened.
230 2055 : bool const parent = parent_token_.stop_requested();
231 2055 : bool const fired = stop_src_.stop_requested();
232 :
233 : // If inner_.await_resume() throws below, request_stop() is
234 : // skipped; the still-armed timeout coroutine is then drained
235 : // by timer_'s destructor rather than by us.
236 2055 : auto r = inner_.await_resume();
237 :
238 : // Cancel whichever is still pending (idempotent)
239 2053 : stop_src_.request_stop();
240 2053 : destroy_parent_cb();
241 :
242 : // Deadline won: stop_src_ is assumed to be the only
243 : // cancellation source, whose only writers are the timer
244 : // coroutine and the parent forwarder, so fired && !parent
245 : // identifies a timeout. A third-party cancellation of the
246 : // inner op (e.g. a socket cancel issued from elsewhere)
247 : // landing in the same window as the deadline firing is
248 : // reported as a timeout.
249 2053 : if (fired && !parent && std::get<0>(r) == capy::cond::canceled)
250 : {
251 24 : std::remove_cvref_t<decltype(r)> t{};
252 24 : std::get<0>(t) = make_error_code(capy::error::timeout);
253 24 : return t;
254 : }
255 2029 : return r;
256 : }
257 :
258 6181 : void destroy_parent_cb() noexcept
259 : {
260 6181 : if (cb_active_)
261 : {
262 2057 : std::launder(reinterpret_cast<stop_cb_type*>(cb_buf_))
263 2057 : ->~stop_cb_type();
264 2057 : cb_active_ = false;
265 : }
266 6181 : }
267 : };
268 :
269 : } // namespace boost::corosio::detail
270 :
271 : #endif
|
