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