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_STREAM_FILE_HPP
11 : #define BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_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/stream_file.hpp>
19 : #include <boost/corosio/file_base.hpp>
20 : #include <boost/corosio/detail/intrusive.hpp>
21 : #include <boost/corosio/detail/dispatch_coro.hpp>
22 : #include <boost/corosio/detail/scheduler_op.hpp>
23 : #include <boost/corosio/detail/thread_pool.hpp>
24 : #include <boost/corosio/detail/scheduler.hpp>
25 : #include <boost/corosio/detail/buffer_param.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/capy/ex/executor_ref.hpp>
30 : #include <boost/capy/error.hpp>
31 : #include <boost/capy/buffers.hpp>
32 :
33 : #include <atomic>
34 : #include <coroutine>
35 : #include <cstddef>
36 : #include <cstdint>
37 : #include <filesystem>
38 : #include <limits>
39 : #include <memory>
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 Stream File Implementation
52 : =================================
53 :
54 : Regular files cannot be monitored by epoll/kqueue/select — the kernel
55 : always reports them as ready. Blocking I/O (pread/pwrite) is dispatched
56 : to a shared thread pool, with completion posted back to the scheduler.
57 :
58 : This follows the same pattern as posix_resolver: pool_work_item for
59 : dispatch, scheduler_op for completion, shared_from_this for lifetime.
60 :
61 : Completion Flow
62 : ---------------
63 : 1. read_some() sets up file_read_op, posts to thread pool
64 : 2. Pool thread runs preadv() (blocking)
65 : 3. Pool thread stores results, posts scheduler_op to scheduler
66 : 4. Scheduler invokes op() which resumes the coroutine
67 :
68 : Single-Inflight Constraint
69 : --------------------------
70 : Only one asynchronous operation may be in flight at a time on a
71 : given file object. Concurrent read and write is not supported
72 : because both share offset_ without synchronization.
73 : */
74 :
75 : namespace boost::corosio::detail {
76 :
77 : struct scheduler;
78 : class posix_stream_file_service;
79 :
80 : /** Stream file implementation for POSIX backends.
81 :
82 : Each instance contains embedded operation objects (read_op_, write_op_)
83 : that are reused across calls. This avoids per-operation heap allocation.
84 : */
85 : class posix_stream_file final
86 : : public stream_file::implementation
87 : , public std::enable_shared_from_this<posix_stream_file>
88 : , public intrusive_list<posix_stream_file>::node
89 : {
90 : friend class posix_stream_file_service;
91 :
92 : public:
93 : static constexpr std::size_t max_buffers = 16;
94 :
95 : /** Operation state for a single file read or write.
96 :
97 : The coroutine, cancellation and keepalive machinery is inherited
98 : from `coro_op`; only the pool-path result state lives here.
99 : */
100 : struct file_op : coro_op
101 : {
102 : // Buffer data (copied from buffer_param at submission time)
103 : iovec iovecs[max_buffers];
104 : int iovec_count = 0;
105 :
106 : // Result storage (populated by worker thread)
107 : int errn = 0;
108 : std::size_t bytes_transferred = 0;
109 :
110 HIT 528 : file_op() = default;
111 :
112 237 : void reset() noexcept
113 : {
114 237 : iovec_count = 0;
115 237 : errn = 0;
116 237 : bytes_transferred = 0;
117 237 : is_read = false;
118 237 : cancelled.store(false, std::memory_order_relaxed);
119 237 : stop_cb.reset();
120 237 : impl_ptr.reset();
121 237 : ec_out = nullptr;
122 237 : bytes_out = nullptr;
123 237 : }
124 :
125 : void operator()() override;
126 : void destroy() override;
127 : };
128 :
129 : /** Pool work item for thread pool dispatch. */
130 : struct pool_op : pool_work_item
131 : {
132 : posix_stream_file* file_ = nullptr;
133 : std::shared_ptr<posix_stream_file> ref_;
134 : };
135 :
136 : explicit posix_stream_file(posix_stream_file_service& svc) noexcept;
137 :
138 : // -- io_stream::implementation --
139 :
140 : std::coroutine_handle<> read_some(
141 : std::coroutine_handle<>,
142 : capy::executor_ref,
143 : buffer_param,
144 : std::stop_token,
145 : std::error_code*,
146 : std::size_t*) override;
147 :
148 : std::coroutine_handle<> write_some(
149 : std::coroutine_handle<>,
150 : capy::executor_ref,
151 : buffer_param,
152 : std::stop_token,
153 : std::error_code*,
154 : std::size_t*) override;
155 :
156 : // -- stream_file::implementation --
157 :
158 793 : native_handle_type native_handle() const noexcept override
159 : {
160 793 : return fd_;
161 : }
162 :
163 763 : void cancel() noexcept override
164 : {
165 763 : read_op_.request_cancel();
166 763 : write_op_.request_cancel();
167 763 : }
168 :
169 : std::uint64_t size() const override;
170 : std::error_code resize(std::uint64_t new_size) noexcept override;
171 : std::error_code sync_data() noexcept override;
172 : std::error_code sync_all() noexcept override;
173 : native_handle_type release() override;
174 : std::error_code assign(native_handle_type handle) noexcept override;
175 : capy::io_result<std::uint64_t>
176 : seek(std::int64_t offset, file_base::seek_basis origin) noexcept override;
177 :
178 : // -- Internal --
179 :
180 : /** Open the file and store the fd. */
181 : std::error_code
182 : open_file(std::filesystem::path const& path, file_base::flags mode);
183 :
184 : /** Close the file descriptor. */
185 : void close_file() noexcept;
186 :
187 : private:
188 : posix_stream_file_service& svc_;
189 : int fd_ = -1;
190 : std::uint64_t offset_ = 0;
191 :
192 : file_op read_op_;
193 : file_op write_op_;
194 : pool_op read_pool_op_;
195 : pool_op write_pool_op_;
196 :
197 : static void do_read_work(pool_work_item*) noexcept;
198 : static void do_write_work(pool_work_item*) noexcept;
199 : };
200 :
201 : // ---------------------------------------------------------------------------
202 : // Inline implementation
203 : // ---------------------------------------------------------------------------
204 :
205 264 : inline posix_stream_file::posix_stream_file(
206 264 : posix_stream_file_service& svc) noexcept
207 264 : : svc_(svc)
208 : {
209 264 : }
210 :
211 : inline std::error_code
212 247 : posix_stream_file::open_file(
213 : std::filesystem::path const& path, file_base::flags mode)
214 : {
215 247 : close_file();
216 :
217 247 : int oflags = 0;
218 :
219 : // Access mode
220 247 : unsigned access = static_cast<unsigned>(mode) & 3u;
221 247 : if (access == static_cast<unsigned>(file_base::read_write))
222 21 : oflags |= O_RDWR;
223 226 : else if (access == static_cast<unsigned>(file_base::write_only))
224 81 : oflags |= O_WRONLY;
225 : else
226 145 : oflags |= O_RDONLY;
227 :
228 : // Creation flags
229 247 : if ((mode & file_base::create) != file_base::flags(0))
230 40 : oflags |= O_CREAT;
231 247 : if ((mode & file_base::exclusive) != file_base::flags(0))
232 2 : oflags |= O_EXCL;
233 247 : if ((mode & file_base::truncate) != file_base::flags(0))
234 17 : oflags |= O_TRUNC;
235 247 : if ((mode & file_base::append) != file_base::flags(0))
236 8 : oflags |= O_APPEND;
237 247 : if ((mode & file_base::sync_all_on_write) != file_base::flags(0))
238 2 : oflags |= O_SYNC;
239 :
240 247 : int fd = ::open(path.c_str(), oflags, 0666);
241 247 : if (fd < 0)
242 9 : return make_err(errno);
243 :
244 238 : fd_ = fd;
245 238 : offset_ = 0;
246 :
247 : // Append mode: position at end-of-file (preadv/pwritev use
248 : // explicit offsets, so O_APPEND alone is not sufficient).
249 238 : if ((mode & file_base::append) != file_base::flags(0))
250 : {
251 : struct stat st;
252 8 : if (::fstat(fd, &st) < 0)
253 : {
254 5 : int err = errno;
255 5 : ::close(fd);
256 5 : fd_ = -1;
257 5 : return make_err(err);
258 : }
259 3 : offset_ = static_cast<std::uint64_t>(st.st_size);
260 : }
261 :
262 : #ifdef POSIX_FADV_SEQUENTIAL
263 233 : ::posix_fadvise(fd_, 0, 0, POSIX_FADV_SEQUENTIAL);
264 : #endif
265 :
266 233 : return {};
267 : }
268 :
269 : inline void
270 1014 : posix_stream_file::close_file() noexcept
271 : {
272 1014 : if (fd_ >= 0)
273 : {
274 237 : ::close(fd_);
275 237 : fd_ = -1;
276 : }
277 1014 : }
278 :
279 : inline std::uint64_t
280 17 : posix_stream_file::size() const
281 : {
282 : struct stat st;
283 17 : if (::fstat(fd_, &st) < 0)
284 5 : throw_system_error(make_err(errno), "stream_file::size");
285 12 : return static_cast<std::uint64_t>(st.st_size);
286 : }
287 :
288 : inline std::error_code
289 12 : posix_stream_file::resize(std::uint64_t new_size) noexcept
290 : {
291 12 : if (new_size >
292 12 : static_cast<std::uint64_t>((std::numeric_limits<off_t>::max)()))
293 2 : return make_err(EOVERFLOW);
294 10 : if (::ftruncate(fd_, static_cast<off_t>(new_size)) < 0)
295 7 : return make_err(errno);
296 3 : return {};
297 : }
298 :
299 : inline std::error_code
300 10 : posix_stream_file::sync_data() noexcept
301 : {
302 : #if BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
303 10 : if (::fdatasync(fd_) < 0)
304 : #else // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
305 : if (::fsync(fd_) < 0)
306 : #endif // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
307 7 : return make_err(errno);
308 3 : return {};
309 : }
310 :
311 : inline std::error_code
312 10 : posix_stream_file::sync_all() noexcept
313 : {
314 10 : if (::fsync(fd_) < 0)
315 7 : return make_err(errno);
316 3 : return {};
317 : }
318 :
319 : inline native_handle_type
320 2 : posix_stream_file::release()
321 : {
322 2 : int fd = fd_;
323 2 : fd_ = -1;
324 2 : offset_ = 0;
325 2 : return fd;
326 : }
327 :
328 : inline std::error_code
329 6 : posix_stream_file::assign(native_handle_type handle) noexcept
330 : {
331 6 : close_file();
332 6 : fd_ = handle;
333 6 : offset_ = 0;
334 6 : return {};
335 : }
336 :
337 : inline capy::io_result<std::uint64_t>
338 30 : posix_stream_file::seek(
339 : std::int64_t offset, file_base::seek_basis origin) noexcept
340 : {
341 : // We track offset_ ourselves (not the kernel fd offset)
342 : // because preadv/pwritev use explicit offsets.
343 : std::int64_t new_pos;
344 :
345 30 : if (origin == file_base::seek_set)
346 : {
347 14 : new_pos = offset;
348 : }
349 16 : else if (origin == file_base::seek_cur)
350 : {
351 5 : new_pos = static_cast<std::int64_t>(offset_) + offset;
352 : }
353 : else
354 : {
355 : struct stat st;
356 11 : if (::fstat(fd_, &st) < 0)
357 5 : return {make_err(errno), 0};
358 6 : new_pos = st.st_size + offset;
359 : }
360 :
361 25 : if (new_pos < 0)
362 6 : return {make_err(EINVAL), 0};
363 19 : if (new_pos >
364 19 : static_cast<std::int64_t>((std::numeric_limits<off_t>::max)()))
365 MIS 0 : return {make_err(EOVERFLOW), 0};
366 :
367 HIT 19 : offset_ = static_cast<std::uint64_t>(new_pos);
368 :
369 19 : return {std::error_code{}, offset_};
370 : }
371 :
372 : // -- file_op completion handler --
373 : // (read_some, write_some, do_read_work, do_write_work are
374 : // defined in posix_stream_file_service.hpp after the service)
375 :
376 : inline void
377 211 : posix_stream_file::file_op::operator()()
378 : {
379 211 : stop_cb.reset();
380 :
381 : // Empty buffers never reach the pool (diverted at initiation), so
382 : // empty_buffer stays false and a 0-byte read is a genuine EOF.
383 408 : decode_io_result(
384 211 : ec_out, bytes_out, cancelled.load(std::memory_order_acquire),
385 211 : errn != 0 ? make_err(errn) : std::error_code{}, is_read,
386 : bytes_transferred, /*empty_buffer=*/false);
387 :
388 : // Move impl_ptr to a local so members remain valid through
389 : // dispatch — impl_ptr may be the last shared_ptr keeping
390 : // the parent posix_stream_file (which embeds this file_op) alive.
391 211 : auto prevent_destroy = std::move(impl_ptr);
392 211 : ex.on_work_finished();
393 211 : cont.h = h;
394 211 : dispatch_coro(ex, cont).resume();
395 211 : }
396 :
397 : inline void
398 2 : posix_stream_file::file_op::destroy()
399 : {
400 2 : stop_cb.reset();
401 2 : auto local_ex = ex;
402 2 : impl_ptr.reset();
403 2 : local_ex.on_work_finished();
404 2 : }
405 :
406 : } // namespace boost::corosio::detail
407 :
408 : #endif // BOOST_COROSIO_POSIX
409 :
410 : #endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_HPP
|
