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/make_err.hpp>
27 : #include <boost/capy/ex/executor_ref.hpp>
28 : #include <boost/capy/error.hpp>
29 : #include <boost/capy/buffers.hpp>
30 :
31 : #include <atomic>
32 : #include <coroutine>
33 : #include <cstddef>
34 : #include <cstdint>
35 : #include <filesystem>
36 : #include <limits>
37 : #include <memory>
38 : #include <mutex>
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 Random-Access File Implementation
51 : ========================================
52 :
53 : Each async read/write heap-allocates an raf_op that serves
54 : as both the thread-pool work item and the scheduler completion
55 : op. This allows unlimited concurrent operations on the same
56 : file object, matching Asio's per-op allocation model.
57 :
58 : The raf_op self-deletes on completion or shutdown.
59 : */
60 :
61 : namespace boost::corosio::detail {
62 :
63 : struct scheduler;
64 : class posix_random_access_file_service;
65 :
66 : /** Random-access file implementation for POSIX backends. */
67 : class posix_random_access_file final
68 : : public random_access_file::implementation
69 : , public std::enable_shared_from_this<posix_random_access_file>
70 : , public intrusive_list<posix_random_access_file>::node
71 : {
72 : friend class posix_random_access_file_service;
73 :
74 : public:
75 : static constexpr std::size_t max_buffers = 16;
76 :
77 : /** Per-operation state, heap-allocated for each async call.
78 :
79 : Inherits from `coro_op` (for scheduler completion plus the shared
80 : coroutine, cancellation and keepalive machinery) and
81 : `pool_work_item` (for thread-pool dispatch). Linked into the
82 : file's outstanding_ops_ list for cancellation tracking. `coro_op`
83 : leads the base list so a `scheduler_op*` round-trips.
84 : */
85 : struct raf_op final
86 : : coro_op
87 : , pool_work_item
88 : , intrusive_list<raf_op>::node
89 : {
90 : iovec iovecs[max_buffers];
91 : int iovec_count = 0;
92 : std::uint64_t offset = 0;
93 :
94 : int errn = 0;
95 : std::size_t bytes_transferred = 0;
96 :
97 : // Raw back-pointer for the typed work; `impl_ptr` is the keepalive.
98 : posix_random_access_file* file_ = nullptr;
99 :
100 : void operator()() override;
101 : void destroy() override;
102 :
103 : /// Thread-pool work function: executes preadv/pwritev.
104 : static void do_work(pool_work_item*) noexcept;
105 : };
106 :
107 : explicit posix_random_access_file(
108 : posix_random_access_file_service& svc) noexcept;
109 :
110 : // -- random_access_file::implementation --
111 :
112 : std::coroutine_handle<> read_some_at(
113 : std::uint64_t offset,
114 : std::coroutine_handle<>,
115 : capy::executor_ref,
116 : buffer_param,
117 : std::stop_token,
118 : std::error_code*,
119 : std::size_t*) override;
120 :
121 : std::coroutine_handle<> write_some_at(
122 : std::uint64_t offset,
123 : std::coroutine_handle<>,
124 : capy::executor_ref,
125 : buffer_param,
126 : std::stop_token,
127 : std::error_code*,
128 : std::size_t*) override;
129 :
130 HIT 658 : native_handle_type native_handle() const noexcept override
131 : {
132 658 : return fd_;
133 : }
134 :
135 296 : void cancel() noexcept override
136 : {
137 296 : std::lock_guard<std::mutex> lock(ops_mutex_);
138 296 : outstanding_ops_.for_each([](raf_op* op) {
139 6 : op->cancelled.store(true, std::memory_order_release);
140 6 : });
141 296 : }
142 :
143 : std::uint64_t size() const override;
144 : std::error_code resize(std::uint64_t new_size) noexcept override;
145 : std::error_code sync_data() noexcept override;
146 : std::error_code sync_all() noexcept override;
147 : native_handle_type release() override;
148 : std::error_code assign(native_handle_type handle) noexcept override;
149 :
150 : std::error_code open_file(
151 : std::filesystem::path const& path, file_base::flags mode);
152 : void close_file() noexcept;
153 :
154 : private:
155 : posix_random_access_file_service& svc_;
156 : int fd_ = -1;
157 : std::mutex ops_mutex_;
158 : intrusive_list<raf_op> outstanding_ops_;
159 : };
160 :
161 : // ---------------------------------------------------------------------------
162 : // Inline implementation
163 : // ---------------------------------------------------------------------------
164 :
165 : inline
166 105 : posix_random_access_file::posix_random_access_file(
167 105 : posix_random_access_file_service& svc) noexcept
168 105 : : svc_(svc)
169 : {
170 105 : }
171 :
172 : inline std::error_code
173 91 : posix_random_access_file::open_file(
174 : std::filesystem::path const& path, file_base::flags mode)
175 : {
176 91 : close_file();
177 :
178 91 : int oflags = 0;
179 :
180 91 : unsigned access = static_cast<unsigned>(mode) & 3u;
181 91 : if (access == static_cast<unsigned>(file_base::read_write))
182 27 : oflags |= O_RDWR;
183 64 : else if (access == static_cast<unsigned>(file_base::write_only))
184 12 : oflags |= O_WRONLY;
185 : else
186 52 : oflags |= O_RDONLY;
187 :
188 91 : if ((mode & file_base::create) != file_base::flags(0))
189 28 : oflags |= O_CREAT;
190 91 : if ((mode & file_base::exclusive) != file_base::flags(0))
191 4 : oflags |= O_EXCL;
192 91 : if ((mode & file_base::truncate) != file_base::flags(0))
193 14 : oflags |= O_TRUNC;
194 91 : 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 91 : int fd = ::open(path.c_str(), oflags, 0666);
199 91 : if (fd < 0)
200 9 : return make_err(errno);
201 :
202 82 : fd_ = fd;
203 :
204 : #ifdef POSIX_FADV_RANDOM
205 82 : ::posix_fadvise(fd_, 0, 0, POSIX_FADV_RANDOM);
206 : #endif
207 :
208 82 : return {};
209 : }
210 :
211 : inline void
212 390 : posix_random_access_file::close_file() noexcept
213 : {
214 390 : if (fd_ >= 0)
215 : {
216 86 : ::close(fd_);
217 86 : fd_ = -1;
218 : }
219 390 : }
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 322 : posix_random_access_file::raf_op::operator()()
284 : {
285 322 : stop_cb.reset();
286 :
287 322 : bool const was_cancelled = cancelled.load(std::memory_order_acquire);
288 :
289 322 : if (ec_out)
290 : {
291 322 : if (was_cancelled)
292 2 : *ec_out = capy::error::canceled;
293 320 : else if (errn != 0)
294 16 : *ec_out = make_err(errn);
295 304 : else if (is_read && bytes_transferred == 0)
296 9 : *ec_out = capy::error::eof;
297 : else
298 295 : *ec_out = {};
299 : }
300 :
301 322 : if (bytes_out)
302 322 : *bytes_out = was_cancelled ? 0 : bytes_transferred;
303 :
304 : {
305 322 : std::lock_guard<std::mutex> lock(file_->ops_mutex_);
306 322 : file_->outstanding_ops_.remove(this);
307 322 : }
308 :
309 322 : impl_ptr.reset();
310 :
311 322 : auto coro = h;
312 322 : ex.on_work_finished();
313 322 : delete this;
314 322 : coro.resume();
315 322 : }
316 :
317 : // -- raf_op shutdown cleanup --
318 :
319 : inline void
320 2 : posix_random_access_file::raf_op::destroy()
321 : {
322 2 : stop_cb.reset();
323 : {
324 2 : std::lock_guard<std::mutex> lock(file_->ops_mutex_);
325 2 : file_->outstanding_ops_.remove(this);
326 2 : }
327 2 : impl_ptr.reset();
328 2 : ex.on_work_finished();
329 2 : delete this;
330 2 : }
331 :
332 : } // namespace boost::corosio::detail
333 :
334 : #endif // BOOST_COROSIO_POSIX
335 :
336 : #endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_FILE_HPP
|
