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

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

99.3% Lines (150 / 151) 72.0% Branches (59 / 82) 100.0% Functions (21 / 21)
[ ] prev/next file n p prev/next uncovered line
posix_stream_file.hpp
f(x) Functions (21)
Function Calls Lines Branches Blocks
boost::corosio::detail::posix_stream_file::~posix_stream_file() :91 1290x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::file_op::~file_op() :105 2580x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::file_op::file_op() :125 2580x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::file_op::reset() :127 646x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::pool_op::pool_op() :149 2580x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::pool_op::~pool_op() :149 2580x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::reuse() :170 64x 100.0% 25.0% 40.0% boost::corosio::detail::posix_stream_file::native_handle() const :198 2884x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::cancel() :203 2503x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::posix_stream_file(boost::corosio::detail::posix_stream_file_service&) :257 1290x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::open_file(std::__1::__fs::filesystem::path const&, boost::corosio::file_base::flags) :264 683x 100.0% 100.0% 95.0% boost::corosio::detail::posix_stream_file::bump_generation() :323 3182x 100.0% 50.0% 66.0% boost::corosio::detail::posix_stream_file::close_file() :330 3176x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::size() const :338 12x 100.0% 75.0% 83.0% boost::corosio::detail::posix_stream_file::resize(unsigned long long) :347 9x 100.0% 87.5% 87.0% boost::corosio::detail::posix_stream_file::sync_data() :358 5x 100.0% 66.7% 85.0% boost::corosio::detail::posix_stream_file::sync_all() :370 5x 100.0% 66.7% 85.0% boost::corosio::detail::posix_stream_file::release() :378 8x 100.0% 100.0% 100.0% boost::corosio::detail::posix_stream_file::seek(long long, boost::corosio::file_base::seek_basis) :395 426x 94.4% 78.6% 81.0% boost::corosio::detail::posix_stream_file::file_op::operator()() :434 617x 100.0% 75.0% 71.0% boost::corosio::detail::posix_stream_file::file_op::destroy() :454 4x 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_STREAM_FILE_HPP
11 #define BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_HPP
12
13 #include <boost/corosio/detail/platform.hpp>
14
15 #if BOOST_COROSIO_POSIX
16
17 #include <boost/corosio/detail/config.hpp>
18 #include <boost/corosio/stream_file.hpp>
19 #include <boost/corosio/file_base.hpp>
20 #include <boost/corosio/detail/intrusive.hpp>
21 #include <boost/corosio/detail/dispatch_coro.hpp>
22 #include <boost/corosio/detail/scheduler_op.hpp>
23 #include <boost/corosio/detail/thread_pool.hpp>
24 #include <boost/corosio/detail/scheduler.hpp>
25 #include <boost/corosio/detail/buffer_param.hpp>
26 #include <boost/corosio/native/detail/coro_op.hpp>
27 #include <boost/corosio/native/detail/coro_op_complete.hpp>
28 #include <boost/corosio/native/detail/make_err.hpp>
29 #include <boost/corosio/native/detail/posix/large_file.hpp>
30 #include <boost/corosio/native/detail/posix/posix_fd_gate.hpp>
31 #include <boost/corosio/native/detail/validate_fd.hpp>
32 #include <boost/capy/continuation.hpp>
33 #include <boost/capy/ex/executor_ref.hpp>
34 #include <boost/capy/error.hpp>
35 #include <boost/capy/buffers.hpp>
36
37 #include <atomic>
38 #include <coroutine>
39 #include <cstddef>
40 #include <cstdint>
41 #include <filesystem>
42 #include <limits>
43 #include <memory>
44 #include <mutex>
45 #include <optional>
46 #include <stop_token>
47 #include <system_error>
48
49 #include <errno.h>
50 #include <fcntl.h>
51 #include <sys/stat.h>
52 #include <sys/uio.h>
53 #include <unistd.h>
54
55 /*
56 POSIX Stream File Implementation
57 =================================
58
59 Regular files cannot be monitored by epoll/kqueue/select — the kernel
60 always reports them as ready. Blocking I/O (pread/pwrite) is dispatched
61 to a shared thread pool, with completion posted back to the scheduler.
62
63 This follows the same pattern as posix_resolver: pool_work_item for
64 dispatch, scheduler_op for completion, an object_ref keepalive on
65 each op for lifetime while its pool work is in flight.
66
67 Completion Flow
68 ---------------
69 1. read_some() sets up file_read_op, posts to thread pool
70 2. Pool thread runs preadv() (blocking)
71 3. Pool thread stores results, posts scheduler_op to scheduler
72 4. Scheduler invokes op() which resumes the coroutine
73
74 Single-Inflight Constraint
75 --------------------------
76 Only one asynchronous operation may be in flight at a time on a
77 given file object. Concurrent read and write is not supported
78 because both share offset_ without synchronization.
79 */
80
81 namespace boost::corosio::detail {
82
83 class scheduler;
84 class posix_stream_file_service;
85
86 /** Stream file implementation for POSIX backends.
87
88 Each instance contains embedded operation objects (read_op_, write_op_)
89 that are reused across calls. This avoids per-operation heap allocation.
90 */
91 class posix_stream_file final
92 : public stream_file::implementation
93 , public intrusive_list<posix_stream_file>::node
94 {
95 friend class posix_stream_file_service;
96
97 public:
98 static constexpr std::size_t max_buffers = 16;
99
100 /** Operation state for a single file read or write.
101
102 The coroutine, cancellation and keepalive machinery is inherited
103 from `coro_op`; only the pool-path result state lives here.
104 */
105 struct file_op : coro_op
106 {
107 // Buffer data (copied from buffer_param at submission time)
108 iovec iovecs[max_buffers];
109 1290x int iovec_count = 0;
110
111 // Snapshotted at submission: assign() may replace fd_ and
112 // offset_ while the worker runs.
113 1290x int fd = -1;
114 1290x std::uint64_t offset = 0;
115 1290x std::uint64_t generation = 0;
116
117 // Result storage (populated by worker thread)
118 1290x int errn = 0;
119 1290x std::size_t bytes_transferred = 0;
120
121 // The op is work on this service's context, which outlives the
122 // file the op is embedded in.
123 1290x posix_stream_file_service* svc = nullptr;
124
125 3870x file_op() = default;
126
127 646x void reset() noexcept
128 {
129 646x iovec_count = 0;
130 646x errn = 0;
131 646x bytes_transferred = 0;
132 646x is_read = false;
133 646x cancelled.store(false, std::memory_order_relaxed);
134 646x stop_cb.reset();
135 646x object_ref_.reset();
136 646x ec_out = nullptr;
137 646x bytes_out = nullptr;
138 646x }
139
140 void operator()() override;
141 void destroy() override;
142 };
143
144 /// Release the work an op holds on its service's context.
145 static void
146 release_work(posix_stream_file_service& svc) noexcept;
147
148 /** Pool work item for thread pool dispatch. */
149 1290x struct pool_op : pool_work_item
150 {
151 1290x posix_stream_file* file_ = nullptr;
152 detail::object_ref ref_;
153 };
154
155 explicit posix_stream_file(posix_stream_file_service& svc) noexcept;
156
157 /// Recycle into the owning service's pool. Defined out-of-line
158 /// after posix_stream_file_service for its complete type.
159 void retire() noexcept override;
160
161 /** Reset op slots for recycling.
162
163 `close_file()` already drove fd_ to its closed value before
164 the refcount reached zero; each op's own `reset()` runs again
165 before its next use, so only the `stop_cb` invariant is worth
166 asserting here.
167
168 @pre refs_ == 0, fd closed, no op in flight.
169 */
170 64x void reuse() noexcept
171 {
172
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 64 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 64x BOOST_COROSIO_ASSERT(fd_ == -1);
173
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 64 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 64x BOOST_COROSIO_ASSERT(gate_.idle());
174
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 64 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 64x BOOST_COROSIO_ASSERT(!read_op_.stop_cb);
175
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 64 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 64x BOOST_COROSIO_ASSERT(!write_op_.stop_cb);
176 64x }
177
178 // -- io_stream::implementation --
179
180 std::coroutine_handle<> read_some(
181 capy::continuation&,
182 capy::executor_ref,
183 buffer_param,
184 std::stop_token,
185 std::error_code*,
186 std::size_t*) override;
187
188 std::coroutine_handle<> write_some(
189 capy::continuation&,
190 capy::executor_ref,
191 buffer_param,
192 std::stop_token,
193 std::error_code*,
194 std::size_t*) override;
195
196 // -- stream_file::implementation --
197
198 2884x native_handle_type native_handle() const noexcept override
199 {
200 2884x return fd_;
201 }
202
203 2503x void cancel() noexcept override
204 {
205 2503x read_op_.request_cancel();
206 2503x write_op_.request_cancel();
207 2503x }
208
209 capy::results<std::uint64_t> size() const noexcept override;
210 std::error_code resize(std::uint64_t new_size) noexcept override;
211 std::error_code sync_data() noexcept override;
212 std::error_code sync_all() noexcept override;
213 capy::results<native_handle_type> release() noexcept override;
214 std::error_code assign(native_handle_type handle) noexcept override;
215 capy::results<std::uint64_t>
216 seek(std::int64_t offset, file_base::seek_basis origin) noexcept override;
217
218 // -- Internal --
219
220 /** Open the file and store the fd. */
221 std::error_code
222 open_file(std::filesystem::path const& path, file_base::flags mode);
223
224 /** Close the file descriptor. */
225 void close_file() noexcept;
226
227 private:
228 posix_stream_file_service& svc_;
229 645x int fd_ = -1;
230 645x std::uint64_t offset_ = 0;
231
232 // Bumped whenever the held file changes. A worker advances offset_
233 // only if this still matches its op's snapshot; the mutex makes the
234 // check and the advance one step against a concurrent assign().
235 std::mutex offset_mutex_;
236 /// Keeps fd_'s number allocated while a pool worker uses it.
237 posix_fd_gate gate_;
238 645x std::uint64_t generation_ = 0;
239
240 file_op read_op_;
241 file_op write_op_;
242 pool_op read_pool_op_;
243 pool_op write_pool_op_;
244
245 void bump_generation() noexcept;
246
247 static void do_read_work(pool_work_item*) noexcept;
248 static void do_write_work(pool_work_item*) noexcept;
249 static void destroy_read_work(pool_work_item*) noexcept;
250 static void destroy_write_work(pool_work_item*) noexcept;
251 };
252
253 // ---------------------------------------------------------------------------
254 // Inline implementation
255 // ---------------------------------------------------------------------------
256
257 3225x inline posix_stream_file::posix_stream_file(
258 posix_stream_file_service& svc) noexcept
259 645x : svc_(svc)
260 1935x {
261 1290x }
262
263 inline std::error_code
264 683x posix_stream_file::open_file(
265 std::filesystem::path const& path, file_base::flags mode)
266 {
267 683x close_file();
268
269 683x int oflags = 0;
270
271 // Access mode
272 683x unsigned access = static_cast<unsigned>(mode) & 3u;
273
2/2
✓ Branch 0 taken 80 times.
✓ Branch 1 taken 603 times.
 683x if (access == static_cast<unsigned>(file_base::read_write))
274 80x oflags |= O_RDWR;
275
2/2
✓ Branch 0 taken 80 times.
✓ Branch 1 taken 523 times.
 603x else if (access == static_cast<unsigned>(file_base::write_only))
276 80x oflags |= O_WRONLY;
277 else
278 523x oflags |= O_RDONLY;
279
280 // Creation flags
281
2/2
✓ Branch 0 taken 589 times.
✓ Branch 1 taken 94 times.
 683x if ((mode & file_base::create) != file_base::flags(0))
282 94x oflags |= O_CREAT;
283
2/2
✓ Branch 0 taken 681 times.
✓ Branch 1 taken 2 times.
 683x if ((mode & file_base::exclusive) != file_base::flags(0))
284 2x oflags |= O_EXCL;
285
2/2
✓ Branch 0 taken 600 times.
✓ Branch 1 taken 83 times.
 683x if ((mode & file_base::truncate) != file_base::flags(0))
286 83x oflags |= O_TRUNC;
287
2/2
✓ Branch 0 taken 678 times.
✓ Branch 1 taken 5 times.
 683x if ((mode & file_base::append) != file_base::flags(0))
288 5x oflags |= O_APPEND;
289
2/2
✓ Branch 0 taken 681 times.
✓ Branch 1 taken 2 times.
 683x if ((mode & file_base::sync_all_on_write) != file_base::flags(0))
290 2x oflags |= O_SYNC;
291
292 683x int fd = ::open(path.c_str(), oflags | large_file_open_flag, 0666);
293
2/2
✓ Branch 0 taken 677 times.
✓ Branch 1 taken 6 times.
 683x if (fd < 0)
294 6x return make_err(errno);
295
296 677x fd_ = fd;
297 677x offset_ = 0;
298 677x gate_.open(fd);
299
300 // Append mode: position at end-of-file (preadv/pwritev use
301 // explicit offsets, so O_APPEND alone is not sufficient).
302
2/2
✓ Branch 0 taken 672 times.
✓ Branch 1 taken 5 times.
 677x if ((mode & file_base::append) != file_base::flags(0))
303 {
304 file_stat_t st;
305
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 2 times.
 5x if (file_fstat(fd, &st) < 0)
306 {
307 2x int err = errno;
308 2x gate_.close();
309 2x fd_ = -1;
310 2x return make_err(err);
311 }
312 3x offset_ = static_cast<std::uint64_t>(st.st_size);
313 3x }
314
315 #ifdef POSIX_FADV_SEQUENTIAL
316 ::posix_fadvise(fd_, 0, 0, POSIX_FADV_SEQUENTIAL);
317 #endif
318
319 675x return {};
320 683x }
321
322 inline void
323 3182x posix_stream_file::bump_generation() noexcept
324 {
325
1/2
✓ Branch 0 taken 3182 times.
✗ Branch 1 not taken.
 3182x std::lock_guard<std::mutex> lock(offset_mutex_);
326 3182x ++generation_;
327 3182x }
328
329 inline void
330 3176x posix_stream_file::close_file() noexcept
331 {
332 3176x bump_generation();
333 3176x gate_.close();
334 3176x fd_ = -1;
335 3176x }
336
337 inline capy::results<std::uint64_t>
338 12x posix_stream_file::size() const noexcept
339 {
340 file_stat_t st;
341
2/2
✓ Branch 0 taken 10 times.
✓ Branch 1 taken 2 times.
 12x if (file_fstat(fd_, &st) < 0)
342
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2x return {make_err(errno), 0};
343 10x return {std::error_code{}, static_cast<std::uint64_t>(st.st_size)};
344 12x }
345
346 inline std::error_code
347 9x posix_stream_file::resize(std::uint64_t new_size) noexcept
348 {
349
4/4
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 7 times.
✓ Branch 2 taken 2 times.
✓ Branch 3 taken 7 times.
 18x if (new_size >
350 9x static_cast<std::uint64_t>((std::numeric_limits<file_off_t>::max)()))
351 2x return make_err(EOVERFLOW);
352
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 4 times.
 7x if (file_ftruncate(fd_, static_cast<file_off_t>(new_size)) < 0)
353
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4x return make_err(errno);
354 3x return {};
355 9x }
356
357 inline std::error_code
358 5x posix_stream_file::sync_data() noexcept
359 {
360 #if BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
361 if (::fdatasync(fd_) < 0)
362 #else // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
363
3/4
✓ Branch 0 taken 5 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✓ Branch 3 taken 3 times.
 5x if (::fsync(fd_) < 0)
364 #endif // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
365
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2x return make_err(errno);
366 3x return {};
367 5x }
368
369 inline std::error_code
370 5x posix_stream_file::sync_all() noexcept
371 {
372
3/4
✓ Branch 0 taken 5 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✓ Branch 3 taken 3 times.
 5x if (::fsync(fd_) < 0)
373
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2x return make_err(errno);
374 3x return {};
375 5x }
376
377 inline capy::results<native_handle_type>
378 8x posix_stream_file::release() noexcept
379 {
380 // A queued op has already copied the fd number; it must not run
381 // after the caller closes it and the number is recycled.
382 8x cancel();
383
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 6 times.
 8x if (!gate_.release())
384 2x return {
385 2x std::make_error_code(std::errc::device_or_resource_busy),
386 invalid_native_handle};
387 6x bump_generation();
388 6x int fd = fd_;
389 6x fd_ = -1;
390 6x offset_ = 0;
391 6x return {std::error_code{}, fd};
392 8x }
393
394 inline capy::results<std::uint64_t>
395 426x posix_stream_file::seek(
396 std::int64_t offset, file_base::seek_basis origin) noexcept
397 {
398 // We track offset_ ourselves (not the kernel fd offset)
399 // because preadv/pwritev use explicit offsets.
400 std::int64_t new_pos;
401
402
2/2
✓ Branch 0 taken 415 times.
✓ Branch 1 taken 11 times.
 426x if (origin == file_base::seek_set)
403 {
404 11x new_pos = offset;
405 11x }
406
2/2
✓ Branch 0 taken 407 times.
✓ Branch 1 taken 8 times.
 415x else if (origin == file_base::seek_cur)
407 {
408 407x new_pos = static_cast<std::int64_t>(offset_) + offset;
409 407x }
410 else
411 {
412 file_stat_t st;
413
2/2
✓ Branch 0 taken 6 times.
✓ Branch 1 taken 2 times.
 8x if (file_fstat(fd_, &st) < 0)
414
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2x return {make_err(errno), 0};
415 6x new_pos = st.st_size + offset;
416 }
417
418
2/2
✓ Branch 0 taken 418 times.
✓ Branch 1 taken 6 times.
 424x if (new_pos < 0)
419 6x return {make_err(EINVAL), 0};
420
2/4
✗ Branch 0 not taken.
✓ Branch 1 taken 418 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 418 times.
 836x if (new_pos >
421 418x static_cast<std::int64_t>((std::numeric_limits<file_off_t>::max)()))
422 ✗ return {make_err(EOVERFLOW), 0};
423
424 418x offset_ = static_cast<std::uint64_t>(new_pos);
425
426 418x return {std::error_code{}, offset_};
427 426x }
428
429 // -- file_op completion handler --
430 // (read_some, write_some, do_read_work, do_write_work are
431 // defined in posix_stream_file_service.hpp after the service)
432
433 inline void
434 617x posix_stream_file::file_op::operator()()
435 {
436 617x stop_cb.reset();
437
438 // Empty buffers never reach the pool (diverted at initiation), so
439 // empty_buffer stays false and a 0-byte read is a genuine EOF.
440 617x decode_results(
441 617x ec_out, bytes_out, cancelled.load(std::memory_order_acquire),
442
2/2
✓ Branch 0 taken 8 times.
✓ Branch 1 taken 609 times.
 617x errn != 0 ? make_err(errn) : std::error_code{}, is_read,
443 617x bytes_transferred, /*empty_buffer=*/false);
444
445 // Move object_ref_ to a local so members remain valid through
446 // dispatch — this may be the last reference keeping the parent
447 // posix_stream_file (which embeds this file_op) alive.
448 617x auto prevent_destroy = std::move(object_ref_);
449 617x release_work(*svc);
450
1/2
✓ Branch 0 taken 617 times.
✗ Branch 1 not taken.
 617x dispatch_coro(ex, *cont).resume();
451 617x }
452
453 inline void
454 4x posix_stream_file::file_op::destroy()
455 {
456 // Copied first: the abandon may unwind the frame at once, destroying
457 // the file, and with it this op.
458 4x auto* s = svc;
459 4x destroy_awaiting();
460 4x release_work(*s);
461 4x }
462
463 } // namespace boost::corosio::detail
464
465 #endif // BOOST_COROSIO_POSIX
466
467 #endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_HPP
468