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

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

100.0% Lines (176/176) 100.0% List of functions (15/15) 77.6% Branches (45/58)
[ ] prev/next file n p prev/next uncovered line
posix_stream_file_service.hpp
f(x) Functions (15)
Function Calls Lines Branches Blocks
boost::corosio::detail::posix_stream_file_service::posix_stream_file_service(boost::capy::execution_context&, boost::corosio::detail::scheduler&) :35 4582x 100.0% – 100.0% boost::corosio::detail::posix_stream_file_service::~posix_stream_file_service() :41 6873x 100.0% – 100.0% boost::corosio::detail::posix_stream_file_service::construct() :47 125x 100.0% 50.0% 42.0% boost::corosio::detail::posix_stream_file_service::destroy(boost::corosio::io_object::implementation*) :61 123x 100.0% – 100.0% boost::corosio::detail::posix_stream_file_service::close(boost::corosio::io_object::handle&) :69 222x 100.0% 50.0% 100.0% boost::corosio::detail::posix_stream_file_service::open_file(boost::corosio::stream_file::implementation&, std::__1::__fs::filesystem::path const&, boost::corosio::file_base::flags) :79 107x 100.0% 100.0% 100.0% boost::corosio::detail::posix_stream_file_service::shutdown() :91 2291x 100.0% 100.0% 100.0% boost::corosio::detail::posix_stream_file_service::destroy_impl(boost::corosio::detail::posix_stream_file&) :103 123x 100.0% 50.0% 50.0% boost::corosio::detail::posix_stream_file_service::post(boost::corosio::detail::scheduler_op*) :110 63x 100.0% – 100.0% boost::corosio::detail::posix_stream_file_service::pool() :138 72x 100.0% – 100.0% boost::corosio::detail::get_stream_file_service(boost::capy::execution_context&, boost::corosio::detail::scheduler&) :154 2291x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::read_some(std::__1::coroutine_handle<void>, boost::capy::executor_ref, boost::corosio::buffer_param, std::__1::stop_token, std::__1::error_code*, unsigned long*) :164 53x 100.0% 100.0% 93.0% boost::corosio::detail::posix_stream_file::do_read_work(boost::corosio::detail::pool_work_item*) :231 38x 100.0% 68.8% 94.0% boost::corosio::detail::posix_stream_file::write_some(std::__1::coroutine_handle<void>, boost::capy::executor_ref, boost::corosio::buffer_param, std::__1::stop_token, std::__1::error_code*, unsigned long*) :266 35x 100.0% 100.0% 93.0% boost::corosio::detail::posix_stream_file::do_write_work(boost::corosio::detail::pool_work_item*) :333 25x 100.0% 71.4% 93.0%
Line Branch TLA 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/thread_pool.hpp>
21
22 #include <mutex>
23 #include <unordered_map>
24
25 namespace boost::corosio::detail {
26
27 /** Stream file service for POSIX backends.
28
29 Owns all posix_stream_file instances. Thread lifecycle is
30 managed by the thread_pool service (shared with resolver).
31 */
32 class BOOST_COROSIO_DECL posix_stream_file_service final : public file_service
33 {
34 public:
35 6873x posix_stream_file_service(capy::execution_context& ctx, scheduler& sched)
36 2291x : sched_(&sched)
37 2291x , pool_(ctx)
38 4582x {
39 4582x }
40
41 6873x ~posix_stream_file_service() override = default;
42
43 posix_stream_file_service(posix_stream_file_service const&) = delete;
44 posix_stream_file_service&
45 operator=(posix_stream_file_service const&) = delete;
46
47 125x io_object::implementation* construct() override
48 {
49 125x auto ptr = std::make_shared<posix_stream_file>(*this);
50 125x auto* impl = ptr.get();
51
52 {
53
1/2
✓ Branch 0 taken 125 times.
✗ Branch 1 not taken.
 125x std::lock_guard<std::mutex> lock(mutex_);
54 125x file_list_.push_back(impl);
55
1/2
✓ Branch 0 taken 125 times.
✗ Branch 1 not taken.
 125x file_ptrs_[impl] = std::move(ptr);
56 125x }
57
58 125x return impl;
59 125x }
60
61 123x void destroy(io_object::implementation* p) override
62 {
63 123x auto& impl = static_cast<posix_stream_file&>(*p);
64 123x impl.cancel();
65 123x impl.close_file();
66 123x destroy_impl(impl);
67 123x }
68
69 222x void close(io_object::handle& h) override
70 {
71
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 222 times.
 222x if (h.get())
72 {
73 222x auto& impl = static_cast<posix_stream_file&>(*h.get());
74 222x impl.cancel();
75 222x impl.close_file();
76 222x }
77 222x }
78
79 107x std::error_code open_file(
80 stream_file::implementation& impl,
81 std::filesystem::path const& path,
82 file_base::flags mode) override
83 {
84 // Unavailable in the unsafe tier: the file thread pool completes
85 // cross-thread, which the lockless scheduler cannot accept.
86
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 105 times.
 107x if (sched_->scheduler_locking_disabled())
87 2x return std::make_error_code(std::errc::operation_not_supported);
88 105x return static_cast<posix_stream_file&>(impl).open_file(path, mode);
89 107x }
90
91 2291x void shutdown() override
92 {
93 2291x std::lock_guard<std::mutex> lock(mutex_);
94
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 2291 times.
 2293x for (auto* impl = file_list_.pop_front(); impl != nullptr;
95 2x impl = file_list_.pop_front())
96 {
97 2x impl->cancel();
98 2x impl->close_file();
99 2x }
100 2291x file_ptrs_.clear();
101 2291x }
102
103 123x void destroy_impl(posix_stream_file& impl)
104 {
105 123x std::lock_guard<std::mutex> lock(mutex_);
106 123x file_list_.remove(&impl);
107
1/2
✓ Branch 0 taken 123 times.
✗ Branch 1 not taken.
 123x file_ptrs_.erase(&impl);
108 123x }
109
110 63x void post(scheduler_op* op)
111 {
112 63x sched_->post(op);
113 63x }
114
115 void work_started() noexcept
116 {
117 sched_->work_started();
118 }
119
120 void work_finished() noexcept
121 {
122 sched_->work_finished();
123 }
124
125 /** Return the thread pool that runs this service's file work.
126
127 The pool's service is created on first use, so this can fail
128 where a plain accessor could not. Its workers start later, on
129 the first post, and a thread the system refuses there is
130 reported by that post rather than thrown here.
131
132 @throws std::bad_alloc If the service cannot be allocated.
133
134 @return The context's shared blocking-I/O pool.
135
136 @see thread_pool_ref::get
137 */
138 72x thread_pool& pool()
139 {
140 72x return pool_.get();
141 }
142
143 private:
144 scheduler* sched_;
145 thread_pool_ref pool_;
146 std::mutex mutex_;
147 intrusive_list<posix_stream_file> file_list_;
148 std::unordered_map<posix_stream_file*, std::shared_ptr<posix_stream_file>>
149 file_ptrs_;
150 };
151
152 /** Get or create the stream file service for the given context. */
153 inline posix_stream_file_service&
154 2291x get_stream_file_service(capy::execution_context& ctx, scheduler& sched)
155 {
156 2291x return ctx.make_service<posix_stream_file_service>(sched);
157 }
158
159 // ---------------------------------------------------------------------------
160 // posix_stream_file inline implementations (require complete service type)
161 // ---------------------------------------------------------------------------
162
163 inline std::coroutine_handle<>
164 53x posix_stream_file::read_some(
165 std::coroutine_handle<> h,
166 capy::executor_ref ex,
167 buffer_param param,
168 std::stop_token token,
169 std::error_code* ec,
170 std::size_t* bytes_out)
171 {
172 53x auto& op = read_op_;
173 53x op.reset();
174 53x op.is_read = true;
175
176 // Closed-object contract outranks the zero-length no-op.
177
2/2
✓ Branch 0 taken 47 times.
✓ Branch 1 taken 6 times.
 53x if (fd_ < 0)
178 {
179 6x *ec = make_error_code(std::errc::bad_file_descriptor);
180 6x *bytes_out = 0;
181 6x op.cont.h = h;
182 6x return dispatch_coro(ex, op.cont);
183 }
184
185 47x capy::mutable_buffer bufs[max_buffers];
186 47x op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
187
188
2/2
✓ Branch 0 taken 45 times.
✓ Branch 1 taken 2 times.
 47x if (op.iovec_count == 0)
189 {
190 2x *ec = {};
191 2x *bytes_out = 0;
192 2x op.cont.h = h;
193 2x return dispatch_coro(ex, op.cont);
194 }
195
196
2/2
✓ Branch 0 taken 45 times.
✓ Branch 1 taken 45 times.
 90x for (int i = 0; i < op.iovec_count; ++i)
197 {
198 45x op.iovecs[i].iov_base = bufs[i].data();
199 45x op.iovecs[i].iov_len = bufs[i].size();
200 45x }
201
202 45x op.h = h;
203 45x op.ex = ex;
204 45x op.ec_out = ec;
205 45x op.bytes_out = bytes_out;
206 45x op.start(token);
207
208 45x op.ex.on_work_started();
209
210 45x read_pool_op_.file_ = this;
211 45x read_pool_op_.ref_ = this->shared_from_this();
212 45x read_pool_op_.func_ = &posix_stream_file::do_read_work;
213
2/2
✓ Branch 0 taken 7 times.
✓ Branch 1 taken 38 times.
 45x if (auto pec = svc_.pool().post(&read_pool_op_))
214 {
215 // The pool is shutting down, or the system refused it a thread.
216 // Nothing of this read went cross-thread, so it answers here
217 // like the closed-descriptor and zero-length exits above rather
218 // than through a completion the scheduler has to carry back.
219 7x read_pool_op_.ref_.reset();
220 7x op.stop_cb.reset();
221 7x op.ex.on_work_finished();
222 7x *ec = pec;
223 7x *bytes_out = 0;
224 7x op.cont.h = h;
225 7x return dispatch_coro(ex, op.cont);
226 }
227 38x return std::noop_coroutine();
228 53x }
229
230 inline void
231 38x posix_stream_file::do_read_work(pool_work_item* w) noexcept
232 {
233 38x auto* pw = static_cast<pool_op*>(w);
234 38x auto* self = pw->file_;
235 38x auto& op = self->read_op_;
236
237
2/2
✓ Branch 0 taken 6 times.
✓ Branch 1 taken 32 times.
 38x if (!op.cancelled.load(std::memory_order_acquire))
238 {
239 ssize_t n;
240 32x do
241 {
242
1/2
✓ Branch 0 taken 32 times.
✗ Branch 1 not taken.
 32x n = ::preadv(
243 32x self->fd_, op.iovecs, op.iovec_count,
244 32x static_cast<off_t>(self->offset_));
245 64x }
246
3/4
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 28 times.
✓ Branch 2 taken 4 times.
✗ Branch 3 not taken.
 32x while (n < 0 && errno == EINTR);
247
248
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 28 times.
 32x if (n >= 0)
249 {
250 28x op.errn = 0;
251 28x op.bytes_transferred = static_cast<std::size_t>(n);
252 28x self->offset_ += static_cast<std::uint64_t>(n);
253 28x }
254 else
255 {
256
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4x op.errn = errno;
257 4x op.bytes_transferred = 0;
258 }
259 32x }
260
261
1/2
✓ Branch 0 taken 38 times.
✗ Branch 1 not taken.
 38x op.impl_ptr = std::move(pw->ref_);
262
1/2
✓ Branch 0 taken 38 times.
✗ Branch 1 not taken.
 38x self->svc_.post(&op);
263 38x }
264
265 inline std::coroutine_handle<>
266 35x posix_stream_file::write_some(
267 std::coroutine_handle<> h,
268 capy::executor_ref ex,
269 buffer_param param,
270 std::stop_token token,
271 std::error_code* ec,
272 std::size_t* bytes_out)
273 {
274 35x auto& op = write_op_;
275 35x op.reset();
276 35x op.is_read = false;
277
278 // Closed-object contract outranks the zero-length no-op.
279
2/2
✓ Branch 0 taken 29 times.
✓ Branch 1 taken 6 times.
 35x if (fd_ < 0)
280 {
281 6x *ec = make_error_code(std::errc::bad_file_descriptor);
282 6x *bytes_out = 0;
283 6x op.cont.h = h;
284 6x return dispatch_coro(ex, op.cont);
285 }
286
287 29x capy::mutable_buffer bufs[max_buffers];
288 29x op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
289
290
2/2
✓ Branch 0 taken 27 times.
✓ Branch 1 taken 2 times.
 29x if (op.iovec_count == 0)
291 {
292 2x *ec = {};
293 2x *bytes_out = 0;
294 2x op.cont.h = h;
295 2x return dispatch_coro(ex, op.cont);
296 }
297
298
2/2
✓ Branch 0 taken 27 times.
✓ Branch 1 taken 27 times.
 54x for (int i = 0; i < op.iovec_count; ++i)
299 {
300 27x op.iovecs[i].iov_base = bufs[i].data();
301 27x op.iovecs[i].iov_len = bufs[i].size();
302 27x }
303
304 27x op.h = h;
305 27x op.ex = ex;
306 27x op.ec_out = ec;
307 27x op.bytes_out = bytes_out;
308 27x op.start(token);
309
310 27x op.ex.on_work_started();
311
312 27x write_pool_op_.file_ = this;
313 27x write_pool_op_.ref_ = this->shared_from_this();
314 27x write_pool_op_.func_ = &posix_stream_file::do_write_work;
315
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 25 times.
 27x if (auto pec = svc_.pool().post(&write_pool_op_))
316 {
317 // The pool is shutting down, or the system refused it a thread.
318 // Nothing of this write went cross-thread, so it answers here
319 // like the closed-descriptor and zero-length exits above rather
320 // than through a completion the scheduler has to carry back.
321 2x write_pool_op_.ref_.reset();
322 2x op.stop_cb.reset();
323 2x op.ex.on_work_finished();
324 2x *ec = pec;
325 2x *bytes_out = 0;
326 2x op.cont.h = h;
327 2x return dispatch_coro(ex, op.cont);
328 }
329 25x return std::noop_coroutine();
330 35x }
331
332 inline void
333 25x posix_stream_file::do_write_work(pool_work_item* w) noexcept
334 {
335 25x auto* pw = static_cast<pool_op*>(w);
336 25x auto* self = pw->file_;
337 25x auto& op = self->write_op_;
338
339
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 23 times.
 25x if (!op.cancelled.load(std::memory_order_acquire))
340 {
341 ssize_t n;
342 23x do
343 {
344
1/2
✓ Branch 0 taken 23 times.
✗ Branch 1 not taken.
 23x n = ::pwritev(
345 23x self->fd_, op.iovecs, op.iovec_count,
346 23x static_cast<off_t>(self->offset_));
347 46x }
348
3/4
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 19 times.
✓ Branch 2 taken 4 times.
✗ Branch 3 not taken.
 23x while (n < 0 && errno == EINTR);
349
350
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 19 times.
 23x if (n >= 0)
351 {
352 19x op.errn = 0;
353 19x op.bytes_transferred = static_cast<std::size_t>(n);
354 19x self->offset_ += static_cast<std::uint64_t>(n);
355 19x }
356 else
357 {
358
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4x op.errn = errno;
359 4x op.bytes_transferred = 0;
360 }
361 23x }
362
363 25x op.impl_ptr = std::move(pw->ref_);
364
1/2
✓ Branch 0 taken 25 times.
✗ Branch 1 not taken.
 25x self->svc_.post(&op);
365 25x }
366
367 } // namespace boost::corosio::detail
368
369 #endif // BOOST_COROSIO_POSIX
370
371 #endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_SERVICE_HPP
372