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