include/boost/capy/io/write_now.hpp

100.0% Lines (77 / 77, 3 excl) 97.0% Functions (64 / 66, 1 excl)

[ ] prev/next file

write_now.hpp

f(x) Functions (67)

Function Calls Lines Blocks

Line	Hits	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_WRITE_NOW_HPP
12		#define BOOST_CAPY_IO_WRITE_NOW_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/buffers/consuming_buffers.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/executor_ref.hpp>
22		#include <boost/capy/ex/frame_allocator.hpp>
23		#include <boost/capy/ex/abandon.hpp>
24		#include <boost/capy/ex/await_context.hpp>
25		#include <boost/capy/results.hpp>
26
27		#include <cstddef>
28		#include <exception>
29		#include <new>
30		#include <stop_token>
31		#include <utility>
32
33		#ifndef BOOST_CAPY_WRITE_NOW_WORKAROUND
34		# if defined(__GNUC__) && !defined(__clang__)
35		# define BOOST_CAPY_WRITE_NOW_WORKAROUND 1
36		# else
37		# define BOOST_CAPY_WRITE_NOW_WORKAROUND 0
38		# endif
39		#endif
40
41		namespace boost {
42		namespace capy {
43
44		/** Eagerly writes complete buffer sequences with frame caching.
45
46		This class wraps a @ref WriteStream and provides an `operator()`
47		that writes an entire buffer sequence, attempting to complete
48		synchronously. If every `write_some` completes without suspending,
49		the entire operation finishes in `await_ready` with no coroutine
50		suspension.
51
52		The class maintains a one-element coroutine frame cache. After
53		the first call, subsequent calls reuse the cached frame memory,
54		avoiding repeated allocation for the internal coroutine.
55
56		@tparam Stream The stream type, must satisfy @ref WriteStream.
57
58		@par Thread Safety
59		Distinct objects: Safe.
60		Shared objects: Unsafe.
61
62		@par Preconditions
63		Only one operation may be outstanding at a time. A new call to
64		`operator()` must not be made until the previous operation has
65		completed (i.e., the returned awaitable is fully consumed).
66
67		Destroying the returned operation object while it is being awaited
68		calls `std::terminate`, except while its chain is unwound after an
69		@ref abandon.
70
71		@par Example
72
73		@code
74		template< WriteStream Stream >
75		task<> send_messages( Stream& stream )
76		{
77		write_now wn( stream );
78		auto [ec1, n1] = co_await wn( make_buffer( "hello" ) );
79		if( ec1 )
80		detail::throw_system_error( ec1 );
81		auto [ec2, n2] = co_await wn( make_buffer( "world" ) );
82		if( ec2 )
83		detail::throw_system_error( ec2 );
84		}
85		@endcode
86
87		@see @ref write, @ref write_some, @ref WriteStream, @ref ConstBufferSequence
88		*/
89		template<class Stream>
90		requires WriteStream<Stream>
91		class write_now
92		{
93		Stream& stream_;
94		void* cached_frame_ = nullptr;
95		std::size_t cached_size_ = 0;
96
97		struct [[nodiscard]] BOOST_CAPY_CORO_AWAIT_ELIDABLE
98		op_type
99		{
100		struct promise_type
101		{
102		results<std::size_t> result_;
103		std::exception_ptr ep_;
104		std::coroutine_handle<> cont_{nullptr};
105		await_context const* env_ = nullptr;
106		bool done_ = false;
107
108		// The awaiting op_type while a parent awaits this frame.
109		op_type* owner_ = nullptr;
110
111	76x	op_type get_return_object()
112		{
113		return op_type{
114		std::coroutine_handle<
115	76x	promise_type>::from_promise(*this)};
116		}
117
118	76x	auto initial_suspend() noexcept
119		{
120		#if BOOST_CAPY_WRITE_NOW_WORKAROUND
121	76x	return std::suspend_always{};
122		#else
123		return std::suspend_never{};
124		#endif
125		}
126
127	72x	auto final_suspend() noexcept
128		{
129		struct awaiter
130		{
131		promise_type* p_;
132
133	2x	bool await_ready() const noexcept
134		{
135	2x	return false;
136		}
137
138	2x	std::coroutine_handle<> await_suspend(std::coroutine_handle<>) const noexcept
139		{
140	2x	p_->done_ = true;
141	2x	if(!p_->cont_)
142	−	return std::noop_coroutine(); // LCOV_EXCL_LINE cont_ always set on this (suspend_always) path
143	2x	return p_->cont_;
144		}
145
146	−	void await_resume() const noexcept {} // LCOV_EXCL_LINE final_suspend awaiter, never resumed
147		};
148	72x	return awaiter{this};
149		}
150
151	50x	void return_value(
152		results<std::size_t> r) noexcept
153		{
154	50x	result_ = r;
155	50x	}
156
157	22x	void unhandled_exception()
158		{
159	22x	ep_ = std::current_exception();
160	22x	}
161
162		std::suspend_always yield_value(int) noexcept
163		{
164		return {};
165		}
166
167		template<class A>
168	93x	auto await_transform(A&& a)
169		{
170		using decayed = std::decay_t<A>;
171		if constexpr (ContextAwaitable<decayed>)
172		{
173		struct wrapper
174		{
175		decayed inner_;
176		promise_type* p_;
177
178	5x	bool await_ready()
179		{
180	5x	return inner_.await_ready();
181		}
182
183		// The inner operation may resume or destroy this
184		// frame on another thread before control returns
185		// here, so the handle must not round-trip through
186		// the frame.
187	5x	auto await_suspend(std::coroutine_handle<> h)
188		{
189	5x	return detail::symmetric_transfer(
190		detail::call_await_suspend(
191		&inner_, h,
192	10x	p_->env_));
193		}
194
195	5x	decltype(auto) await_resume()
196		{
197		// The resumption may arrive on another thread.
198	5x	set_current_frame_allocator(
199	5x	p_->env_->frame_allocator);
200	5x	return inner_.await_resume();
201		}
202		};
203		return wrapper{
204	93x	std::forward<A>(a), this};
205		}
206		else
207		{
208		return std::forward<A>(a);
209		}
210		}
211
212		static void*
213	76x	operator new(
214		std::size_t size,
215		write_now& self,
216		auto&)
217		{
218	76x	if(self.cached_frame_ &&
219	6x	self.cached_size_ >= size)
220	5x	return self.cached_frame_;
221	71x	void* p = ::operator new(size);
222	71x	if(self.cached_frame_)
223	1x	::operator delete(self.cached_frame_);
224	71x	self.cached_frame_ = p;
225	71x	self.cached_size_ = size;
226	71x	return p;
227		}
228
229		static void
230	76x	operator delete(void*, std::size_t) noexcept
231		{
232	76x	}
233		};
234
235		std::coroutine_handle<promise_type> h_;
236
237	150x	~op_type()
238		{
239	150x	if(h_)
240		{
241		// Only the unwind of an abandoned chain destroys a frame
242		// that is still awaited.
243	76x	auto& p = h_.promise();
244	78x	if(!p.done_ && p.owner_ &&
245	2x	!(p.env_->owner && p.env_->owner->abandoned()))
246	−	std::terminate(); // LCOV_EXCL_LINE reached only in a death test's child process
247	76x	h_.destroy();
248		}
249	150x	}
250
251		op_type(op_type const&) = delete;
252		op_type& operator=(op_type const&) = delete;
253
254	74x	op_type(op_type&& other) noexcept
255	74x	: h_(std::exchange(other.h_, nullptr))
256		{
257	74x	}
258
259		op_type& operator=(op_type&&) = delete;
260
261	74x	bool await_ready() const noexcept
262		{
263	74x	return h_.promise().done_;
264		}
265
266	74x	std::coroutine_handle<> await_suspend(
267		std::coroutine_handle<> cont,
268		await_context const* env)
269		{
270	74x	auto& p = h_.promise();
271	74x	p.cont_ = cont;
272	74x	p.env_ = env;
273	74x	p.owner_ = this;
274	74x	return h_;
275		}
276
277	72x	[[nodiscard]] results<std::size_t> await_resume()
278		{
279	72x	auto& p = h_.promise();
280	72x	p.owner_ = nullptr;
281	72x	if(p.ep_)
282	22x	std::rethrow_exception(p.ep_);
283	50x	return p.result_;
284		}
285
286		private:
287	76x	explicit op_type(
288		std::coroutine_handle<promise_type> h)
289	76x	: h_(h)
290		{
291	76x	}
292		};
293
294		public:
295		/** Destructor. Frees the cached coroutine frame. */
296	70x	~write_now()
297		{
298	70x	if(cached_frame_)
299	70x	::operator delete(cached_frame_);
300	70x	}
301
302		/** Construct from a stream reference.
303
304		@param s The stream to write to. Must outlive this object.
305		*/
306		explicit
307	70x	write_now(Stream& s) noexcept
308	70x	: stream_(s)
309		{
310	70x	}
311
312		write_now(write_now const&) = delete;
313		write_now& operator=(write_now const&) = delete;
314
315		/** Eagerly write the entire buffer sequence.
316
317		Writes data to the stream by calling `write_some` repeatedly
318		until the entire buffer sequence is written or an error
319		occurs. The operation attempts to complete synchronously:
320		if every `write_some` completes without suspending, the
321		entire operation finishes in `await_ready`.
322
323		When the fast path cannot complete, the coroutine suspends
324		and continues asynchronously. The internal coroutine frame
325		is cached and reused across calls.
326
327		@param buffers The buffer sequence to write. Passed by
328		value to ensure the sequence lives in the coroutine
329		frame across suspension points.
330
331		@return An awaitable that await-returns `(error_code,std::size_t)`.
332		On success, `n` equals `buffer_size(buffers)`. On
333		error, `n` is the number of bytes written before the
334		error. Compare error codes to conditions:
335		@li `cond::canceled` - Operation was cancelled
336		@li `std::errc::broken_pipe` - Peer closed connection
337
338		@par Example
339
340		@code
341		write_now wn( stream );
342		auto [ec, n] = co_await wn( make_buffer( body ) );
343		if( ec )
344		detail::throw_system_error( ec );
345		@endcode
346
347		@see @ref write, @ref write_some, @ref WriteStream
348		*/
349		// GCC falsely warns that the coroutine promise's
350		// placement operator new(size_t, write_now&, auto&)
351		// mismatches operator delete(void*, size_t). Per the
352		// standard, coroutine deallocation lookup is separate.
353		#if defined(__GNUC__) && !defined(__clang__)
354		#pragma GCC diagnostic push
355		#pragma GCC diagnostic ignored "-Wmismatched-new-delete"
356		#endif
357
358		#if BOOST_CAPY_WRITE_NOW_WORKAROUND
359		template<ConstBufferSequence Buffers>
360		op_type
361	76x	operator()(Buffers buffers)
362		{
363		std::size_t const total_size = buffer_size(buffers);
364		std::size_t total_written = 0;
365		consuming_buffers cb(buffers);
366		while(total_written < total_size)
367		{
368		auto r =
369		co_await stream_.write_some(cb.data());
370		cb.consume(std::get<1>(r));
371		total_written += std::get<1>(r);
372		if(std::get<0>(r))
373		co_return results<std::size_t>{
374		std::get<0>(r), total_written};
375		}
376		co_return results<std::size_t>{
377		std::error_code(), total_written};
378	152x	}
379		#else
380		template<ConstBufferSequence Buffers>
381		op_type
382		operator()(Buffers buffers)
383		{
384		std::size_t const total_size = buffer_size(buffers);
385		std::size_t total_written = 0;
386
387		// GCC ICE in expand_expr_real_1 (expr.cc:11376)
388		// when the buffer slice spans a co_yield, so
389		// the GCC path uses a separate simple coroutine.
390		consuming_buffers cb(buffers);
391		while(total_written < total_size)
392		{
393		auto inner = stream_.write_some(cb.data());
394		if(!inner.await_ready())
395		break;
396		auto r = inner.await_resume();
397		if(std::get<0>(r))
398		co_return results<std::size_t>{
399		std::get<0>(r), total_written};
400		cb.consume(std::get<1>(r));
401		total_written += std::get<1>(r);
402		}
403
404		if(total_written >= total_size)
405		co_return results<std::size_t>{
406		std::error_code(), total_written};
407
408		co_yield 0;
409
410		while(total_written < total_size)
411		{
412		auto r =
413		co_await stream_.write_some(cb.data());
414		cb.consume(std::get<1>(r));
415		total_written += std::get<1>(r);
416		if(std::get<0>(r))
417		co_return results<std::size_t>{
418		std::get<0>(r), total_written};
419		}
420		co_return results<std::size_t>{
421		std::error_code(), total_written};
422		}
423		#endif
424
425		#if defined(__GNUC__) && !defined(__clang__)
426		#pragma GCC diagnostic pop
427		#endif
428		};
429
430		template<WriteStream S>
431		write_now(S&) -> write_now<S>;
432
433		} // namespace capy
434		} // namespace boost
435
436		#endif
437