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

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

99.3% Lines (143/144) 100.0% List of functions (20/20) 81.7% Branches (67/82)
[ ] prev/next file n p prev/next uncovered line
posix_stream_file.hpp
f(x) Functions (20)
Function Calls Lines Branches Blocks
boost::corosio::detail::posix_stream_file::~posix_stream_file() :84 250x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::file_op::~file_op() :99 500x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::file_op::file_op() :109 500x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::file_op::reset() :111 88x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::pool_op::pool_op() :129 500x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::pool_op::~pool_op() :129 500x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::native_handle() const :157 382x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::cancel() :162 349x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::posix_stream_file(boost::corosio::detail::posix_stream_file_service&) :204 250x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::open_file(std::__1::__fs::filesystem::path const&, boost::corosio::file_base::flags) :211 105x 100.0% 100.0% 100.0% boost::corosio::detail::posix_stream_file::close_file() :269 458x 100.0% 75.0% 80.0% boost::corosio::detail::posix_stream_file::size() const :279 14x 100.0% 100.0% 100.0% boost::corosio::detail::posix_stream_file::resize(unsigned long long) :288 9x 100.0% 80.0% 88.0% boost::corosio::detail::posix_stream_file::sync_data() :299 7x 100.0% 66.7% 85.0% boost::corosio::detail::posix_stream_file::sync_all() :311 7x 100.0% 66.7% 85.0% boost::corosio::detail::posix_stream_file::release() :319 2x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::assign(int) :328 6x 100.0% – 100.0% boost::corosio::detail::posix_stream_file::seek(long long, boost::corosio::file_base::seek_basis) :337 24x 94.4% 75.0% 88.0% boost::corosio::detail::posix_stream_file::file_op::operator()() :376 61x 100.0% 77.8% 90.0% boost::corosio::detail::posix_stream_file::file_op::destroy() :407 2x 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/make_err.hpp>
28 #include <boost/capy/ex/executor_ref.hpp>
29 #include <boost/capy/error.hpp>
30 #include <boost/capy/buffers.hpp>
31
32 #include <atomic>
33 #include <coroutine>
34 #include <cstddef>
35 #include <cstdint>
36 #include <filesystem>
37 #include <limits>
38 #include <memory>
39 #include <optional>
40 #include <stop_token>
41 #include <system_error>
42
43 #include <errno.h>
44 #include <fcntl.h>
45 #include <sys/stat.h>
46 #include <sys/uio.h>
47 #include <unistd.h>
48
49 /*
50 POSIX Stream File Implementation
51 =================================
52
53 Regular files cannot be monitored by epoll/kqueue/select — the kernel
54 always reports them as ready. Blocking I/O (pread/pwrite) is dispatched
55 to a shared thread pool, with completion posted back to the scheduler.
56
57 This follows the same pattern as posix_resolver: pool_work_item for
58 dispatch, scheduler_op for completion, shared_from_this for lifetime.
59
60 Completion Flow
61 ---------------
62 1. read_some() sets up file_read_op, posts to thread pool
63 2. Pool thread runs preadv() (blocking)
64 3. Pool thread stores results, posts scheduler_op to scheduler
65 4. Scheduler invokes op() which resumes the coroutine
66
67 Single-Inflight Constraint
68 --------------------------
69 Only one asynchronous operation may be in flight at a time on a
70 given file object. Concurrent read and write is not supported
71 because both share offset_ without synchronization.
72 */
73
74 namespace boost::corosio::detail {
75
76 struct scheduler;
77 class posix_stream_file_service;
78
79 /** Stream file implementation for POSIX backends.
80
81 Each instance contains embedded operation objects (read_op_, write_op_)
82 that are reused across calls. This avoids per-operation heap allocation.
83 */
84 class posix_stream_file final
85 : public stream_file::implementation
86 , public std::enable_shared_from_this<posix_stream_file>
87 , public intrusive_list<posix_stream_file>::node
88 {
89 friend class posix_stream_file_service;
90
91 public:
92 static constexpr std::size_t max_buffers = 16;
93
94 /** Operation state for a single file read or write.
95
96 The coroutine, cancellation and keepalive machinery is inherited
97 from `coro_op`; only the pool-path result state lives here.
98 */
99 struct file_op : coro_op
100 {
101 // Buffer data (copied from buffer_param at submission time)
102 iovec iovecs[max_buffers];
103 250x int iovec_count = 0;
104
105 // Result storage (populated by worker thread)
106 250x int errn = 0;
107 250x std::size_t bytes_transferred = 0;
108
109 750x file_op() = default;
110
111 88x void reset() noexcept
112 {
113 88x iovec_count = 0;
114 88x errn = 0;
115 88x bytes_transferred = 0;
116 88x is_read = false;
117 88x cancelled.store(false, std::memory_order_relaxed);
118 88x stop_cb.reset();
119 88x impl_ptr.reset();
120 88x ec_out = nullptr;
121 88x bytes_out = nullptr;
122 88x }
123
124 void operator()() override;
125 void destroy() override;
126 };
127
128 /** Pool work item for thread pool dispatch. */
129 250x struct pool_op : pool_work_item
130 {
131 250x posix_stream_file* file_ = nullptr;
132 std::shared_ptr<posix_stream_file> ref_;
133 };
134
135 explicit posix_stream_file(posix_stream_file_service& svc) noexcept;
136
137 // -- io_stream::implementation --
138
139 std::coroutine_handle<> read_some(
140 std::coroutine_handle<>,
141 capy::executor_ref,
142 buffer_param,
143 std::stop_token,
144 std::error_code*,
145 std::size_t*) override;
146
147 std::coroutine_handle<> write_some(
148 std::coroutine_handle<>,
149 capy::executor_ref,
150 buffer_param,
151 std::stop_token,
152 std::error_code*,
153 std::size_t*) override;
154
155 // -- stream_file::implementation --
156
157 382x native_handle_type native_handle() const noexcept override
158 {
159 382x return fd_;
160 }
161
162 349x void cancel() noexcept override
163 {
164 349x read_op_.request_cancel();
165 349x write_op_.request_cancel();
166 349x }
167
168 std::uint64_t size() const override;
169 std::error_code resize(std::uint64_t new_size) noexcept override;
170 std::error_code sync_data() noexcept override;
171 std::error_code sync_all() noexcept override;
172 native_handle_type release() override;
173 std::error_code assign(native_handle_type handle) noexcept override;
174 capy::io_result<std::uint64_t>
175 seek(std::int64_t offset, file_base::seek_basis origin) noexcept override;
176
177 // -- Internal --
178
179 /** Open the file and store the fd. */
180 std::error_code
181 open_file(std::filesystem::path const& path, file_base::flags mode);
182
183 /** Close the file descriptor. */
184 void close_file() noexcept;
185
186 private:
187 posix_stream_file_service& svc_;
188 125x int fd_ = -1;
189 125x std::uint64_t offset_ = 0;
190
191 file_op read_op_;
192 file_op write_op_;
193 pool_op read_pool_op_;
194 pool_op write_pool_op_;
195
196 static void do_read_work(pool_work_item*) noexcept;
197 static void do_write_work(pool_work_item*) noexcept;
198 };
199
200 // ---------------------------------------------------------------------------
201 // Inline implementation
202 // ---------------------------------------------------------------------------
203
204 625x inline posix_stream_file::posix_stream_file(
205 posix_stream_file_service& svc) noexcept
206 125x : svc_(svc)
207 500x {
208 250x }
209
210 inline std::error_code
211 105x posix_stream_file::open_file(
212 std::filesystem::path const& path, file_base::flags mode)
213 {
214 105x close_file();
215
216 105x int oflags = 0;
217
218 // Access mode
219 105x unsigned access = static_cast<unsigned>(mode) & 3u;
220
2/2
✓ Branch 0 taken 12 times.
✓ Branch 1 taken 93 times.
 105x if (access == static_cast<unsigned>(file_base::read_write))
221 12x oflags |= O_RDWR;
222
2/2
✓ Branch 0 taken 28 times.
✓ Branch 1 taken 65 times.
 93x else if (access == static_cast<unsigned>(file_base::write_only))
223 28x oflags |= O_WRONLY;
224 else
225 65x oflags |= O_RDONLY;
226
227 // Creation flags
228
2/2
✓ Branch 0 taken 77 times.
✓ Branch 1 taken 28 times.
 105x if ((mode & file_base::create) != file_base::flags(0))
229 28x oflags |= O_CREAT;
230
2/2
✓ Branch 0 taken 103 times.
✓ Branch 1 taken 2 times.
 105x if ((mode & file_base::exclusive) != file_base::flags(0))
231 2x oflags |= O_EXCL;
232
2/2
✓ Branch 0 taken 88 times.
✓ Branch 1 taken 17 times.
 105x if ((mode & file_base::truncate) != file_base::flags(0))
233 17x oflags |= O_TRUNC;
234
2/2
✓ Branch 0 taken 100 times.
✓ Branch 1 taken 5 times.
 105x if ((mode & file_base::append) != file_base::flags(0))
235 5x oflags |= O_APPEND;
236
2/2
✓ Branch 0 taken 103 times.
✓ Branch 1 taken 2 times.
 105x if ((mode & file_base::sync_all_on_write) != file_base::flags(0))
237 2x oflags |= O_SYNC;
238
239 105x int fd = ::open(path.c_str(), oflags, 0666);
240
2/2
✓ Branch 0 taken 99 times.
✓ Branch 1 taken 6 times.
 105x if (fd < 0)
241 6x return make_err(errno);
242
243 99x fd_ = fd;
244 99x offset_ = 0;
245
246 // Append mode: position at end-of-file (preadv/pwritev use
247 // explicit offsets, so O_APPEND alone is not sufficient).
248
2/2
✓ Branch 0 taken 94 times.
✓ Branch 1 taken 5 times.
 99x if ((mode & file_base::append) != file_base::flags(0))
249 {
250 struct stat st;
251
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 2 times.
 5x if (::fstat(fd, &st) < 0)
252 {
253 2x int err = errno;
254 2x ::close(fd);
255 2x fd_ = -1;
256 2x return make_err(err);
257 }
258 3x offset_ = static_cast<std::uint64_t>(st.st_size);
259 3x }
260
261 #ifdef POSIX_FADV_SEQUENTIAL
262 ::posix_fadvise(fd_, 0, 0, POSIX_FADV_SEQUENTIAL);
263 #endif
264
265 97x return {};
266 105x }
267
268 inline void
269 458x posix_stream_file::close_file() noexcept
270 {
271
2/2
✓ Branch 0 taken 357 times.
✓ Branch 1 taken 101 times.
 458x if (fd_ >= 0)
272 {
273
1/2
✓ Branch 0 taken 101 times.
✗ Branch 1 not taken.
 101x ::close(fd_);
274 101x fd_ = -1;
275 101x }
276 458x }
277
278 inline std::uint64_t
279 14x posix_stream_file::size() const
280 {
281 struct stat st;
282
2/2
✓ Branch 0 taken 12 times.
✓ Branch 1 taken 2 times.
 14x if (::fstat(fd_, &st) < 0)
283 2x throw_system_error(make_err(errno), "stream_file::size");
284 12x return static_cast<std::uint64_t>(st.st_size);
285 }
286
287 inline std::error_code
288 9x posix_stream_file::resize(std::uint64_t new_size) noexcept
289 {
290
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 >
291 9x static_cast<std::uint64_t>((std::numeric_limits<off_t>::max)()))
292 2x return make_err(EOVERFLOW);
293
3/4
✓ Branch 0 taken 7 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 4 times.
✓ Branch 3 taken 3 times.
 7x if (::ftruncate(fd_, static_cast<off_t>(new_size)) < 0)
294
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4x return make_err(errno);
295 3x return {};
296 9x }
297
298 inline std::error_code
299 7x posix_stream_file::sync_data() noexcept
300 {
301 #if BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
302 if (::fdatasync(fd_) < 0)
303 #else // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
304
3/4
✓ Branch 0 taken 7 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 4 times.
✓ Branch 3 taken 3 times.
 7x if (::fsync(fd_) < 0)
305 #endif // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
306
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4x return make_err(errno);
307 3x return {};
308 7x }
309
310 inline std::error_code
311 7x posix_stream_file::sync_all() noexcept
312 {
313
3/4
✓ Branch 0 taken 7 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 4 times.
✓ Branch 3 taken 3 times.
 7x if (::fsync(fd_) < 0)
314
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4x return make_err(errno);
315 3x return {};
316 7x }
317
318 inline native_handle_type
319 2x posix_stream_file::release()
320 {
321 2x int fd = fd_;
322 2x fd_ = -1;
323 2x offset_ = 0;
324 2x return fd;
325 }
326
327 inline std::error_code
328 6x posix_stream_file::assign(native_handle_type handle) noexcept
329 {
330 6x close_file();
331 6x fd_ = handle;
332 6x offset_ = 0;
333 6x return {};
334 }
335
336 inline capy::io_result<std::uint64_t>
337 24x posix_stream_file::seek(
338 std::int64_t offset, file_base::seek_basis origin) noexcept
339 {
340 // We track offset_ ourselves (not the kernel fd offset)
341 // because preadv/pwritev use explicit offsets.
342 std::int64_t new_pos;
343
344
2/2
✓ Branch 0 taken 13 times.
✓ Branch 1 taken 11 times.
 24x if (origin == file_base::seek_set)
345 {
346 11x new_pos = offset;
347 11x }
348
2/2
✓ Branch 0 taken 5 times.
✓ Branch 1 taken 8 times.
 13x else if (origin == file_base::seek_cur)
349 {
350 5x new_pos = static_cast<std::int64_t>(offset_) + offset;
351 5x }
352 else
353 {
354 struct stat st;
355
3/4
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✓ Branch 3 taken 6 times.
 8x if (::fstat(fd_, &st) < 0)
356
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2x return {make_err(errno), 0};
357 6x new_pos = st.st_size + offset;
358 }
359
360
2/2
✓ Branch 0 taken 16 times.
✓ Branch 1 taken 6 times.
 22x if (new_pos < 0)
361 6x return {make_err(EINVAL), 0};
362
2/4
✗ Branch 0 not taken.
✓ Branch 1 taken 16 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 16 times.
 32x if (new_pos >
363 16x static_cast<std::int64_t>((std::numeric_limits<off_t>::max)()))
364 ✗ return {make_err(EOVERFLOW), 0};
365
366 16x offset_ = static_cast<std::uint64_t>(new_pos);
367
368 16x return {std::error_code{}, offset_};
369 24x }
370
371 // -- file_op completion handler --
372 // (read_some, write_some, do_read_work, do_write_work are
373 // defined in posix_stream_file_service.hpp after the service)
374
375 inline void
376 61x posix_stream_file::file_op::operator()()
377 {
378 61x stop_cb.reset();
379
380 61x bool const was_cancelled = cancelled.load(std::memory_order_acquire);
381
382
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 61 times.
 61x if (ec_out)
383 {
384
2/2
✓ Branch 0 taken 6 times.
✓ Branch 1 taken 55 times.
 61x if (was_cancelled)
385 6x *ec_out = capy::error::canceled;
386
2/2
✓ Branch 0 taken 8 times.
✓ Branch 1 taken 47 times.
 55x else if (errn != 0)
387 8x *ec_out = make_err(errn);
388
4/4
✓ Branch 0 taken 28 times.
✓ Branch 1 taken 19 times.
✓ Branch 2 taken 23 times.
✓ Branch 3 taken 5 times.
 47x else if (is_read && bytes_transferred == 0)
389 5x *ec_out = capy::error::eof;
390 else
391 42x *ec_out = {};
392 61x }
393
394
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 61 times.
 61x if (bytes_out)
395
2/2
✓ Branch 0 taken 6 times.
✓ Branch 1 taken 55 times.
 61x *bytes_out = was_cancelled ? 0 : bytes_transferred;
396
397 // Move impl_ptr to a local so members remain valid through
398 // dispatch — impl_ptr may be the last shared_ptr keeping
399 // the parent posix_stream_file (which embeds this file_op) alive.
400 61x auto prevent_destroy = std::move(impl_ptr);
401 61x ex.on_work_finished();
402 61x cont.h = h;
403
2/4
✓ Branch 0 taken 61 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 61 times.
✗ Branch 3 not taken.
 61x dispatch_coro(ex, cont).resume();
404 61x }
405
406 inline void
407 2x posix_stream_file::file_op::destroy()
408 {
409 2x stop_cb.reset();
410 2x auto local_ex = ex;
411 2x impl_ptr.reset();
412 2x local_ex.on_work_finished();
413 2x }
414
415 } // namespace boost::corosio::detail
416
417 #endif // BOOST_COROSIO_POSIX
418
419 #endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_HPP
420