. /include/boost/capy /io /any_write_stream.hpp
100.00% Lines (130/130) 100.00% Functions (25/25)
[ ] prev/next file n p prev/next uncovered line
TLA Baseline Branch
Line Hits Code Line Hits Code
1   // 1   //
2   // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com) 2   // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
3   // Copyright (c) 2026 Michael Vandeberg 3   // Copyright (c) 2026 Michael Vandeberg
4   // 4   //
5   // Distributed under the Boost Software License, Version 1.0. (See accompanying 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) 6   // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7   // 7   //
8   // Official repository: https://github.com/cppalliance/capy 8   // Official repository: https://github.com/cppalliance/capy
9   // 9   //
10   10  
11   #ifndef BOOST_CAPY_IO_ANY_WRITE_STREAM_HPP 11   #ifndef BOOST_CAPY_IO_ANY_WRITE_STREAM_HPP
12   #define BOOST_CAPY_IO_ANY_WRITE_STREAM_HPP 12   #define BOOST_CAPY_IO_ANY_WRITE_STREAM_HPP
13   13  
14   #include <boost/capy/detail/config.hpp> 14   #include <boost/capy/detail/config.hpp>
15   #include <boost/capy/detail/await_suspend_helper.hpp> 15   #include <boost/capy/detail/await_suspend_helper.hpp>
16   #include <boost/capy/buffers.hpp> 16   #include <boost/capy/buffers.hpp>
17   #include <boost/capy/detail/buffer_array.hpp> 17   #include <boost/capy/detail/buffer_array.hpp>
18   #include <boost/capy/concept/context_awaitable.hpp> 18   #include <boost/capy/concept/context_awaitable.hpp>
19   #include <boost/capy/concept/write_stream.hpp> 19   #include <boost/capy/concept/write_stream.hpp>
20   #include <coroutine> 20   #include <coroutine>
21   #include <boost/capy/ex/await_context.hpp> 21   #include <boost/capy/ex/await_context.hpp>
22   #include <boost/capy/results.hpp> 22   #include <boost/capy/results.hpp>
23   23  
24   #include <concepts> 24   #include <concepts>
25   #include <coroutine> 25   #include <coroutine>
26   #include <cstddef> 26   #include <cstddef>
27   #include <exception> 27   #include <exception>
28   #include <new> 28   #include <new>
29   #include <span> 29   #include <span>
30   #include <stop_token> 30   #include <stop_token>
31   #include <system_error> 31   #include <system_error>
32   #include <utility> 32   #include <utility>
33   33  
34   namespace boost { 34   namespace boost {
35   namespace capy { 35   namespace capy {
36   36  
37   /** Dispatches `write_some` through a type-erased vtable, using preallocated awaitable storage. 37   /** Dispatches `write_some` through a type-erased vtable, using preallocated awaitable storage.
38   38  
39   This class provides type erasure for any type satisfying the 39   This class provides type erasure for any type satisfying the
40   @ref WriteStream concept, enabling runtime polymorphism for 40   @ref WriteStream concept, enabling runtime polymorphism for
41   write operations. It uses cached awaitable storage to achieve 41   write operations. It uses cached awaitable storage to achieve
42   zero steady-state allocation after construction. 42   zero steady-state allocation after construction.
43   43  
44   The wrapper supports two construction modes: 44   The wrapper supports two construction modes:
45   - **Owning**: Pass by value to transfer ownership. The wrapper 45   - **Owning**: Pass by value to transfer ownership. The wrapper
46   allocates storage and owns the stream. 46   allocates storage and owns the stream.
47   - **Reference**: Pass a pointer to wrap without ownership. The 47   - **Reference**: Pass a pointer to wrap without ownership. The
48   pointed-to stream must outlive this wrapper. 48   pointed-to stream must outlive this wrapper.
49   49  
50   @par Awaitable Preallocation 50   @par Awaitable Preallocation
51   The constructor preallocates storage for the type-erased awaitable. 51   The constructor preallocates storage for the type-erased awaitable.
52   This reserves all virtual address space at server startup 52   This reserves all virtual address space at server startup
53   so memory usage can be measured up front, rather than 53   so memory usage can be measured up front, rather than
54   allocating piecemeal as traffic arrives. 54   allocating piecemeal as traffic arrives.
55   55  
56   @par Immediate Completion 56   @par Immediate Completion
57   Operations complete immediately without suspending when the 57   Operations complete immediately without suspending when the
58   buffer sequence is empty, or when the underlying stream's 58   buffer sequence is empty, or when the underlying stream's
59   awaitable reports readiness via `await_ready`. 59   awaitable reports readiness via `await_ready`.
60   60  
61   @par Thread Safety 61   @par Thread Safety
62   Not thread-safe. Concurrent operations on the same wrapper 62   Not thread-safe. Concurrent operations on the same wrapper
63   are undefined behavior. 63   are undefined behavior.
64   64  
65   @par Example 65   @par Example
66   @par !example example 66   @par !example example
67   67  
68   68  
69   @see any_read_stream, any_stream, WriteStream 69   @see any_read_stream, any_stream, WriteStream
70   */ 70   */
71   class any_write_stream 71   class any_write_stream
72   { 72   {
73   struct vtable; 73   struct vtable;
74   74  
75   template<WriteStream S> 75   template<WriteStream S>
76   struct vtable_for_impl; 76   struct vtable_for_impl;
77   77  
78   // ordered for cache line coherence 78   // ordered for cache line coherence
79   void* stream_ = nullptr; 79   void* stream_ = nullptr;
80   vtable const* vt_ = nullptr; 80   vtable const* vt_ = nullptr;
81   void* cached_awaitable_ = nullptr; 81   void* cached_awaitable_ = nullptr;
82   void* storage_ = nullptr; 82   void* storage_ = nullptr;
83   bool awaitable_active_ = false; 83   bool awaitable_active_ = false;
84   // The flag of the awaitable whose operation is active, cleared 84   // The flag of the awaitable whose operation is active, cleared
85   // when the stream ends that operation itself. 85   // when the stream ends that operation itself.
86   bool* owner_ = nullptr; 86   bool* owner_ = nullptr;
87   87  
88   void 88   void
HITCBC 89   208 disown() noexcept 89   208 disown() noexcept
90   { 90   {
HITCBC 91   208 if(owner_) 91   208 if(owner_)
HITCBC 92   6 *std::exchange(owner_, nullptr) = false; 92   6 *std::exchange(owner_, nullptr) = false;
HITCBC 93   208 } 93   208 }
94   94  
95   public: 95   public:
96   /** Destructor. 96   /** Destructor.
97   97  
98   Ends any operation in flight, then destroys the owned stream 98   Ends any operation in flight, then destroys the owned stream
99   (if any) and releases the cached awaitable storage. 99   (if any) and releases the cached awaitable storage.
100   */ 100   */
101   ~any_write_stream(); 101   ~any_write_stream();
102   102  
103   /** Construct a default instance. 103   /** Construct a default instance.
104   104  
105   Constructs an empty wrapper. @ref has_value and `operator bool` 105   Constructs an empty wrapper. @ref has_value and `operator bool`
106   report the empty state; calling @ref write_some before the 106   report the empty state; calling @ref write_some before the
107   wrapper holds a stream is undefined behavior. 107   wrapper holds a stream is undefined behavior.
108   */ 108   */
HITCBC 109   10 any_write_stream() = default; 109   10 any_write_stream() = default;
110   110  
111   /** Non-copyable. 111   /** Non-copyable.
112   112  
113   The awaitable cache is per-instance and cannot be shared. 113   The awaitable cache is per-instance and cannot be shared.
114   114  
115   @param other The wrapper that would be copied. 115   @param other The wrapper that would be copied.
116   */ 116   */
117   any_write_stream(any_write_stream const& other) = delete; 117   any_write_stream(any_write_stream const& other) = delete;
118   118  
119   /** Copy assignment is disabled. 119   /** Copy assignment is disabled.
120   120  
121   The awaitable cache is per-instance and cannot be shared. 121   The awaitable cache is per-instance and cannot be shared.
122   122  
123   @param other The wrapper that would be assigned from. 123   @param other The wrapper that would be assigned from.
124   124  
125   @return A reference to `*this`. 125   @return A reference to `*this`.
126   */ 126   */
127   any_write_stream& operator=(any_write_stream const& other) = delete; 127   any_write_stream& operator=(any_write_stream const& other) = delete;
128   128  
129   /** Construct by moving. 129   /** Construct by moving.
130   130  
131   Transfers ownership of the wrapped stream (if owned) and 131   Transfers ownership of the wrapped stream (if owned) and
132   cached awaitable storage from `other`. After the move, `other` is 132   cached awaitable storage from `other`. After the move, `other` is
133   in a default-constructed state. 133   in a default-constructed state.
134   134  
135   @param other The wrapper to move from. 135   @param other The wrapper to move from.
136   */ 136   */
HITCBC 137   2 any_write_stream(any_write_stream&& other) noexcept 137   2 any_write_stream(any_write_stream&& other) noexcept
HITCBC 138   2 : stream_(std::exchange(other.stream_, nullptr)) 138   2 : stream_(std::exchange(other.stream_, nullptr))
HITCBC 139   2 , vt_(std::exchange(other.vt_, nullptr)) 139   2 , vt_(std::exchange(other.vt_, nullptr))
HITCBC 140   2 , cached_awaitable_(std::exchange(other.cached_awaitable_, nullptr)) 140   2 , cached_awaitable_(std::exchange(other.cached_awaitable_, nullptr))
HITCBC 141   2 , storage_(std::exchange(other.storage_, nullptr)) 141   2 , storage_(std::exchange(other.storage_, nullptr))
HITCBC 142   2 , awaitable_active_(std::exchange(other.awaitable_active_, false)) 142   2 , awaitable_active_(std::exchange(other.awaitable_active_, false))
143   { 143   {
HITCBC 144   2 other.disown(); 144   2 other.disown();
HITCBC 145   2 } 145   2 }
146   146  
147   /** Assign by moving. 147   /** Assign by moving.
148   148  
149   Destroys any owned stream and releases existing resources, 149   Destroys any owned stream and releases existing resources,
150   then transfers ownership from `other`. 150   then transfers ownership from `other`.
151   151  
152   @param other The wrapper to move from. 152   @param other The wrapper to move from.
153   @return Reference to this wrapper. 153   @return Reference to this wrapper.
154   */ 154   */
155   any_write_stream& 155   any_write_stream&
156   operator=(any_write_stream&& other) noexcept; 156   operator=(any_write_stream&& other) noexcept;
157   157  
158   /** Construct by taking ownership of a WriteStream. 158   /** Construct by taking ownership of a WriteStream.
159   159  
160   Allocates storage and moves the stream into this wrapper. 160   Allocates storage and moves the stream into this wrapper.
161   The wrapper owns the stream and destroys it. 161   The wrapper owns the stream and destroys it.
162   162  
163   @param s The stream to take ownership of. 163   @param s The stream to take ownership of.
164   */ 164   */
165   template<WriteStream S> 165   template<WriteStream S>
166   requires (!std::same_as<std::decay_t<S>, any_write_stream>) 166   requires (!std::same_as<std::decay_t<S>, any_write_stream>)
167   any_write_stream(S s); 167   any_write_stream(S s);
168   168  
169   /** Construct by wrapping a WriteStream without ownership. 169   /** Construct by wrapping a WriteStream without ownership.
170   170  
171   Wraps the given stream by pointer. The stream must remain 171   Wraps the given stream by pointer. The stream must remain
172   valid for the lifetime of this wrapper. 172   valid for the lifetime of this wrapper.
173   173  
174   @param s Pointer to the stream to wrap. 174   @param s Pointer to the stream to wrap.
175   */ 175   */
176   template<WriteStream S> 176   template<WriteStream S>
177   any_write_stream(S* s); 177   any_write_stream(S* s);
178   178  
179   /** Check if the wrapper contains a valid stream. 179   /** Check if the wrapper contains a valid stream.
180   180  
181   @return `true` if wrapping a stream, `false` if default-constructed 181   @return `true` if wrapping a stream, `false` if default-constructed
182   or moved-from. 182   or moved-from.
183   */ 183   */
184   bool 184   bool
HITCBC 185   26 has_value() const noexcept 185   26 has_value() const noexcept
186   { 186   {
HITCBC 187   26 return stream_ != nullptr; 187   26 return stream_ != nullptr;
188   } 188   }
189   189  
190   /** Check if the wrapper contains a valid stream. 190   /** Check if the wrapper contains a valid stream.
191   191  
192   @return `true` if wrapping a stream, `false` if default-constructed 192   @return `true` if wrapping a stream, `false` if default-constructed
193   or moved-from. 193   or moved-from.
194   */ 194   */
195   explicit 195   explicit
HITCBC 196   3 operator bool() const noexcept 196   3 operator bool() const noexcept
197   { 197   {
HITCBC 198   3 return has_value(); 198   3 return has_value();
199   } 199   }
200   200  
201   /** Initiate an asynchronous write operation. 201   /** Initiate an asynchronous write operation.
202   202  
203   Writes data from the provided buffer sequence. The operation 203   Writes data from the provided buffer sequence. The operation
204   completes when at least one byte is written, or an error 204   completes when at least one byte is written, or an error
205   occurs. 205   occurs.
206   206  
207   @param buffers The buffer sequence containing data to write. 207   @param buffers The buffer sequence containing data to write.
208   Passed by value to ensure the sequence lives in the 208   Passed by value to ensure the sequence lives in the
209   coroutine frame across suspension points. 209   coroutine frame across suspension points.
210   210  
211   @return An awaitable that await-returns `(error_code,std::size_t)`. 211   @return An awaitable that await-returns `(error_code,std::size_t)`.
212   212  
213   @par Immediate Completion 213   @par Immediate Completion
214   The operation completes immediately without suspending 214   The operation completes immediately without suspending
215   the calling coroutine when: 215   the calling coroutine when:
216   @li The buffer sequence is empty, returning `{error_code{}, 0}`. 216   @li The buffer sequence is empty, returning `{error_code{}, 0}`.
217   @li The underlying stream's awaitable reports immediate 217   @li The underlying stream's awaitable reports immediate
218   readiness via `await_ready`. 218   readiness via `await_ready`.
219   219  
220   @note This is a partial operation and may not process the 220   @note This is a partial operation and may not process the
221   entire buffer sequence. Use the composed @ref write algorithm 221   entire buffer sequence. Use the composed @ref write algorithm
222   for guaranteed complete transfer. 222   for guaranteed complete transfer.
223   223  
224   @par Preconditions 224   @par Preconditions
225   The wrapper must contain a valid stream (`has_value() == true`). 225   The wrapper must contain a valid stream (`has_value() == true`).
226   226  
227   @par After an Error 227   @par After an Error
228   A subsequent call is permitted. The wrapper forwards directly 228   A subsequent call is permitted. The wrapper forwards directly
229   to the underlying stream, imposing no stricter rule than 229   to the underlying stream, imposing no stricter rule than
230   @ref WriteStream. 230   @ref WriteStream.
231   */ 231   */
232   template<ConstBufferSequence CB> 232   template<ConstBufferSequence CB>
233   auto 233   auto
234   write_some(CB buffers); 234   write_some(CB buffers);
235   235  
236   protected: 236   protected:
237   /** End the operation in flight, if any. 237   /** End the operation in flight, if any.
238   238  
239   Destroys the wrapped stream's active operation, which may refer 239   Destroys the wrapped stream's active operation, which may refer
240   to the stream, so an owner must call this before destroying the 240   to the stream, so an owner must call this before destroying the
241   stream. The awaitable that started the operation no longer ends 241   stream. The awaitable that started the operation no longer ends
242   it. 242   it.
243   */ 243   */
244   void 244   void
245   end_operation() noexcept; 245   end_operation() noexcept;
246   246  
247   /** Rebind to a new stream after move. 247   /** Rebind to a new stream after move.
248   248  
249   Updates the internal pointer to reference a new stream object. 249   Updates the internal pointer to reference a new stream object.
250   Used by owning wrappers after move assignment when the owned 250   Used by owning wrappers after move assignment when the owned
251   object has moved to a new location. 251   object has moved to a new location.
252   252  
253   @param new_stream The new stream to bind to. Must be the same 253   @param new_stream The new stream to bind to. Must be the same
254   type as the original stream. 254   type as the original stream.
255   255  
256   @note Terminates if called with a stream of different type 256   @note Terminates if called with a stream of different type
257   than the original. 257   than the original.
258   */ 258   */
259   template<WriteStream S> 259   template<WriteStream S>
260   void 260   void
261   rebind(S& new_stream) noexcept 261   rebind(S& new_stream) noexcept
262   { 262   {
263   if(vt_ != &vtable_for_impl<S>::value) 263   if(vt_ != &vtable_for_impl<S>::value)
264   std::terminate(); 264   std::terminate();
265   stream_ = &new_stream; 265   stream_ = &new_stream;
266   } 266   }
267   }; 267   };
268   268  
269   struct any_write_stream::vtable 269   struct any_write_stream::vtable
270   { 270   {
271   // ordered by call frequency for cache line coherence 271   // ordered by call frequency for cache line coherence
272   void (*construct_awaitable)( 272   void (*construct_awaitable)(
273   void* stream, 273   void* stream,
274   void* storage, 274   void* storage,
275   std::span<const_buffer const> buffers); 275   std::span<const_buffer const> buffers);
276   bool (*await_ready)(void*); 276   bool (*await_ready)(void*);
277   std::coroutine_handle<> (*await_suspend)(void*, std::coroutine_handle<>, await_context const*); 277   std::coroutine_handle<> (*await_suspend)(void*, std::coroutine_handle<>, await_context const*);
278   results<std::size_t> (*await_resume)(void*); 278   results<std::size_t> (*await_resume)(void*);
279   void (*destroy_awaitable)(void*) noexcept; 279   void (*destroy_awaitable)(void*) noexcept;
280   std::size_t awaitable_size; 280   std::size_t awaitable_size;
281   std::size_t awaitable_align; 281   std::size_t awaitable_align;
282   void (*destroy)(void*) noexcept; 282   void (*destroy)(void*) noexcept;
283   }; 283   };
284   284  
285   template<WriteStream S> 285   template<WriteStream S>
286   struct any_write_stream::vtable_for_impl 286   struct any_write_stream::vtable_for_impl
287   { 287   {
288   using Awaitable = decltype(std::declval<S&>().write_some( 288   using Awaitable = decltype(std::declval<S&>().write_some(
289   std::span<const_buffer const>{})); 289   std::span<const_buffer const>{}));
290   290  
291   static void 291   static void
HITCBC 292   3 do_destroy_impl(void* stream) noexcept 292   3 do_destroy_impl(void* stream) noexcept
293   { 293   {
HITCBC 294   3 static_cast<S*>(stream)->~S(); 294   3 static_cast<S*>(stream)->~S();
HITCBC 295   3 } 295   3 }
296   296  
297   static void 297   static void
HITCBC 298   86 construct_awaitable_impl( 298   86 construct_awaitable_impl(
299   void* stream, 299   void* stream,
300   void* storage, 300   void* storage,
301   std::span<const_buffer const> buffers) 301   std::span<const_buffer const> buffers)
302   { 302   {
HITCBC 303   86 auto& s = *static_cast<S*>(stream); 303   86 auto& s = *static_cast<S*>(stream);
HITCBC 304   86 ::new(storage) Awaitable(s.write_some(buffers)); 304   86 ::new(storage) Awaitable(s.write_some(buffers));
HITCBC 305   86 } 305   86 }
306   306  
307   static constexpr vtable value = { 307   static constexpr vtable value = {
308   &construct_awaitable_impl, 308   &construct_awaitable_impl,
HITCBC 309   86 +[](void* p) { 309   86 +[](void* p) {
HITCBC 310   86 return static_cast<Awaitable*>(p)->await_ready(); 310   86 return static_cast<Awaitable*>(p)->await_ready();
311   }, 311   },
HITCBC 312   74 +[](void* p, std::coroutine_handle<> h, await_context const* env) { 312   74 +[](void* p, std::coroutine_handle<> h, await_context const* env) {
HITCBC 313   74 return detail::call_await_suspend( 313   74 return detail::call_await_suspend(
HITCBC 314   74 static_cast<Awaitable*>(p), h, env); 314   74 static_cast<Awaitable*>(p), h, env);
315   }, 315   },
HITCBC 316   77 +[](void* p) { 316   77 +[](void* p) {
HITCBC 317   77 return static_cast<Awaitable*>(p)->await_resume(); 317   77 return static_cast<Awaitable*>(p)->await_resume();
318   }, 318   },
HITCBC 319   95 +[](void* p) noexcept { 319   95 +[](void* p) noexcept {
HITCBC 320   12 static_cast<Awaitable*>(p)->~Awaitable(); 320   12 static_cast<Awaitable*>(p)->~Awaitable();
321   }, 321   },
322   sizeof(Awaitable), 322   sizeof(Awaitable),
323   alignof(Awaitable), 323   alignof(Awaitable),
324   &do_destroy_impl 324   &do_destroy_impl
325   }; 325   };
326   }; 326   };
327   327  
328   inline void 328   inline void
HITCBC 329   187 any_write_stream::end_operation() noexcept 329   187 any_write_stream::end_operation() noexcept
330   { 330   {
HITCBC 331   187 disown(); 331   187 disown();
HITCBC 332   187 if(awaitable_active_) 332   187 if(awaitable_active_)
333   { 333   {
HITCBC 334   6 vt_->destroy_awaitable(cached_awaitable_); 334   6 vt_->destroy_awaitable(cached_awaitable_);
HITCBC 335   6 awaitable_active_ = false; 335   6 awaitable_active_ = false;
336   } 336   }
HITCBC 337   187 } 337   187 }
338   338  
339   inline 339   inline
HITCBC 340   125 any_write_stream::~any_write_stream() 340   125 any_write_stream::~any_write_stream()
341   { 341   {
HITCBC 342   125 end_operation(); 342   125 end_operation();
HITCBC 343   125 if(cached_awaitable_) 343   125 if(cached_awaitable_)
HITCBC 344   101 ::operator delete(cached_awaitable_); 344   101 ::operator delete(cached_awaitable_);
HITCBC 345   125 if(storage_) 345   125 if(storage_)
346   { 346   {
HITCBC 347   2 vt_->destroy(stream_); 347   2 vt_->destroy(stream_);
HITCBC 348   2 ::operator delete(storage_); 348   2 ::operator delete(storage_);
349   } 349   }
HITCBC 350   125 } 350   125 }
351   351  
352   inline any_write_stream& 352   inline any_write_stream&
HITCBC 353   19 any_write_stream::operator=(any_write_stream&& other) noexcept 353   19 any_write_stream::operator=(any_write_stream&& other) noexcept
354   { 354   {
HITCBC 355   19 if(this != &other) 355   19 if(this != &other)
356   { 356   {
HITCBC 357   19 end_operation(); 357   19 end_operation();
HITCBC 358   19 other.disown(); 358   19 other.disown();
HITCBC 359   19 if(cached_awaitable_) 359   19 if(cached_awaitable_)
HITCBC 360   7 ::operator delete(cached_awaitable_); 360   7 ::operator delete(cached_awaitable_);
HITCBC 361   19 if(storage_) 361   19 if(storage_)
362   { 362   {
HITCBC 363   1 vt_->destroy(stream_); 363   1 vt_->destroy(stream_);
HITCBC 364   1 ::operator delete(storage_); 364   1 ::operator delete(storage_);
365   } 365   }
HITCBC 366   19 stream_ = std::exchange(other.stream_, nullptr); 366   19 stream_ = std::exchange(other.stream_, nullptr);
HITCBC 367   19 vt_ = std::exchange(other.vt_, nullptr); 367   19 vt_ = std::exchange(other.vt_, nullptr);
HITCBC 368   19 cached_awaitable_ = std::exchange(other.cached_awaitable_, nullptr); 368   19 cached_awaitable_ = std::exchange(other.cached_awaitable_, nullptr);
HITCBC 369   19 storage_ = std::exchange(other.storage_, nullptr); 369   19 storage_ = std::exchange(other.storage_, nullptr);
HITCBC 370   19 awaitable_active_ = std::exchange(other.awaitable_active_, false); 370   19 awaitable_active_ = std::exchange(other.awaitable_active_, false);
371   } 371   }
HITCBC 372   19 return *this; 372   19 return *this;
373   } 373   }
374   374  
375   template<WriteStream S> 375   template<WriteStream S>
376   requires (!std::same_as<std::decay_t<S>, any_write_stream>) 376   requires (!std::same_as<std::decay_t<S>, any_write_stream>)
HITCBC 377   4 any_write_stream::any_write_stream(S s) 377   4 any_write_stream::any_write_stream(S s)
HITCBC 378   4 : vt_(&vtable_for_impl<S>::value) 378   4 : vt_(&vtable_for_impl<S>::value)
379   { 379   {
380   struct guard { 380   struct guard {
381   any_write_stream* self; 381   any_write_stream* self;
382   bool committed = false; 382   bool committed = false;
HITCBC 383   4 ~guard() { 383   4 ~guard() {
HITCBC 384   4 if(!committed && self->storage_) { 384   4 if(!committed && self->storage_) {
HITCBC 385   1 if(self->stream_) 385   1 if(self->stream_)
386   self->vt_->destroy(self->stream_); // LCOV_EXCL_LINE OOM rollback: only when the cached-awaitable allocation throws 386   self->vt_->destroy(self->stream_); // LCOV_EXCL_LINE OOM rollback: only when the cached-awaitable allocation throws
HITCBC 387   1 ::operator delete(self->storage_); 387   1 ::operator delete(self->storage_);
HITCBC 388   1 self->storage_ = nullptr; 388   1 self->storage_ = nullptr;
HITCBC 389   1 self->stream_ = nullptr; 389   1 self->stream_ = nullptr;
390   } 390   }
HITCBC 391   4 } 391   4 }
HITCBC 392   4 } g{this}; 392   4 } g{this};
393   393  
HITCBC 394   4 storage_ = ::operator new(sizeof(S)); 394   4 storage_ = ::operator new(sizeof(S));
HITCBC 395   4 stream_ = ::new(storage_) S(std::move(s)); 395   4 stream_ = ::new(storage_) S(std::move(s));
396   396  
397   // Preallocate the awaitable storage 397   // Preallocate the awaitable storage
HITCBC 398   3 cached_awaitable_ = ::operator new(vt_->awaitable_size); 398   3 cached_awaitable_ = ::operator new(vt_->awaitable_size);
399   399  
HITCBC 400   3 g.committed = true; 400   3 g.committed = true;
HITCBC 401   4 } 401   4 }
402   402  
403   template<WriteStream S> 403   template<WriteStream S>
HITCBC 404   105 any_write_stream::any_write_stream(S* s) 404   105 any_write_stream::any_write_stream(S* s)
HITCBC 405   105 : stream_(s) 405   105 : stream_(s)
HITCBC 406   105 , vt_(&vtable_for_impl<S>::value) 406   105 , vt_(&vtable_for_impl<S>::value)
407   { 407   {
408   // Preallocate the awaitable storage 408   // Preallocate the awaitable storage
HITCBC 409   105 cached_awaitable_ = ::operator new(vt_->awaitable_size); 409   105 cached_awaitable_ = ::operator new(vt_->awaitable_size);
HITCBC 410   105 } 410   105 }
411   411  
412   template<ConstBufferSequence CB> 412   template<ConstBufferSequence CB>
413   auto 413   auto
HITCBC 414   91 any_write_stream::write_some(CB buffers) 414   91 any_write_stream::write_some(CB buffers)
415   { 415   {
416   struct awaitable 416   struct awaitable
417   { 417   {
418   any_write_stream* self_; 418   any_write_stream* self_;
419   detail::const_buffer_array<detail::max_iovec_> ba_; 419   detail::const_buffer_array<detail::max_iovec_> ba_;
420   // Set while this awaitable, not another on the same stream, 420   // Set while this awaitable, not another on the same stream,
421   // holds the stream's inner awaitable. 421   // holds the stream's inner awaitable.
422   bool owns_ = false; 422   bool owns_ = false;
423   423  
424   // A chain unwound while suspended here never reaches 424   // A chain unwound while suspended here never reaches
425   // await_resume, so the inner awaitable is destroyed here. 425   // await_resume, so the inner awaitable is destroyed here.
HITCBC 426   178 ~awaitable() 426   178 ~awaitable()
427   { 427   {
HITCBC 428   178 if(owns_) 428   178 if(owns_)
429   { 429   {
HITCBC 430   3 self_->vt_->destroy_awaitable(self_->cached_awaitable_); 430   3 self_->vt_->destroy_awaitable(self_->cached_awaitable_);
HITCBC 431   3 self_->awaitable_active_ = false; 431   3 self_->awaitable_active_ = false;
HITCBC 432   3 self_->owner_ = nullptr; 432   3 self_->owner_ = nullptr;
433   } 433   }
HITCBC 434   178 } 434   178 }
435   435  
HITCBC 436   91 awaitable( 436   91 awaitable(
437   any_write_stream* self, 437   any_write_stream* self,
438   CB const& buffers) noexcept 438   CB const& buffers) noexcept
HITCBC 439   91 : self_(self) 439   91 : self_(self)
HITCBC 440   91 , ba_(buffers) 440   91 , ba_(buffers)
441   { 441   {
HITCBC 442   91 } 442   91 }
443   443  
HITCBC 444   87 awaitable(awaitable&& other) noexcept 444   87 awaitable(awaitable&& other) noexcept
HITCBC 445   87 : self_(std::exchange(other.self_, nullptr)) 445   87 : self_(std::exchange(other.self_, nullptr))
HITCBC 446   87 , ba_(other.ba_) 446   87 , ba_(other.ba_)
HITCBC 447   87 , owns_(std::exchange(other.owns_, false)) 447   87 , owns_(std::exchange(other.owns_, false))
448   { 448   {
HITCBC 449   87 if(owns_) 449   87 if(owns_)
HITCBC 450   1 self_->owner_ = &owns_; 450   1 self_->owner_ = &owns_;
HITCBC 451   87 } 451   87 }
452   452  
453   bool 453   bool
HITCBC 454   90 await_ready() 454   90 await_ready()
455   { 455   {
456   // An empty write never touches the underlying stream 456   // An empty write never touches the underlying stream
HITCBC 457   90 if(ba_.to_span().empty()) 457   90 if(ba_.to_span().empty())
HITCBC 458   4 return true; 458   4 return true;
459   459  
HITCBC 460   86 self_->vt_->construct_awaitable( 460   86 self_->vt_->construct_awaitable(
HITCBC 461   86 self_->stream_, 461   86 self_->stream_,
HITCBC 462   86 self_->cached_awaitable_, 462   86 self_->cached_awaitable_,
HITCBC 463   86 ba_.to_span()); 463   86 ba_.to_span());
HITCBC 464   86 self_->awaitable_active_ = true; 464   86 self_->awaitable_active_ = true;
HITCBC 465   86 owns_ = true; 465   86 owns_ = true;
HITCBC 466   86 self_->owner_ = &owns_; 466   86 self_->owner_ = &owns_;
467   467  
HITCBC 468   172 return self_->vt_->await_ready( 468   172 return self_->vt_->await_ready(
HITCBC 469   86 self_->cached_awaitable_); 469   86 self_->cached_awaitable_);
470   } 470   }
471   471  
472   std::coroutine_handle<> 472   std::coroutine_handle<>
HITCBC 473   74 await_suspend(std::coroutine_handle<> h, await_context const* env) 473   74 await_suspend(std::coroutine_handle<> h, await_context const* env)
474   { 474   {
HITCBC 475   74 return self_->vt_->await_suspend( 475   74 return self_->vt_->await_suspend(
HITCBC 476   74 self_->cached_awaitable_, h, env); 476   74 self_->cached_awaitable_, h, env);
477   } 477   }
478   478  
479   [[nodiscard]] results<std::size_t> 479   [[nodiscard]] results<std::size_t>
HITCBC 480   81 await_resume() 480   81 await_resume()
481   { 481   {
HITCBC 482   81 if(!owns_) 482   81 if(!owns_)
HITCBC 483   4 return {std::error_code(), 0}; 483   4 return {std::error_code(), 0};
484   struct guard { 484   struct guard {
485   awaitable* a; 485   awaitable* a;
HITCBC 486   77 ~guard() { 486   77 ~guard() {
HITCBC 487   77 a->self_->vt_->destroy_awaitable( 487   77 a->self_->vt_->destroy_awaitable(
HITCBC 488   77 a->self_->cached_awaitable_); 488   77 a->self_->cached_awaitable_);
HITCBC 489   77 a->self_->awaitable_active_ = false; 489   77 a->self_->awaitable_active_ = false;
HITCBC 490   77 a->self_->owner_ = nullptr; 490   77 a->self_->owner_ = nullptr;
HITCBC 491   77 a->owns_ = false; 491   77 a->owns_ = false;
HITCBC 492   77 } 492   77 }
HITCBC 493   77 } g{this}; 493   77 } g{this};
HITCBC 494   77 return self_->vt_->await_resume( 494   77 return self_->vt_->await_resume(
HITCBC 495   77 self_->cached_awaitable_); 495   77 self_->cached_awaitable_);
HITCBC 496   77 } 496   77 }
497   }; 497   };
HITCBC 498   91 return awaitable{this, buffers}; 498   91 return awaitable{this, buffers};
499   } 499   }
500   500  
501   } // namespace capy 501   } // namespace capy
502   } // namespace boost 502   } // namespace boost
503   503  
504   #endif 504   #endif