TLA Line data Source code
1 : //
2 : // Copyright (c) 2026 Michael Vandeberg
3 : //
4 : // Distributed under the Boost Software License, Version 1.0. (See accompanying
5 : // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6 : //
7 : // Official repository: https://github.com/cppalliance/corosio
8 : //
9 :
10 : #ifndef BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_FILE_HPP
11 : #define BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_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/random_access_file.hpp>
19 : #include <boost/corosio/file_base.hpp>
20 : #include <boost/corosio/detail/intrusive.hpp>
21 : #include <boost/corosio/detail/scheduler_op.hpp>
22 : #include <boost/corosio/detail/thread_pool.hpp>
23 : #include <boost/corosio/detail/scheduler.hpp>
24 : #include <boost/corosio/detail/buffer_param.hpp>
25 : #include <boost/corosio/detail/dispatch_coro.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/ex/abandon.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 <mutex>
44 : #include <optional>
45 : #include <stop_token>
46 : #include <system_error>
47 : #include <utility>
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 Random-Access File Implementation
57 : ========================================
58 :
59 : Each async read/write acquires an raf_op that serves as both the
60 : thread-pool work item and the scheduler completion op. This
61 : allows unlimited concurrent operations on the same file object.
62 :
63 : Ops recycle through the file's free_ops_ list on completion or
64 : shutdown, so steady-state reads and writes allocate nothing; the
65 : file destructor frees the recycled storage.
66 : */
67 :
68 : namespace boost::corosio::detail {
69 :
70 : class scheduler;
71 : class posix_random_access_file_service;
72 :
73 : /** Random-access file implementation for POSIX backends. */
74 : class posix_random_access_file final
75 : : public random_access_file::implementation
76 : , public intrusive_list<posix_random_access_file>::node
77 : {
78 : friend class posix_random_access_file_service;
79 :
80 : public:
81 : static constexpr std::size_t max_buffers = 16;
82 :
83 : /** Per-operation state, acquired from the file's free list.
84 :
85 : Inherits from `coro_op` (for scheduler completion plus the shared
86 : coroutine, cancellation and keepalive machinery) and
87 : `pool_work_item` (for thread-pool dispatch). The intrusive hook
88 : links it into `outstanding_ops_` while in flight and `free_ops_`
89 : once recycled — membership is strictly sequential, so one hook
90 : serves both under `ops_mutex_`. `coro_op` leads the base list so
91 : a `scheduler_op*` round-trips.
92 : */
93 : struct raf_op final
94 : : coro_op
95 : , pool_work_item
96 : , intrusive_list<raf_op>::node
97 : {
98 : iovec iovecs[max_buffers];
99 : int iovec_count = 0;
100 : std::uint64_t offset = 0;
101 : // Snapshotted at submission: assign() may replace fd_ while the
102 : // worker runs.
103 : int fd = -1;
104 :
105 : int errn = 0;
106 : std::size_t bytes_transferred = 0;
107 :
108 : // Raw back-pointer for the typed work; `object_ref_` is the keepalive.
109 : posix_random_access_file* file_ = nullptr;
110 :
111 : void operator()() override;
112 : void destroy() override;
113 :
114 : /// Recycle an op the pool refused, leaving its coroutine to
115 : /// the initiator that answers it.
116 : void discard() noexcept;
117 :
118 : /// Thread-pool work function: executes preadv/pwritev.
119 : static void do_work(pool_work_item*) noexcept;
120 :
121 : /// Destroy an op the pool shut down before running.
122 : static void destroy_work(pool_work_item*) noexcept;
123 :
124 : /// Return this op to its file's free list.
125 : void recycle() noexcept;
126 : };
127 :
128 : /// Release the work an op holds on its service's context.
129 : static void
130 : release_work(posix_random_access_file_service& svc) noexcept;
131 :
132 : explicit posix_random_access_file(
133 : posix_random_access_file_service& svc) noexcept;
134 :
135 : /// Recycle into the owning service's pool. Defined out-of-line
136 : /// after posix_random_access_file_service for its complete type.
137 : void retire() noexcept override;
138 :
139 : /** Reset for recycling.
140 :
141 : `close_file()` already drove fd_ to its closed value before
142 : the refcount reached zero. Every `raf_op` holds its own
143 : `object_ref` while in flight, so the refcount cannot reach
144 : zero while one is outstanding — asserting `outstanding_ops_`
145 : is empty is therefore a precondition check, not a defensive
146 : one. `free_ops_` deliberately survives recycling: the storage
147 : belongs to this impl and only the destructor frees it.
148 :
149 : @pre refs_ == 0, fd closed, no op in flight.
150 : */
151 MIS 0 : void reuse() noexcept
152 : {
153 0 : BOOST_COROSIO_ASSERT(fd_ == -1);
154 0 : BOOST_COROSIO_ASSERT(gate_.idle());
155 0 : BOOST_COROSIO_ASSERT(outstanding_ops_.empty());
156 0 : }
157 :
158 HIT 488 : ~posix_random_access_file() override
159 244 : {
160 664 : while (auto* op = free_ops_.pop_front())
161 420 : delete op;
162 488 : }
163 :
164 : // -- random_access_file::implementation --
165 :
166 : std::coroutine_handle<> read_some_at(
167 : std::uint64_t offset,
168 : capy::continuation&,
169 : capy::executor_ref,
170 : buffer_param,
171 : std::stop_token,
172 : std::error_code*,
173 : std::size_t*) override;
174 :
175 : std::coroutine_handle<> write_some_at(
176 : std::uint64_t offset,
177 : capy::continuation&,
178 : capy::executor_ref,
179 : buffer_param,
180 : std::stop_token,
181 : std::error_code*,
182 : std::size_t*) override;
183 :
184 1246 : native_handle_type native_handle() const noexcept override
185 : {
186 1246 : return fd_;
187 : }
188 :
189 714 : void cancel() noexcept override
190 : {
191 714 : std::lock_guard<std::mutex> lock(ops_mutex_);
192 714 : outstanding_ops_.for_each([](raf_op* op) {
193 23 : op->cancelled.store(true, std::memory_order_release);
194 23 : });
195 714 : }
196 :
197 : std::uint64_t size() const override;
198 : std::error_code resize(std::uint64_t new_size) noexcept override;
199 : std::error_code sync_data() noexcept override;
200 : std::error_code sync_all() noexcept override;
201 : native_handle_type release() override;
202 : std::error_code assign(native_handle_type handle) noexcept override;
203 :
204 : std::error_code
205 : open_file(std::filesystem::path const& path, file_base::flags mode);
206 : void close_file() noexcept;
207 :
208 : private:
209 : /** Pop a recycled op, or allocate on the cold path.
210 :
211 : Per-use fields are filled by the caller; the fields `prepare`-
212 : style reuse must not inherit from the previous run (`errn`,
213 : `bytes_transferred`) are reset here. `start()` resets the
214 : cancellation machinery.
215 : */
216 515 : raf_op* acquire_op()
217 : {
218 : raf_op* op;
219 : {
220 515 : std::lock_guard<std::mutex> lock(ops_mutex_);
221 515 : op = free_ops_.pop_front();
222 515 : }
223 515 : if (!op)
224 420 : op = new raf_op();
225 515 : op->errn = 0;
226 515 : op->bytes_transferred = 0;
227 515 : return op;
228 : }
229 :
230 : posix_random_access_file_service& svc_;
231 : int fd_ = -1;
232 : /// Keeps fd_'s number allocated while a pool worker uses it.
233 : posix_fd_gate gate_;
234 : std::mutex ops_mutex_;
235 : intrusive_list<raf_op> outstanding_ops_;
236 : intrusive_list<raf_op> free_ops_;
237 : };
238 :
239 : // ---------------------------------------------------------------------------
240 : // Inline implementation
241 : // ---------------------------------------------------------------------------
242 :
243 244 : inline posix_random_access_file::posix_random_access_file(
244 244 : posix_random_access_file_service& svc) noexcept
245 244 : : svc_(svc)
246 : {
247 244 : }
248 :
249 : inline std::error_code
250 224 : posix_random_access_file::open_file(
251 : std::filesystem::path const& path, file_base::flags mode)
252 : {
253 224 : close_file();
254 :
255 224 : int oflags = 0;
256 :
257 224 : unsigned access = static_cast<unsigned>(mode) & 3u;
258 224 : if (access == static_cast<unsigned>(file_base::read_write))
259 37 : oflags |= O_RDWR;
260 187 : else if (access == static_cast<unsigned>(file_base::write_only))
261 69 : oflags |= O_WRONLY;
262 : else
263 118 : oflags |= O_RDONLY;
264 :
265 224 : if ((mode & file_base::create) != file_base::flags(0))
266 28 : oflags |= O_CREAT;
267 224 : if ((mode & file_base::exclusive) != file_base::flags(0))
268 4 : oflags |= O_EXCL;
269 224 : if ((mode & file_base::truncate) != file_base::flags(0))
270 14 : oflags |= O_TRUNC;
271 224 : if ((mode & file_base::sync_all_on_write) != file_base::flags(0))
272 2 : oflags |= O_SYNC;
273 : // Note: no O_APPEND for random access files
274 :
275 224 : int fd = ::open(path.c_str(), oflags | large_file_open_flag, 0666);
276 224 : if (fd < 0)
277 9 : return make_err(errno);
278 :
279 215 : fd_ = fd;
280 215 : gate_.open(fd);
281 :
282 : #ifdef POSIX_FADV_RANDOM
283 215 : ::posix_fadvise(fd_, 0, 0, POSIX_FADV_RANDOM);
284 : #endif
285 :
286 215 : return {};
287 : }
288 :
289 : inline void
290 929 : posix_random_access_file::close_file() noexcept
291 : {
292 929 : gate_.close();
293 929 : fd_ = -1;
294 929 : }
295 :
296 : inline std::uint64_t
297 11 : posix_random_access_file::size() const
298 : {
299 : file_stat_t st;
300 11 : if (file_fstat(fd_, &st) < 0)
301 5 : throw_system_error(make_err(errno), "random_access_file::size");
302 6 : return static_cast<std::uint64_t>(st.st_size);
303 : }
304 :
305 : inline std::error_code
306 13 : posix_random_access_file::resize(std::uint64_t new_size) noexcept
307 : {
308 13 : if (new_size >
309 13 : static_cast<std::uint64_t>((std::numeric_limits<file_off_t>::max)()))
310 2 : return make_err(EOVERFLOW);
311 11 : if (file_ftruncate(fd_, static_cast<file_off_t>(new_size)) < 0)
312 7 : return make_err(errno);
313 4 : return {};
314 : }
315 :
316 : inline std::error_code
317 7 : posix_random_access_file::sync_data() noexcept
318 : {
319 : #if BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
320 7 : if (::fdatasync(fd_) < 0)
321 : #else // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
322 : if (::fsync(fd_) < 0)
323 : #endif // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
324 5 : return make_err(errno);
325 2 : return {};
326 : }
327 :
328 : inline std::error_code
329 7 : posix_random_access_file::sync_all() noexcept
330 : {
331 7 : if (::fsync(fd_) < 0)
332 5 : return make_err(errno);
333 2 : return {};
334 : }
335 :
336 : inline native_handle_type
337 5 : posix_random_access_file::release()
338 : {
339 : // A queued op has already copied the fd number; it must not run
340 : // after the caller closes it and the number is recycled.
341 5 : cancel();
342 5 : if (!gate_.release())
343 MIS 0 : throw_system_error(
344 0 : std::make_error_code(std::errc::device_or_resource_busy),
345 : "random_access_file::release");
346 HIT 5 : int fd = fd_;
347 5 : fd_ = -1;
348 5 : return fd;
349 : }
350 :
351 : // read_some_at, write_some_at are defined in
352 : // posix_random_access_file_service.hpp after the service.
353 :
354 : // -- raf_op completion handler (scheduler thread) --
355 :
356 : inline void
357 505 : posix_random_access_file::raf_op::operator()()
358 : {
359 505 : stop_cb.reset();
360 :
361 : // Empty buffers never reach the pool (diverted at initiation), so
362 : // empty_buffer stays false and a 0-byte read is a genuine EOF.
363 926 : decode_results(
364 505 : ec_out, bytes_out, cancelled.load(std::memory_order_acquire),
365 505 : errn != 0 ? make_err(errn) : std::error_code{}, is_read,
366 : bytes_transferred, /*empty_buffer=*/false);
367 :
368 : // Copy out everything needed after recycling: once this op is on
369 : // free_ops_ a concurrent initiation may pop and refill it. The
370 : // keepalive drops after the push so a final release never runs
371 : // under ops_mutex_.
372 505 : auto keep = std::move(object_ref_);
373 505 : auto* c = cont;
374 505 : auto exec = ex;
375 505 : auto& svc = file_->svc_;
376 : {
377 505 : std::lock_guard<std::mutex> lock(file_->ops_mutex_);
378 505 : file_->outstanding_ops_.remove(this);
379 505 : file_->free_ops_.push_front(this);
380 505 : }
381 505 : keep.reset();
382 505 : release_work(svc);
383 505 : dispatch_coro(exec, *c).resume();
384 505 : }
385 :
386 : // -- raf_op shutdown cleanup --
387 :
388 : inline void
389 10 : posix_random_access_file::raf_op::recycle() noexcept
390 : {
391 10 : std::lock_guard<std::mutex> lock(file_->ops_mutex_);
392 10 : file_->outstanding_ops_.remove(this);
393 10 : file_->free_ops_.push_front(this);
394 10 : }
395 :
396 : inline void
397 6 : posix_random_access_file::raf_op::destroy()
398 : {
399 6 : stop_cb.reset();
400 6 : auto keep = std::move(object_ref_);
401 6 : auto& svc = file_->svc_;
402 6 : auto* c = std::exchange(cont, nullptr);
403 6 : recycle();
404 : // The frame may own the file, so it goes while the keepalive
405 : // still holds the implementation the op was recycled into.
406 6 : if (c && c->h)
407 6 : capy::abandon(*c);
408 6 : keep.reset();
409 6 : release_work(svc);
410 6 : }
411 :
412 : inline void
413 4 : posix_random_access_file::raf_op::discard() noexcept
414 : {
415 4 : stop_cb.reset();
416 4 : auto keep = std::move(object_ref_);
417 4 : auto& svc = file_->svc_;
418 4 : recycle();
419 4 : keep.reset();
420 4 : release_work(svc);
421 4 : }
422 :
423 : inline void
424 6 : posix_random_access_file::raf_op::destroy_work(pool_work_item* w) noexcept
425 : {
426 6 : static_cast<raf_op*>(w)->destroy();
427 6 : }
428 :
429 : } // namespace boost::corosio::detail
430 :
431 : #endif // BOOST_COROSIO_POSIX
432 :
433 : #endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_FILE_HPP
|
