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

95.1% Lines (195 / 205) 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 Blocks

Line	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	639x	explicit posix_stream_file_service(capy::execution_context& ctx)
39	1917x	: sched_(&get_scheduler(ctx))
40	639x	, pool_(ctx)
41		{
42	639x	}
43
44	1278x	~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	759x	io_object::implementation* construct() override
51		{
52	759x	return object_pool_.acquire(*this);
53		}
54
55	759x	void destroy(io_object::implementation* p) override
56		{
57	759x	auto& impl = static_cast<posix_stream_file&>(*p);
58	759x	impl.cancel();
59	759x	impl.close_file();
60	759x	release(&impl);
61	759x	}
62
63	1882x	void close(io_object::handle& h) override
64		{
65	1882x	if (h.get())
66		{
67	1882x	auto& impl = static_cast<posix_stream_file&>(*h.get());
68	1882x	impl.cancel();
69	1882x	impl.close_file();
70		}
71	1882x	}
72
73	738x	std::error_code open_file(
74		stream_file::implementation& impl,
75		std::filesystem::path const& path,
76		file_base::flags mode) override
77		{
78	738x	if (pool_unavailable())
79	2x	return std::make_error_code(std::errc::operation_not_supported);
80	736x	return static_cast<posix_stream_file&>(impl).open_file(path, mode);
81		}
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	1151x	bool pool_unavailable() const noexcept
89		{
90	1151x	return sched_->scheduler_locking_disabled();
91		}
92
93	639x	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	639x	std::vector<posix_stream_file*> live;
101	639x	object_pool_.shutdown(
102	639x	[&](posix_stream_file* f)
103		{
104	4x	acquire(f);
105	4x	live.push_back(f);
106	4x	});
107	643x	for (auto* f : live)
108		{
109	4x	f->cancel();
110	4x	f->close_file();
111	4x	release(f);
112		}
113	639x	}
114
115	659x	void post(scheduler_op* op)
116		{
117	659x	sched_->post(op);
118	659x	}
119
120	671x	void work_started() noexcept
121		{
122	671x	sched_->work_started();
123	671x	}
124
125	671x	void work_finished() noexcept
126		{
127	671x	sched_->work_finished();
128	671x	}
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	671x	thread_pool& pool()
144		{
145	671x	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	589x	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	589x	auto& op = read_op_;
168	589x	op.reset();
169	589x	op.is_read = true;
170
171		// Closed-object contract outranks the zero-length no-op.
172	589x	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	583x	capy::mutable_buffer bufs[max_buffers];
180	583x	op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
181
182	583x	if (op.iovec_count == 0)
183		{
184	2x	*ec = {};
185	2x	*bytes_out = 0;
186	2x	return dispatch_coro(ex, cont);
187		}
188
189	1162x	for (int i = 0; i < op.iovec_count; ++i)
190		{
191	581x	op.iovecs[i].iov_base = bufs[i].data();
192	581x	op.iovecs[i].iov_len = bufs[i].size();
193		}
194
195	581x	op.cont = &cont;
196	581x	op.ex = ex;
197	581x	op.ec_out = ec;
198	581x	op.bytes_out = bytes_out;
199	581x	op.start(token);
200
201	581x	op.fd = fd_;
202	581x	op.offset = offset_;
203	581x	op.generation = generation_;
204	581x	op.svc = &svc_;
205
206	581x	svc_.work_started();
207
208	581x	read_pool_op_.file_ = this;
209	581x	read_pool_op_.ref_ = detail::object_ref(this);
210	581x	read_pool_op_.func_ = &posix_stream_file::do_read_work;
211	581x	read_pool_op_.destroy_ = &posix_stream_file::destroy_read_work;
212	581x	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	6x	read_pool_op_.ref_.reset();
219	6x	op.stop_cb.reset();
220	6x	svc_.work_finished();
221	6x	*ec = pec;
222	6x	*bytes_out = 0;
223	6x	return dispatch_coro(ex, cont);
224		}
225	575x	return std::noop_coroutine();
226		}
227
228		inline void
229	571x	posix_stream_file::do_read_work(pool_work_item* w) noexcept
230		{
231	571x	auto* pw = static_cast<pool_op*>(w);
232	571x	auto* self = pw->file_;
233	571x	auto& op = self->read_op_;
234
235	571x	if (!op.cancelled.load(std::memory_order_acquire))
236		{
237	135x	if (!self->gate_.enter(op.fd))
238		{
239		// Closed or replaced since this was queued.
240	✗	op.request_cancel();
241		}
242	135x	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	✗	self->gate_.leave();
247		}
248		else
249		{
250		ssize_t n;
251		do
252		{
253	135x	n = file_preadv(
254	135x	op.fd, op.iovecs, op.iovec_count,
255	135x	static_cast<file_off_t>(op.offset));
256		}
257	135x	while (n < 0 && errno == EINTR);
258
259	135x	if (n >= 0)
260		{
261	128x	op.errn = 0;
262	128x	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	128x	std::lock_guard<std::mutex> lock(self->offset_mutex_);
267	128x	if (self->generation_ == op.generation)
268	122x	self->offset_ += static_cast<std::uint64_t>(n);
269	128x	}
270		else
271		{
272	7x	op.errn = errno;
273	7x	op.bytes_transferred = 0;
274		}
275	135x	self->gate_.leave();
276		}
277		}
278
279	571x	op.object_ref_ = std::move(pw->ref_);
280	571x	self->svc_.post(&op);
281	571x	}
282
283		inline std::coroutine_handle<>
284	98x	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	98x	auto& op = write_op_;
293	98x	op.reset();
294	98x	op.is_read = false;
295
296		// Closed-object contract outranks the zero-length no-op.
297	98x	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	92x	capy::mutable_buffer bufs[max_buffers];
305	92x	op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
306
307	92x	if (op.iovec_count == 0)
308		{
309	2x	*ec = {};
310	2x	*bytes_out = 0;
311	2x	return dispatch_coro(ex, cont);
312		}
313
314	180x	for (int i = 0; i < op.iovec_count; ++i)
315		{
316	90x	op.iovecs[i].iov_base = bufs[i].data();
317	90x	op.iovecs[i].iov_len = bufs[i].size();
318		}
319
320	90x	op.cont = &cont;
321	90x	op.ex = ex;
322	90x	op.ec_out = ec;
323	90x	op.bytes_out = bytes_out;
324	90x	op.start(token);
325
326	90x	op.fd = fd_;
327	90x	op.offset = offset_;
328	90x	op.generation = generation_;
329	90x	op.svc = &svc_;
330
331	90x	svc_.work_started();
332
333	90x	write_pool_op_.file_ = this;
334	90x	write_pool_op_.ref_ = detail::object_ref(this);
335	90x	write_pool_op_.func_ = &posix_stream_file::do_write_work;
336	90x	write_pool_op_.destroy_ = &posix_stream_file::destroy_write_work;
337	90x	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	88x	return std::noop_coroutine();
351		}
352
353		inline void
354	88x	posix_stream_file::do_write_work(pool_work_item* w) noexcept
355		{
356	88x	auto* pw = static_cast<pool_op*>(w);
357	88x	auto* self = pw->file_;
358	88x	auto& op = self->write_op_;
359
360	88x	if (!op.cancelled.load(std::memory_order_acquire))
361		{
362	37x	if (!self->gate_.enter(op.fd))
363		{
364		// Closed or replaced since this was queued.
365	✗	op.request_cancel();
366		}
367	37x	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		do
377		{
378	37x	n = file_pwritev(
379	37x	op.fd, op.iovecs, op.iovec_count,
380	37x	static_cast<file_off_t>(op.offset));
381		}
382	37x	while (n < 0 && errno == EINTR);
383
384	37x	if (n >= 0)
385		{
386	30x	op.errn = 0;
387	30x	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	30x	std::lock_guard<std::mutex> lock(self->offset_mutex_);
392	30x	if (self->generation_ == op.generation)
393	25x	self->offset_ += static_cast<std::uint64_t>(n);
394	30x	}
395		else
396		{
397	7x	op.errn = errno;
398	7x	op.bytes_transferred = 0;
399		}
400	37x	self->gate_.leave();
401		}
402		}
403
404	88x	op.object_ref_ = std::move(pw->ref_);
405	88x	self->svc_.post(&op);
406	88x	}
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	413x	posix_stream_file::assign(native_handle_type handle) noexcept
428		{
429		// The public assign() guarantees the object is closed.
430	413x	if (svc_.pool_unavailable())
431	2x	return std::make_error_code(std::errc::operation_not_supported);
432	411x	if (auto ec = validate_file_fd(handle))
433	4x	return ec;
434
435	407x	fd_ = handle;
436	407x	offset_ = 0;
437	407x	gate_.open(handle);
438	407x	return {};
439		}
440
441		inline void
442	759x	posix_stream_file::retire() noexcept
443		{
444	759x	svc_.object_pool_.recycle(this);
445	759x	}
446
447		inline void
448	663x	posix_stream_file::release_work(posix_stream_file_service& svc) noexcept
449		{
450	663x	svc.work_finished();
451	663x	}
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