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

95.8% Lines (183 / 191) 74.3% Branches (55 / 74) 100.0% Functions (19 / 19)
[ ] prev/next file n p prev/next uncovered line
posix_random_access_file_service.hpp
f(x) Functions (19)
Function Calls Lines Branches Blocks
boost::corosio::detail::posix_random_access_file_service::posix_random_access_file_service(boost::capy::execution_context&) :36 374x 100.0% 50.0% 100.0% boost::corosio::detail::posix_random_access_file_service::~posix_random_access_file_service() :42 561x 100.0% – 100.0% boost::corosio::detail::posix_random_access_file_service::construct() :49 242x 100.0% – 100.0% boost::corosio::detail::posix_random_access_file_service::destroy(boost::corosio::io_object::implementation*) :54 242x 100.0% – 100.0% boost::corosio::detail::posix_random_access_file_service::close(boost::corosio::io_object::handle&) :62 449x 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) :72 219x 83.3% 50.0% 75.0% boost::corosio::detail::posix_random_access_file_service::pool_unavailable() const :88 228x 100.0% – 100.0% boost::corosio::detail::posix_random_access_file_service::shutdown() :93 187x 100.0% 50.0% 75.0% boost::corosio::detail::posix_random_access_file_service::shutdown()::'lambda'(boost::corosio::detail::posix_random_access_file*)::operator()(boost::corosio::detail::posix_random_access_file*) const :102 8x 100.0% 100.0% 100.0% boost::corosio::detail::posix_random_access_file_service::post(boost::corosio::detail::scheduler_op*) :115 492x 100.0% – 100.0% boost::corosio::detail::posix_random_access_file_service::work_started() :120 504x 100.0% – 100.0% boost::corosio::detail::posix_random_access_file_service::work_finished() :125 504x 100.0% – 100.0% boost::corosio::detail::posix_random_access_file_service::pool() :143 504x 100.0% – 100.0% boost::corosio::detail::posix_random_access_file::read_some_at(unsigned long long, boost::capy::continuation&, boost::capy::executor_ref, boost::corosio::detail::buffer_param, std::__1::stop_token, std::__1::error_code*, unsigned long*) :159 389x 100.0% 78.6% 95.0% boost::corosio::detail::posix_random_access_file::write_some_at(unsigned long long, boost::capy::continuation&, boost::capy::executor_ref, boost::corosio::detail::buffer_param, std::__1::stop_token, std::__1::error_code*, unsigned long*) :232 125x 100.0% 78.6% 95.0% boost::corosio::detail::posix_random_access_file::raf_op::do_work(boost::corosio::detail::pool_work_item*) :307 492x 84.1% 75.0% 89.0% boost::corosio::detail::posix_random_access_file::assign(int) :379 9x 100.0% 100.0% 100.0% boost::corosio::detail::posix_random_access_file::retire() :393 242x 100.0% 50.0% 66.0% boost::corosio::detail::posix_random_access_file::release_work(boost::corosio::detail::posix_random_access_file_service&) :399 504x 100.0% – 100.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/object_pool.hpp>
21 #include <boost/corosio/detail/object_ref.hpp>
22 #include <boost/corosio/detail/thread_pool.hpp>
23
24 #include <limits>
25 #include <vector>
26
27 namespace boost::corosio::detail {
28
29 /** Random-access file service for POSIX backends. */
30 class BOOST_COROSIO_DECL posix_random_access_file_service final
31 : public random_access_file_service
32 {
33 friend class posix_random_access_file;
34
35 public:
36 374x explicit posix_random_access_file_service(capy::execution_context& ctx)
37
1/2
✓ Branch 0 taken 187 times.
✗ Branch 1 not taken.
 187x : sched_(&get_scheduler(ctx))
38 187x , pool_(ctx)
39 374x {
40 374x }
41
42 561x ~posix_random_access_file_service() override = default;
43
44 posix_random_access_file_service(posix_random_access_file_service const&) =
45 delete;
46 posix_random_access_file_service&
47 operator=(posix_random_access_file_service const&) = delete;
48
49 242x io_object::implementation* construct() override
50 {
51 242x return object_pool_.acquire(*this);
52 }
53
54 242x void destroy(io_object::implementation* p) override
55 {
56 242x auto& impl = static_cast<posix_random_access_file&>(*p);
57 242x impl.cancel();
58 242x impl.close_file();
59 242x release(&impl);
60 242x }
61
62 449x void close(io_object::handle& h) override
63 {
64
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 449 times.
 449x if (h.get())
65 {
66 449x auto& impl = static_cast<posix_random_access_file&>(*h.get());
67 449x impl.cancel();
68 449x impl.close_file();
69 449x }
70 449x }
71
72 219x std::error_code open_file(
73 random_access_file::implementation& impl,
74 std::filesystem::path const& path,
75 file_base::flags mode) override
76 {
77
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 219 times.
 219x if (pool_unavailable())
78 ✗ return std::make_error_code(std::errc::operation_not_supported);
79 438x return static_cast<posix_random_access_file&>(impl).open_file(
80 219x path, mode);
81 219x }
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 228x bool pool_unavailable() const noexcept
89 {
90 228x return sched_->scheduler_locking_disabled();
91 }
92
93 187x 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 187x std::vector<posix_random_access_file*> live;
101
2/4
✓ Branch 0 taken 187 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 187 times.
✗ Branch 3 not taken.
 374x object_pool_.shutdown(
102 195x [&](posix_random_access_file* f)
103 {
104 8x acquire(f);
105 8x live.push_back(f);
106 8x });
107
2/2
✓ Branch 0 taken 8 times.
✓ Branch 1 taken 187 times.
 195x for (auto* f : live)
108 {
109 8x f->cancel();
110 8x f->close_file();
111 8x release(f);
112 }
113 187x }
114
115 492x void post(scheduler_op* op)
116 {
117 492x sched_->post(op);
118 492x }
119
120 504x void work_started() noexcept
121 {
122 504x sched_->work_started();
123 504x }
124
125 504x void work_finished() noexcept
126 {
127 504x sched_->work_finished();
128 504x }
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 504x thread_pool& pool()
144 {
145 504x return pool_.get();
146 }
147
148 private:
149 scheduler* sched_;
150 thread_pool_ref pool_;
151 object_pool<posix_random_access_file> object_pool_;
152 };
153
154 // ---------------------------------------------------------------------------
155 // posix_random_access_file inline implementations (require complete service)
156 // ---------------------------------------------------------------------------
157
158 inline std::coroutine_handle<>
159 389x posix_random_access_file::read_some_at(
160 std::uint64_t offset,
161 capy::continuation& cont,
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 // Closed-object contract outranks the zero-length no-op.
169
2/2
✓ Branch 0 taken 385 times.
✓ Branch 1 taken 4 times.
 389x if (fd_ < 0)
170 {
171 4x *ec = make_error_code(std::errc::bad_file_descriptor);
172 4x *bytes_out = 0;
173 4x return cont.h;
174 }
175
176 385x capy::mutable_buffer bufs[max_buffers];
177 385x auto count = param.copy_to(bufs, max_buffers);
178
179
2/2
✓ Branch 0 taken 383 times.
✓ Branch 1 taken 2 times.
 385x if (count == 0)
180 {
181 2x *ec = {};
182 2x *bytes_out = 0;
183 2x return cont.h;
184 }
185
186 383x auto* op = acquire_op();
187 383x op->is_read = true;
188 383x op->offset = offset;
189 383x op->fd = fd_;
190
191 383x op->iovec_count = static_cast<int>(count);
192
2/2
✓ Branch 0 taken 383 times.
✓ Branch 1 taken 383 times.
 766x for (int i = 0; i < op->iovec_count; ++i)
193 {
194 383x op->iovecs[i].iov_base = bufs[i].data();
195 383x op->iovecs[i].iov_len = bufs[i].size();
196 383x }
197
198 383x op->cont = &cont;
199 383x op->ex = ex;
200 383x op->ec_out = ec;
201 383x op->bytes_out = bytes_out;
202 383x op->file_ = this;
203 383x op->object_ref_ = detail::object_ref(this);
204 383x op->start(token);
205
206 383x svc_.work_started();
207
208 {
209 383x std::lock_guard<std::mutex> lock(ops_mutex_);
210 383x outstanding_ops_.push_back(op);
211 383x }
212
213
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 383 times.
 383x static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
214
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 383 times.
 383x static_cast<pool_work_item*>(op)->destroy_ = &raf_op::destroy_work;
215
3/4
✗ Branch 0 not taken.
✓ Branch 1 taken 383 times.
✓ Branch 2 taken 2 times.
✓ Branch 3 taken 381 times.
 383x if (auto pec = svc_.pool().post(static_cast<pool_work_item*>(op)))
216 {
217 // The pool is shutting down, or the system refused it a thread.
218 // Nothing of this read went cross-thread, so it answers here
219 // like the closed-descriptor and zero-length exits above rather
220 // than through a completion the scheduler has to carry back.
221 // The op never reached the queue: it unlinks, unwinds the
222 // work count and recycles, and the coroutine resumes below.
223 2x op->discard();
224 2x *ec = pec;
225 2x *bytes_out = 0;
226 2x return cont.h;
227 }
228 381x return std::noop_coroutine();
229 389x }
230
231 inline std::coroutine_handle<>
232 125x posix_random_access_file::write_some_at(
233 std::uint64_t offset,
234 capy::continuation& cont,
235 capy::executor_ref ex,
236 buffer_param param,
237 std::stop_token token,
238 std::error_code* ec,
239 std::size_t* bytes_out)
240 {
241 // Closed-object contract outranks the zero-length no-op.
242
2/2
✓ Branch 0 taken 123 times.
✓ Branch 1 taken 2 times.
 125x if (fd_ < 0)
243 {
244 2x *ec = make_error_code(std::errc::bad_file_descriptor);
245 2x *bytes_out = 0;
246 2x return cont.h;
247 }
248
249 123x capy::mutable_buffer bufs[max_buffers];
250 123x auto count = param.copy_to(bufs, max_buffers);
251
252
2/2
✓ Branch 0 taken 121 times.
✓ Branch 1 taken 2 times.
 123x if (count == 0)
253 {
254 2x *ec = {};
255 2x *bytes_out = 0;
256 2x return cont.h;
257 }
258
259 121x auto* op = acquire_op();
260 121x op->is_read = false;
261 121x op->offset = offset;
262 121x op->fd = fd_;
263
264 121x op->iovec_count = static_cast<int>(count);
265
2/2
✓ Branch 0 taken 121 times.
✓ Branch 1 taken 121 times.
 242x for (int i = 0; i < op->iovec_count; ++i)
266 {
267 121x op->iovecs[i].iov_base = bufs[i].data();
268 121x op->iovecs[i].iov_len = bufs[i].size();
269 121x }
270
271 121x op->cont = &cont;
272 121x op->ex = ex;
273 121x op->ec_out = ec;
274 121x op->bytes_out = bytes_out;
275 121x op->file_ = this;
276 121x op->object_ref_ = detail::object_ref(this);
277 121x op->start(token);
278
279 121x svc_.work_started();
280
281 {
282 121x std::lock_guard<std::mutex> lock(ops_mutex_);
283 121x outstanding_ops_.push_back(op);
284 121x }
285
286
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 121 times.
 121x static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
287
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 121 times.
 121x static_cast<pool_work_item*>(op)->destroy_ = &raf_op::destroy_work;
288
3/4
✗ Branch 0 not taken.
✓ Branch 1 taken 121 times.
✓ Branch 2 taken 2 times.
✓ Branch 3 taken 119 times.
 121x if (auto pec = svc_.pool().post(static_cast<pool_work_item*>(op)))
289 {
290 // The pool is shutting down, or the system refused it a thread.
291 // Nothing of this write went cross-thread, so it answers here
292 // like the closed-descriptor and zero-length exits above rather
293 // than through a completion the scheduler has to carry back.
294 // The op never reached the queue: it unlinks, unwinds the
295 // work count and recycles, and the coroutine resumes below.
296 2x op->discard();
297 2x *ec = pec;
298 2x *bytes_out = 0;
299 2x return cont.h;
300 }
301 119x return std::noop_coroutine();
302 125x }
303
304 // -- raf_op thread-pool work function --
305
306 inline void
307 492x posix_random_access_file::raf_op::do_work(pool_work_item* w) noexcept
308 {
309
1/2
✓ Branch 0 taken 492 times.
✗ Branch 1 not taken.
 492x auto* op = static_cast<raf_op*>(w);
310 492x auto* self = op->file_;
311
312
2/2
✓ Branch 0 taken 101 times.
✓ Branch 1 taken 391 times.
 492x if (op->cancelled.load(std::memory_order_acquire))
313 {
314 101x op->errn = ECANCELED;
315 101x op->bytes_transferred = 0;
316 101x }
317 else if (
318
4/4
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 389 times.
✓ Branch 2 taken 2 times.
✓ Branch 3 taken 389 times.
 782x op->offset >
319 391x static_cast<std::uint64_t>(std::numeric_limits<file_off_t>::max()))
320 {
321 2x op->errn = EOVERFLOW;
322 2x op->bytes_transferred = 0;
323 2x }
324
1/2
✓ Branch 0 taken 389 times.
✗ Branch 1 not taken.
 389x else if (!self->gate_.enter(op->fd))
325 {
326 // Closed or replaced since this was queued.
327 ✗ op->errn = ECANCELED;
328 ✗ op->bytes_transferred = 0;
329 ✗ }
330
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 389 times.
 389x else if (op->cancelled.load(std::memory_order_acquire))
331 {
332 // Closed and reopened onto the same number since the check
333 // above: the gate admits the number, but not this op.
334 ✗ self->gate_.leave();
335 ✗ op->errn = ECANCELED;
336 ✗ op->bytes_transferred = 0;
337 ✗ }
338 else
339 {
340 ssize_t n;
341
2/2
✓ Branch 0 taken 321 times.
✓ Branch 1 taken 68 times.
 389x if (op->is_read)
342 {
343 321x do
344 {
345 321x n = file_preadv(
346 321x op->fd, op->iovecs, op->iovec_count,
347 321x static_cast<file_off_t>(op->offset));
348 642x }
349
3/4
✓ Branch 0 taken 317 times.
✓ Branch 1 taken 4 times.
✓ Branch 2 taken 4 times.
✗ Branch 3 not taken.
 321x while (n < 0 && errno == EINTR);
350 321x }
351 else
352 {
353 68x do
354 {
355 68x n = file_pwritev(
356 68x op->fd, op->iovecs, op->iovec_count,
357 68x static_cast<file_off_t>(op->offset));
358 136x }
359
3/4
✓ Branch 0 taken 64 times.
✓ Branch 1 taken 4 times.
✓ Branch 2 taken 4 times.
✗ Branch 3 not taken.
 68x while (n < 0 && errno == EINTR);
360 }
361
362
2/2
✓ Branch 0 taken 8 times.
✓ Branch 1 taken 381 times.
 389x if (n >= 0)
363 {
364 381x op->errn = 0;
365 381x op->bytes_transferred = static_cast<std::size_t>(n);
366 381x }
367 else
368 {
369
1/2
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
 8x op->errn = errno;
370 8x op->bytes_transferred = 0;
371 }
372 389x self->gate_.leave();
373 }
374
375
1/2
✓ Branch 0 taken 492 times.
✗ Branch 1 not taken.
 492x self->svc_.post(static_cast<scheduler_op*>(op));
376 492x }
377
378 inline std::error_code
379 9x posix_random_access_file::assign(native_handle_type handle) noexcept
380 {
381 // The public assign() guarantees the object is closed.
382
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 7 times.
 9x if (svc_.pool_unavailable())
383 2x return std::make_error_code(std::errc::operation_not_supported);
384
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 3 times.
 7x if (auto ec = validate_file_fd(handle))
385 4x return ec;
386
387 3x fd_ = handle;
388 3x gate_.open(handle);
389 3x return {};
390 9x }
391
392 inline void
393 242x posix_random_access_file::retire() noexcept
394 {
395
1/2
✓ Branch 0 taken 242 times.
✗ Branch 1 not taken.
 242x svc_.object_pool_.recycle(this);
396 242x }
397
398 inline void
399 504x posix_random_access_file::release_work(
400 posix_random_access_file_service& svc) noexcept
401 {
402 504x svc.work_finished();
403 504x }
404
405 } // namespace boost::corosio::detail
406
407 #endif // BOOST_COROSIO_POSIX
408
409 #endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_FILE_SERVICE_HPP
410