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