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 682 : native_handle_type native_handle() const noexcept override
131 : {
132 682 : return fd_;
133 : }
134 :
135 320 : void cancel() noexcept override
136 : {
137 320 : std::lock_guard<std::mutex> lock(ops_mutex_);
138 320 : outstanding_ops_.for_each([](raf_op* op) {
139 6 : op->cancelled.store(true, std::memory_order_release);
140 6 : });
141 320 : }
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
151 : open_file(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 113 : inline posix_random_access_file::posix_random_access_file(
166 113 : posix_random_access_file_service& svc) noexcept
167 113 : : svc_(svc)
168 : {
169 113 : }
170 :
171 : inline std::error_code
172 99 : posix_random_access_file::open_file(
173 : std::filesystem::path const& path, file_base::flags mode)
174 : {
175 99 : close_file();
176 :
177 99 : int oflags = 0;
178 :
179 99 : unsigned access = static_cast<unsigned>(mode) & 3u;
180 99 : if (access == static_cast<unsigned>(file_base::read_write))
181 31 : oflags |= O_RDWR;
182 68 : else if (access == static_cast<unsigned>(file_base::write_only))
183 14 : oflags |= O_WRONLY;
184 : else
185 54 : oflags |= O_RDONLY;
186 :
187 99 : if ((mode & file_base::create) != file_base::flags(0))
188 28 : oflags |= O_CREAT;
189 99 : if ((mode & file_base::exclusive) != file_base::flags(0))
190 4 : oflags |= O_EXCL;
191 99 : if ((mode & file_base::truncate) != file_base::flags(0))
192 14 : oflags |= O_TRUNC;
193 99 : if ((mode & file_base::sync_all_on_write) != file_base::flags(0))
194 2 : oflags |= O_SYNC;
195 : // Note: no O_APPEND for random access files
196 :
197 99 : int fd = ::open(path.c_str(), oflags, 0666);
198 99 : if (fd < 0)
199 9 : return make_err(errno);
200 :
201 90 : fd_ = fd;
202 :
203 : #ifdef POSIX_FADV_RANDOM
204 90 : ::posix_fadvise(fd_, 0, 0, POSIX_FADV_RANDOM);
205 : #endif
206 :
207 90 : return {};
208 : }
209 :
210 : inline void
211 422 : posix_random_access_file::close_file() noexcept
212 : {
213 422 : if (fd_ >= 0)
214 : {
215 94 : ::close(fd_);
216 94 : fd_ = -1;
217 : }
218 422 : }
219 :
220 : inline std::uint64_t
221 13 : posix_random_access_file::size() const
222 : {
223 : struct stat st;
224 13 : if (::fstat(fd_, &st) < 0)
225 5 : throw_system_error(make_err(errno), "random_access_file::size");
226 8 : return static_cast<std::uint64_t>(st.st_size);
227 : }
228 :
229 : inline std::error_code
230 13 : posix_random_access_file::resize(std::uint64_t new_size) noexcept
231 : {
232 13 : if (new_size >
233 13 : static_cast<std::uint64_t>((std::numeric_limits<off_t>::max)()))
234 2 : return make_err(EOVERFLOW);
235 11 : if (::ftruncate(fd_, static_cast<off_t>(new_size)) < 0)
236 7 : return make_err(errno);
237 4 : return {};
238 : }
239 :
240 : inline std::error_code
241 9 : posix_random_access_file::sync_data() noexcept
242 : {
243 : #if BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
244 9 : if (::fdatasync(fd_) < 0)
245 : #else // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
246 : if (::fsync(fd_) < 0)
247 : #endif // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
248 7 : return make_err(errno);
249 2 : return {};
250 : }
251 :
252 : inline std::error_code
253 9 : posix_random_access_file::sync_all() noexcept
254 : {
255 9 : if (::fsync(fd_) < 0)
256 7 : return make_err(errno);
257 2 : return {};
258 : }
259 :
260 : inline native_handle_type
261 3 : posix_random_access_file::release()
262 : {
263 3 : int fd = fd_;
264 3 : fd_ = -1;
265 3 : return fd;
266 : }
267 :
268 : inline std::error_code
269 7 : posix_random_access_file::assign(native_handle_type handle) noexcept
270 : {
271 7 : close_file();
272 7 : fd_ = handle;
273 7 : return {};
274 : }
275 :
276 : // read_some_at, write_some_at are defined in
277 : // posix_random_access_file_service.hpp after the service.
278 :
279 : // -- raf_op completion handler (scheduler thread) --
280 :
281 : inline void
282 330 : posix_random_access_file::raf_op::operator()()
283 : {
284 330 : stop_cb.reset();
285 :
286 330 : bool const was_cancelled = cancelled.load(std::memory_order_acquire);
287 :
288 330 : if (ec_out)
289 : {
290 330 : if (was_cancelled)
291 10 : *ec_out = capy::error::canceled;
292 320 : else if (errn != 0)
293 16 : *ec_out = make_err(errn);
294 304 : else if (is_read && bytes_transferred == 0)
295 9 : *ec_out = capy::error::eof;
296 : else
297 295 : *ec_out = {};
298 : }
299 :
300 330 : if (bytes_out)
301 330 : *bytes_out = was_cancelled ? 0 : bytes_transferred;
302 :
303 : {
304 330 : std::lock_guard<std::mutex> lock(file_->ops_mutex_);
305 330 : file_->outstanding_ops_.remove(this);
306 330 : }
307 :
308 330 : impl_ptr.reset();
309 :
310 330 : auto coro = h;
311 330 : ex.on_work_finished();
312 330 : delete this;
313 330 : coro.resume();
314 330 : }
315 :
316 : // -- raf_op shutdown cleanup --
317 :
318 : inline void
319 6 : posix_random_access_file::raf_op::destroy()
320 : {
321 6 : stop_cb.reset();
322 : {
323 6 : std::lock_guard<std::mutex> lock(file_->ops_mutex_);
324 6 : file_->outstanding_ops_.remove(this);
325 6 : }
326 6 : impl_ptr.reset();
327 6 : ex.on_work_finished();
328 6 : delete this;
329 6 : }
330 :
331 : } // namespace boost::corosio::detail
332 :
333 : #endif // BOOST_COROSIO_POSIX
334 :
335 : #endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_FILE_HPP
|
