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