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/io_awaitable.hpp>
19 : #include <boost/capy/concept/write_stream.hpp>
20 : #include <coroutine>
21 : #include <boost/capy/ex/io_env.hpp>
22 : #include <boost/capy/io_result.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 : @code
67 : // Owning - takes ownership of the stream
68 : any_write_stream owning_stream(socket{ioc});
69 :
70 : // Reference - wraps without ownership
71 : socket sock(ioc);
72 : any_write_stream ref_stream(&sock);
73 :
74 : char data[] = "hello";
75 : const_buffer buf(data, sizeof(data));
76 : auto [ec, n] = co_await owning_stream.write_some(std::span(&buf, 1));
77 : @endcode
78 :
79 : @see any_read_stream, any_stream, WriteStream
80 : */
81 : class any_write_stream
82 : {
83 : struct vtable;
84 :
85 : template<WriteStream S>
86 : struct vtable_for_impl;
87 :
88 : // ordered for cache line coherence
89 : void* stream_ = nullptr;
90 : vtable const* vt_ = nullptr;
91 : void* cached_awaitable_ = nullptr;
92 : void* storage_ = nullptr;
93 : bool awaitable_active_ = false;
94 :
95 : public:
96 : /** Destructor.
97 :
98 : Destroys the owned stream (if any) and releases the cached
99 : 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 HIT 4 : 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 : }
145 :
146 : /** Assign by moving.
147 :
148 : Destroys any owned stream and releases existing resources,
149 : then transfers ownership from `other`.
150 :
151 : @param other The wrapper to move from.
152 : @return Reference to this wrapper.
153 : */
154 : any_write_stream&
155 : operator=(any_write_stream&& other) noexcept;
156 :
157 : /** Construct by taking ownership of a WriteStream.
158 :
159 : Allocates storage and moves the stream into this wrapper.
160 : The wrapper owns the stream and destroys it.
161 :
162 : @param s The stream to take ownership of.
163 : */
164 : template<WriteStream S>
165 : requires (!std::same_as<std::decay_t<S>, any_write_stream>)
166 : any_write_stream(S s);
167 :
168 : /** Construct by wrapping a WriteStream without ownership.
169 :
170 : Wraps the given stream by pointer. The stream must remain
171 : valid for the lifetime of this wrapper.
172 :
173 : @param s Pointer to the stream to wrap.
174 : */
175 : template<WriteStream S>
176 : any_write_stream(S* s);
177 :
178 : /** Check if the wrapper contains a valid stream.
179 :
180 : @return `true` if wrapping a stream, `false` if default-constructed
181 : or moved-from.
182 : */
183 : bool
184 26 : has_value() const noexcept
185 : {
186 26 : return stream_ != nullptr;
187 : }
188 :
189 : /** Check if the wrapper contains a valid stream.
190 :
191 : @return `true` if wrapping a stream, `false` if default-constructed
192 : or moved-from.
193 : */
194 : explicit
195 3 : operator bool() const noexcept
196 : {
197 3 : return has_value();
198 : }
199 :
200 : /** Initiate an asynchronous write operation.
201 :
202 : Writes data from the provided buffer sequence. The operation
203 : completes when at least one byte is written, or an error
204 : occurs.
205 :
206 : @param buffers The buffer sequence containing data to write.
207 : Passed by value to ensure the sequence lives in the
208 : coroutine frame across suspension points.
209 :
210 : @return An awaitable that await-returns `(error_code,std::size_t)`.
211 :
212 : @par Immediate Completion
213 : The operation completes immediately without suspending
214 : the calling coroutine when:
215 : @li The buffer sequence is empty, returning `{error_code{}, 0}`.
216 : @li The underlying stream's awaitable reports immediate
217 : readiness via `await_ready`.
218 :
219 : @note This is a partial operation and may not process the
220 : entire buffer sequence. Use the composed @ref write algorithm
221 : for guaranteed complete transfer.
222 :
223 : @par Preconditions
224 : The wrapper must contain a valid stream (`has_value() == true`).
225 :
226 : @par After an Error
227 : A subsequent call is permitted. The wrapper forwards directly
228 : to the underlying stream, imposing no stricter rule than
229 : @ref WriteStream.
230 : */
231 : template<ConstBufferSequence CB>
232 : auto
233 : write_some(CB buffers);
234 :
235 : protected:
236 : /** Rebind to a new stream after move.
237 :
238 : Updates the internal pointer to reference a new stream object.
239 : Used by owning wrappers after move assignment when the owned
240 : object has moved to a new location.
241 :
242 : @param new_stream The new stream to bind to. Must be the same
243 : type as the original stream.
244 :
245 : @note Terminates if called with a stream of different type
246 : than the original.
247 : */
248 : template<WriteStream S>
249 : void
250 : rebind(S& new_stream) noexcept
251 : {
252 : if(vt_ != &vtable_for_impl<S>::value)
253 : std::terminate();
254 : stream_ = &new_stream;
255 : }
256 : };
257 :
258 : struct any_write_stream::vtable
259 : {
260 : // ordered by call frequency for cache line coherence
261 : void (*construct_awaitable)(
262 : void* stream,
263 : void* storage,
264 : std::span<const_buffer const> buffers);
265 : bool (*await_ready)(void*);
266 : std::coroutine_handle<> (*await_suspend)(void*, std::coroutine_handle<>, io_env const*);
267 : io_result<std::size_t> (*await_resume)(void*);
268 : void (*destroy_awaitable)(void*) noexcept;
269 : std::size_t awaitable_size;
270 : std::size_t awaitable_align;
271 : void (*destroy)(void*) noexcept;
272 : };
273 :
274 : template<WriteStream S>
275 : struct any_write_stream::vtable_for_impl
276 : {
277 : using Awaitable = decltype(std::declval<S&>().write_some(
278 : std::span<const_buffer const>{}));
279 :
280 : static void
281 3 : do_destroy_impl(void* stream) noexcept
282 : {
283 3 : static_cast<S*>(stream)->~S();
284 3 : }
285 :
286 : static void
287 89 : construct_awaitable_impl(
288 : void* stream,
289 : void* storage,
290 : std::span<const_buffer const> buffers)
291 : {
292 89 : auto& s = *static_cast<S*>(stream);
293 89 : ::new(storage) Awaitable(s.write_some(buffers));
294 89 : }
295 :
296 : static constexpr vtable value = {
297 : &construct_awaitable_impl,
298 89 : +[](void* p) {
299 89 : return static_cast<Awaitable*>(p)->await_ready();
300 : },
301 79 : +[](void* p, std::coroutine_handle<> h, io_env const* env) {
302 79 : return detail::call_await_suspend(
303 79 : static_cast<Awaitable*>(p), h, env);
304 : },
305 87 : +[](void* p) {
306 87 : return static_cast<Awaitable*>(p)->await_resume();
307 : },
308 91 : +[](void* p) noexcept {
309 26 : static_cast<Awaitable*>(p)->~Awaitable();
310 : },
311 : sizeof(Awaitable),
312 : alignof(Awaitable),
313 : &do_destroy_impl
314 : };
315 : };
316 :
317 : inline
318 117 : any_write_stream::~any_write_stream()
319 : {
320 117 : if(storage_)
321 : {
322 2 : vt_->destroy(stream_);
323 2 : ::operator delete(storage_);
324 : }
325 117 : if(cached_awaitable_)
326 : {
327 102 : if(awaitable_active_)
328 1 : vt_->destroy_awaitable(cached_awaitable_);
329 102 : ::operator delete(cached_awaitable_);
330 : }
331 117 : }
332 :
333 : inline any_write_stream&
334 10 : any_write_stream::operator=(any_write_stream&& other) noexcept
335 : {
336 10 : if(this != &other)
337 : {
338 10 : if(storage_)
339 : {
340 1 : vt_->destroy(stream_);
341 1 : ::operator delete(storage_);
342 : }
343 10 : if(cached_awaitable_)
344 : {
345 4 : if(awaitable_active_)
346 1 : vt_->destroy_awaitable(cached_awaitable_);
347 4 : ::operator delete(cached_awaitable_);
348 : }
349 10 : stream_ = std::exchange(other.stream_, nullptr);
350 10 : vt_ = std::exchange(other.vt_, nullptr);
351 10 : cached_awaitable_ = std::exchange(other.cached_awaitable_, nullptr);
352 10 : storage_ = std::exchange(other.storage_, nullptr);
353 10 : awaitable_active_ = std::exchange(other.awaitable_active_, false);
354 : }
355 10 : return *this;
356 : }
357 :
358 : template<WriteStream S>
359 : requires (!std::same_as<std::decay_t<S>, any_write_stream>)
360 4 : any_write_stream::any_write_stream(S s)
361 4 : : vt_(&vtable_for_impl<S>::value)
362 : {
363 : struct guard {
364 : any_write_stream* self;
365 : bool committed = false;
366 4 : ~guard() {
367 4 : if(!committed && self->storage_) {
368 1 : if(self->stream_)
369 : self->vt_->destroy(self->stream_); // LCOV_EXCL_LINE OOM rollback: only when the cached-awaitable allocation throws
370 1 : ::operator delete(self->storage_);
371 1 : self->storage_ = nullptr;
372 1 : self->stream_ = nullptr;
373 : }
374 4 : }
375 4 : } g{this};
376 :
377 4 : storage_ = ::operator new(sizeof(S));
378 4 : stream_ = ::new(storage_) S(std::move(s));
379 :
380 : // Preallocate the awaitable storage
381 3 : cached_awaitable_ = ::operator new(vt_->awaitable_size);
382 :
383 3 : g.committed = true;
384 4 : }
385 :
386 : template<WriteStream S>
387 103 : any_write_stream::any_write_stream(S* s)
388 103 : : stream_(s)
389 103 : , vt_(&vtable_for_impl<S>::value)
390 : {
391 : // Preallocate the awaitable storage
392 103 : cached_awaitable_ = ::operator new(vt_->awaitable_size);
393 103 : }
394 :
395 : template<ConstBufferSequence CB>
396 : auto
397 93 : any_write_stream::write_some(CB buffers)
398 : {
399 : struct awaitable
400 : {
401 : any_write_stream* self_;
402 : detail::const_buffer_array<detail::max_iovec_> ba_;
403 :
404 93 : awaitable(
405 : any_write_stream* self,
406 : CB const& buffers) noexcept
407 93 : : self_(self)
408 93 : , ba_(buffers)
409 : {
410 93 : }
411 :
412 : bool
413 93 : await_ready() const noexcept
414 : {
415 93 : return ba_.to_span().empty();
416 : }
417 :
418 : std::coroutine_handle<>
419 89 : await_suspend(std::coroutine_handle<> h, io_env const* env)
420 : {
421 89 : self_->vt_->construct_awaitable(
422 89 : self_->stream_,
423 89 : self_->cached_awaitable_,
424 89 : ba_.to_span());
425 89 : self_->awaitable_active_ = true;
426 :
427 89 : if(self_->vt_->await_ready(self_->cached_awaitable_))
428 10 : return h;
429 :
430 79 : return self_->vt_->await_suspend(
431 79 : self_->cached_awaitable_, h, env);
432 : }
433 :
434 : io_result<std::size_t>
435 91 : await_resume()
436 : {
437 91 : if(!self_->awaitable_active_)
438 4 : return {{}, 0};
439 : struct guard {
440 : any_write_stream* self;
441 87 : ~guard() {
442 87 : self->vt_->destroy_awaitable(self->cached_awaitable_);
443 87 : self->awaitable_active_ = false;
444 87 : }
445 87 : } g{self_};
446 87 : return self_->vt_->await_resume(
447 87 : self_->cached_awaitable_);
448 87 : }
449 : };
450 93 : return awaitable{this, buffers};
451 : }
452 :
453 : } // namespace capy
454 : } // namespace boost
455 :
456 : #endif
|
