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

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

89.6% Lines (147/164) 100.0% List of functions (16/16) 53.2% Branches (33/62)
[ ] 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 2662x 100.0% 50.0% 100.0% boost::corosio::detail::posix_stream_file_service::~posix_stream_file_service() :43 3993x 100.0% – 100.0% boost::corosio::detail::posix_stream_file_service::construct() :48 78x 100.0% 50.0% 42.0% boost::corosio::detail::posix_stream_file_service::destroy(boost::corosio::io_object::implementation*) :62 78x 100.0% – 100.0% boost::corosio::detail::posix_stream_file_service::close(boost::corosio::io_object::handle&) :70 134x 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 62x 100.0% 100.0% 100.0% boost::corosio::detail::posix_stream_file_service::shutdown() :92 1331x 55.6% 50.0% 60.0% boost::corosio::detail::posix_stream_file_service::destroy_impl(boost::corosio::detail::posix_stream_file&) :104 78x 100.0% 50.0% 50.0% boost::corosio::detail::posix_stream_file_service::post(boost::corosio::detail::scheduler_op*) :111 32x 100.0% – 100.0% boost::corosio::detail::posix_stream_file_service::pool() :126 32x 100.0% – 100.0% boost::corosio::detail::posix_stream_file_service::get_or_create_pool(boost::capy::execution_context&) :132 1331x 83.3% 50.0% 75.0% boost::corosio::detail::get_stream_file_service(boost::capy::execution_context&, boost::corosio::detail::scheduler&) :150 1331x 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 18x 87.1% 83.3% 84.0% boost::corosio::detail::posix_stream_file::do_read_work(boost::corosio::detail::pool_work_item*) :210 16x 90.5% 44.4% 70.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*) :244 18x 87.1% 83.3% 84.0% boost::corosio::detail::posix_stream_file::do_write_work(boost::corosio::detail::pool_work_item*) :294 16x 90.5% 37.5% 68.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 2662x posix_stream_file_service(
37 capy::execution_context& ctx, scheduler& sched)
38 1331x : sched_(&sched)
39
1/2
✓ Branch 0 taken 1331 times.
✗ Branch 1 not taken.
 1331x , pool_(get_or_create_pool(ctx))
40 2662x {
41 2662x }
42
43 3993x ~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 78x io_object::implementation* construct() override
49 {
50 78x auto ptr = std::make_shared<posix_stream_file>(*this);
51 78x auto* impl = ptr.get();
52
53 {
54
1/2
✓ Branch 0 taken 78 times.
✗ Branch 1 not taken.
 78x std::lock_guard<std::mutex> lock(mutex_);
55 78x file_list_.push_back(impl);
56
1/2
✓ Branch 0 taken 78 times.
✗ Branch 1 not taken.
 78x file_ptrs_[impl] = std::move(ptr);
57 78x }
58
59 78x return impl;
60 78x }
61
62 78x void destroy(io_object::implementation* p) override
63 {
64 78x auto& impl = static_cast<posix_stream_file&>(*p);
65 78x impl.cancel();
66 78x impl.close_file();
67 78x destroy_impl(impl);
68 78x }
69
70 134x void close(io_object::handle& h) override
71 {
72
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 134 times.
 134x if (h.get())
73 {
74 134x auto& impl = static_cast<posix_stream_file&>(*h.get());
75 134x impl.cancel();
76 134x impl.close_file();
77 134x }
78 134x }
79
80 62x 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 60 times.
 62x if (sched_->scheduler_locking_disabled())
