include/boost/corosio/native/detail/posix/posix_stream_file_service.hpp

95.5% Lines (213 / 223) 79.7% Branches (59 / 74) 95.5% Functions (21 / 22)

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

posix_stream_file_service.hpp

f(x) Functions (22)

Function Calls Lines Branches Blocks

Line	Branch	Hits	Source Code
1			//
2			// Copyright (c) 2026 Michael Vandeberg
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 BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_SERVICE_HPP
11			#define BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_SERVICE_HPP
12
13			#include <boost/corosio/detail/platform.hpp>
14
15			#if BOOST_COROSIO_POSIX
16
17			#include <boost/corosio/native/detail/posix/posix_stream_file.hpp>
18			#include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp>
19			#include <boost/corosio/detail/file_service.hpp>
20			#include <boost/corosio/detail/object_pool.hpp>
21			#include <boost/corosio/detail/object_ref.hpp>
22			#include <boost/corosio/detail/thread_pool.hpp>
23
24			#include <vector>
25
26			namespace boost::corosio::detail {
27
28			/** Stream file service for POSIX backends.
29
30			Owns all posix_stream_file instances. Thread lifecycle is
31			managed by the thread_pool service (shared with resolver).
32			*/
33			class BOOST_COROSIO_DECL posix_stream_file_service final : public file_service
34			{
35			friend class posix_stream_file;
36
37			public:
38		1178x	explicit posix_stream_file_service(capy::execution_context& ctx)
39	1/2 ✓ Branch 0 taken 589 times. ✗ Branch 1 not taken.	589x	: sched_(&get_scheduler(ctx))
40		589x	, pool_(ctx)
41		1178x	{
42		1178x	}
43
44		1767x	~posix_stream_file_service() override = default;
45
46			posix_stream_file_service(posix_stream_file_service const&) = delete;
47			posix_stream_file_service&
48			operator=(posix_stream_file_service const&) = delete;
49
50		709x	io_object::implementation* construct() override
51			{
52		709x	return object_pool_.acquire(*this);
53			}
54
55		709x	void destroy(io_object::implementation* p) override
56			{
57		709x	auto& impl = static_cast<posix_stream_file&>(*p);
58		709x	impl.cancel();
59		709x	impl.close_file();
60		709x	release(&impl);
61		709x	}
62
63		1780x	void close(io_object::handle& h) override
64			{
65	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 1780 times.	1780x	if (h.get())
66			{
67		1780x	auto& impl = static_cast<posix_stream_file&>(*h.get());
68		1780x	impl.cancel();
69		1780x	impl.close_file();
70		1780x	}
71		1780x	}
72
73		685x	std::error_code open_file(
74			stream_file::implementation& impl,
75			std::filesystem::path const& path,
76			file_base::flags mode) override
77			{
78	2/2 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 683 times.	685x	if (pool_unavailable())
79		2x	return std::make_error_code(std::errc::operation_not_supported);
80		683x	return static_cast<posix_stream_file&>(impl).open_file(path, mode);
81		685x	}
82
83			/** Check whether file I/O is unavailable on this context.
84
85			The file thread pool completes cross-thread, which the
86			lockless scheduler of the unsafe tier cannot accept.
87			*/
88		1095x	bool pool_unavailable() const noexcept
89			{
90		1095x	return sched_->scheduler_locking_disabled();
91			}
92
93		589x	void shutdown() override
94			{
95			// See uring_socket_service_base::shutdown(): snapshot under an
96			// acquired reference, cancel+close without the pool lock held;
97			// shutdown() sets shutting-down and takes the snapshot in one
98			// critical section, so a close that drops the last ref deletes
99			// rather than recycles.
100		589x	std::vector<posix_stream_file*> live;
101	2/4 ✓ Branch 0 taken 589 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 589 times. ✗ Branch 3 not taken.	1178x	object_pool_.shutdown(
102		593x	[&](posix_stream_file* f)
103			{
104		4x	acquire(f);
105		4x	live.push_back(f);
106		4x	});
107	2/2 ✓ Branch 0 taken 4 times. ✓ Branch 1 taken 589 times.	593x	for (auto* f : live)
108			{
109		4x	f->cancel();
110		4x	f->close_file();
111		4x	release(f);
112			}
113		589x	}
114
115		617x	void post(scheduler_op* op)
116			{
117		617x	sched_->post(op);
118		617x	}
119
120		630x	void work_started() noexcept
121			{
122		630x	sched_->work_started();
123		630x	}
124
125		630x	void work_finished() noexcept
126			{
127		630x	sched_->work_finished();
128		630x	}
129
130			/** Return the thread pool that runs this service's file work.
131
132			The pool's service is created on first use, so this can fail
133			where a plain accessor could not. Its workers start later, on
134			the first post, and a thread the system refuses there is
135			reported by that post rather than thrown here.
136
137			@throws std::bad_alloc If the service cannot be allocated.
138
139			@return The context's shared blocking-I/O pool.
140
141			@see thread_pool_ref::get
142			*/
143		630x	thread_pool& pool()
144			{
145		630x	return pool_.get();
146			}
147
148			private:
149			scheduler* sched_;
150			thread_pool_ref pool_;
151			object_pool<posix_stream_file> object_pool_;
152			};
153
154			// ---------------------------------------------------------------------------
155			// posix_stream_file inline implementations (require complete service type)
156			// ---------------------------------------------------------------------------
157
158			inline std::coroutine_handle<>
159		557x	posix_stream_file::read_some(
160			capy::continuation& cont,
161			capy::executor_ref ex,
162			buffer_param param,
163			std::stop_token token,
164			std::error_code* ec,
165			std::size_t* bytes_out)
166			{
167		557x	auto& op = read_op_;
168		557x	op.reset();
169		557x	op.is_read = true;
170
171			// Closed-object contract outranks the zero-length no-op.
172	2/2 ✓ Branch 0 taken 551 times. ✓ Branch 1 taken 6 times.	557x	if (fd_ < 0)
173			{
174		6x	*ec = make_error_code(std::errc::bad_file_descriptor);
175		6x	*bytes_out = 0;
176		6x	return dispatch_coro(ex, cont);
177			}
178
179		551x	capy::mutable_buffer bufs[max_buffers];
180		551x	op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
181
182	2/2 ✓ Branch 0 taken 549 times. ✓ Branch 1 taken 2 times.	551x	if (op.iovec_count == 0)
183			{
184		2x	*ec = {};
185		2x	*bytes_out = 0;
186		2x	return dispatch_coro(ex, cont);
187			}
188
189	2/2 ✓ Branch 0 taken 549 times. ✓ Branch 1 taken 549 times.	1098x	for (int i = 0; i < op.iovec_count; ++i)
190			{
191		549x	op.iovecs[i].iov_base = bufs[i].data();
192		549x	op.iovecs[i].iov_len = bufs[i].size();
193		549x	}
194
195		549x	op.cont = &cont;
196		549x	op.ex = ex;
197		549x	op.ec_out = ec;
198		549x	op.bytes_out = bytes_out;
199		549x	op.start(token);
200
201		549x	op.fd = fd_;
202		549x	op.offset = offset_;
203		549x	op.generation = generation_;
204		549x	op.svc = &svc_;
205
206		549x	svc_.work_started();
207
208		549x	read_pool_op_.file_ = this;
209		549x	read_pool_op_.ref_ = detail::object_ref(this);
210		549x	read_pool_op_.func_ = &posix_stream_file::do_read_work;
211		549x	read_pool_op_.destroy_ = &posix_stream_file::destroy_read_work;
212	2/2 ✓ Branch 0 taken 7 times. ✓ Branch 1 taken 542 times.	549x	if (auto pec = svc_.pool().post(&read_pool_op_))
213			{
214			// The pool is shutting down, or the system refused it a thread.
215			// Nothing of this read went cross-thread, so it answers here
216			// like the closed-descriptor and zero-length exits above rather
217			// than through a completion the scheduler has to carry back.
218		7x	read_pool_op_.ref_.reset();
219		7x	op.stop_cb.reset();
220		7x	svc_.work_finished();
221		7x	*ec = pec;
222		7x	*bytes_out = 0;
223		7x	return dispatch_coro(ex, cont);
224			}
225		542x	return std::noop_coroutine();
226		557x	}
227
228			inline void
229		538x	posix_stream_file::do_read_work(pool_work_item* w) noexcept
230			{
231		538x	auto* pw = static_cast<pool_op*>(w);
232		538x	auto* self = pw->file_;
233		538x	auto& op = self->read_op_;
234
235	2/2 ✓ Branch 0 taken 376 times. ✓ Branch 1 taken 162 times.	538x	if (!op.cancelled.load(std::memory_order_acquire))
236			{
237	2/2 ✓ Branch 0 taken 161 times. ✓ Branch 1 taken 1 time.	162x	if (!self->gate_.enter(op.fd))
238			{
239			// Closed or replaced since this was queued.
240		1x	op.request_cancel();
241		1x	}
242	2/2 ✓ Branch 0 taken 4 times. ✓ Branch 1 taken 157 times.	161x	else if (op.cancelled.load(std::memory_order_acquire))
243			{
244			// Closed and reopened onto the same number since the check
245			// above: the gate admits the number, but not this op.
246		4x	self->gate_.leave();
247		4x	}
248			else
249			{
250			ssize_t n;
251		157x	do
252			{
253		157x	n = file_preadv(
254		157x	op.fd, op.iovecs, op.iovec_count,
255		157x	static_cast<file_off_t>(op.offset));
256		314x	}
257	3/4 ✓ Branch 0 taken 153 times. ✓ Branch 1 taken 4 times. ✓ Branch 2 taken 4 times. ✗ Branch 3 not taken.	157x	while (n < 0 && errno == EINTR);
258
259	2/2 ✓ Branch 0 taken 153 times. ✓ Branch 1 taken 4 times.	157x	if (n >= 0)
260			{
261		153x	op.errn = 0;
262		153x	op.bytes_transferred = static_cast<std::size_t>(n);
263
264			// The file may have been replaced while this ran; its
265			// position belongs to whatever it now holds.
266	1/2 ✓ Branch 0 taken 153 times. ✗ Branch 1 not taken.	153x	std::lock_guard<std::mutex> lock(self->offset_mutex_);
267	2/2 ✓ Branch 0 taken 23 times. ✓ Branch 1 taken 130 times.	153x	if (self->generation_ == op.generation)
268		130x	self->offset_ += static_cast<std::uint64_t>(n);
269		153x	}
270			else
271			{
272	1/2 ✓ Branch 0 taken 4 times. ✗ Branch 1 not taken.	4x	op.errn = errno;
273		4x	op.bytes_transferred = 0;
274			}
275		157x	self->gate_.leave();
276			}
277		162x	}
278
279		538x	op.object_ref_ = std::move(pw->ref_);
280	1/2 ✓ Branch 0 taken 538 times. ✗ Branch 1 not taken.	538x	self->svc_.post(&op);
281		538x	}
282
283			inline std::coroutine_handle<>
284		89x	posix_stream_file::write_some(
285			capy::continuation& cont,
286			capy::executor_ref ex,
287			buffer_param param,
288			std::stop_token token,
289			std::error_code* ec,
290			std::size_t* bytes_out)
291			{
292		89x	auto& op = write_op_;
293		89x	op.reset();
294		89x	op.is_read = false;
295
296			// Closed-object contract outranks the zero-length no-op.
297	2/2 ✓ Branch 0 taken 83 times. ✓ Branch 1 taken 6 times.	89x	if (fd_ < 0)
298			{
299		6x	*ec = make_error_code(std::errc::bad_file_descriptor);
300		6x	*bytes_out = 0;
301		6x	return dispatch_coro(ex, cont);
302			}
303
304		83x	capy::mutable_buffer bufs[max_buffers];
305		83x	op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
306
307	2/2 ✓ Branch 0 taken 81 times. ✓ Branch 1 taken 2 times.	83x	if (op.iovec_count == 0)
308			{
309		2x	*ec = {};
310		2x	*bytes_out = 0;
311		2x	return dispatch_coro(ex, cont);
312			}
313
314	2/2 ✓ Branch 0 taken 81 times. ✓ Branch 1 taken 81 times.	162x	for (int i = 0; i < op.iovec_count; ++i)
315			{
316		81x	op.iovecs[i].iov_base = bufs[i].data();
317		81x	op.iovecs[i].iov_len = bufs[i].size();
318		81x	}
319
320		81x	op.cont = &cont;
321		81x	op.ex = ex;
322		81x	op.ec_out = ec;
323		81x	op.bytes_out = bytes_out;
324		81x	op.start(token);
325
326		81x	op.fd = fd_;
327		81x	op.offset = offset_;
328		81x	op.generation = generation_;
329		81x	op.svc = &svc_;
330
331		81x	svc_.work_started();
332
333		81x	write_pool_op_.file_ = this;
334		81x	write_pool_op_.ref_ = detail::object_ref(this);
335		81x	write_pool_op_.func_ = &posix_stream_file::do_write_work;
336		81x	write_pool_op_.destroy_ = &posix_stream_file::destroy_write_work;
337	2/2 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 79 times.	81x	if (auto pec = svc_.pool().post(&write_pool_op_))
338			{
339			// The pool is shutting down, or the system refused it a thread.
340			// Nothing of this write went cross-thread, so it answers here
341			// like the closed-descriptor and zero-length exits above rather
342			// than through a completion the scheduler has to carry back.
343		2x	write_pool_op_.ref_.reset();
344		2x	op.stop_cb.reset();
345		2x	svc_.work_finished();
346		2x	*ec = pec;
347		2x	*bytes_out = 0;
348		2x	return dispatch_coro(ex, cont);
349			}
350		79x	return std::noop_coroutine();
351		89x	}
352
353			inline void
354		79x	posix_stream_file::do_write_work(pool_work_item* w) noexcept
355			{
356		79x	auto* pw = static_cast<pool_op*>(w);
357		79x	auto* self = pw->file_;
358		79x	auto& op = self->write_op_;
359
360	2/2 ✓ Branch 0 taken 52 times. ✓ Branch 1 taken 27 times.	79x	if (!op.cancelled.load(std::memory_order_acquire))
361			{
362	1/2 ✓ Branch 0 taken 27 times. ✗ Branch 1 not taken.	27x	if (!self->gate_.enter(op.fd))
363			{
364			// Closed or replaced since this was queued.
365		✗	op.request_cancel();
366		✗	}
367	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 27 times.	27x	else if (op.cancelled.load(std::memory_order_acquire))
368			{
369			// Closed and reopened onto the same number since the check
370			// above: the gate admits the number, but not this op.
371		✗	self->gate_.leave();
372		✗	}
373			else
374			{
375			ssize_t n;
376		27x	do
377			{
378		27x	n = file_pwritev(
379		27x	op.fd, op.iovecs, op.iovec_count,
380		27x	static_cast<file_off_t>(op.offset));
381		54x	}
382	3/4 ✓ Branch 0 taken 23 times. ✓ Branch 1 taken 4 times. ✓ Branch 2 taken 4 times. ✗ Branch 3 not taken.	27x	while (n < 0 && errno == EINTR);
383
384	2/2 ✓ Branch 0 taken 23 times. ✓ Branch 1 taken 4 times.	27x	if (n >= 0)
385			{
386		23x	op.errn = 0;
387		23x	op.bytes_transferred = static_cast<std::size_t>(n);
388
389			// The file may have been replaced while this ran; its
390			// position belongs to whatever it now holds.
391	1/2 ✓ Branch 0 taken 23 times. ✗ Branch 1 not taken.	23x	std::lock_guard<std::mutex> lock(self->offset_mutex_);
392	2/2 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 21 times.	23x	if (self->generation_ == op.generation)
393		21x	self->offset_ += static_cast<std::uint64_t>(n);
394		23x	}
395			else
396			{
397	1/2 ✓ Branch 0 taken 4 times. ✗ Branch 1 not taken.	4x	op.errn = errno;
398		4x	op.bytes_transferred = 0;
399			}
400		27x	self->gate_.leave();
401			}
402		27x	}
403
404		79x	op.object_ref_ = std::move(pw->ref_);
405	1/2 ✓ Branch 0 taken 79 times. ✗ Branch 1 not taken.	79x	self->svc_.post(&op);
406		79x	}
407
408			inline void
409		4x	posix_stream_file::destroy_read_work(pool_work_item* w) noexcept
410			{
411		4x	auto* pw = static_cast<pool_op*>(w);
412		4x	auto* self = pw->file_;
413		4x	self->read_op_.object_ref_ = std::move(pw->ref_);
414		4x	self->read_op_.destroy();
415		4x	}
416
417			inline void
418		✗	posix_stream_file::destroy_write_work(pool_work_item* w) noexcept
419			{
420		✗	auto* pw = static_cast<pool_op*>(w);
421		✗	auto* self = pw->file_;
422		✗	self->write_op_.object_ref_ = std::move(pw->ref_);
423		✗	self->write_op_.destroy();
424		✗	}
425
426			inline std::error_code
427		410x	posix_stream_file::assign(native_handle_type handle) noexcept
428			{
429			// The public assign() guarantees the object is closed.
430	2/2 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 408 times.	410x	if (svc_.pool_unavailable())
431		2x	return std::make_error_code(std::errc::operation_not_supported);
432	2/2 ✓ Branch 0 taken 4 times. ✓ Branch 1 taken 404 times.	408x	if (auto ec = validate_file_fd(handle))
433		4x	return ec;
434
435		404x	fd_ = handle;
436		404x	offset_ = 0;
437		404x	gate_.open(handle);
438		404x	return {};
439		410x	}
440
441			inline void
442		709x	posix_stream_file::retire() noexcept
443			{
444	1/2 ✓ Branch 0 taken 709 times. ✗ Branch 1 not taken.	709x	svc_.object_pool_.recycle(this);
445		709x	}
446
447			inline void
448		621x	posix_stream_file::release_work(posix_stream_file_service& svc) noexcept
449			{
450		621x	svc.work_finished();
451		621x	}
452
453			} // namespace boost::corosio::detail
454
455			#endif // BOOST_COROSIO_POSIX
456
457			#endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_SERVICE_HPP
458