. /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/io_awaitable.hpp> 18   #include <boost/capy/concept/io_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/io_env.hpp> 21   #include <boost/capy/ex/io_env.hpp>
22   #include <boost/capy/io_result.hpp> 22   #include <boost/capy/io_result.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
  85 + // when the stream ends that operation itself.
  86 + bool* owner_ = nullptr;
  87 +
  88 + void
HITGNC   89 + 208 disown() noexcept
  90 + {
HITGNC   91 + 208 if(owner_)
HITGNC   92 + 6 *std::exchange(owner_, nullptr) = false;
HITGNC   93 + 208 }
84   94  
85   public: 95   public:
86   /** Destructor. 96   /** Destructor.
87   97  
88 - Destroys the owned stream (if any) and releases the cached 98 + Ends any operation in flight, then destroys the owned stream
89 - awaitable storage. 99 + (if any) and releases the cached awaitable storage.
90   */ 100   */
91   ~any_write_stream(); 101   ~any_write_stream();
92   102  
93   /** Construct a default instance. 103   /** Construct a default instance.
94   104  
95   Constructs an empty wrapper. @ref has_value and `operator bool` 105   Constructs an empty wrapper. @ref has_value and `operator bool`
96   report the empty state; calling @ref write_some before the 106   report the empty state; calling @ref write_some before the
97   wrapper holds a stream is undefined behavior. 107   wrapper holds a stream is undefined behavior.
98   */ 108   */
HITCBC 99   4 any_write_stream() = default; 109   10 any_write_stream() = default;
100   110  
101   /** Non-copyable. 111   /** Non-copyable.
102   112  
103   The awaitable cache is per-instance and cannot be shared. 113   The awaitable cache is per-instance and cannot be shared.
104   114  
105   @param other The wrapper that would be copied. 115   @param other The wrapper that would be copied.
106   */ 116   */
107   any_write_stream(any_write_stream const& other) = delete; 117   any_write_stream(any_write_stream const& other) = delete;
108   118  
109   /** Copy assignment is disabled. 119   /** Copy assignment is disabled.
110   120  
111   The awaitable cache is per-instance and cannot be shared. 121   The awaitable cache is per-instance and cannot be shared.
112   122  
113   @param other The wrapper that would be assigned from. 123   @param other The wrapper that would be assigned from.
114   124  
115   @return A reference to `*this`. 125   @return A reference to `*this`.
116   */ 126   */
117   any_write_stream& operator=(any_write_stream const& other) = delete; 127   any_write_stream& operator=(any_write_stream const& other) = delete;
118   128  
119   /** Construct by moving. 129   /** Construct by moving.
120   130  
121   Transfers ownership of the wrapped stream (if owned) and 131   Transfers ownership of the wrapped stream (if owned) and
122   cached awaitable storage from `other`. After the move, `other` is 132   cached awaitable storage from `other`. After the move, `other` is
123   in a default-constructed state. 133   in a default-constructed state.
124   134  
125   @param other The wrapper to move from. 135   @param other The wrapper to move from.
126   */ 136   */
HITCBC 127   2 any_write_stream(any_write_stream&& other) noexcept 137   2 any_write_stream(any_write_stream&& other) noexcept
HITCBC 128   2 : stream_(std::exchange(other.stream_, nullptr)) 138   2 : stream_(std::exchange(other.stream_, nullptr))
HITCBC 129   2 , vt_(std::exchange(other.vt_, nullptr)) 139   2 , vt_(std::exchange(other.vt_, nullptr))
HITCBC 130   2 , cached_awaitable_(std::exchange(other.cached_awaitable_, nullptr)) 140   2 , cached_awaitable_(std::exchange(other.cached_awaitable_, nullptr))
HITCBC 131   2 , storage_(std::exchange(other.storage_, nullptr)) 141   2 , storage_(std::exchange(other.storage_, nullptr))
HITCBC 132   2 , awaitable_active_(std::exchange(other.awaitable_active_, false)) 142   2 , awaitable_active_(std::exchange(other.awaitable_active_, false))
133   { 143   {
HITGNC   144 + 2 other.disown();
HITCBC 134   2 } 145   2 }
135   146  
136   /** Assign by moving. 147   /** Assign by moving.
137   148  
138   Destroys any owned stream and releases existing resources, 149   Destroys any owned stream and releases existing resources,
139   then transfers ownership from `other`. 150   then transfers ownership from `other`.
140   151  
141   @param other The wrapper to move from. 152   @param other The wrapper to move from.
142   @return Reference to this wrapper. 153   @return Reference to this wrapper.
143   */ 154   */
144   any_write_stream& 155   any_write_stream&
145   operator=(any_write_stream&& other) noexcept; 156   operator=(any_write_stream&& other) noexcept;
146   157  
147   /** Construct by taking ownership of a WriteStream. 158   /** Construct by taking ownership of a WriteStream.
148   159  
149   Allocates storage and moves the stream into this wrapper. 160   Allocates storage and moves the stream into this wrapper.
150   The wrapper owns the stream and destroys it. 161   The wrapper owns the stream and destroys it.
151   162  
152   @param s The stream to take ownership of. 163   @param s The stream to take ownership of.
153   */ 164   */
154   template<WriteStream S> 165   template<WriteStream S>
155   requires (!std::same_as<std::decay_t<S>, any_write_stream>) 166   requires (!std::same_as<std::decay_t<S>, any_write_stream>)
156   any_write_stream(S s); 167   any_write_stream(S s);
157   168  
158   /** Construct by wrapping a WriteStream without ownership. 169   /** Construct by wrapping a WriteStream without ownership.
159   170  
160   Wraps the given stream by pointer. The stream must remain 171   Wraps the given stream by pointer. The stream must remain
161   valid for the lifetime of this wrapper. 172   valid for the lifetime of this wrapper.
162   173  
163   @param s Pointer to the stream to wrap. 174   @param s Pointer to the stream to wrap.
164   */ 175   */
165   template<WriteStream S> 176   template<WriteStream S>
166   any_write_stream(S* s); 177   any_write_stream(S* s);
167   178  
168   /** Check if the wrapper contains a valid stream. 179   /** Check if the wrapper contains a valid stream.
169   180  
170   @return `true` if wrapping a stream, `false` if default-constructed 181   @return `true` if wrapping a stream, `false` if default-constructed
171   or moved-from. 182   or moved-from.
172   */ 183   */
173   bool 184   bool
HITCBC 174   26 has_value() const noexcept 185   26 has_value() const noexcept
175   { 186   {
HITCBC 176   26 return stream_ != nullptr; 187   26 return stream_ != nullptr;
177   } 188   }
178   189  
179   /** Check if the wrapper contains a valid stream. 190   /** Check if the wrapper contains a valid stream.
180   191  
181   @return `true` if wrapping a stream, `false` if default-constructed 192   @return `true` if wrapping a stream, `false` if default-constructed
182   or moved-from. 193   or moved-from.
183   */ 194   */
184   explicit 195   explicit
HITCBC 185   3 operator bool() const noexcept 196   3 operator bool() const noexcept
186   { 197   {
HITCBC 187   3 return has_value(); 198   3 return has_value();
188   } 199   }
189   200  
190   /** Initiate an asynchronous write operation. 201   /** Initiate an asynchronous write operation.
191   202  
192   Writes data from the provided buffer sequence. The operation 203   Writes data from the provided buffer sequence. The operation
193   completes when at least one byte is written, or an error 204   completes when at least one byte is written, or an error
194   occurs. 205   occurs.
195   206  
196   @param buffers The buffer sequence containing data to write. 207   @param buffers The buffer sequence containing data to write.
197   Passed by value to ensure the sequence lives in the 208   Passed by value to ensure the sequence lives in the
198   coroutine frame across suspension points. 209   coroutine frame across suspension points.
199   210  
200   @return An awaitable that await-returns `(error_code,std::size_t)`. 211   @return An awaitable that await-returns `(error_code,std::size_t)`.
201   212  
202   @par Immediate Completion 213   @par Immediate Completion
203   The operation completes immediately without suspending 214   The operation completes immediately without suspending
204   the calling coroutine when: 215   the calling coroutine when:
205   @li The buffer sequence is empty, returning `{error_code{}, 0}`. 216   @li The buffer sequence is empty, returning `{error_code{}, 0}`.
206   @li The underlying stream's awaitable reports immediate 217   @li The underlying stream's awaitable reports immediate
207   readiness via `await_ready`. 218   readiness via `await_ready`.
208   219  
209   @note This is a partial operation and may not process the 220   @note This is a partial operation and may not process the
210   entire buffer sequence. Use the composed @ref write algorithm 221   entire buffer sequence. Use the composed @ref write algorithm
211   for guaranteed complete transfer. 222   for guaranteed complete transfer.
212   223  
213   @par Preconditions 224   @par Preconditions
214   The wrapper must contain a valid stream (`has_value() == true`). 225   The wrapper must contain a valid stream (`has_value() == true`).
215   226  
216   @par After an Error 227   @par After an Error
217   A subsequent call is permitted. The wrapper forwards directly 228   A subsequent call is permitted. The wrapper forwards directly
218   to the underlying stream, imposing no stricter rule than 229   to the underlying stream, imposing no stricter rule than
219   @ref WriteStream. 230   @ref WriteStream.
220   */ 231   */
221   template<ConstBufferSequence CB> 232   template<ConstBufferSequence CB>
222   auto 233   auto
223   write_some(CB buffers); 234   write_some(CB buffers);
224   235  
225   protected: 236   protected:
  237 + /** End the operation in flight, if any.
  238 +
  239 + Destroys the wrapped stream's active operation, which may refer
  240 + to the stream, so an owner must call this before destroying the
  241 + stream. The awaitable that started the operation no longer ends
  242 + it.
  243 + */
  244 + void
  245 + end_operation() noexcept;
  246 +
