TLA Line data Source code
1 : //
2 : // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
3 : // Copyright (c) 2026 Michael Vandeberg
4 : //
5 : // Distributed under the Boost Software License, Version 1.0. (See accompanying
6 : // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7 : //
8 : // Official repository: https://github.com/cppalliance/capy
9 : //
10 :
11 : #ifndef BOOST_CAPY_IO_WRITE_NOW_HPP
12 : #define BOOST_CAPY_IO_WRITE_NOW_HPP
13 :
14 : #include <boost/capy/detail/config.hpp>
15 : #include <boost/capy/detail/await_suspend_helper.hpp>
16 : #include <boost/capy/buffers.hpp>
17 : #include <boost/capy/buffers/consuming_buffers.hpp>
18 : #include <boost/capy/concept/context_awaitable.hpp>
19 : #include <boost/capy/concept/write_stream.hpp>
20 : #include <coroutine>
21 : #include <boost/capy/ex/executor_ref.hpp>
22 : #include <boost/capy/ex/frame_allocator.hpp>
23 : #include <boost/capy/ex/abandon.hpp>
24 : #include <boost/capy/ex/await_context.hpp>
25 : #include <boost/capy/results.hpp>
26 :
27 : #include <cstddef>
28 : #include <exception>
29 : #include <new>
30 : #include <stop_token>
31 : #include <utility>
32 :
33 : #ifndef BOOST_CAPY_WRITE_NOW_WORKAROUND
34 : # if defined(__GNUC__) && !defined(__clang__)
35 : # define BOOST_CAPY_WRITE_NOW_WORKAROUND 1
36 : # else
37 : # define BOOST_CAPY_WRITE_NOW_WORKAROUND 0
38 : # endif
39 : #endif
40 :
41 : namespace boost {
42 : namespace capy {
43 :
44 : /** Eagerly writes complete buffer sequences with frame caching.
45 :
46 : This class wraps a @ref WriteStream and provides an `operator()`
47 : that writes an entire buffer sequence, attempting to complete
48 : synchronously. If every `write_some` completes without suspending,
49 : the entire operation finishes in `await_ready` with no coroutine
50 : suspension.
51 :
52 : The class maintains a one-element coroutine frame cache. After
53 : the first call, subsequent calls reuse the cached frame memory,
54 : avoiding repeated allocation for the internal coroutine.
55 :
56 : @tparam Stream The stream type, must satisfy @ref WriteStream.
57 :
58 : @par Thread Safety
59 : Distinct objects: Safe.
60 : Shared objects: Unsafe.
61 :
62 : @par Preconditions
63 : Only one operation may be outstanding at a time. A new call to
64 : `operator()` must not be made until the previous operation has
65 : completed (i.e., the returned awaitable is fully consumed).
66 :
67 : Destroying the returned operation object while it is being awaited
68 : calls `std::terminate`, except while its chain is unwound after an
69 : @ref abandon.
70 :
71 : @par Example
72 :
73 : @code
74 : template< WriteStream Stream >
75 : task<> send_messages( Stream& stream )
76 : {
77 : write_now wn( stream );
78 : auto [ec1, n1] = co_await wn( make_buffer( "hello" ) );
79 : if( ec1 )
80 : detail::throw_system_error( ec1 );
81 : auto [ec2, n2] = co_await wn( make_buffer( "world" ) );
82 : if( ec2 )
83 : detail::throw_system_error( ec2 );
84 : }
85 : @endcode
86 :
87 : @see @ref write, @ref write_some, @ref WriteStream, @ref ConstBufferSequence
88 : */
89 : template<class Stream>
90 : requires WriteStream<Stream>
91 : class write_now
92 : {
93 : Stream& stream_;
94 : void* cached_frame_ = nullptr;
95 : std::size_t cached_size_ = 0;
96 :
97 : struct [[nodiscard]] BOOST_CAPY_CORO_AWAIT_ELIDABLE
98 : op_type
99 : {
100 : struct promise_type
101 : {
102 : results<std::size_t> result_;
103 : std::exception_ptr ep_;
104 : std::coroutine_handle<> cont_{nullptr};
105 : await_context const* env_ = nullptr;
106 : bool done_ = false;
107 :
108 : // The awaiting op_type while a parent awaits this frame.
109 : op_type* owner_ = nullptr;
110 :
111 HIT 76 : op_type get_return_object()
112 : {
113 : return op_type{
114 : std::coroutine_handle<
115 76 : promise_type>::from_promise(*this)};
116 : }
117 :
118 76 : auto initial_suspend() noexcept
119 : {
120 : #if BOOST_CAPY_WRITE_NOW_WORKAROUND
121 76 : return std::suspend_always{};
122 : #else
123 : return std::suspend_never{};
124 : #endif
125 : }
126 :
127 72 : auto final_suspend() noexcept
128 : {
129 : struct awaiter
130 : {
131 : promise_type* p_;
132 :
133 72 : bool await_ready() const noexcept
134 : {
135 72 : return false;
136 : }
137 :
138 72 : std::coroutine_handle<> await_suspend(std::coroutine_handle<>) const noexcept
139 : {
140 72 : p_->done_ = true;
141 72 : if(!p_->cont_)
142 : return std::noop_coroutine(); // LCOV_EXCL_LINE cont_ always set on this (suspend_always) path
143 72 : return p_->cont_;
144 : }
145 :
146 : void await_resume() const noexcept {} // LCOV_EXCL_LINE final_suspend awaiter, never resumed
147 : };
148 72 : return awaiter{this};
149 : }
150 :
151 50 : void return_value(
152 : results<std::size_t> r) noexcept
153 : {
154 50 : result_ = r;
155 50 : }
156 :
157 22 : void unhandled_exception()
158 : {
159 22 : ep_ = std::current_exception();
160 22 : }
161 :
162 : std::suspend_always yield_value(int) noexcept
163 : {
164 : return {};
165 : }
166 :
167 : template<class A>
168 93 : auto await_transform(A&& a)
169 : {
170 : using decayed = std::decay_t<A>;
171 : if constexpr (ContextAwaitable<decayed>)
172 : {
173 : struct wrapper
174 : {
175 : decayed inner_;
176 : promise_type* p_;
177 :
178 93 : bool await_ready()
179 : {
180 93 : return inner_.await_ready();
181 : }
182 :
183 : // The inner operation may resume or destroy this
184 : // frame on another thread before control returns
185 : // here, so the handle must not round-trip through
186 : // the frame.
187 93 : auto await_suspend(std::coroutine_handle<> h)
188 : {
189 93 : return detail::symmetric_transfer(
190 : detail::call_await_suspend(
191 : &inner_, h,
192 186 : p_->env_));
193 : }
194 :
195 91 : decltype(auto) await_resume()
196 : {
197 : // The resumption may arrive on another thread.
198 91 : set_current_frame_allocator(
199 91 : p_->env_->frame_allocator);
200 91 : return inner_.await_resume();
201 : }
202 : };
203 : return wrapper{
204 93 : std::forward<A>(a), this};
205 : }
206 : else
207 : {
208 : return std::forward<A>(a);
209 : }
210 : }
211 :
212 : static void*
213 76 : operator new(
214 : std::size_t size,
215 : write_now& self,
216 : auto&)
217 : {
218 76 : if(self.cached_frame_ &&
219 6 : self.cached_size_ >= size)
220 5 : return self.cached_frame_;
221 71 : void* p = ::operator new(size);
222 71 : if(self.cached_frame_)
223 1 : ::operator delete(self.cached_frame_);
224 71 : self.cached_frame_ = p;
225 71 : self.cached_size_ = size;
226 71 : return p;
227 : }
228 :
229 : static void
230 76 : operator delete(void*, std::size_t) noexcept
231 : {
232 76 : }
233 : };
234 :
235 : std::coroutine_handle<promise_type> h_;
236 :
237 150 : ~op_type()
238 : {
239 150 : if(h_)
240 : {
241 : // Only the unwind of an abandoned chain destroys a frame
242 : // that is still awaited.
243 76 : auto& p = h_.promise();
244 78 : if(!p.done_ && p.owner_ &&
245 2 : !(p.env_->owner && p.env_->owner->abandoned()))
246 : std::terminate(); // LCOV_EXCL_LINE reached only in a death test's child process
247 76 : h_.destroy();
248 : }
249 150 : }
250 :
251 : op_type(op_type const&) = delete;
252 : op_type& operator=(op_type const&) = delete;
253 :
254 74 : op_type(op_type&& other) noexcept
255 74 : : h_(std::exchange(other.h_, nullptr))
256 : {
257 74 : }
258 :
259 : op_type& operator=(op_type&&) = delete;
260 :
261 74 : bool await_ready() const noexcept
262 : {
263 74 : return h_.promise().done_;
264 : }
265 :
266 74 : std::coroutine_handle<> await_suspend(
267 : std::coroutine_handle<> cont,
268 : await_context const* env)
269 : {
270 74 : auto& p = h_.promise();
271 74 : p.cont_ = cont;
272 74 : p.env_ = env;
273 74 : p.owner_ = this;
274 74 : return h_;
275 : }
276 :
277 72 : [[nodiscard]] results<std::size_t> await_resume()
278 : {
279 72 : auto& p = h_.promise();
280 72 : p.owner_ = nullptr;
281 72 : if(p.ep_)
282 22 : std::rethrow_exception(p.ep_);
283 50 : return p.result_;
284 : }
285 :
286 : private:
287 76 : explicit op_type(
288 : std::coroutine_handle<promise_type> h)
289 76 : : h_(h)
290 : {
291 76 : }
292 : };
293 :
294 : public:
295 : /** Destructor. Frees the cached coroutine frame. */
296 70 : ~write_now()
297 : {
298 70 : if(cached_frame_)
299 70 : ::operator delete(cached_frame_);
300 70 : }
301 :
302 : /** Construct from a stream reference.
303 :
304 : @param s The stream to write to. Must outlive this object.
305 : */
306 : explicit
307 70 : write_now(Stream& s) noexcept
308 70 : : stream_(s)
309 : {
310 70 : }
311 :
312 : write_now(write_now const&) = delete;
313 : write_now& operator=(write_now const&) = delete;
314 :
315 : /** Eagerly write the entire buffer sequence.
316 :
317 : Writes data to the stream by calling `write_some` repeatedly
318 : until the entire buffer sequence is written or an error
319 : occurs. The operation attempts to complete synchronously:
320 : if every `write_some` completes without suspending, the
321 : entire operation finishes in `await_ready`.
322 :
323 : When the fast path cannot complete, the coroutine suspends
324 : and continues asynchronously. The internal coroutine frame
325 : is cached and reused across calls.
326 :
327 : @param buffers The buffer sequence to write. Passed by
328 : value to ensure the sequence lives in the coroutine
329 : frame across suspension points.
330 :
331 : @return An awaitable that await-returns `(error_code,std::size_t)`.
332 : On success, `n` equals `buffer_size(buffers)`. On
333 : error, `n` is the number of bytes written before the
334 : error. Compare error codes to conditions:
335 : @li `cond::canceled` - Operation was cancelled
336 : @li `std::errc::broken_pipe` - Peer closed connection
337 :
338 : @par Example
339 :
340 : @code
341 : write_now wn( stream );
342 : auto [ec, n] = co_await wn( make_buffer( body ) );
343 : if( ec )
344 : detail::throw_system_error( ec );
345 : @endcode
346 :
347 : @see @ref write, @ref write_some, @ref WriteStream
348 : */
349 : // GCC falsely warns that the coroutine promise's
350 : // placement operator new(size_t, write_now&, auto&)
351 : // mismatches operator delete(void*, size_t). Per the
352 : // standard, coroutine deallocation lookup is separate.
353 : #if defined(__GNUC__) && !defined(__clang__)
354 : #pragma GCC diagnostic push
355 : #pragma GCC diagnostic ignored "-Wmismatched-new-delete"
356 : #endif
357 :
358 : #if BOOST_CAPY_WRITE_NOW_WORKAROUND
359 : template<ConstBufferSequence Buffers>
360 : op_type
361 76 : operator()(Buffers buffers)
362 : {
363 : std::size_t const total_size = buffer_size(buffers);
364 : std::size_t total_written = 0;
365 : consuming_buffers cb(buffers);
366 : while(total_written < total_size)
367 : {
368 : auto r =
369 : co_await stream_.write_some(cb.data());
370 : cb.consume(std::get<1>(r));
371 : total_written += std::get<1>(r);
372 : if(std::get<0>(r))
373 : co_return results<std::size_t>{
374 : std::get<0>(r), total_written};
375 : }
376 : co_return results<std::size_t>{
377 : std::error_code(), total_written};
378 152 : }
379 : #else
380 : template<ConstBufferSequence Buffers>
381 : op_type
382 : operator()(Buffers buffers)
383 : {
384 : std::size_t const total_size = buffer_size(buffers);
385 : std::size_t total_written = 0;
386 :
387 : // GCC ICE in expand_expr_real_1 (expr.cc:11376)
388 : // when the buffer slice spans a co_yield, so
389 : // the GCC path uses a separate simple coroutine.
390 : consuming_buffers cb(buffers);
391 : while(total_written < total_size)
392 : {
393 : auto inner = stream_.write_some(cb.data());
394 : if(!inner.await_ready())
395 : break;
396 : auto r = inner.await_resume();
397 : if(std::get<0>(r))
398 : co_return results<std::size_t>{
399 : std::get<0>(r), total_written};
400 : cb.consume(std::get<1>(r));
401 : total_written += std::get<1>(r);
402 : }
403 :
404 : if(total_written >= total_size)
405 : co_return results<std::size_t>{
406 : std::error_code(), total_written};
407 :
408 : co_yield 0;
409 :
410 : while(total_written < total_size)
411 : {
412 : auto r =
413 : co_await stream_.write_some(cb.data());
414 : cb.consume(std::get<1>(r));
415 : total_written += std::get<1>(r);
416 : if(std::get<0>(r))
417 : co_return results<std::size_t>{
418 : std::get<0>(r), total_written};
419 : }
420 : co_return results<std::size_t>{
421 : std::error_code(), total_written};
422 : }
423 : #endif
424 :
425 : #if defined(__GNUC__) && !defined(__clang__)
426 : #pragma GCC diagnostic pop
427 : #endif
428 : };
429 :
430 : template<WriteStream S>
431 : write_now(S&) -> write_now<S>;
432 :
433 : } // namespace capy
434 : } // namespace boost
435 :
436 : #endif
|
