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