226   /** Rebind to a new stream after move. 247   /** Rebind to a new stream after move.
227   248  
228   Updates the internal pointer to reference a new stream object. 249   Updates the internal pointer to reference a new stream object.
229   Used by owning wrappers after move assignment when the owned 250   Used by owning wrappers after move assignment when the owned
230   object has moved to a new location. 251   object has moved to a new location.
231   252  
232   @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
233   type as the original stream. 254   type as the original stream.
234   255  
235   @note Terminates if called with a stream of different type 256   @note Terminates if called with a stream of different type
236   than the original. 257   than the original.
237   */ 258   */
238   template<WriteStream S> 259   template<WriteStream S>
239   void 260   void
240   rebind(S& new_stream) noexcept 261   rebind(S& new_stream) noexcept
241   { 262   {
242   if(vt_ != &vtable_for_impl<S>::value) 263   if(vt_ != &vtable_for_impl<S>::value)
243   std::terminate(); 264   std::terminate();
244   stream_ = &new_stream; 265   stream_ = &new_stream;
245   } 266   }
246   }; 267   };
247   268  
248   struct any_write_stream::vtable 269   struct any_write_stream::vtable
249   { 270   {
250   // ordered by call frequency for cache line coherence 271   // ordered by call frequency for cache line coherence
251   void (*construct_awaitable)( 272   void (*construct_awaitable)(
252   void* stream, 273   void* stream,
253   void* storage, 274   void* storage,
254   std::span<const_buffer const> buffers); 275   std::span<const_buffer const> buffers);
255   bool (*await_ready)(void*); 276   bool (*await_ready)(void*);
256   std::coroutine_handle<> (*await_suspend)(void*, std::coroutine_handle<>, io_env const*); 277   std::coroutine_handle<> (*await_suspend)(void*, std::coroutine_handle<>, io_env const*);
257   io_result<std::size_t> (*await_resume)(void*); 278   io_result<std::size_t> (*await_resume)(void*);
258   void (*destroy_awaitable)(void*) noexcept; 279   void (*destroy_awaitable)(void*) noexcept;
259   std::size_t awaitable_size; 280   std::size_t awaitable_size;
260   std::size_t awaitable_align; 281   std::size_t awaitable_align;
261   void (*destroy)(void*) noexcept; 282   void (*destroy)(void*) noexcept;
262   }; 283   };
263   284  
264   template<WriteStream S> 285   template<WriteStream S>
265   struct any_write_stream::vtable_for_impl 286   struct any_write_stream::vtable_for_impl
266   { 287   {
267   using Awaitable = decltype(std::declval<S&>().write_some( 288   using Awaitable = decltype(std::declval<S&>().write_some(
268   std::span<const_buffer const>{})); 289   std::span<const_buffer const>{}));
269   290  
270   static void 291   static void
HITCBC 271   3 do_destroy_impl(void* stream) noexcept 292   3 do_destroy_impl(void* stream) noexcept
272   { 293   {
HITCBC 273   3 static_cast<S*>(stream)->~S(); 294   3 static_cast<S*>(stream)->~S();
HITCBC 274   3 } 295   3 }
275   296  
276   static void 297   static void
HITCBC 277   79 construct_awaitable_impl( 298   86 construct_awaitable_impl(
278   void* stream, 299   void* stream,
279   void* storage, 300   void* storage,
280   std::span<const_buffer const> buffers) 301   std::span<const_buffer const> buffers)
281   { 302   {
HITCBC 282   79 auto& s = *static_cast<S*>(stream); 303   86 auto& s = *static_cast<S*>(stream);
HITCBC 283   79 ::new(storage) Awaitable(s.write_some(buffers)); 304   86 ::new(storage) Awaitable(s.write_some(buffers));
HITCBC 284   79 } 305   86 }
285   306  
286   static constexpr vtable value = { 307   static constexpr vtable value = {
287   &construct_awaitable_impl, 308   &construct_awaitable_impl,
HITCBC 288   79 +[](void* p) { 309   86 +[](void* p) {
HITCBC 289   79 return static_cast<Awaitable*>(p)->await_ready(); 310   86 return static_cast<Awaitable*>(p)->await_ready();
290   }, 311   },
HITCBC 291   68 +[](void* p, std::coroutine_handle<> h, io_env const* env) { 312   74 +[](void* p, std::coroutine_handle<> h, io_env const* env) {
HITCBC 292   68 return detail::call_await_suspend( 313   74 return detail::call_await_suspend(
HITCBC 293   68 static_cast<Awaitable*>(p), h, env); 314   74 static_cast<Awaitable*>(p), h, env);
294   }, 315   },
HITCBC 295   77 +[](void* p) { 316   77 +[](void* p) {
HITCBC 296   77 return static_cast<Awaitable*>(p)->await_resume(); 317   77 return static_cast<Awaitable*>(p)->await_resume();
297   }, 318   },
HITCBC 298   81 +[](void* p) noexcept { 319   95 +[](void* p) noexcept {
HITCBC 299   15 static_cast<Awaitable*>(p)->~Awaitable(); 320   12 static_cast<Awaitable*>(p)->~Awaitable();
300   }, 321   },
301   sizeof(Awaitable), 322   sizeof(Awaitable),
302   alignof(Awaitable), 323   alignof(Awaitable),
303   &do_destroy_impl 324   &do_destroy_impl
304   }; 325   };
305   }; 326   };
306   327  
  328 + inline void
