. /include/boost/capy /io /any_write_sink.hpp
0.00% Lines (0/0) 0.00% Functions (0/0)
[ ] prev/next file n p prev/next uncovered line
TLA Baseline Branch
Line Hits Code Line Hits 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_ANY_WRITE_SINK_HPP  
11 - #define BOOST_CAPY_IO_ANY_WRITE_SINK_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/detail/buffer_array.hpp>  
17 - #include <boost/capy/buffers/buffer_param.hpp>  
18 - #include <boost/capy/concept/io_awaitable.hpp>  
19 - #include <boost/capy/concept/write_sink.hpp>  
20 - #include <coroutine>  
21 - #include <boost/capy/ex/io_env.hpp>  
22 - #include <boost/capy/io_result.hpp>  
23 - #include <boost/capy/io_task.hpp>  
24 -  
25 - #include <concepts>  
26 - #include <coroutine>  
27 - #include <cstddef>  
28 - #include <exception>  
29 - #include <new>  
30 - #include <span>  
31 - #include <stop_token>  
32 - #include <system_error>  
33 - #include <utility>  
34 -  
35 - namespace boost {  
36 - namespace capy {  
37 -  
38 - /** Type-erased wrapper for any WriteSink.  
39 -  
40 - This class provides type erasure for any type satisfying the  
41 - @ref WriteSink concept, enabling runtime polymorphism for  
42 - sink write operations. It uses cached awaitable storage to achieve  
43 - zero steady-state allocation after construction.  
44 -  
45 - The wrapper supports two construction modes:  
46 - - **Owning**: Pass by value to transfer ownership. The wrapper  
47 - allocates storage and owns the sink.  
48 - - **Reference**: Pass a pointer to wrap without ownership. The  
49 - pointed-to sink must outlive this wrapper.  
50 -  
51 - @par Awaitable Preallocation  
52 - The constructor preallocates storage for the type-erased awaitable.  
53 - This reserves all virtual address space at server startup  
54 - so memory usage can be measured up front, rather than  
55 - allocating piecemeal as traffic arrives.  
56 -  
57 - @par Immediate Completion  
58 - Operations complete immediately without suspending when the  
59 - buffer sequence is empty, or when the underlying sink's  
60 - awaitable reports readiness via `await_ready`.  
61 -  
62 - @par Thread Safety  
63 - Not thread-safe. Concurrent operations on the same wrapper  
64 - are undefined behavior.  
65 -  
66 - @par Example  
67 - @code  
68 - // Owning - takes ownership of the sink  
69 - any_write_sink ws(some_sink{args...});  
70 -  
71 - // Reference - wraps without ownership  
72 - some_sink sink;  
73 - any_write_sink ws(&sink);  
74 -  
75 - const_buffer buf(data, size);  
76 - auto [ec, n] = co_await ws.write(std::span(&buf, 1));  
77 - auto [ec2] = co_await ws.write_eof();  
78 - @endcode  
79 -  
80 - @see any_write_stream, WriteSink  
81 - */  
82 - class any_write_sink  
83 - {  
84 - struct vtable;  
85 - struct write_awaitable_ops;  
86 - struct eof_awaitable_ops;  
87 -  
88 - template<WriteSink S>  
89 - struct vtable_for_impl;  
90 -  
91 - void* sink_ = nullptr;  
92 - vtable const* vt_ = nullptr;  
93 - void* cached_awaitable_ = nullptr;  
94 - void* storage_ = nullptr;  
95 - write_awaitable_ops const* active_write_ops_ = nullptr;  
96 - eof_awaitable_ops const* active_eof_ops_ = nullptr;  
97 -  
98 - public:  
99 - /** Destructor.  
100 -  
101 - Destroys the owned sink (if any) and releases the cached  
102 - awaitable storage.  
103 - */  
104 - ~any_write_sink();  
105 -  
106 - /** Construct a default instance.  
107 -  
108 - Constructs an empty wrapper. Operations on a default-constructed  
109 - wrapper result in undefined behavior.  
110 - */  
111 - any_write_sink() = default;  
112 -  
113 - /** Non-copyable.  
114 -  
115 - The awaitable cache is per-instance and cannot be shared.  
116 - */  
117 - any_write_sink(any_write_sink const&) = delete;  
118 - any_write_sink& operator=(any_write_sink const&) = delete;  
119 -  
120 - /** Construct by moving.  
121 -  
122 - Transfers ownership of the wrapped sink (if owned) and  
123 - cached awaitable storage from `other`. After the move, `other` is  
124 - in a default-constructed state.  
125 -  
126 - @param other The wrapper to move from.  
127 - */  
DCB 128 - 1 any_write_sink(any_write_sink&& other) noexcept  
DCB 129 - 1 : sink_(std::exchange(other.sink_, nullptr))  
DCB 130 - 1 , vt_(std::exchange(other.vt_, nullptr))  
DCB 131 - 1 , cached_awaitable_(std::exchange(other.cached_awaitable_, nullptr))  
DCB 132 - 1 , storage_(std::exchange(other.storage_, nullptr))  
DCB 133 - 1 , active_write_ops_(std::exchange(other.active_write_ops_, nullptr))  
DCB 134 - 1 , active_eof_ops_(std::exchange(other.active_eof_ops_, nullptr))  
135 - {  
DCB 136 - 1 }  
137 -  
138 - /** Assign by moving.  
139 -  
140 - Destroys any owned sink and releases existing resources,  
141 - then transfers ownership from `other`.  
142 -  
143 - @param other The wrapper to move from.  
144 - @return Reference to this wrapper.  
145 - */  
146 - any_write_sink&  
147 - operator=(any_write_sink&& other) noexcept;  
148 -  
149 - /** Construct by taking ownership of a WriteSink.  
150 -  
151 - Allocates storage and moves the sink into this wrapper.  
152 - The wrapper owns the sink and will destroy it.  
153 -  
154 - @param s The sink to take ownership of.  
155 - */  
156 - template<WriteSink S>  
157 - requires (!std::same_as<std::decay_t<S>, any_write_sink>)  
158 - any_write_sink(S s);  
159 -  
160 - /** Construct by wrapping a WriteSink without ownership.  
161 -  
162 - Wraps the given sink by pointer. The sink must remain  
163 - valid for the lifetime of this wrapper.  
164 -  
165 - @param s Pointer to the sink to wrap.  
166 - */  
167 - template<WriteSink S>  
168 - any_write_sink(S* s);  
169 -  
170 - /** Check if the wrapper contains a valid sink.  
171 -  
172 - @return `true` if wrapping a sink, `false` if default-constructed  
173 - or moved-from.  
174 - */  
175 - bool  
DCB 176 - 18 has_value() const noexcept  
177 - {  
DCB 178 - 18 return sink_ != nullptr;  
179 - }  
180 -  
181 - /** Check if the wrapper contains a valid sink.  
182 -  
183 - @return `true` if wrapping a sink, `false` if default-constructed  
184 - or moved-from.  
185 - */  
186 - explicit  
DCB 187 - 2 operator bool() const noexcept  
188 - {  
DCB 189 - 2 return has_value();  
190 - }  
191 -  
192 - /** Initiate a partial write operation.  
193 -  
194 - Attempt to write up to `buffer_size( buffers )` bytes from  
195 - the provided buffer sequence. May consume less than the  
196 - full sequence.  
197 -  
198 - @param buffers The buffer sequence containing data to write.  
199 -  
200 - @return An awaitable that await-returns `(error_code,std::size_t)`.  
201 -  
202 - @par Immediate Completion  
203 - The operation completes immediately without suspending  
204 - the calling coroutine when:  
205 - @li The buffer sequence is empty, returning `{error_code{}, 0}`.  
206 - @li The underlying sink's awaitable reports immediate  
207 - readiness via `await_ready`.  
208 -  
209 - @note This is a partial operation and may not process the  
210 - entire buffer sequence. Use @ref write for guaranteed  
211 - complete transfer.  
212 -  
213 - @par Preconditions  
214 - The wrapper must contain a valid sink (`has_value() == true`).  
215 - */  
216 - template<ConstBufferSequence CB>  
217 - auto  
218 - write_some(CB buffers);  
219 -  
220 - /** Initiate a complete write operation.  
221 -  
222 - Writes data from the provided buffer sequence. The operation  
223 - completes when all bytes have been consumed, or an error  
224 - occurs. Forwards to the underlying sink's `write` operation,  
225 - windowed through @ref buffer_param when the sequence exceeds  
226 - the per-call buffer limit.  
227 -  
228 - @param buffers The buffer sequence containing data to write.  
229 -  
230 - @return An awaitable that await-returns `(error_code,std::size_t)`.  
231 -  
232 - @par Immediate Completion  
233 - The operation completes immediately without suspending  
234 - the calling coroutine when:  
235 - @li The buffer sequence is empty, returning `{error_code{}, 0}`.  
236 - @li Every underlying `write` call completes  
237 - immediately (the wrapped sink reports readiness  
238 - via `await_ready` on each iteration).  
239 -  
240 - @par Preconditions  
241 - The wrapper must contain a valid sink (`has_value() == true`).  
242 - */  
243 - template<ConstBufferSequence CB>  
244 - io_task<std::size_t>  
245 - write(CB buffers);  
246 -  
247 - /** Atomically write data and signal end-of-stream.  
248 -  
249 - Writes all data from the buffer sequence and then signals  
250 - end-of-stream. The implementation decides how to partition  
251 - the data across calls to the underlying sink's @ref write  
252 - and `write_eof`. When the caller's buffer sequence is  
253 - non-empty, the final call to the underlying sink is always  
254 - `write_eof` with a non-empty buffer sequence. When the  
255 - caller's buffer sequence is empty, only `write_eof()` with  
256 - no data is called.  
257 -  
258 - @param buffers The buffer sequence containing data to write.  
259 -  
260 - @return An awaitable that await-returns `(error_code,std::size_t)`.  
261 -  
262 - @par Immediate Completion  
263 - The operation completes immediately without suspending  
264 - the calling coroutine when:  
265 - @li The buffer sequence is empty. Only the @ref write_eof()  
266 - call is performed.  
267 - @li All underlying operations complete immediately (the  
268 - wrapped sink reports readiness via `await_ready`).  
269 -  
270 - @par Preconditions  
271 - The wrapper must contain a valid sink (`has_value() == true`).  
272 - */  
273 - template<ConstBufferSequence CB>  
274 - io_task<std::size_t>  
275 - write_eof(CB buffers);  
276 -  
277 - /** Signal end of data.  
278 -  
279 - Indicates that no more data will be written to the sink.  
280 - The operation completes when the sink is finalized, or  
281 - an error occurs.  
282 -  
283 - @return An awaitable that await-returns `(error_code)`.  
284 -  
285 - @par Immediate Completion  
286 - The operation completes immediately without suspending  
287 - the calling coroutine when the underlying sink's awaitable  
288 - reports immediate readiness via `await_ready`.  
289 -  
290 - @par Preconditions  
291 - The wrapper must contain a valid sink (`has_value() == true`).  
292 - */  
293 - auto  
294 - write_eof();  
295 -  
296 - protected:  
297 - /** Rebind to a new sink after move.  
298 -  
299 - Updates the internal pointer to reference a new sink object.  
300 - Used by owning wrappers after move assignment when the owned  
301 - object has moved to a new location.  
302 -  
303 - @param new_sink The new sink to bind to. Must be the same  
304 - type as the original sink.  
305 -  
306 - @note Terminates if called with a sink of different type  
307 - than the original.  
308 - */  
309 - template<WriteSink S>  
310 - void  
311 - rebind(S& new_sink) noexcept  
312 - {  
313 - if(vt_ != &vtable_for_impl<S>::value)  
314 - std::terminate();  
315 - sink_ = &new_sink;  
316 - }  
317 -  
318 - private:  
319 - auto  
320 - write_some_(std::span<const_buffer const> buffers);  
321 -  
322 - auto  
323 - write_(std::span<const_buffer const> buffers);  
324 -  
325 - auto  
326 - write_eof_buffers_(std::span<const_buffer const> buffers);  
327 - };  
328 -  
329 - struct any_write_sink::write_awaitable_ops  
330 - {  
331 - bool (*await_ready)(void*);  
332 - std::coroutine_handle<> (*await_suspend)(void*, std::coroutine_handle<>, io_env const*);  
333 - io_result<std::size_t> (*await_resume)(void*);  
334 - void (*destroy)(void*) noexcept;  
335 - };  
336 -  
337 - struct any_write_sink::eof_awaitable_ops  
338 - {  
339 - bool (*await_ready)(void*);  
340 - std::coroutine_handle<> (*await_suspend)(void*, std::coroutine_handle<>, io_env const*);  
341 - io_result<> (*await_resume)(void*);  
342 - void (*destroy)(void*) noexcept;  
343 - };  
344 -  
345 - struct any_write_sink::vtable  
346 - {  
347 - write_awaitable_ops const* (*construct_write_some_awaitable)(  
348 - void* sink,  
349 - void* storage,  
350 - std::span<const_buffer const> buffers);  
351 - write_awaitable_ops const* (*construct_write_awaitable)(  
352 - void* sink,  
353 - void* storage,  
354 - std::span<const_buffer const> buffers);  
355 - write_awaitable_ops const* (*construct_write_eof_buffers_awaitable)(  
356 - void* sink,  
357 - void* storage,  
358 - std::span<const_buffer const> buffers);  
359 - eof_awaitable_ops const* (*construct_eof_awaitable)(  
360 - void* sink,  
361 - void* storage);  
362 - std::size_t awaitable_size;  
363 - std::size_t awaitable_align;  
364 - void (*destroy)(void*) noexcept;  
365 - };  
366 -  
367 - template<WriteSink S>  
368 - struct any_write_sink::vtable_for_impl  
369 - {  
370 - using WriteSomeAwaitable = decltype(std::declval<S&>().write_some(  
371 - std::span<const_buffer const>{}));  
372 - using WriteAwaitable = decltype(std::declval<S&>().write(  
373 - std::span<const_buffer const>{}));  
374 - using WriteEofBuffersAwaitable = decltype(std::declval<S&>().write_eof(  
375 - std::span<const_buffer const>{}));  
376 - using EofAwaitable = decltype(std::declval<S&>().write_eof());  
377 -  
378 - static void  
DCB 379 - 8 do_destroy_impl(void* sink) noexcept  
380 - {  
DCB 381 - 8 static_cast<S*>(sink)->~S();  
DCB 382 - 8 }  
383 -  
384 - static write_awaitable_ops const*  
DCB 385 - 41 construct_write_some_awaitable_impl(  
386 - void* sink,  
387 - void* storage,  
388 - std::span<const_buffer const> buffers)  
389 - {  
DCB 390 - 41 auto& s = *static_cast<S*>(sink);  
DCB 391 - 41 ::new(storage) WriteSomeAwaitable(s.write_some(buffers));  
392 -  
393 - static constexpr write_awaitable_ops ops = {  
DCB 394 - 41 +[](void* p) {  
DCB 395 - 41 return static_cast<WriteSomeAwaitable*>(p)->await_ready();  
396 - },  
DCB 397 - 41 +[](void* p, std::coroutine_handle<> h, io_env const* env) {  
DCB 398 - 41 return detail::call_await_suspend(  
DCB 399 - 41 static_cast<WriteSomeAwaitable*>(p), h, env);  
400 - },  
DCB 401 - 39 +[](void* p) {  
DCB 402 - 39 return static_cast<WriteSomeAwaitable*>(p)->await_resume();  
403 - },  
DCB 404 - 43 +[](void* p) noexcept {  
DCB 405 - 2 static_cast<WriteSomeAwaitable*>(p)->~WriteSomeAwaitable();  
406 - }  
407 - };  
DCB 408 - 41 return &ops;  
409 - }  
410 -  
411 - static write_awaitable_ops const*  
DCB 412 - 79 construct_write_awaitable_impl(  
413 - void* sink,  
414 - void* storage,  
415 - std::span<const_buffer const> buffers)  
416 - {  
DCB 417 - 79 auto& s = *static_cast<S*>(sink);  
DCB 418 - 79 ::new(storage) WriteAwaitable(s.write(buffers));  
419 -  
420 - static constexpr write_awaitable_ops ops = {  
DCB 421 - 79 +[](void* p) {  
DCB 422 - 79 return static_cast<WriteAwaitable*>(p)->await_ready();  
423 - },  
DCB 424 - 79 +[](void* p, std::coroutine_handle<> h, io_env const* env) {  
DCB 425 - 79 return detail::call_await_suspend(  
DCB 426 - 79 static_cast<WriteAwaitable*>(p), h, env);  
427 - },  
DCB 428 - 79 +[](void* p) {  
DCB 429 - 79 return static_cast<WriteAwaitable*>(p)->await_resume();  
430 - },  
DCB 431 - 79 +[](void* p) noexcept {  
432 - static_cast<WriteAwaitable*>(p)->~WriteAwaitable(); // LCOV_EXCL_LINE runs via destroy tests; gcov miscounts fn-ptr thunk  
433 - }  
434 - };  
DCB 435 - 79 return &ops;  
436 - }  
437 -  
438 - static write_awaitable_ops const*  
DCB 439 - 17 construct_write_eof_buffers_awaitable_impl(  
440 - void* sink,  
441 - void* storage,  
442 - std::span<const_buffer const> buffers)  
443 - {  
DCB 444 - 17 auto& s = *static_cast<S*>(sink);  
DCB 445 - 17 ::new(storage) WriteEofBuffersAwaitable(s.write_eof(buffers));  
446 -  
447 - static constexpr write_awaitable_ops ops = {  
DCB 448 - 17 +[](void* p) {  
DCB 449 - 17 return static_cast<WriteEofBuffersAwaitable*>(p)->await_ready();  
450 - },  
DCB 451 - 17 +[](void* p, std::coroutine_handle<> h, io_env const* env) {  
DCB 452 - 17 return detail::call_await_suspend(  
DCB 453 - 17 static_cast<WriteEofBuffersAwaitable*>(p), h, env);  
454 - },  
DCB 455 - 17 +[](void* p) {  
DCB 456 - 17 return static_cast<WriteEofBuffersAwaitable*>(p)->await_resume();  
457 - },  
DCB 458 - 17 +[](void* p) noexcept {  
459 - static_cast<WriteEofBuffersAwaitable*>(p)->~WriteEofBuffersAwaitable(); // LCOV_EXCL_LINE runs via destroy tests; gcov miscounts fn-ptr thunk  
460 - }  
461 - };  
DCB 462 - 17 return &ops;  
463 - }  
464 -  
465 - static eof_awaitable_ops const*  
DCB 466 - 19 construct_eof_awaitable_impl(  
467 - void* sink,  
468 - void* storage)  
469 - {  
DCB 470 - 19 auto& s = *static_cast<S*>(sink);  
DCB 471 - 19 ::new(storage) EofAwaitable(s.write_eof());  
472 -  
473 - static constexpr eof_awaitable_ops ops = {  
DCB 474 - 19 +[](void* p) {  
DCB 475 - 19 return static_cast<EofAwaitable*>(p)->await_ready();  
476 - },  
DCB 477 - 19 +[](void* p, std::coroutine_handle<> h, io_env const* env) {  
DCB 478 - 19 return detail::call_await_suspend(  
DCB 479 - 19 static_cast<EofAwaitable*>(p), h, env);  
480 - },  
DCB 481 - 17 +[](void* p) {  
DCB 482 - 17 return static_cast<EofAwaitable*>(p)->await_resume();  
483 - },  
DCB 484 - 21 +[](void* p) noexcept {  
DCB 485 - 2 static_cast<EofAwaitable*>(p)->~EofAwaitable();  
486 - }  
487 - };  
DCB 488 - 19 return &ops;  
489 - }  
490 -  
491 - static constexpr std::size_t max4(  
492 - std::size_t a, std::size_t b,  
493 - std::size_t c, std::size_t d) noexcept  
494 - {  
495 - std::size_t ab = a > b ? a : b;  
496 - std::size_t cd = c > d ? c : d;  
497 - return ab > cd ? ab : cd;  
498 - }  
499 -  
500 - static constexpr std::size_t max_awaitable_size =  
501 - max4(sizeof(WriteSomeAwaitable),  
502 - sizeof(WriteAwaitable),  
503 - sizeof(WriteEofBuffersAwaitable),  
504 - sizeof(EofAwaitable));  
505 -  
506 - static constexpr std::size_t max_awaitable_align =  
507 - max4(alignof(WriteSomeAwaitable),  
508 - alignof(WriteAwaitable),  
509 - alignof(WriteEofBuffersAwaitable),  
510 - alignof(EofAwaitable));  
511 -  
512 - static constexpr vtable value = {  
513 - &construct_write_some_awaitable_impl,  
514 - &construct_write_awaitable_impl,  
515 - &construct_write_eof_buffers_awaitable_impl,  
516 - &construct_eof_awaitable_impl,  
517 - max_awaitable_size,  
518 - max_awaitable_align,  
519 - &do_destroy_impl  
520 - };  
521 - };  
522 -  
523 - inline  
DCB 524 - 134 any_write_sink::~any_write_sink()  
525 - {  
DCB 526 - 134 if(storage_)  
527 - {  
DCB 528 - 7 vt_->destroy(sink_);  
DCB 529 - 7 ::operator delete(storage_);  
530 - }  
DCB 531 - 134 if(cached_awaitable_)  
532 - {  
DCB 533 - 126 if(active_write_ops_)  
DCB 534 - 1 active_write_ops_->destroy(cached_awaitable_);  
DCB 535 - 125 else if(active_eof_ops_)  
DCB 536 - 1 active_eof_ops_->destroy(cached_awaitable_);  
DCB 537 - 126 ::operator delete(cached_awaitable_);  
538 - }  
DCB 539 - 134 }  
540 -  
541 - inline any_write_sink&  
DCB 542 - 4 any_write_sink::operator=(any_write_sink&& other) noexcept  
543 - {  
DCB 544 - 4 if(this != &other)  
545 - {  
DCB 546 - 4 if(storage_)  
547 - {  
DCB 548 - 1 vt_->destroy(sink_);  
DCB 549 - 1 ::operator delete(storage_);  
550 - }  
DCB 551 - 4 if(cached_awaitable_)  
552 - {  
DCB 553 - 3 if(active_write_ops_)  
DCB 554 - 1 active_write_ops_->destroy(cached_awaitable_);  
DCB 555 - 2 else if(active_eof_ops_)  
DCB 556 - 1 active_eof_ops_->destroy(cached_awaitable_);  
DCB 557 - 3 ::operator delete(cached_awaitable_);  
558 - }  
DCB 559 - 4 sink_ = std::exchange(other.sink_, nullptr);  
DCB 560 - 4 vt_ = std::exchange(other.vt_, nullptr);  
DCB 561 - 4 cached_awaitable_ = std::exchange(other.cached_awaitable_, nullptr);  
DCB 562 - 4 storage_ = std::exchange(other.storage_, nullptr);  
DCB 563 - 4 active_write_ops_ = std::exchange(other.active_write_ops_, nullptr);  
DCB 564 - 4 active_eof_ops_ = std::exchange(other.active_eof_ops_, nullptr);  
565 - }  
DCB 566 - 4 return *this;  
567 - }  
568 -  
569 - template<WriteSink S>  
570 - requires (!std::same_as<std::decay_t<S>, any_write_sink>)  
DCB 571 - 9 any_write_sink::any_write_sink(S s)  
DCB 572 - 9 : vt_(&vtable_for_impl<S>::value)  
573 - {  
574 - struct guard {  
575 - any_write_sink* self;  
576 - bool committed = false;  
DCB 577 - 9 ~guard() {  
DCB 578 - 9 if(!committed && self->storage_) {  
DCB 579 - 1 if(self->sink_)  
580 - self->vt_->destroy(self->sink_); // LCOV_EXCL_LINE OOM rollback: only when the cached-awaitable allocation throws  
DCB 581 - 1 ::operator delete(self->storage_);  
DCB 582 - 1 self->storage_ = nullptr;  
DCB 583 - 1 self->sink_ = nullptr;  
584 - }  
DCB 585 - 9 }  
DCB 586 - 9 } g{this};  
587 -  
DCB 588 - 9 storage_ = ::operator new(sizeof(S));  
DCB 589 - 9 sink_ = ::new(storage_) S(std::move(s));  
590 -  
591 - // Preallocate the awaitable storage (sized for max of write/eof)  
DCB 592 - 8 cached_awaitable_ = ::operator new(vt_->awaitable_size);  
593 -  
DCB 594 - 8 g.committed = true;  
DCB 595 - 9 }  
596 -  
597 - template<WriteSink S>  
DCB 598 - 121 any_write_sink::any_write_sink(S* s)  
DCB 599 - 121 : sink_(s)  
DCB 600 - 121 , vt_(&vtable_for_impl<S>::value)  
601 - {  
602 - // Preallocate the awaitable storage (sized for max of write/eof)  
DCB 603 - 121 cached_awaitable_ = ::operator new(vt_->awaitable_size);  
DCB 604 - 121 }  
605 -  
606 - inline auto  
607 - any_write_sink::write_some_(  
608 - std::span<const_buffer const> buffers)  
609 - {  
610 - struct awaitable  
611 - {  
612 - any_write_sink* self_;  
613 - std::span<const_buffer const> buffers_;  
614 -  
615 - bool  
616 - await_ready() const noexcept  
617 - {  
618 - return false;  
619 - }  
620 -  
621 - std::coroutine_handle<>  
622 - await_suspend(std::coroutine_handle<> h, io_env const* env)  
623 - {  
624 - self_->active_write_ops_ = self_->vt_->construct_write_some_awaitable(  
625 - self_->sink_,  
626 - self_->cached_awaitable_,  
627 - buffers_);  
628 -  
629 - if(self_->active_write_ops_->await_ready(self_->cached_awaitable_))  
630 - return h;  
631 -  
632 - return self_->active_write_ops_->await_suspend(  
633 - self_->cached_awaitable_, h, env);  
634 - }  
635 -  
636 - io_result<std::size_t>  
637 - await_resume()  
638 - {  
639 - struct guard {  
640 - any_write_sink* self;  
641 - ~guard() {  
642 - self->active_write_ops_->destroy(self->cached_awaitable_);  
643 - self->active_write_ops_ = nullptr;  
644 - }  
645 - } g{self_};  
646 - return self_->active_write_ops_->await_resume(  
647 - self_->cached_awaitable_);  
648 - }  
649 - };  
650 - return awaitable{this, buffers};  
651 - }  
652 -  
653 - inline auto  
DCB 654 - 79 any_write_sink::write_(  
655 - std::span<const_buffer const> buffers)  
656 - {  
657 - struct awaitable  
658 - {  
659 - any_write_sink* self_;  
660 - std::span<const_buffer const> buffers_;  
661 -  
662 - bool  
DCB 663 - 79 await_ready() const noexcept  
664 - {  
DCB 665 - 79 return false;  
666 - }  
667 -  
668 - std::coroutine_handle<>  
DCB 669 - 79 await_suspend(std::coroutine_handle<> h, io_env const* env)  
670 - {  
DCB 671 - 158 self_->active_write_ops_ = self_->vt_->construct_write_awaitable(  
DCB 672 - 79 self_->sink_,  
DCB 673 - 79 self_->cached_awaitable_,  
674 - buffers_);  
675 -  
DCB 676 - 79 if(self_->active_write_ops_->await_ready(self_->cached_awaitable_))  
DUB 677 - ✗ return h;  
678 -  
DCB 679 - 79 return self_->active_write_ops_->await_suspend(  
DCB 680 - 79 self_->cached_awaitable_, h, env);  
681 - }  
682 -  
683 - io_result<std::size_t>  
DCB 684 - 79 await_resume()  
685 - {  
686 - struct guard {  
687 - any_write_sink* self;  
DCB 688 - 79 ~guard() {  
DCB 689 - 79 self->active_write_ops_->destroy(self->cached_awaitable_);  
DCB 690 - 79 self->active_write_ops_ = nullptr;  
DCB 691 - 79 }  
DCB 692 - 79 } g{self_};  
DCB 693 - 79 return self_->active_write_ops_->await_resume(  
DCB 694 - 137 self_->cached_awaitable_);  
DCB 695 - 79 }  
696 - };  
DCB 697 - 79 return awaitable{this, buffers};  
698 - }  
699 -  
700 - inline auto  
DCB 701 - 19 any_write_sink::write_eof()  
702 - {  
703 - struct awaitable  
704 - {  
705 - any_write_sink* self_;  
706 -  
707 - bool  
DCB 708 - 19 await_ready() const noexcept  
709 - {  
DCB 710 - 19 return false;  
711 - }  
712 -  
713 - std::coroutine_handle<>  
DCB 714 - 19 await_suspend(std::coroutine_handle<> h, io_env const* env)  
715 - {  
716 - // Construct the underlying awaitable into cached storage  
DCB 717 - 38 self_->active_eof_ops_ = self_->vt_->construct_eof_awaitable(  
DCB 718 - 19 self_->sink_,  
DCB 719 - 19 self_->cached_awaitable_);  
720 -  
721 - // Check if underlying is immediately ready  
DCB 722 - 19 if(self_->active_eof_ops_->await_ready(self_->cached_awaitable_))  
DUB 723 - ✗ return h;  
724 -  
725 - // Forward to underlying awaitable  
DCB 726 - 19 return self_->active_eof_ops_->await_suspend(  
DCB 727 - 19 self_->cached_awaitable_, h, env);  
728 - }  
729 -  
730 - io_result<>  
DCB 731 - 17 await_resume()  
732 - {  
733 - struct guard {  
734 - any_write_sink* self;  
DCB 735 - 17 ~guard() {  
DCB 736 - 17 self->active_eof_ops_->destroy(self->cached_awaitable_);  
DCB 737 - 17 self->active_eof_ops_ = nullptr;  
DCB 738 - 17 }  
DCB 739 - 17 } g{self_};  
DCB 740 - 17 return self_->active_eof_ops_->await_resume(  
DCB 741 - 29 self_->cached_awaitable_);  
DCB 742 - 17 }  
743 - };  
DCB 744 - 19 return awaitable{this};  
745 - }  
746 -  
747 - inline auto  
DCB 748 - 17 any_write_sink::write_eof_buffers_(  
749 - std::span<const_buffer const> buffers)  
750 - {  
751 - struct awaitable  
752 - {  
753 - any_write_sink* self_;  
754 - std::span<const_buffer const> buffers_;  
755 -  
756 - bool  
DCB 757 - 17 await_ready() const noexcept  
758 - {  
DCB 759 - 17 return false;  
760 - }  
761 -  
762 - std::coroutine_handle<>  
DCB 763 - 17 await_suspend(std::coroutine_handle<> h, io_env const* env)  
764 - {  
DCB 765 - 34 self_->active_write_ops_ =  
DCB 766 - 34 self_->vt_->construct_write_eof_buffers_awaitable(  
DCB 767 - 17 self_->sink_,  
DCB 768 - 17 self_->cached_awaitable_,  
769 - buffers_);  
770 -  
DCB 771 - 17 if(self_->active_write_ops_->await_ready(self_->cached_awaitable_))  
DUB 772 - ✗ return h;  
773 -  
DCB 774 - 17 return self_->active_write_ops_->await_suspend(  
DCB 775 - 17 self_->cached_awaitable_, h, env);  
776 - }  
777 -  
778 - io_result<std::size_t>  
DCB 779 - 17 await_resume()  
780 - {  
781 - struct guard {  
782 - any_write_sink* self;  
DCB 783 - 17 ~guard() {  
DCB 784 - 17 self->active_write_ops_->destroy(self->cached_awaitable_);  
DCB 785 - 17 self->active_write_ops_ = nullptr;  
DCB 786 - 17 }  
DCB 787 - 17 } g{self_};  
DCB 788 - 17 return self_->active_write_ops_->await_resume(  
DCB 789 - 29 self_->cached_awaitable_);  
DCB 790 - 17 }  
791 - };  
DCB 792 - 17 return awaitable{this, buffers};  
793 - }  
794 -  
795 - template<ConstBufferSequence CB>  
796 - auto  
DCB 797 - 43 any_write_sink::write_some(CB buffers)  
798 - {  
799 - struct awaitable  
800 - {  
801 - any_write_sink* self_;  
802 - detail::const_buffer_array<detail::max_iovec_> ba_;  
803 -  
DCB 804 - 43 awaitable(  
805 - any_write_sink* self,  
806 - CB const& buffers)  
DCB 807 - 43 : self_(self)  
DCB 808 - 43 , ba_(buffers)  
809 - {  
DCB 810 - 43 }  
811 -  
812 - bool  
DCB 813 - 43 await_ready() const noexcept  
814 - {  
DCB 815 - 43 return ba_.to_span().empty();  
816 - }  
817 -  
818 - std::coroutine_handle<>  
DCB 819 - 41 await_suspend(std::coroutine_handle<> h, io_env const* env)  
820 - {  
DCB 821 - 41 self_->active_write_ops_ = self_->vt_->construct_write_some_awaitable(  
DCB 822 - 41 self_->sink_,  
DCB 823 - 41 self_->cached_awaitable_,  
DCB 824 - 41 ba_.to_span());  
825 -  
DCB 826 - 41 if(self_->active_write_ops_->await_ready(self_->cached_awaitable_))  
DUB 827 - ✗ return h;  
828 -  
DCB 829 - 41 return self_->active_write_ops_->await_suspend(  
DCB 830 - 41 self_->cached_awaitable_, h, env);  
831 - }  
832 -  
833 - io_result<std::size_t>  
DCB 834 - 41 await_resume()  
835 - {  
DCB 836 - 41 if(ba_.to_span().empty())  
DCB 837 - 2 return {{}, 0};  
838 -  
839 - struct guard {  
840 - any_write_sink* self;  
DCB 841 - 39 ~guard() {  
DCB 842 - 39 self->active_write_ops_->destroy(self->cached_awaitable_);  
DCB 843 - 39 self->active_write_ops_ = nullptr;  
DCB 844 - 39 }  
DCB 845 - 39 } g{self_};  
DCB 846 - 39 return self_->active_write_ops_->await_resume(  
DCB 847 - 39 self_->cached_awaitable_);  
DCB 848 - 39 }  
849 - };  
DCB 850 - 43 return awaitable{this, buffers};  
851 - }  
852 -  
853 - template<ConstBufferSequence CB>  
854 - io_task<std::size_t>  
DCB 855 - 69 any_write_sink::write(CB buffers)  
856 - {  
857 - buffer_param<CB> bp(buffers);  
858 - std::size_t total = 0;  
859 -  
860 - for(;;)  
861 - {  
862 - auto bufs = bp.data();  
863 - if(bufs.empty())  
864 - break;  
865 -  
866 - auto [ec, n] = co_await write_(bufs);  
867 - total += n;  
868 - if(ec)  
869 - co_return {ec, total};  
870 - bp.consume(n);  
871 - }  
872 -  
873 - co_return {{}, total};  
DCB 874 - 138 }  
875 -  
876 - template<ConstBufferSequence CB>  
877 - io_task<std::size_t>  
DCB 878 - 27 any_write_sink::write_eof(CB buffers)  
879 - {  
880 - const_buffer_param<CB> bp(buffers);  
881 - std::size_t total = 0;  
882 -  
883 - for(;;)  
884 - {  
885 - auto bufs = bp.data();  
886 - if(bufs.empty())  
887 - {  
888 - auto [ec] = co_await write_eof();  
889 - co_return {ec, total};  
890 - }  
891 -  
892 - if(! bp.more())  
893 - {  
894 - // Last window — send atomically with EOF  
895 - auto [ec, n] = co_await write_eof_buffers_(bufs);  
896 - total += n;  
897 - co_return {ec, total};  
898 - }  
899 -  
900 - auto [ec, n] = co_await write_(bufs);  
901 - total += n;  
902 - if(ec)  
903 - co_return {ec, total};  
904 - bp.consume(n);  
905 - }  
DCB 906 - 54 }  
907 -  
908 - } // namespace capy  
909 - } // namespace boost  
910 -  
911 - #endif