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