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