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