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/corosio/native/detail/posix/large_file.hpp>
30 : #include <boost/corosio/native/detail/posix/posix_fd_gate.hpp>
31 : #include <boost/corosio/native/detail/validate_fd.hpp>
32 : #include <boost/capy/continuation.hpp>
33 : #include <boost/capy/ex/executor_ref.hpp>
34 : #include <boost/capy/error.hpp>
35 : #include <boost/capy/buffers.hpp>
36 :
37 : #include <atomic>
38 : #include <coroutine>
39 : #include <cstddef>
40 : #include <cstdint>
41 : #include <filesystem>
42 : #include <limits>
43 : #include <memory>
44 : #include <mutex>
45 : #include <optional>
46 : #include <stop_token>
47 : #include <system_error>
48 :
49 : #include <errno.h>
50 : #include <fcntl.h>
51 : #include <sys/stat.h>
52 : #include <sys/uio.h>
53 : #include <unistd.h>
54 :
55 : /*
56 : POSIX Stream File Implementation
57 : =================================
58 :
59 : Regular files cannot be monitored by epoll/kqueue/select — the kernel
60 : always reports them as ready. Blocking I/O (pread/pwrite) is dispatched
61 : to a shared thread pool, with completion posted back to the scheduler.
62 :
63 : This follows the same pattern as posix_resolver: pool_work_item for
64 : dispatch, scheduler_op for completion, an object_ref keepalive on
65 : each op for lifetime while its pool work is in flight.
66 :
67 : Completion Flow
68 : ---------------
69 : 1. read_some() sets up file_read_op, posts to thread pool
70 : 2. Pool thread runs preadv() (blocking)
71 : 3. Pool thread stores results, posts scheduler_op to scheduler
72 : 4. Scheduler invokes op() which resumes the coroutine
73 :
74 : Single-Inflight Constraint
75 : --------------------------
76 : Only one asynchronous operation may be in flight at a time on a
77 : given file object. Concurrent read and write is not supported
78 : because both share offset_ without synchronization.
79 : */
80 :
81 : namespace boost::corosio::detail {
82 :
83 : class scheduler;
84 : class posix_stream_file_service;
85 :
86 : /** Stream file implementation for POSIX backends.
87 :
88 : Each instance contains embedded operation objects (read_op_, write_op_)
89 : that are reused across calls. This avoids per-operation heap allocation.
90 : */
91 : class posix_stream_file final
92 : : public stream_file::implementation
93 : , public intrusive_list<posix_stream_file>::node
94 : {
95 : friend class posix_stream_file_service;
96 :
97 : public:
98 : static constexpr std::size_t max_buffers = 16;
99 :
100 : /** Operation state for a single file read or write.
101 :
102 : The coroutine, cancellation and keepalive machinery is inherited
103 : from `coro_op`; only the pool-path result state lives here.
104 : */
105 : struct file_op : coro_op
106 : {
107 : // Buffer data (copied from buffer_param at submission time)
108 : iovec iovecs[max_buffers];
109 : int iovec_count = 0;
110 :
111 : // Snapshotted at submission: assign() may replace fd_ and
112 : // offset_ while the worker runs.
113 : int fd = -1;
114 : std::uint64_t offset = 0;
115 : std::uint64_t generation = 0;
116 :
117 : // Result storage (populated by worker thread)
118 : int errn = 0;
119 : std::size_t bytes_transferred = 0;
120 :
121 : // The op is work on this service's context, which outlives the
122 : // file the op is embedded in.
123 : posix_stream_file_service* svc = nullptr;
124 :
125 HIT 1390 : file_op() = default;
126 :
127 687 : void reset() noexcept
128 : {
129 687 : iovec_count = 0;
130 687 : errn = 0;
131 687 : bytes_transferred = 0;
132 687 : is_read = false;
133 687 : cancelled.store(false, std::memory_order_relaxed);
134 687 : stop_cb.reset();
135 687 : object_ref_.reset();
136 687 : ec_out = nullptr;
137 687 : bytes_out = nullptr;
138 687 : }
139 :
140 : void operator()() override;
141 : void destroy() override;
142 : };
143 :
144 : /// Release the work an op holds on its service's context.
145 : static void
146 : release_work(posix_stream_file_service& svc) noexcept;
147 :
148 : /** Pool work item for thread pool dispatch. */
149 : struct pool_op : pool_work_item
150 : {
151 : posix_stream_file* file_ = nullptr;
152 : detail::object_ref ref_;
153 : };
154 :
155 : explicit posix_stream_file(posix_stream_file_service& svc) noexcept;
156 :
157 : /// Recycle into the owning service's pool. Defined out-of-line
158 : /// after posix_stream_file_service for its complete type.
159 : void retire() noexcept override;
160 :
161 : /** Reset op slots for recycling.
162 :
163 : `close_file()` already drove fd_ to its closed value before
164 : the refcount reached zero; each op's own `reset()` runs again
165 : before its next use, so only the `stop_cb` invariant is worth
166 : asserting here.
167 :
168 : @pre refs_ == 0, fd closed, no op in flight.
169 : */
170 64 : void reuse() noexcept
171 : {
172 64 : BOOST_COROSIO_ASSERT(fd_ == -1);
173 64 : BOOST_COROSIO_ASSERT(gate_.idle());
174 64 : BOOST_COROSIO_ASSERT(!read_op_.stop_cb);
175 64 : BOOST_COROSIO_ASSERT(!write_op_.stop_cb);
176 64 : }
177 :
178 : // -- io_stream::implementation --
179 :
180 : std::coroutine_handle<> read_some(
181 : capy::continuation&,
182 : capy::executor_ref,
183 : buffer_param,
184 : std::stop_token,
185 : std::error_code*,
186 : std::size_t*) override;
187 :
188 : std::coroutine_handle<> write_some(
189 : capy::continuation&,
190 : capy::executor_ref,
191 : buffer_param,
192 : std::stop_token,
193 : std::error_code*,
194 : std::size_t*) override;
195 :
196 : // -- stream_file::implementation --
197 :
198 3016 : native_handle_type native_handle() const noexcept override
199 : {
200 3016 : return fd_;
201 : }
202 :
203 2651 : void cancel() noexcept override
204 : {
205 2651 : read_op_.request_cancel();
206 2651 : write_op_.request_cancel();
207 2651 : }
208 :
209 : std::uint64_t size() const override;
210 : std::error_code resize(std::uint64_t new_size) noexcept override;
211 : std::error_code sync_data() noexcept override;
212 : std::error_code sync_all() noexcept override;
213 : native_handle_type release() override;
214 : std::error_code assign(native_handle_type handle) noexcept override;
215 : capy::results<std::uint64_t>
216 : seek(std::int64_t offset, file_base::seek_basis origin) noexcept override;
217 :
218 : // -- Internal --
219 :
220 : /** Open the file and store the fd. */
221 : std::error_code
222 : open_file(std::filesystem::path const& path, file_base::flags mode);
223 :
224 : /** Close the file descriptor. */
225 : void close_file() noexcept;
226 :
227 : private:
228 : posix_stream_file_service& svc_;
229 : int fd_ = -1;
230 : std::uint64_t offset_ = 0;
231 :
232 : // Bumped whenever the held file changes. A worker advances offset_
233 : // only if this still matches its op's snapshot; the mutex makes the
234 : // check and the advance one step against a concurrent assign().
235 : std::mutex offset_mutex_;
236 : /// Keeps fd_'s number allocated while a pool worker uses it.
237 : posix_fd_gate gate_;
238 : std::uint64_t generation_ = 0;
239 :
240 : file_op read_op_;
241 : file_op write_op_;
242 : pool_op read_pool_op_;
243 : pool_op write_pool_op_;
244 :
245 : void bump_generation() noexcept;
246 :
247 : static void do_read_work(pool_work_item*) noexcept;
248 : static void do_write_work(pool_work_item*) noexcept;
249 : static void destroy_read_work(pool_work_item*) noexcept;
250 : static void destroy_write_work(pool_work_item*) noexcept;
251 : };
252 :
253 : // ---------------------------------------------------------------------------
254 : // Inline implementation
255 : // ---------------------------------------------------------------------------
256 :
257 695 : inline posix_stream_file::posix_stream_file(
258 695 : posix_stream_file_service& svc) noexcept
259 695 : : svc_(svc)
260 : {
261 695 : }
262 :
263 : inline std::error_code
264 736 : posix_stream_file::open_file(
265 : std::filesystem::path const& path, file_base::flags mode)
266 : {
267 736 : close_file();
268 :
269 736 : int oflags = 0;
270 :
271 : // Access mode
272 736 : unsigned access = static_cast<unsigned>(mode) & 3u;
273 736 : if (access == static_cast<unsigned>(file_base::read_write))
274 89 : oflags |= O_RDWR;
275 647 : else if (access == static_cast<unsigned>(file_base::write_only))
276 86 : oflags |= O_WRONLY;
277 : else
278 561 : oflags |= O_RDONLY;
279 :
280 : // Creation flags
281 736 : if ((mode & file_base::create) != file_base::flags(0))
282 106 : oflags |= O_CREAT;
283 736 : if ((mode & file_base::exclusive) != file_base::flags(0))
284 2 : oflags |= O_EXCL;
285 736 : if ((mode & file_base::truncate) != file_base::flags(0))
286 83 : oflags |= O_TRUNC;
287 736 : if ((mode & file_base::append) != file_base::flags(0))
288 8 : oflags |= O_APPEND;
289 736 : if ((mode & file_base::sync_all_on_write) != file_base::flags(0))
290 2 : oflags |= O_SYNC;
291 :
292 736 : int fd = ::open(path.c_str(), oflags | large_file_open_flag, 0666);
293 736 : if (fd < 0)
294 9 : return make_err(errno);
295 :
296 727 : fd_ = fd;
297 727 : offset_ = 0;
298 727 : gate_.open(fd);
299 :
300 : // Append mode: position at end-of-file (preadv/pwritev use
301 : // explicit offsets, so O_APPEND alone is not sufficient).
302 727 : if ((mode & file_base::append) != file_base::flags(0))
303 : {
304 : file_stat_t st;
305 8 : if (file_fstat(fd, &st) < 0)
306 : {
307 5 : int err = errno;
308 5 : gate_.close();
309 5 : fd_ = -1;
310 5 : return make_err(err);
311 : }
312 3 : offset_ = static_cast<std::uint64_t>(st.st_size);
313 : }
314 :
315 : #ifdef POSIX_FADV_SEQUENTIAL
316 722 : ::posix_fadvise(fd_, 0, 0, POSIX_FADV_SEQUENTIAL);
317 : #endif
318 :
319 722 : return {};
320 : }
321 :
322 : inline void
323 3385 : posix_stream_file::bump_generation() noexcept
324 : {
325 3385 : std::lock_guard<std::mutex> lock(offset_mutex_);
326 3385 : ++generation_;
327 3385 : }
328 :
329 : inline void
330 3381 : posix_stream_file::close_file() noexcept
331 : {
332 3381 : bump_generation();
333 3381 : gate_.close();
334 3381 : fd_ = -1;
335 3381 : }
336 :
337 : inline std::uint64_t
338 15 : posix_stream_file::size() const
339 : {
340 : file_stat_t st;
341 15 : if (file_fstat(fd_, &st) < 0)
342 5 : throw_system_error(make_err(errno), "stream_file::size");
343 10 : return static_cast<std::uint64_t>(st.st_size);
344 : }
345 :
346 : inline std::error_code
347 12 : posix_stream_file::resize(std::uint64_t new_size) noexcept
348 : {
349 12 : if (new_size >
350 12 : static_cast<std::uint64_t>((std::numeric_limits<file_off_t>::max)()))
351 2 : return make_err(EOVERFLOW);
352 10 : if (file_ftruncate(fd_, static_cast<file_off_t>(new_size)) < 0)
353 7 : return make_err(errno);
354 3 : return {};
355 : }
356 :
357 : inline std::error_code
358 8 : posix_stream_file::sync_data() noexcept
359 : {
360 : #if BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
361 8 : if (::fdatasync(fd_) < 0)
362 : #else // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
363 : if (::fsync(fd_) < 0)
364 : #endif // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
365 5 : return make_err(errno);
366 3 : return {};
367 : }
368 :
369 : inline std::error_code
370 8 : posix_stream_file::sync_all() noexcept
371 : {
372 8 : if (::fsync(fd_) < 0)
373 5 : return make_err(errno);
374 3 : return {};
375 : }
376 :
377 : inline native_handle_type
378 4 : posix_stream_file::release()
379 : {
380 : // A queued op has already copied the fd number; it must not run
381 : // after the caller closes it and the number is recycled.
382 4 : cancel();
383 4 : if (!gate_.release())
384 MIS 0 : throw_system_error(
385 0 : std::make_error_code(std::errc::device_or_resource_busy),
386 : "stream_file::release");
387 HIT 4 : bump_generation();
388 4 : int fd = fd_;
389 4 : fd_ = -1;
390 4 : offset_ = 0;
391 4 : return fd;
392 : }
393 :
394 : inline capy::results<std::uint64_t>
395 435 : posix_stream_file::seek(
396 : std::int64_t offset, file_base::seek_basis origin) noexcept
397 : {
398 : // We track offset_ ourselves (not the kernel fd offset)
399 : // because preadv/pwritev use explicit offsets.
400 : std::int64_t new_pos;
401 :
402 435 : if (origin == file_base::seek_set)
403 : {
404 14 : new_pos = offset;
405 : }
406 421 : else if (origin == file_base::seek_cur)
407 : {
408 410 : new_pos = static_cast<std::int64_t>(offset_) + offset;
409 : }
410 : else
411 : {
412 : file_stat_t st;
413 11 : if (file_fstat(fd_, &st) < 0)
414 5 : return {make_err(errno), 0};
415 6 : new_pos = st.st_size + offset;
416 : }
417 :
418 430 : if (new_pos < 0)
419 6 : return {make_err(EINVAL), 0};
420 424 : if (new_pos >
421 424 : static_cast<std::int64_t>((std::numeric_limits<file_off_t>::max)()))
422 MIS 0 : return {make_err(EOVERFLOW), 0};
423 :
424 HIT 424 : offset_ = static_cast<std::uint64_t>(new_pos);
425 :
426 424 : return {std::error_code{}, offset_};
427 : }
428 :
429 : // -- file_op completion handler --
430 : // (read_some, write_some, do_read_work, do_write_work are
431 : // defined in posix_stream_file_service.hpp after the service)
432 :
433 : inline void
434 659 : posix_stream_file::file_op::operator()()
435 : {
436 659 : stop_cb.reset();
437 :
438 : // Empty buffers never reach the pool (diverted at initiation), so
439 : // empty_buffer stays false and a 0-byte read is a genuine EOF.
440 1304 : decode_results(
441 659 : ec_out, bytes_out, cancelled.load(std::memory_order_acquire),
442 659 : errn != 0 ? make_err(errn) : std::error_code{}, is_read,
443 : bytes_transferred, /*empty_buffer=*/false);
444 :
445 : // Move object_ref_ to a local so members remain valid through
446 : // dispatch — this may be the last reference keeping the parent
447 : // posix_stream_file (which embeds this file_op) alive.
448 659 : auto prevent_destroy = std::move(object_ref_);
449 659 : release_work(*svc);
450 659 : dispatch_coro(ex, *cont).resume();
451 659 : }
452 :
453 : inline void
454 4 : posix_stream_file::file_op::destroy()
455 : {
456 : // Copied first: the abandon may unwind the frame at once, destroying
457 : // the file, and with it this op.
458 4 : auto* s = svc;
459 4 : destroy_awaiting();
460 4 : release_work(*s);
461 4 : }
462 :
463 : } // namespace boost::corosio::detail
464 :
465 : #endif // BOOST_COROSIO_POSIX
466 :
467 : #endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_HPP
|
