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

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

92.5% Lines (161/174) 100.0% List of functions (16/16) 68.2% Branches (45/66)
[ ] prev/next file n p prev/next uncovered line
posix_stream_file_service.hpp
f(x) Functions (16)
Function Calls Lines Branches Blocks
boost::corosio::detail::posix_stream_file_service::posix_stream_file_service(boost::capy::execution_context&, boost::corosio::detail::scheduler&) :36 3888x 100.0% 50.0% 100.0% boost::corosio::detail::posix_stream_file_service::~posix_stream_file_service() :43 5832x 100.0% – 100.0% boost::corosio::detail::posix_stream_file_service::construct() :48 95x 100.0% 50.0% 42.0% boost::corosio::detail::posix_stream_file_service::destroy(boost::corosio::io_object::implementation*) :62 95x 100.0% – 100.0% boost::corosio::detail::posix_stream_file_service::close(boost::corosio::io_object::handle&) :70 168x 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) :80 75x 100.0% 100.0% 100.0% boost::corosio::detail::posix_stream_file_service::shutdown() :92 1944x 55.6% 50.0% 60.0% boost::corosio::detail::posix_stream_file_service::destroy_impl(boost::corosio::detail::posix_stream_file&) :104 95x 100.0% 50.0% 50.0% boost::corosio::detail::posix_stream_file_service::post(boost::corosio::detail::scheduler_op*) :111 39x 100.0% – 100.0% boost::corosio::detail::posix_stream_file_service::pool() :126 39x 100.0% – 100.0% boost::corosio::detail::posix_stream_file_service::get_or_create_pool(boost::capy::execution_context&) :132 1944x 83.3% 50.0% 75.0% boost::corosio::detail::get_stream_file_service(boost::capy::execution_context&, boost::corosio::detail::scheduler&) :150 1944x 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*) :160 28x 88.9% 87.5% 86.0% boost::corosio::detail::posix_stream_file::do_read_work(boost::corosio::detail::pool_work_item*) :219 20x 100.0% 66.7% 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*) :253 27x 88.9% 87.5% 86.0% boost::corosio::detail::posix_stream_file::do_write_work(boost::corosio::detail::pool_work_item*) :312 19x 100.0% 62.5% 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
33 : public file_service
34 {
35 public:
36 3888x posix_stream_file_service(
37 capy::execution_context& ctx, scheduler& sched)
38 1944x : sched_(&sched)
39
1/2
✓ Branch 0 taken 1944 times.
✗ Branch 1 not taken.
 1944x , pool_(get_or_create_pool(ctx))
40 3888x {
41 3888x }
42
43 5832x ~posix_stream_file_service() override = default;
44
45 posix_stream_file_service(posix_stream_file_service const&) = delete;
46 posix_stream_file_service& operator=(posix_stream_file_service const&) = delete;
47
48 95x io_object::implementation* construct() override
49 {
50 95x auto ptr = std::make_shared<posix_stream_file>(*this);
51 95x auto* impl = ptr.get();
52
53 {
54
1/2
✓ Branch 0 taken 95 times.
✗ Branch 1 not taken.
 95x std::lock_guard<std::mutex> lock(mutex_);
55 95x file_list_.push_back(impl);
56
1/2
✓ Branch 0 taken 95 times.
✗ Branch 1 not taken.
 95x file_ptrs_[impl] = std::move(ptr);
57 95x }
58
59 95x return impl;
60 95x }
61
62 95x void destroy(io_object::implementation* p) override
63 {
64 95x auto& impl = static_cast<posix_stream_file&>(*p);
65 95x impl.cancel();
66 95x impl.close_file();
67 95x destroy_impl(impl);
68 95x }
69
70 168x void close(io_object::handle& h) override
71 {
72
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 168 times.
 168x if (h.get())
73 {
74 168x auto& impl = static_cast<posix_stream_file&>(*h.get());
75 168x impl.cancel();
76 168x impl.close_file();
77 168x }
78 168x }
79
80 75x std::error_code open_file(
81 stream_file::implementation& impl,
82 std::filesystem::path const& path,
83 file_base::flags mode) override
84 {
85 // Unavailable in the unsafe tier: the file thread pool completes
86 // cross-thread, which the lockless scheduler cannot accept.
87
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 73 times.
 75x if (sched_->scheduler_locking_disabled())
88 2x return std::make_error_code(std::errc::operation_not_supported);
89 73x return static_cast<posix_stream_file&>(impl).open_file(path, mode);
90 75x }
91
92 1944x void shutdown() override
93 {
94 1944x std::lock_guard<std::mutex> lock(mutex_);
95
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1944 times.
 1944x for (auto* impl = file_list_.pop_front(); impl != nullptr;
96 ✗ impl = file_list_.pop_front())
97 {
98 ✗ impl->cancel();
99 ✗ impl->close_file();
100 ✗ }
101 1944x file_ptrs_.clear();
102 1944x }
103
104 95x void destroy_impl(posix_stream_file& impl)
105 {
106 95x std::lock_guard<std::mutex> lock(mutex_);
107 95x file_list_.remove(&impl);
108
1/2
✓ Branch 0 taken 95 times.
✗ Branch 1 not taken.
 95x file_ptrs_.erase(&impl);
109 95x }
110
111 39x void post(scheduler_op* op)
112 {
113 39x sched_->post(op);
114 39x }
115
116 void work_started() noexcept
117 {
118 sched_->work_started();
119 }
120
121 void work_finished() noexcept
122 {
123 sched_->work_finished();
124 }
125
126 39x thread_pool& pool() noexcept
127 {
128 39x return pool_;
129 }
130
131 private:
132 1944x static thread_pool& get_or_create_pool(capy::execution_context& ctx)
133 {
134 1944x auto* p = ctx.find_service<thread_pool>();
135
1/2
✓ Branch 0 taken 1944 times.
✗ Branch 1 not taken.
 1944x if (p)
136 1944x return *p;
137 ✗ return ctx.make_service<thread_pool>();
138 1944x }
139
140 scheduler* sched_;
141 thread_pool& pool_;
142 std::mutex mutex_;
143 intrusive_list<posix_stream_file> file_list_;
144 std::unordered_map<posix_stream_file*, std::shared_ptr<posix_stream_file>>
145 file_ptrs_;
146 };
147
148 /** Get or create the stream file service for the given context. */
149 inline posix_stream_file_service&
150 1944x get_stream_file_service(capy::execution_context& ctx, scheduler& sched)
151 {
152 1944x return ctx.make_service<posix_stream_file_service>(sched);
153 }
154
155 // ---------------------------------------------------------------------------
156 // posix_stream_file inline implementations (require complete service type)
157 // ---------------------------------------------------------------------------
158
159 inline std::coroutine_handle<>
160 28x posix_stream_file::read_some(
161 std::coroutine_handle<> h,
162 capy::executor_ref ex,
163 buffer_param param,
164 std::stop_token token,
165 std::error_code* ec,
166 std::size_t* bytes_out)
167 {
168 28x auto& op = read_op_;
169 28x op.reset();
170 28x op.is_read = true;
171
172 // Closed-object contract outranks the zero-length no-op.
173
2/2
✓ Branch 0 taken 22 times.
✓ Branch 1 taken 6 times.
 28x if (fd_ < 0)
174 {
175 6x *ec = make_error_code(std::errc::bad_file_descriptor);
176 6x *bytes_out = 0;
177 6x op.cont.h = h;
178 6x return dispatch_coro(ex, op.cont);
179 }
180
181 22x capy::mutable_buffer bufs[max_buffers];
182 22x op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
183
184
2/2
✓ Branch 0 taken 20 times.
✓ Branch 1 taken 2 times.
 22x if (op.iovec_count == 0)
185 {
186 2x *ec = {};
187 2x *bytes_out = 0;
188 2x op.cont.h = h;
189 2x return dispatch_coro(ex, op.cont);
190 }
191
192
2/2
✓ Branch 0 taken 20 times.
✓ Branch 1 taken 20 times.
 40x for (int i = 0; i < op.iovec_count; ++i)
193 {
194 20x op.iovecs[i].iov_base = bufs[i].data();
195 20x op.iovecs[i].iov_len = bufs[i].size();
196 20x }
197
198 20x op.h = h;
199 20x op.ex = ex;
200 20x op.ec_out = ec;
201 20x op.bytes_out = bytes_out;
202 20x op.start(token);
203
204 20x op.ex.on_work_started();
205
206 20x read_pool_op_.file_ = this;
207 20x read_pool_op_.ref_ = this->shared_from_this();
208 20x read_pool_op_.func_ = &posix_stream_file::do_read_work;
209
1/2
✓ Branch 0 taken 20 times.
✗ Branch 1 not taken.
 20x if (!svc_.pool().post(&read_pool_op_))
210 {
211 ✗ op.impl_ref = std::move(read_pool_op_.ref_);
212 ✗ op.cancelled.store(true, std::memory_order_release);
213 ✗ svc_.post(&read_op_);
214 ✗ }
215 20x return std::noop_coroutine();
216 28x }
217
218 inline void
219 20x posix_stream_file::do_read_work(pool_work_item* w) noexcept
220 {
221 20x auto* pw = static_cast<pool_op*>(w);
222 20x auto* self = pw->file_;
223 20x auto& op = self->read_op_;
224
225
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 18 times.
 20x if (!op.cancelled.load(std::memory_order_acquire))
226 {
227 ssize_t n;
228 18x do
229 {
230
2/4
✓ Branch 0 taken 18 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 18 times.
✗ Branch 3 not taken.
 36x n = ::preadv(self->fd_, op.iovecs, op.iovec_count,
231 18x static_cast<off_t>(self->offset_));
232 36x }
233
3/4
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 16 times.
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
 18x while (n < 0 && errno == EINTR);
234
235
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 16 times.
 18x if (n >= 0)
236 {
237 16x op.errn = 0;
238 16x op.bytes_transferred = static_cast<std::size_t>(n);
239 16x self->offset_ += static_cast<std::uint64_t>(n);
240 16x }
241 else
242 {
243
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2x op.errn = errno;
244 2x op.bytes_transferred = 0;
245 }
246 18x }
247
248
1/2
✓ Branch 0 taken 20 times.
✗ Branch 1 not taken.
 20x op.impl_ref = std::move(pw->ref_);
249
1/2
✓ Branch 0 taken 20 times.
✗ Branch 1 not taken.
 20x self->svc_.post(&op);
250 20x }
251
252 inline std::coroutine_handle<>
253 27x posix_stream_file::write_some(
254 std::coroutine_handle<> h,
255 capy::executor_ref ex,
256 buffer_param param,
257 std::stop_token token,
258 std::error_code* ec,
259 std::size_t* bytes_out)
260 {
261 27x auto& op = write_op_;
262 27x op.reset();
263 27x op.is_read = false;
264
265 // Closed-object contract outranks the zero-length no-op.
266
2/2
✓ Branch 0 taken 21 times.
✓ Branch 1 taken 6 times.
 27x if (fd_ < 0)
267 {
268 6x *ec = make_error_code(std::errc::bad_file_descriptor);
269 6x *bytes_out = 0;
270 6x op.cont.h = h;
271 6x return dispatch_coro(ex, op.cont);
272 }
273
274 21x capy::mutable_buffer bufs[max_buffers];
275 21x op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
276
277
2/2
✓ Branch 0 taken 19 times.
✓ Branch 1 taken 2 times.
 21x if (op.iovec_count == 0)
278 {
279 2x *ec = {};
280 2x *bytes_out = 0;
281 2x op.cont.h = h;
282 2x return dispatch_coro(ex, op.cont);
283 }
284
285
2/2
✓ Branch 0 taken 19 times.
✓ Branch 1 taken 19 times.
 38x for (int i = 0; i < op.iovec_count; ++i)
286 {
287 19x op.iovecs[i].iov_base = bufs[i].data();
288 19x op.iovecs[i].iov_len = bufs[i].size();
289 19x }
290
291 19x op.h = h;
292 19x op.ex = ex;
293 19x op.ec_out = ec;
294 19x op.bytes_out = bytes_out;
295 19x op.start(token);
296
297 19x op.ex.on_work_started();
298
299 19x write_pool_op_.file_ = this;
300 19x write_pool_op_.ref_ = this->shared_from_this();
301 19x write_pool_op_.func_ = &posix_stream_file::do_write_work;
302
1/2
✓ Branch 0 taken 19 times.
✗ Branch 1 not taken.
 19x if (!svc_.pool().post(&write_pool_op_))
303 {
304 ✗ op.impl_ref = std::move(write_pool_op_.ref_);
305 ✗ op.cancelled.store(true, std::memory_order_release);
306 ✗ svc_.post(&write_op_);
307 ✗ }
308 19x return std::noop_coroutine();
309 27x }
310
311 inline void
312 19x posix_stream_file::do_write_work(pool_work_item* w) noexcept
313 {
314 19x auto* pw = static_cast<pool_op*>(w);
315 19x auto* self = pw->file_;
316 19x auto& op = self->write_op_;
317
318
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 19 times.
 19x if (!op.cancelled.load(std::memory_order_acquire))
319 {
320 ssize_t n;
321 19x do
322 {
323
2/4
✓ Branch 0 taken 19 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 19 times.
✗ Branch 3 not taken.
 38x n = ::pwritev(self->fd_, op.iovecs, op.iovec_count,
324 19x static_cast<off_t>(self->offset_));
325 38x }
326
3/4
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 17 times.
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
 19x while (n < 0 && errno == EINTR);
327
328
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 17 times.
 19x if (n >= 0)
329 {
330 17x op.errn = 0;
331 17x op.bytes_transferred = static_cast<std::size_t>(n);
332 17x self->offset_ += static_cast<std::uint64_t>(n);
333 17x }
334 else
335 {
336
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2x op.errn = errno;
337 2x op.bytes_transferred = 0;
338 }
339 19x }
340
341 19x op.impl_ref = std::move(pw->ref_);
342
1/2
✓ Branch 0 taken 19 times.
✗ Branch 1 not taken.
 19x self->svc_.post(&op);
343 19x }
344
345 } // namespace boost::corosio::detail
346
347 #endif // BOOST_COROSIO_POSIX
348
349 #endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_SERVICE_HPP
350