88 2x return std::make_error_code(std::errc::operation_not_supported);
89 60x return static_cast<posix_stream_file&>(impl).open_file(path, mode);
90 62x }
91
92 1331x void shutdown() override
93 {
94 1331x std::lock_guard<std::mutex> lock(mutex_);
95
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1331 times.
 1331x 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 1331x file_ptrs_.clear();
102 1331x }
103
104 78x void destroy_impl(posix_stream_file& impl)
105 {
106 78x std::lock_guard<std::mutex> lock(mutex_);
107 78x file_list_.remove(&impl);
108
1/2
✓ Branch 0 taken 78 times.
✗ Branch 1 not taken.
 78x file_ptrs_.erase(&impl);
109 78x }
110
111 32x void post(scheduler_op* op)
112 {
113 32x sched_->post(op);
114 32x }
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 32x thread_pool& pool() noexcept
127 {
128 32x return pool_;
129 }
130
131 private:
132 1331x static thread_pool& get_or_create_pool(capy::execution_context& ctx)
133 {
134 1331x auto* p = ctx.find_service<thread_pool>();
135
1/2
✓ Branch 0 taken 1331 times.
✗ Branch 1 not taken.
 1331x if (p)
136 1331x return *p;
137 ✗ return ctx.make_service<thread_pool>();
138 1331x }
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 1331x get_stream_file_service(capy::execution_context& ctx, scheduler& sched)
151 {
152 1331x 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 18x 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 18x auto& op = read_op_;
169 18x op.reset();
170 18x op.is_read = true;
171
172 18x capy::mutable_buffer bufs[max_buffers];
173 18x op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
174
175
2/2
✓ Branch 0 taken 16 times.
✓ Branch 1 taken 2 times.
 18x if (op.iovec_count == 0)
176 {
177 2x *ec = {};
178 2x *bytes_out = 0;
179 2x op.cont.h = h;
180 2x return dispatch_coro(ex, op.cont);
181 }
182
183
2/2
✓ Branch 0 taken 16 times.
✓ Branch 1 taken 16 times.
 32x for (int i = 0; i < op.iovec_count; ++i)
184 {
185 16x op.iovecs[i].iov_base = bufs[i].data();
186 16x op.iovecs[i].iov_len = bufs[i].size();
187 16x }
188
189 16x op.h = h;
190 16x op.ex = ex;
191 16x op.ec_out = ec;
192 16x op.bytes_out = bytes_out;
193 16x op.start(token);
194
195 16x op.ex.on_work_started();
196
197 16x read_pool_op_.file_ = this;
198 16x read_pool_op_.ref_ = this->shared_from_this();
199 16x read_pool_op_.func_ = &posix_stream_file::do_read_work;
200
1/2
✓ Branch 0 taken 16 times.
✗ Branch 1 not taken.
 16x if (!svc_.pool().post(&read_pool_op_))
201 {
202 ✗ op.impl_ref = std::move(read_pool_op_.ref_);
203 ✗ op.cancelled.store(true, std::memory_order_release);
204 ✗ svc_.post(&read_op_);
205 ✗ }
206 16x return std::noop_coroutine();
207 18x }
208
209 inline void
210 16x posix_stream_file::do_read_work(pool_work_item* w) noexcept
211 {
212 16x auto* pw = static_cast<pool_op*>(w);
213 16x auto* self = pw->file_;
214 16x auto& op = self->read_op_;
215
216
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 14 times.
 16x if (!op.cancelled.load(std::memory_order_acquire))
217 {
218 ssize_t n;
219 14x do
220 {
221
2/4
✓ Branch 0 taken 14 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 14 times.
✗ Branch 3 not taken.
 28x n = ::preadv(self->fd_, op.iovecs, op.iovec_count,
222 14x static_cast<off_t>(self->offset_));
223 28x }
224
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 14 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 14x while (n < 0 && errno == EINTR);
225
226
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 14 times.
 14x if (n >= 0)
227 {
228 14x op.errn = 0;
229 14x op.bytes_transferred = static_cast<std::size_t>(n);
230 14x self->offset_ += static_cast<std::uint64_t>(n);
231 14x }
232 else
233 {
234 ✗ op.errn = errno;
235 ✗ op.bytes_transferred = 0;
236 }
237 14x }
238
239
1/2
✓ Branch 0 taken 16 times.
✗ Branch 1 not taken.
 16x op.impl_ref = std::move(pw->ref_);
240
1/2
✓ Branch 0 taken 16 times.
✗ Branch 1 not taken.
 16x self->svc_.post(&op);
241 16x }
242
243 inline std::coroutine_handle<>
244 18x posix_stream_file::write_some(
245 std::coroutine_handle<> h,
246 capy::executor_ref ex,
247 buffer_param param,
248 std::stop_token token,
249 std::error_code* ec,
250 std::size_t* bytes_out)
251 {
252 18x auto& op = write_op_;
253 18x op.reset();
254 18x op.is_read = false;
255
256 18x capy::mutable_buffer bufs[max_buffers];
257 18x op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
258
259
2/2
✓ Branch 0 taken 16 times.
✓ Branch 1 taken 2 times.
 18x if (op.iovec_count == 0)
260 {
261 2x *ec = {};
262 2x *bytes_out = 0;
263 2x op.cont.h = h;
264 2x return dispatch_coro(ex, op.cont);
265 }
266
267
2/2
✓ Branch 0 taken 16 times.
✓ Branch 1 taken 16 times.
 32x for (int i = 0; i < op.iovec_count; ++i)
268 {
269 16x op.iovecs[i].iov_base = bufs[i].data();
270 16x op.iovecs[i].iov_len = bufs[i].size();
271 16x }
272
273 16x op.h = h;
274 16x op.ex = ex;
275 16x op.ec_out = ec;
276 16x op.bytes_out = bytes_out;
277 16x op.start(token);
278
279 16x op.ex.on_work_started();
280
281 16x write_pool_op_.file_ = this;
282 16x write_pool_op_.ref_ = this->shared_from_this();
283 16x write_pool_op_.func_ = &posix_stream_file::do_write_work;
284
1/2
✓ Branch 0 taken 16 times.
✗ Branch 1 not taken.
 16x if (!svc_.pool().post(&write_pool_op_))
285 {
286 ✗ op.impl_ref = std::move(write_pool_op_.ref_);
287 ✗ op.cancelled.store(true, std::memory_order_release);
288 ✗ svc_.post(&write_op_);
289 ✗ }
290 16x return std::noop_coroutine();
291 18x }
292
293 inline void
294 16x posix_stream_file::do_write_work(pool_work_item* w) noexcept
295 {
296 16x auto* pw = static_cast<pool_op*>(w);
297 16x auto* self = pw->file_;
298 16x auto& op = self->write_op_;
299
300
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 16 times.
 16x if (!op.cancelled.load(std::memory_order_acquire))
301 {
302 ssize_t n;
303 16x do
304 {
305
2/4
✓ Branch 0 taken 16 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 16 times.
✗ Branch 3 not taken.
 32x n = ::pwritev(self->fd_, op.iovecs, op.iovec_count,
306 16x static_cast<off_t>(self->offset_));
307 32x }
308
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 16 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 16x while (n < 0 && errno == EINTR);
309
310
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 16 times.
 16x if (n >= 0)
311 {
312 16x op.errn = 0;
313 16x op.bytes_transferred = static_cast<std::size_t>(n);
314 16x self->offset_ += static_cast<std::uint64_t>(n);
315 16x }
316 else
317 {
318 ✗ op.errn = errno;
319 ✗ op.bytes_transferred = 0;
320 }
321 16x }
322
323 16x op.impl_ref = std::move(pw->ref_);
324
1/2
✓ Branch 0 taken 16 times.
✗ Branch 1 not taken.
 16x self->svc_.post(&op);
325 16x }
326
327 } // namespace boost::corosio::detail
328
329 #endif // BOOST_COROSIO_POSIX
330
331 #endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_SERVICE_HPP
332