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