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/native/detail/coro_op.hpp>
26 : #include <boost/corosio/native/detail/coro_op_complete.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 <mutex>
40 : #include <optional>
41 : #include <stop_token>
42 : #include <system_error>
43 :
44 : #include <errno.h>
45 : #include <fcntl.h>
46 : #include <sys/stat.h>
47 : #include <sys/uio.h>
48 : #include <unistd.h>
49 :
50 : /*
51 : POSIX Random-Access File Implementation
52 : ========================================
53 :
54 : Each async read/write heap-allocates an raf_op that serves
55 : as both the thread-pool work item and the scheduler completion
56 : op. This allows unlimited concurrent operations on the same
57 : file object, matching Asio's per-op allocation model.
58 :
59 : The raf_op self-deletes on completion or shutdown.
60 : */
61 :
62 : namespace boost::corosio::detail {
63 :
64 : struct scheduler;
65 : class posix_random_access_file_service;
66 :
67 : /** Random-access file implementation for POSIX backends. */
68 : class posix_random_access_file final
69 : : public random_access_file::implementation
70 : , public std::enable_shared_from_this<posix_random_access_file>
71 : , public intrusive_list<posix_random_access_file>::node
72 : {
73 : friend class posix_random_access_file_service;
74 :
75 : public:
76 : static constexpr std::size_t max_buffers = 16;
77 :
78 : /** Per-operation state, heap-allocated for each async call.
79 :
80 : Inherits from `coro_op` (for scheduler completion plus the shared
81 : coroutine, cancellation and keepalive machinery) and
82 : `pool_work_item` (for thread-pool dispatch). Linked into the
83 : file's outstanding_ops_ list for cancellation tracking. `coro_op`
84 : leads the base list so a `scheduler_op*` round-trips.
85 : */
86 : struct raf_op final
87 : : coro_op
88 : , pool_work_item
89 : , intrusive_list<raf_op>::node
90 : {
91 : iovec iovecs[max_buffers];
92 : int iovec_count = 0;
93 : std::uint64_t offset = 0;
94 :
95 : int errn = 0;
96 : std::size_t bytes_transferred = 0;
97 :
98 : // Raw back-pointer for the typed work; `impl_ptr` is the keepalive.
99 : posix_random_access_file* file_ = nullptr;
100 :
101 : void operator()() override;
102 : void destroy() override;
103 :
104 : /// Thread-pool work function: executes preadv/pwritev.
105 : static void do_work(pool_work_item*) noexcept;
106 : };
107 :
108 : explicit posix_random_access_file(
109 : posix_random_access_file_service& svc) noexcept;
110 :
111 : // -- random_access_file::implementation --
112 :
113 : std::coroutine_handle<> read_some_at(
114 : std::uint64_t offset,
115 : std::coroutine_handle<>,
116 : capy::executor_ref,
117 : buffer_param,
118 : std::stop_token,
119 : std::error_code*,
120 : std::size_t*) override;
121 :
122 : std::coroutine_handle<> write_some_at(
123 : std::uint64_t offset,
124 : std::coroutine_handle<>,
125 : capy::executor_ref,
126 : buffer_param,
127 : std::stop_token,
128 : std::error_code*,
129 : std::size_t*) override;
130 :
131 HIT 1082 : native_handle_type native_handle() const noexcept override
132 : {
133 1082 : return fd_;
134 : }
135 :
136 620 : void cancel() noexcept override
137 : {
138 620 : std::lock_guard<std::mutex> lock(ops_mutex_);
139 620 : outstanding_ops_.for_each([](raf_op* op) {
140 6 : op->cancelled.store(true, std::memory_order_release);
141 6 : });
142 620 : }
143 :
144 : std::uint64_t size() const override;
145 : std::error_code resize(std::uint64_t new_size) noexcept override;
146 : std::error_code sync_data() noexcept override;
147 : std::error_code sync_all() noexcept override;
148 : native_handle_type release() override;
149 : std::error_code assign(native_handle_type handle) noexcept override;
150 :
151 : std::error_code
152 : open_file(std::filesystem::path const& path, file_base::flags mode);
153 : void close_file() noexcept;
154 :
155 : private:
156 : posix_random_access_file_service& svc_;
157 : int fd_ = -1;
158 : std::mutex ops_mutex_;
159 : intrusive_list<raf_op> outstanding_ops_;
160 : };
161 :
162 : // ---------------------------------------------------------------------------
163 : // Inline implementation
164 : // ---------------------------------------------------------------------------
165 :
166 213 : inline posix_random_access_file::posix_random_access_file(
167 213 : posix_random_access_file_service& svc) noexcept
168 213 : : svc_(svc)
169 : {
170 213 : }
171 :
172 : inline std::error_code
173 199 : posix_random_access_file::open_file(
174 : std::filesystem::path const& path, file_base::flags mode)
175 : {
176 199 : close_file();
177 :
178 199 : int oflags = 0;
179 :
180 199 : unsigned access = static_cast<unsigned>(mode) & 3u;
181 199 : if (access == static_cast<unsigned>(file_base::read_write))
182 31 : oflags |= O_RDWR;
183 168 : else if (access == static_cast<unsigned>(file_base::write_only))
184 64 : oflags |= O_WRONLY;
185 : else
186 104 : oflags |= O_RDONLY;
187 :
188 199 : if ((mode & file_base::create) != file_base::flags(0))
189 28 : oflags |= O_CREAT;
190 199 : if ((mode & file_base::exclusive) != file_base::flags(0))
191 4 : oflags |= O_EXCL;
192 199 : if ((mode & file_base::truncate) != file_base::flags(0))
193 14 : oflags |= O_TRUNC;
194 199 : if ((mode & file_base::sync_all_on_write) != file_base::flags(0))
195 2 : oflags |= O_SYNC;
196 : // Note: no O_APPEND for random access files
197 :
198 199 : int fd = ::open(path.c_str(), oflags, 0666);
199 199 : if (fd < 0)
200 9 : return make_err(errno);
201 :
202 190 : fd_ = fd;
203 :
204 : #ifdef POSIX_FADV_RANDOM
205 190 : ::posix_fadvise(fd_, 0, 0, POSIX_FADV_RANDOM);
206 : #endif
207 :
208 190 : return {};
209 : }
210 :
211 : inline void
212 822 : posix_random_access_file::close_file() noexcept
213 : {
214 822 : if (fd_ >= 0)
215 : {
216 194 : ::close(fd_);
217 194 : fd_ = -1;
218 : }
219 822 : }
220 :
221 : inline std::uint64_t
222 13 : posix_random_access_file::size() const
223 : {
224 : struct stat st;
225 13 : if (::fstat(fd_, &st) < 0)
226 5 : throw_system_error(make_err(errno), "random_access_file::size");
227 8 : return static_cast<std::uint64_t>(st.st_size);
228 : }
229 :
230 : inline std::error_code
231 13 : posix_random_access_file::resize(std::uint64_t new_size) noexcept
232 : {
233 13 : if (new_size >
234 13 : static_cast<std::uint64_t>((std::numeric_limits<off_t>::max)()))
235 2 : return make_err(EOVERFLOW);
236 11 : if (::ftruncate(fd_, static_cast<off_t>(new_size)) < 0)
237 7 : return make_err(errno);
238 4 : return {};
239 : }
240 :
241 : inline std::error_code
242 9 : posix_random_access_file::sync_data() noexcept
243 : {
244 : #if BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
245 9 : if (::fdatasync(fd_) < 0)
246 : #else // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
247 : if (::fsync(fd_) < 0)
248 : #endif // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
249 7 : return make_err(errno);
250 2 : return {};
251 : }
252 :
253 : inline std::error_code
254 9 : posix_random_access_file::sync_all() noexcept
255 : {
256 9 : if (::fsync(fd_) < 0)
257 7 : return make_err(errno);
258 2 : return {};
259 : }
260 :
261 : inline native_handle_type
262 3 : posix_random_access_file::release()
263 : {
264 3 : int fd = fd_;
265 3 : fd_ = -1;
266 3 : return fd;
267 : }
268 :
269 : inline std::error_code
270 7 : posix_random_access_file::assign(native_handle_type handle) noexcept
271 : {
272 7 : close_file();
273 7 : fd_ = handle;
274 7 : return {};
275 : }
276 :
277 : // read_some_at, write_some_at are defined in
278 : // posix_random_access_file_service.hpp after the service.
279 :
280 : // -- raf_op completion handler (scheduler thread) --
281 :
282 : inline void
283 420 : posix_random_access_file::raf_op::operator()()
284 : {
285 420 : stop_cb.reset();
286 :
287 : // Empty buffers never reach the pool (diverted at initiation), so
288 : // empty_buffer stays false and a 0-byte read is a genuine EOF.
289 773 : decode_io_result(
290 420 : ec_out, bytes_out, cancelled.load(std::memory_order_acquire),
291 420 : errn != 0 ? make_err(errn) : std::error_code{}, is_read,
292 : bytes_transferred, /*empty_buffer=*/false);
293 :
294 : {
295 420 : std::lock_guard<std::mutex> lock(file_->ops_mutex_);
296 420 : file_->outstanding_ops_.remove(this);
297 420 : }
298 :
299 420 : impl_ptr.reset();
300 :
301 420 : auto coro = h;
302 420 : ex.on_work_finished();
303 420 : delete this;
304 420 : coro.resume();
305 420 : }
306 :
307 : // -- raf_op shutdown cleanup --
308 :
309 : inline void
310 6 : posix_random_access_file::raf_op::destroy()
311 : {
312 6 : stop_cb.reset();
313 : {
314 6 : std::lock_guard<std::mutex> lock(file_->ops_mutex_);
315 6 : file_->outstanding_ops_.remove(this);
316 6 : }
317 6 : impl_ptr.reset();
318 6 : ex.on_work_finished();
319 6 : delete this;
320 6 : }
321 :
322 : } // namespace boost::corosio::detail
323 :
324 : #endif // BOOST_COROSIO_POSIX
325 :
326 : #endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_FILE_HPP
|
