src/corosio/src/tls/detail/engine_driver.hpp

97.3% Lines (107 / 110) 90.5% Branches (38 / 42) 100.0% Functions (34 / 34)

[ ] prev/next file n p prev/next uncovered line

engine_driver.hpp

f(x) Functions (34)

Function Calls Lines Branches Blocks

Line	Branch	Hits	Source Code
1			//
2			// Copyright (c) 2026 Steve Gerbino
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/corosio
8			//
9
10			#ifndef SRC_TLS_DETAIL_ENGINE_DRIVER_HPP
11			#define SRC_TLS_DETAIL_ENGINE_DRIVER_HPP
12
13			#include <boost/corosio/tls_context.hpp>
14			#include <boost/corosio/tls_stream.hpp>
15
16			#include "src/tls/detail/engine_types.hpp"
17
18			#include <boost/capy/buffers.hpp>
19			#include <boost/capy/detail/buffer_array.hpp>
20			#include <boost/capy/ex/async_mutex.hpp>
21			#include <boost/capy/io/any_stream.hpp>
22			#include <boost/capy/results.hpp>
23			#include <boost/capy/task.hpp>
24			#include <boost/capy/write.hpp>
25
26			#include <atomic>
27			#include <concepts>
28			#include <cstddef>
29			#include <cstdint>
30			#include <cstring>
31			#include <limits>
32			#include <string>
33			#include <string_view>
34			#include <system_error>
35			#include <tuple>
36			#include <utility>
37			#include <vector>
38
39			// This header is instantiated in both backends' TUs, whose vendor
40			// headers cannot coexist (WolfSSL's OpenSSL-compat layer clashes
41			// with genuine OpenSSL declarations on some builds); it must stay
42			// vendor-free.
43			#ifdef OPENSSL_VERSION_NUMBER
44			#error engine_driver.hpp must stay vendor-header-free
45			#endif
46			#ifdef WOLFSSL_VERSION
47			#error engine_driver.hpp must stay vendor-header-free
48			#endif
49
50			/*
51			TLS Driver Architecture
52
53			TLS layer wrapping an underlying stream (via any_stream). One
54			read_some and one write_some may be in flight concurrently.
55
56			Engine / Driver Split. The engine (each backend's detail/engine.hpp)
57			owns the TLS session, its byte interface, and every library-result-
58			to-error-code decision; it is synchronous and transport-free. This
59			driver owns the transport, the per-direction claims, and all
60			buffering, and loops on engine verdicts:
61
62			1. Call eng_.perform(op, data, len)
63			2. output_then_retry: flush pending ciphertext, then retry
64			3. input: feed transport bytes into the engine, then retry
65			4. done / output_then_done: flush if pending, report the
66			(already mapped) ec
67
68			Renegotiation causes cross-direction I/O: a read may need to
69			write handshake data, a write may need to read. Each operation
70			services whatever direction the engine requests.
71
72			Full-Duplex Concurrency. rd_cm_ and wr_cm_ are separate transport
73			claims: a read parked on the network cannot stall a concurrent
74			write's flush, and vice versa. A cross-direction request (e.g. a
75			read needing to flush queued handshake output) takes the other
76			direction's claim itself.
77
78			read_gen_ counts successful deposits into the engine. Each engine
79			call snapshots it before running; if fill_input later finds the
80			generation has moved (a concurrent operation's flush suspension let
81			a deposit land in the meantime) it retries the engine instead of
82			issuing a redundant transport read.
83
84			The transport reads ciphertext straight into the engine's input
85			staging via input_area(), so no driver-side buffer or deposit copy
86			sits on the read path. The write path stays copy-based on purpose:
87			ciphertext drained from the engine into out_buf_[0, out_len_) is
88			decoupled from the engine's shared output staging, so a concurrent
89			reader can still emit its own records (an alert, a key update) while
90			a write is in flight. out_len_ holds only the abandoned tail a failed
91			or canceled write kept for retry; the length being written is a local,
92			never published, so a concurrent flush probe cannot mistake in-flight
93			bytes for a tail and park on the write claim behind a healthy write.
94
95			A read or write that fully satisfies the caller's buffer still owes
96			a trailing flush (queued handshake or session-ticket output must
97			reach the peer before the call can report success). If that flush
98			fails, the transfer already succeeded and the stream contract
99			forbids reporting both a full count and an error in the same
100			result, so the error is stashed in pending_flush_ec_ and surfaced
101			by the next read, write, or shutdown call instead.
102			*/
103
104			namespace boost::corosio {
105
106			namespace detail {
107
108			/** The engine surface `engine_driver` drives.
109
110			A synchronous, transport-free TLS record engine: `perform`
111			advances one operation and returns a fully mapped verdict, while
112			`put_input` / `get_output` shuttle wire bytes. The handshake
113			hooks (`check_context`, `check_session`, `prepare`) let each
114			backend gate and configure a handshake at the protocol-mandated
115			points without the driver knowing the backend's session model.
116			*/
117			template<class Engine>
118			concept tls_engine = requires(
119			Engine& e,
120			Engine const& ce,
121			engine_op op,
122			void* buf,
123			unsigned char const* in,
124			unsigned char* out,
125			std::size_t n,
126			tls_context const& ctx,
127			tls_role role,
128			std::string const& hostname,
129			std::string& alpn) {
130			{ e.perform(op, buf, n) } -> std::same_as<engine_result>;
131			{ e.put_input(in, n) } -> std::same_as<std::size_t>;
132			{ e.input_area() } -> std::same_as<std::pair<unsigned char*, std::size_t>>;
133			{ e.input_committed(n) };
134			{ ce.pending_output() } -> std::same_as<std::size_t>;
135			{ e.get_output(out, n) } -> std::same_as<std::size_t>;
136			{ ce.received_shutdown() } -> std::same_as<bool>;
137			{ ce.capture_alpn(alpn) };
138			{ e.reset() };
139			{ ce.check_context() } -> std::same_as<std::error_code>;
140			{ ce.check_session() } -> std::same_as<std::error_code>;
141			{ e.prepare(ctx, role, hostname) } -> std::same_as<std::error_code>;
142			};
143
144			/** Coroutine driver shared by every TLS backend.
145
146			Hosts the transport, the per-direction claims, the buffering, and
147			the four operation loops; the engine type supplies all TLS
148			mechanics. Each backend's stream impl instantiates this template
149			with its own engine so the driver logic exists exactly once.
150
151			@par Thread Safety
152			Distinct objects: Safe.@n
153			Shared objects: Unsafe.
154			*/
155			template<tls_engine Engine>
156			class engine_driver
157			{
158			// Large enough to hold the largest possible TLS record.
159			static constexpr std::size_t buffer_size_ = std::size_t{17} * 1024;
160
161			// Operations hold a reference across their suspensions, so the
162			// driver outlives the stream that created it until they finish.
163			std::atomic<std::size_t> refs_{1};
164
165			// Set when the stream is destroyed: no transport call may follow.
166			std::atomic<bool> orphaned_{false};
167
168			// Transport calls in flight. Once orphaned, the owned transport is
169			// destroyed when this reaches zero: right away for a stream with no
170			// call parked, so it never outlives its context waiting on an
171			// operation nobody awaits.
172			std::atomic<std::size_t> calls_{0};
173			std::atomic<bool> transport_gone_{false};
174
175			// The transport in owning mode; s_ refers to it without owning.
176			tls_owned_transport owned_;
177			capy::any_stream s_;
178			tls_context ctx_;
179			Engine eng_;
180
181			// A handshake was attempted (successfully or not); the stream
182			// must be reset before the next handshake.
183			bool used_ = false;
184
185			// Per-stream SNI/verification hostname, set via set_hostname().
186			std::string hostname_;
187
188			// ALPN protocol negotiated during the handshake (empty if none).
189			std::string alpn_selected_;
190
191			std::vector<char> out_buf_;
192
193			// One transport claim per direction so a parked reader cannot block
194			// a writer's flush. read_gen_ counts deposits into the engine: each
195			// operation captures it at its engine call and passes it to
196			// fill_input, which retries the engine instead of issuing a stale
197			// transport read when a deposit landed anywhere after the engine's
198			// verdict (including during an intervening flush suspension).
199			capy::async_mutex rd_cm_;
200			capy::async_mutex wr_cm_;
201			std::uint64_t read_gen_ = 0;
202
203			// Engine ciphertext drained but not yet written: the abandoned
204			// tail a failed or canceled transport write kept for retry.
205			// Zero while a write is in flight (that length is a local).
206			std::size_t out_len_ = 0;
207
208			// A trailing flush that fails after the engine already accepted a
209			// full payload cannot be reported alongside n == size (the stream
210			// contract forbids error + full transfer, and capy::write's
211			// composed loop discards ec exactly in that case); the error is
212			// held here and surfaced by the next operation instead.
213			std::error_code pending_flush_ec_;
214
215	1/1 ✓ Branch 5 → 6 taken 69114 times.	69114x	capy::task<std::error_code> flush_output()
216			{
217			if (eng_.pending_output() == 0 && out_len_ == 0)
218			co_return std::error_code{};
219
220			auto [lec] = co_await wr_cm_.lock();
221			if (lec)
222			co_return lec;
223			capy::async_mutex::lock_guard wr_guard(&wr_cm_);
224
225			// Drain under the claim so records reach the wire in engine
226			// order even when both directions produce output.
227			while (eng_.pending_output() > 0 \|\| out_len_ > 0)
228			{
229			if (orphaned())
230			co_return make_error_code(capy::error::canceled);
231
232			// Take the abandoned tail (from an earlier failed write in
233			// this loop or a prior flush) into a local and append fresh
234			// engine ciphertext behind it, preserving record order.
235			// out_len_ drops to zero for the duration of the write: the
236			// in-flight length lives only in `n`, so a concurrent flush
237			// probe never sees these bytes as a tail and never parks on
238			// the write claim behind a healthy write.
239			std::size_t n = out_len_;
240			out_len_ = 0;
241			if (n < out_buf_.size())
242			n += eng_.get_output(
243			reinterpret_cast<unsigned char*>(out_buf_.data()) + n,
244			out_buf_.size() - n);
245			// The loop guard just confirmed pending bytes exist, so a
246			// drain failure here is unreachable in practice; fail loudly
247			// rather than silently drop already-accepted ciphertext.
248			if (n == 0) // LCOV_EXCL_LINE unreachable: pending bytes confirmed
249			co_return make_error_code(
250			std::errc::
251			no_buffer_space); // LCOV_EXCL_LINE unreachable: transport returned 0 with no error unreachable: pending bytes confirmed
252			// capy::write's loop, with the stream's destruction checked
253			// between partial writes: once orphaned, the transport is
254			// no longer this driver's to write to.
255			std::error_code ec;
256			std::size_t wn = 0;
257			while (wn < n)
258			{
259			transport_call call(*this);
260			if (!call)
261			{
262			ec = make_error_code(capy::error::canceled);
263			break;
264			}
265			auto [wec, k] = co_await s_.write_some(
266			capy::const_buffer(out_buf_.data() + wn, n - wn));
267			call.end();
268			wn += k;
269			if (wn < n && wec)
270			{
271			ec = wec;
272			break;
273			}
274			if (wn < n && orphaned())
275			{
276			ec = make_error_code(capy::error::canceled);
277			break;
278			}
279			}
280			if (ec)
281			{
282			// wn bytes already reached the peer; keep only the unsent
283			// remainder so a post-cancellation flush retry resends
284			// neither the delivered prefix nor loses the rest.
285			std::memmove(out_buf_.data(), out_buf_.data() + wn, n - wn);
286			out_len_ = n - wn;
287			co_return ec;
288			}
289			}
290			co_return std::error_code{};
291		138228x	}
292
293			// Flushes on a fatal/alert exit path: the engine may have queued a
294			// close_notify or alert that should still reach the peer, but a
295			// caller already reporting the triggering error cannot also report
296			// this flush's own failure. flush_output() itself no-ops when
297			// nothing is pending.
298	1/1 ✓ Branch 5 → 6 taken 20 times.	20x	capy::task<void> best_effort_flush()
299			{
300			std::ignore = co_await flush_output();
301		40x	}
302
303			// gen is the caller's read_gen_ snapshot from its engine call; a
304			// later snapshot would miss deposits landing while the caller's
305			// flush_output was suspended.
306	1/1 ✓ Branch 5 → 6 taken 34333 times.	34333x	capy::task<std::error_code> fill_input(std::uint64_t gen)
307			{
308			// Input already arrived since the engine's verdict; retry the
309			// engine rather than queue behind a re-parked reader.
310			if (read_gen_ != gen)
311			co_return std::error_code{};
312
313			auto [lec] = co_await rd_cm_.lock();
314			if (lec)
315			co_return lec;
316			capy::async_mutex::lock_guard rd_guard(&rd_cm_);
317
318			// Input arrived while we queued for the claim; the caller's
319			// engine retry consumes it.
320			if (read_gen_ != gen)
321			co_return std::error_code{};
322
323			if (orphaned())
324			co_return make_error_code(capy::error::canceled);
325
326			// Read the transport straight into the engine's input staging:
327			// input_area() hands back the contiguous writable run, so no
328			// staging buffer or deposit copy sits between the socket and the
329			// engine.
330			auto [dst, cap] = eng_.input_area();
331			// The staging is full while the caller still wants input: its own
332			// engine retry must decrypt the staged record to free space, so
333			// report success to re-run it rather than issue a zero-length read.
334			if (cap == 0)
335			co_return std::error_code{};
336
337			transport_call call(*this);
338			if (!call)
339			co_return make_error_code(capy::error::canceled);
340			auto [ec, n] = co_await s_.read_some(capy::mutable_buffer(dst, cap));
341			call.end();
342
343			// The stream went away while the read was parked; its result
344			// belongs to no one.
345			if (orphaned())
346			co_return make_error_code(capy::error::canceled);
347
348			// ReadStream permits n>0 alongside ec (IOCP forwards
349			// bytes_transferred on failed completions; a canceled read can
350			// likewise deliver bytes); commit whatever arrived before
351			// surfacing ec, or the record stream loses wire bytes the
352			// transport already handed over.
353			if (n > 0)
354			{
355			eng_.input_committed(n);
356			++read_gen_;
357			co_return ec;
358			}
359
360			if (ec)
361			co_return ec;
362
363			// The transport delivered nothing without an error, so it cannot
364			// make progress: fail loudly rather than spin the engine's input
365			// retry against a staging that will never fill.
366			co_return make_error_code(
367			std::errc::
368			no_buffer_space); // LCOV_EXCL_LINE unreachable: staging cannot stay empty
369		68666x	}
370
371		6556x	void destroy_transport() noexcept
372			{
373	2/2 ✓ Branch 3 → 4 taken 3278 times. ✓ Branch 3 → 5 taken 3278 times.	6556x	if (transport_gone_.exchange(true, std::memory_order_acq_rel))
374		3278x	return;
375		3278x	auto const owned = std::exchange(owned_, tls_owned_transport{});
376	2/2 ✓ Branch 6 → 7 taken 6 times. ✓ Branch 6 → 8 taken 3272 times.	3278x	if (owned.destroy)
377		6x	owned.destroy(owned.p);
378			}
379
380			// Return false once orphaned: the transport may already be gone.
381			// The orphan check follows the count's increment, pairing with
382			// orphan()'s store-then-load, so one side always sees the other.
383		69064x	bool begin_transport_call() noexcept
384			{
385		69064x	calls_.fetch_add(1, std::memory_order_seq_cst);
386	1/2 ✓ Branch 5 → 6 taken 69064 times. ✗ Branch 5 → 7 not taken.	69064x	if (!orphaned_.load(std::memory_order_seq_cst))
387		69064x	return true;
388		✗	end_transport_call();
389		✗	return false;
390			}
391
392		72342x	void end_transport_call() noexcept
393			{
394	6/6 ✓ Branch 4 → 5 taken 72204 times. ✓ Branch 4 → 8 taken 138 times. ✓ Branch 6 → 7 taken 3278 times. ✓ Branch 6 → 8 taken 68926 times. ✓ Branch 9 → 10 taken 3278 times. ✓ Branch 9 → 11 taken 69064 times.	216888x	if (calls_.fetch_sub(1, std::memory_order_seq_cst) == 1 &&
395		72204x	orphaned_.load(std::memory_order_seq_cst))
396		3278x	destroy_transport();
397		72342x	}
398
399			// A transport call that also ends when its frame is unwound while
400			// parked, so the count never pins an orphaned transport.
401			class transport_call
402			{
403			engine_driver* d_;
404
405			public:
406		69062x	~transport_call()
407			{
408		69062x	end();
409		69062x	}
410
411		69062x	explicit transport_call(engine_driver& d) noexcept
412	1/2 ✓ Branch 3 → 4 taken 69062 times. ✗ Branch 3 → 5 not taken.	69062x	: d_(d.begin_transport_call() ? &d : nullptr)
413			{
414		69062x	}
415
416			transport_call(transport_call const&) = delete;
417			transport_call& operator=(transport_call const&) = delete;
418
419		69062x	explicit operator bool() const noexcept
420			{
421		69062x	return d_ != nullptr;
422			}
423
424		138123x	void end() noexcept
425			{
426	2/2 ✓ Branch 2 → 3 taken 69062 times. ✓ Branch 2 → 6 taken 69061 times.	138123x	if (d_)
427		69062x	std::exchange(d_, nullptr)->end_transport_call();
428		138123x	}
429			};
430
431			// A prior read/write already reported its full transfer as success;
432			// the trailing flush error it deferred surfaces on the next call
433			// instead. Each entry point takes it exactly once.
434		62701x	std::error_code take_pending_flush_ec() noexcept
435			{
436		62701x	std::error_code ec = pending_flush_ec_;
437		62701x	pending_flush_ec_ = {};
438		62701x	return ec;
439			}
440
441			public:
442			/** Construct a driver over a transport stream.
443
444			The driver starts with one reference, the creating stream's.
445
446			@param s The transport.
447			@param owned The transport `s` refers to, when the stream owns
448			it; the driver destroys it on @ref orphan.
449			@param ctx The TLS context handed to the engine's handshake
450			preparation.
451			*/
452		3278x	engine_driver(
453			capy::any_stream s, tls_owned_transport owned, tls_context ctx)
454		3278x	: owned_(owned)
455		3278x	, s_(std::move(s))
456		6556x	, ctx_(std::move(ctx))
457			{
458	1/1 ✓ Branch 17 → 18 taken 3278 times.	3278x	out_buf_.resize(buffer_size_);
459		3278x	}
460
461		3278x	~engine_driver()
462			{
463		3278x	destroy_transport();
464		3278x	}
465
466			engine_driver(engine_driver const&) = delete;
467			engine_driver& operator=(engine_driver const&) = delete;
468
469			/// Add a reference for an operation in flight.
470		65075x	void add_ref() noexcept
471			{
472		65075x	refs_.fetch_add(1, std::memory_order_relaxed);
473		65075x	}
474
475			/// Drop a reference; return true if it was the last.
476		68353x	bool release_ref() noexcept
477			{
478		136706x	return refs_.fetch_sub(1, std::memory_order_acq_rel) == 1;
479			}
480
481			/** Detach the driver from its destroyed stream.
482
483			Operations still in flight finish as canceled without touching
484			the transport again. An owned transport is destroyed now when
485			no transport call is parked on it; otherwise it is told to
486			cancel, if it can, and destroyed when the last parked call
487			returns.
488			*/
489		3278x	void orphan() noexcept
490			{
491			// Counted as a call itself, so a parked call returning meanwhile
492			// cannot destroy the transport under the cancel.
493		3278x	calls_.fetch_add(1, std::memory_order_seq_cst);
494		3278x	orphaned_.store(true, std::memory_order_seq_cst);
495	2/2 ✓ Branch 5 → 6 taken 4 times. ✓ Branch 5 → 7 taken 3274 times.	3278x	if (owned_.cancel)
496		4x	owned_.cancel(owned_.p);
497		3278x	end_transport_call();
498		3278x	}
499
500			/// Cancel the owned transport's pending operations, if it can.
501		2x	void cancel_transport() noexcept
502			{
503	1/2 ✗ Branch 3 → 4 not taken. ✓ Branch 3 → 5 taken 2 times.	2x	if (!begin_transport_call())
504		✗	return;
505	2/2 ✓ Branch 5 → 6 taken 1 time. ✓ Branch 5 → 7 taken 1 time.	2x	if (owned_.cancel)
506		1x	owned_.cancel(owned_.p);
507		2x	end_transport_call();
508			}
509
510			/// Return true once the stream has been destroyed.
511		103394x	bool orphaned() const noexcept
512			{
513		103394x	return orphaned_.load(std::memory_order_acquire);
514			}
515
516			/// Return the transport.
517		2x	capy::any_stream& stream() noexcept
518			{
519		2x	return s_;
520			}
521
522			/// Return the transport.
523			capy::any_stream const& stream() const noexcept
524			{
525			return s_;
526			}
527
528			/// Return the engine for backend-specific setup.
529			Engine& engine() noexcept
530			{
531			return eng_;
532			}
533
534			/// Return the TLS context this driver was constructed with.
535			tls_context const& context() const noexcept
536			{
537			return ctx_;
538			}
539
540			/// Set the hostname applied to the next client handshake.
541		13x	void set_hostname(std::string_view hostname)
542			{
543		13x	hostname_ = hostname;
544		13x	}
545
546			/// Return the ALPN protocol negotiated by the last handshake.
547		3x	std::string_view alpn_protocol() const noexcept
548			{
549		3x	return alpn_selected_;
550			}
551
552			/// Reset the engine and every per-connection driver state.
553		65x	void reset()
554			{
555		65x	eng_.reset();
556
557		65x	out_len_ = 0;
558
559		65x	alpn_selected_.clear();
560		65x	pending_flush_ec_ = {};
561		65x	used_ = false;
562		65x	}
563
564	1/1 ✓ Branch 6 → 7 taken 31323 times.	31323x	capy::task<capy::results<std::size_t>> do_read_some(
565			capy::detail::mutable_buffer_array<capy::detail::max_iovec_> buffers)
566			{
567			if (auto ec = take_pending_flush_ec())
568			co_return {ec, 0};
569
570			std::error_code ec;
571			std::size_t total_read = 0;
572			std::size_t const bufs_size = capy::buffer_size(buffers);
573
574			for (auto& buf : buffers)
575			{
576			char* const dest = static_cast<char*>(buf.data());
577			// Clamp for the engines' int-sized transfer API: a >=2 GiB
578			// buffer becomes a partial read, never a truncated length.
579			int const remaining = static_cast<int>(
580			(std::min)(buf.size(),
581			std::size_t((std::numeric_limits<int>::max)())));
582			if (remaining == 0)
583			continue;
584
585			// Exits by co_return: success after the first transferred
586			// chunk, or any terminal error.
587			while (true)
588			{
589			auto const gen = read_gen_;
590			auto r = eng_.perform(
591			engine_op::read, dest, static_cast<std::size_t>(remaining));
592
593			if (r.ec)
594			{
595			// Terminal, already mapped by the engine. eof (a
596			// received close_notify) arrives as a plain done:
597			// it queues no output, so no flush is needed.
598			if (r.want == engine_want::output_then_done)
599			co_await best_effort_flush();
600			co_return {r.ec, total_read};
601			}
602
603			if (r.want == engine_want::done \|\|
604			r.want == engine_want::output_then_done)
605			{
606			total_read += r.bytes;
607
608			// r.bytes > 0 already satisfies ReadStream's "at
609			// least one byte transferred" success condition;
610			// report now rather than loop for more (another
611			// engine call could park on input). out_len_ > 0
612			// covers a retained tail the engine cannot see.
613			if (r.want == engine_want::output_then_done \|\| out_len_ > 0)
614			ec = co_await flush_output();
615			if (ec && total_read == bufs_size)
616			{
617			// First failure wins: concurrent directions share
618			// one transport failure domain; the earliest
619			// error is the meaningful one.
620			if (!pending_flush_ec_)
621			pending_flush_ec_ = ec;
622			ec = {};
623			}
624			co_return {ec, total_read};
625			}
626
627			if (r.want == engine_want::output_then_retry)
628			{
629			ec = co_await flush_output();
630			if (ec)
631			co_return {ec, total_read};
632			continue;
633			}
634
635			// want == input. Flush first: a retained tail (an
636			// earlier failed write) may hold bytes the peer needs
637			// before it will send more; no-op when nothing pends.
638			ec = co_await flush_output();
639			if (ec)
640			co_return {ec, total_read};
641
642			ec = co_await fill_input(gen);
643			if (ec)
644			{
645			ec = map_fill_error(
646			engine_op::read, ec, eng_.received_shutdown());
647			co_return {ec, total_read};
648			}
649			}
650			}
651
652			co_return {std::error_code{}, total_read};
653		62646x	}
654
655	1/1 ✓ Branch 6 → 7 taken 31223 times.	31223x	capy::task<capy::results<std::size_t>> do_write_some(
656			capy::detail::const_buffer_array<capy::detail::max_iovec_> buffers)
657			{
658			if (auto ec = take_pending_flush_ec())
659			co_return {ec, 0};
660
661			std::error_code ec;
662			std::size_t total_written = 0;
663			std::size_t const bufs_size = capy::buffer_size(buffers);
664
665			for (auto const& buf : buffers)
666			{
667			// The engine only reads through this pointer for a write
668			// op; the cast satisfies perform's single transfer
669			// signature.
670			void* const src = const_cast<void*>(buf.data());
671			// Clamp for the engines' int-sized transfer API: a >=2 GiB
672			// buffer becomes a partial write, never a truncated length.
673			int const remaining = static_cast<int>(
674			(std::min)(buf.size(),
675			std::size_t((std::numeric_limits<int>::max)())));
676			if (remaining == 0)
677			continue;
678
679			// Exits by co_return: success after the first transferred
680			// chunk, or any terminal error.
681			while (true)
682			{
683			auto const gen = read_gen_;
684			auto r = eng_.perform(
685			engine_op::write, src, static_cast<std::size_t>(remaining));
686
687			if (r.ec)
688			{
689			// Terminal, already mapped. eof means the peer's
690			// close_notify WAS received (an announced close);
691			// it arrives as a plain done and needs no flush.
692			if (r.want == engine_want::output_then_done)
693			co_await best_effort_flush();
694			co_return {r.ec, total_written};
695			}
696
697			if (r.want == engine_want::done \|\|
698			r.want == engine_want::output_then_done)
699			{
700			total_written += r.bytes;
701
702			// r.bytes > 0 already satisfies WriteStream's "at
703			// least one byte transferred" success condition;
704			// report now rather than loop for more. out_len_ > 0
705			// covers a retained tail the engine cannot see.
706			if (r.want == engine_want::output_then_done \|\| out_len_ > 0)
707			ec = co_await flush_output();
708			if (ec && total_written == bufs_size)
709			{
710			// First failure wins: concurrent directions share
711			// one transport failure domain; the earliest
712			// error is the meaningful one.
713			if (!pending_flush_ec_)
714			pending_flush_ec_ = ec;
715			ec = {};
716			}
717			co_return {ec, total_written};
718			}
719
720			if (r.want == engine_want::output_then_retry)
721			{
722			ec = co_await flush_output();
723			if (ec)
724			co_return {ec, total_written};
725			continue;
726			}
727
728			// want == input: a write can need a read mid-rekey; the
729			// transport eof this surfaces means the same thing it
730			// means on the read path, so map it the same way. Flush
731			// first for the same retained-tail reason as the read
732			// path.
733			ec = co_await flush_output();
734			if (ec)
735			co_return {ec, total_written};
736
737			ec = co_await fill_input(gen);
738			if (ec)
739			{
740			ec = map_fill_error(
741			engine_op::write, ec, eng_.received_shutdown());
742			co_return {ec, total_written};
743			}
744			}
745			}
746
747			co_return {std::error_code{}, total_written};
748		62446x	}
749
750	1/1 ✓ Branch 5 → 6 taken 2372 times.	2372x	capy::task<capy::results<>> do_handshake(tls_role role)
751			{
752			// Refuse the handshake while the engine's configuration is
753			// unusable, before consuming any per-connection state.
754			if (auto cec = eng_.check_context())
755			co_return {cec};
756
757			if (used_)
758			reset();
759
760			// reset() may not have restored a usable session; refuse to
761			// hand out a handshake on it.
762			if (auto sec = eng_.check_session())
763			co_return {sec};
764
765			// A failed attempt leaves the session in a dead state; any
766			// attempt, not just a completed handshake, consumes the stream
767			// so the next handshake() starts fresh.
768			used_ = true;
769
770			if (auto pec = eng_.prepare(ctx_, role, hostname_))
771			co_return {pec};
772
773			auto const op = role == tls_role::client ? engine_op::handshake_client
774			: engine_op::handshake_server;
775
776			std::error_code ec;
777
778			while (true)
779			{
780			auto const gen = read_gen_;
781			auto r = eng_.perform(op, nullptr, 0);
782
783			if (r.ec)
784			{
785			if (r.want == engine_want::output_then_done)
786			co_await best_effort_flush();
787			co_return {r.ec};
788			}
789
790			if (r.want == engine_want::done \|\|
791			r.want == engine_want::output_then_done)
792			{
793			eng_.capture_alpn(alpn_selected_);
794			ec = co_await flush_output();
795			co_return {ec};
796			}
797
798			if (r.want == engine_want::output_then_retry)
799			{
800			// Must flush (e.g. ClientHello) before reading the
801			// peer's reply.
802			ec = co_await flush_output();
803			if (ec)
804			co_return {ec};
805			continue;
806			}
807
808			// want == input. The flush is a no-op unless a retained
809			// tail pends; the fill error passes through map_fill_error
810			// unchanged (no close is clean before the session is
811			// established).
812			ec = co_await flush_output();
813			if (ec)
814			co_return {ec};
815
816			ec = co_await fill_input(gen);
817			if (ec)
818			{
819			ec = map_fill_error(op, ec, eng_.received_shutdown());
820			co_return {ec};
821			}
822			}
823		4744x	}
824
825	1/1 ✓ Branch 5 → 6 taken 155 times.	155x	capy::task<capy::results<>> do_shutdown()
826			{
827			if (auto ec = take_pending_flush_ec())
828			co_return {ec};
829
830			std::error_code ec;
831
832			while (true)
833			{
834			auto const gen = read_gen_;
835			auto r = eng_.perform(engine_op::shutdown, nullptr, 0);
836
837			if (r.ec)
838			{
839			if (r.want == engine_want::output_then_done)
840			co_await best_effort_flush();
841			co_return {r.ec};
842			}
843
844			if (r.want == engine_want::done \|\|
845			r.want == engine_want::output_then_done)
846			{
847			// Covers both the bidirectional-complete result and the
848			// engine's shutdown-after-close success mapping.
849			ec = co_await flush_output();
850			co_return {ec};
851			}
852
853			if (r.want == engine_want::output_then_retry)
854			{
855			// Sends our close_notify before parking for the peer's.
856			ec = co_await flush_output();
857			if (ec)
858			co_return {ec};
859			continue;
860			}
861
862			// want == input: awaiting the peer's close_notify. It may
863			// already have been deposited (and consumed by a concurrent
864			// reader) during a flush; fill_input then returns without
865			// reading and the loop retries the engine instead of parking.
866			ec = co_await flush_output();
867			if (ec)
868			co_return {ec};
869
870			ec = co_await fill_input(gen);
871			if (ec)
872			{
873			ec = map_fill_error(
874			engine_op::shutdown, ec, eng_.received_shutdown());
875			co_return {ec};
876			}
877			}
878		310x	}
879			};
880
881			/** A reference to a TLS stream's driver held by an operation.
882
883			Taken when the operation is created, so a stream destroyed or
884			moved before the operation is awaited or finishes leaves its
885			driver alive until the operation lets go.
886
887			@tparam Impl The stream's implementation, derived from
888			@ref engine_driver.
889			*/
890			template<class Impl>
891			class driver_ref
892			{
893			Impl* p_;
894
895			public:
896			/// Construct, adding a reference to @p p.
897		65075x	explicit driver_ref(Impl* p) noexcept : p_(p)
898			{
899		65075x	p_->add_ref();
900		65075x	}
901
902			/// Construct by taking @p other's reference.
903		65075x	driver_ref(driver_ref&& other) noexcept
904		65075x	: p_(std::exchange(other.p_, nullptr))
905			{
906		65075x	}
907
908			driver_ref(driver_ref const&) = delete;
909			driver_ref& operator=(driver_ref const&) = delete;
910			driver_ref& operator=(driver_ref&&) = delete;
911
912			/// Destroy, deleting the driver if this was its last reference.
913		130150x	~driver_ref()
914			{
915	6/6 ✓ Branch 2 → 3 taken 65075 times. ✓ Branch 2 → 6 taken 65075 times. ✓ Branch 4 → 5 taken 6 times. ✓ Branch 4 → 6 taken 65069 times. ✓ Branch 7 → 8 taken 6 times. ✓ Branch 7 → 11 taken 130144 times.	130150x	if (p_ && p_->release_ref())
916	1/2 ✓ Branch 8 → 9 taken 6 times. ✗ Branch 8 → 11 not taken.	6x	delete p_;
917		130150x	}
918
919			/// Return the driver.
920		65073x	Impl* operator->() const noexcept
921			{
922		65073x	return p_;
923			}
924			};
925
926			/// Read through a driver an operation keeps alive.
927			template<class Impl>
928			capy::task<capy::results<std::size_t>>
929	1/1 ✓ Branch 7 → 8 taken 31325 times.	31325x	driver_read_some(
930			driver_ref<Impl> d,
931			capy::detail::mutable_buffer_array<capy::detail::max_iovec_> buffers)
932			{
933			co_return co_await d->do_read_some(buffers);
934		62650x	}
935
936			/// Write through a driver an operation keeps alive.
937			template<class Impl>
938			capy::task<capy::results<std::size_t>>
939	1/1 ✓ Branch 7 → 8 taken 31223 times.	31223x	driver_write_some(
940			driver_ref<Impl> d,
941			capy::detail::const_buffer_array<capy::detail::max_iovec_> buffers)
942			{
943			co_return co_await d->do_write_some(buffers);
944		62446x	}
945
946			/// Handshake through a driver an operation keeps alive.
947			template<class Impl>
948			capy::task<capy::results<>>
949	1/1 ✓ Branch 6 → 7 taken 2372 times.	2372x	driver_handshake(driver_ref<Impl> d, tls_role role)
950			{
951			co_return co_await d->do_handshake(role);
952		4744x	}
953
954			/// Shut down through a driver an operation keeps alive.
955			template<class Impl>
956			capy::task<capy::results<>>
957	1/1 ✓ Branch 6 → 7 taken 155 times.	155x	driver_shutdown(driver_ref<Impl> d)
958			{
959			co_return co_await d->do_shutdown();
960		310x	}
961
962			} // namespace detail
963
964			} // namespace boost::corosio
965
966			#endif
967