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