HITGNC   329 + 187 any_write_stream::end_operation() noexcept
  330 + {
HITGNC   331 + 187 disown();
HITGNC   332 + 187 if(awaitable_active_)
  333 + {
HITGNC   334 + 6 vt_->destroy_awaitable(cached_awaitable_);
HITGNC   335 + 6 awaitable_active_ = false;
  336 + }
HITGNC   337 + 187 }
  338 +
307   inline 339   inline
HITCBC 308   108 any_write_stream::~any_write_stream() 340   125 any_write_stream::~any_write_stream()
309   { 341   {
HITGNC   342 + 125 end_operation();
HITGNC   343 + 125 if(cached_awaitable_)
HITGNC   344 + 101 ::operator delete(cached_awaitable_);
HITCBC 310   108 if(storage_) 345   125 if(storage_)
311   { 346   {
HITCBC 312   2 vt_->destroy(stream_); 347   2 vt_->destroy(stream_);
HITCBC 313   2 ::operator delete(storage_); 348   2 ::operator delete(storage_);
314 - if(cached_awaitable_)  
DCB 315 - 108 {  
316 - if(awaitable_active_)  
DCB 317 - 93 vt_->destroy_awaitable(cached_awaitable_);  
DCB 318 - 1 ::operator delete(cached_awaitable_);  
DCB 319 - 93 }  
320   } 349   }
HITCBC 321   108 } 350   125 }
322   351  
323   inline any_write_stream& 352   inline any_write_stream&
HITCBC 324   10 any_write_stream::operator=(any_write_stream&& other) noexcept 353   19 any_write_stream::operator=(any_write_stream&& other) noexcept
325   { 354   {
HITCBC 326   10 if(this != &other) 355   19 if(this != &other)
327   { 356   {
HITGNC   357 + 19 end_operation();
HITGNC   358 + 19 other.disown();
HITGNC   359 + 19 if(cached_awaitable_)
HITGNC   360 + 7 ::operator delete(cached_awaitable_);
HITCBC 328   10 if(storage_) 361   19 if(storage_)
329   { 362   {
HITCBC 330   1 vt_->destroy(stream_); 363   1 vt_->destroy(stream_);
HITCBC 331   1 ::operator delete(storage_); 364   1 ::operator delete(storage_);
332 - if(cached_awaitable_)  
DCB 333 - 10 {  
334 - if(awaitable_active_)  
DCB 335 - 4 vt_->destroy_awaitable(cached_awaitable_);  
DCB 336 - 1 ::operator delete(cached_awaitable_);  
DCB 337 - 4 }  
338   } 365   }
HITCBC 339   10 stream_ = std::exchange(other.stream_, nullptr); 366   19 stream_ = std::exchange(other.stream_, nullptr);
HITCBC 340   10 vt_ = std::exchange(other.vt_, nullptr); 367   19 vt_ = std::exchange(other.vt_, nullptr);
HITCBC 341   10 cached_awaitable_ = std::exchange(other.cached_awaitable_, nullptr); 368   19 cached_awaitable_ = std::exchange(other.cached_awaitable_, nullptr);
HITCBC 342   10 storage_ = std::exchange(other.storage_, nullptr); 369   19 storage_ = std::exchange(other.storage_, nullptr);
HITCBC 343   10 awaitable_active_ = std::exchange(other.awaitable_active_, false); 370   19 awaitable_active_ = std::exchange(other.awaitable_active_, false);
344   } 371   }
HITCBC 345   10 return *this; 372   19 return *this;
346   } 373   }
347   374  
348   template<WriteStream S> 375   template<WriteStream S>
349   requires (!std::same_as<std::decay_t<S>, any_write_stream>) 376   requires (!std::same_as<std::decay_t<S>, any_write_stream>)
HITCBC 350   4 any_write_stream::any_write_stream(S s) 377   4 any_write_stream::any_write_stream(S s)
HITCBC 351   4 : vt_(&vtable_for_impl<S>::value) 378   4 : vt_(&vtable_for_impl<S>::value)
352   { 379   {
353   struct guard { 380   struct guard {
354   any_write_stream* self; 381   any_write_stream* self;
355   bool committed = false; 382   bool committed = false;
HITCBC 356   4 ~guard() { 383   4 ~guard() {
HITCBC 357   4 if(!committed && self->storage_) { 384   4 if(!committed && self->storage_) {
HITCBC 358   1 if(self->stream_) 385   1 if(self->stream_)
359   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 360   1 ::operator delete(self->storage_); 387   1 ::operator delete(self->storage_);
HITCBC 361   1 self->storage_ = nullptr; 388   1 self->storage_ = nullptr;
HITCBC 362   1 self->stream_ = nullptr; 389   1 self->stream_ = nullptr;
363   } 390   }
HITCBC 364   4 } 391   4 }
HITCBC 365   4 } g{this}; 392   4 } g{this};
366   393  
HITCBC 367   4 storage_ = ::operator new(sizeof(S)); 394   4 storage_ = ::operator new(sizeof(S));
HITCBC 368   4 stream_ = ::new(storage_) S(std::move(s)); 395   4 stream_ = ::new(storage_) S(std::move(s));
369   396  
370   // Preallocate the awaitable storage 397   // Preallocate the awaitable storage
HITCBC 371   3 cached_awaitable_ = ::operator new(vt_->awaitable_size); 398   3 cached_awaitable_ = ::operator new(vt_->awaitable_size);
372   399  
HITCBC 373   3 g.committed = true; 400   3 g.committed = true;
HITCBC 374   4 } 401   4 }
375   402  
376   template<WriteStream S> 403   template<WriteStream S>
HITCBC 377   94 any_write_stream::any_write_stream(S* s) 404   105 any_write_stream::any_write_stream(S* s)
HITCBC 378   94 : stream_(s) 405   105 : stream_(s)
HITCBC 379   94 , vt_(&vtable_for_impl<S>::value) 406   105 , vt_(&vtable_for_impl<S>::value)
380   { 407   {
381   // Preallocate the awaitable storage 408   // Preallocate the awaitable storage
HITCBC 382   94 cached_awaitable_ = ::operator new(vt_->awaitable_size); 409   105 cached_awaitable_ = ::operator new(vt_->awaitable_size);
HITCBC 383   94 } 410   105 }
384   411  
385   template<ConstBufferSequence CB> 412   template<ConstBufferSequence CB>
386   auto 413   auto
HITCBC 387   83 any_write_stream::write_some(CB buffers) 414   91 any_write_stream::write_some(CB buffers)
388   { 415   {
389   struct awaitable 416   struct awaitable
390   { 417   {
391   any_write_stream* self_; 418   any_write_stream* self_;
392   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,
  421 + // holds the stream's inner awaitable.
  422 + bool owns_ = false;
  423 +
  424 + // A chain unwound while suspended here never reaches
  425 + // await_resume, so the inner awaitable is destroyed here.
HITGNC   426 + 178 ~awaitable()
  427 + {
HITGNC   428 + 178 if(owns_)
  429 + {
HITGNC   430 + 3 self_->vt_->destroy_awaitable(self_->cached_awaitable_);
HITGNC   431 + 3 self_->awaitable_active_ = false;
HITGNC   432 + 3 self_->owner_ = nullptr;
  433 + }
HITGNC   434 + 178 }
393   435  
HITCBC 394   83 awaitable( 436   91 awaitable(
395   any_write_stream* self, 437   any_write_stream* self,
396   CB const& buffers) noexcept 438   CB const& buffers) noexcept
HITCBC 397   83 : self_(self) 439   91 : self_(self)
HITCBC 398   83 , ba_(buffers) 440   91 , ba_(buffers)
399   { 441   {
HITCBC 400   83 } 442   91 }
401   443  
HITGNC   444 + 87 awaitable(awaitable&& other) noexcept
HITGNC   445 + 87 : self_(std::exchange(other.self_, nullptr))
HITGNC   446 + 87 , ba_(other.ba_)
HITGNC   447 + 87 , owns_(std::exchange(other.owns_, false))
  448 + {
HITGNC   449 + 87 if(owns_)
HITGNC   450 + 1 self_->owner_ = &owns_;
HITGNC   451 + 87 }
  452 +
402   bool 453   bool
HITCBC 403   83 await_ready() 454   90 await_ready()
404   { 455   {
405   // An empty write never touches the underlying stream 456   // An empty write never touches the underlying stream
HITCBC 406   83 if(ba_.to_span().empty()) 457   90 if(ba_.to_span().empty())
HITCBC 407   4 return true; 458   4 return true;
408   459  
HITCBC 409   79 self_->vt_->construct_awaitable( 460   86 self_->vt_->construct_awaitable(
HITCBC 410   79 self_->stream_, 461   86 self_->stream_,
HITCBC 411   79 self_->cached_awaitable_, 462   86 self_->cached_awaitable_,
HITCBC 412   79 ba_.to_span()); 463   86 ba_.to_span());
HITCBC 413   79 self_->awaitable_active_ = true; 464   86 self_->awaitable_active_ = true;
HITGNC   465 + 86 owns_ = true;
HITGNC   466 + 86 self_->owner_ = &owns_;
414   467  
HITCBC 415   158 return self_->vt_->await_ready( 468   172 return self_->vt_->await_ready(
HITCBC 416   79 self_->cached_awaitable_); 469   86 self_->cached_awaitable_);
417   } 470   }
418   471  
419   std::coroutine_handle<> 472   std::coroutine_handle<>
HITCBC 420   68 await_suspend(std::coroutine_handle<> h, io_env const* env) 473   74 await_suspend(std::coroutine_handle<> h, io_env const* env)
421   { 474   {
HITCBC 422   68 return self_->vt_->await_suspend( 475   74 return self_->vt_->await_suspend(
HITCBC 423   68 self_->cached_awaitable_, h, env); 476   74 self_->cached_awaitable_, h, env);
424   } 477   }
425   478  
426   [[nodiscard]] io_result<std::size_t> 479   [[nodiscard]] io_result<std::size_t>
HITCBC 427   81 await_resume() 480   81 await_resume()
428   { 481   {
HITCBC 429 - 81 if(!self_->awaitable_active_) 482 + 81 if(!owns_)
HITCBC 430   4 return {std::error_code(), 0}; 483   4 return {std::error_code(), 0};
431   struct guard { 484   struct guard {
432 - any_write_stream* self; 485 + awaitable* a;
HITCBC 433   77 ~guard() { 486   77 ~guard() {
HITCBC 434 - 77 self->vt_->destroy_awaitable(self->cached_awaitable_); 487 + 77 a->self_->vt_->destroy_awaitable(
HITCBC 435 - 77 self->awaitable_active_ = false; 488 + 77 a->self_->cached_awaitable_);
HITGNC   489 + 77 a->self_->awaitable_active_ = false;
HITGNC   490 + 77 a->self_->owner_ = nullptr;
HITGNC   491 + 77 a->owns_ = false;
HITCBC 436   77 } 492   77 }
HITCBC 437 - 77 } g{self_}; 493 + 77 } g{this};
HITCBC 438   77 return self_->vt_->await_resume( 494   77 return self_->vt_->await_resume(
HITCBC 439   77 self_->cached_awaitable_); 495   77 self_->cached_awaitable_);
HITCBC 440   77 } 496   77 }
441   }; 497   };
HITCBC 442   83 return awaitable{this, buffers}; 498   91 return awaitable{this, buffers};
443   } 499   }
444   500  
445   } // namespace capy 501   } // namespace capy
446   } // namespace boost 502   } // namespace boost
447   503  
448   #endif 504   #endif