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

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

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