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_SERVICE_HPP
11 : #define BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_SERVICE_HPP
12 :
13 : #include <boost/corosio/detail/platform.hpp>
14 :
15 : #if BOOST_COROSIO_POSIX
16 :
17 : #include <boost/corosio/native/detail/posix/posix_stream_file.hpp>
18 : #include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp>
19 : #include <boost/corosio/detail/file_service.hpp>
20 : #include <boost/corosio/detail/object_pool.hpp>
21 : #include <boost/corosio/detail/object_ref.hpp>
22 : #include <boost/corosio/detail/thread_pool.hpp>
23 :
24 : #include <vector>
25 :
26 : namespace boost::corosio::detail {
27 :
28 : /** Stream file service for POSIX backends.
29 :
30 : Owns all posix_stream_file instances. Thread lifecycle is
31 : managed by the thread_pool service (shared with resolver).
32 : */
33 : class BOOST_COROSIO_DECL posix_stream_file_service final : public file_service
34 : {
35 : friend class posix_stream_file;
36 :
37 : public:
38 HIT 639 : explicit posix_stream_file_service(capy::execution_context& ctx)
39 1917 : : sched_(&get_scheduler(ctx))
40 639 : , pool_(ctx)
41 : {
42 639 : }
43 :
44 1278 : ~posix_stream_file_service() override = default;
45 :
46 : posix_stream_file_service(posix_stream_file_service const&) = delete;
47 : posix_stream_file_service&
48 : operator=(posix_stream_file_service const&) = delete;
49 :
50 759 : io_object::implementation* construct() override
51 : {
52 759 : return object_pool_.acquire(*this);
53 : }
54 :
55 759 : void destroy(io_object::implementation* p) override
56 : {
57 759 : auto& impl = static_cast<posix_stream_file&>(*p);
58 759 : impl.cancel();
59 759 : impl.close_file();
60 759 : release(&impl);
61 759 : }
62 :
63 1882 : void close(io_object::handle& h) override
64 : {
65 1882 : if (h.get())
66 : {
67 1882 : auto& impl = static_cast<posix_stream_file&>(*h.get());
68 1882 : impl.cancel();
69 1882 : impl.close_file();
70 : }
71 1882 : }
72 :
73 738 : std::error_code open_file(
74 : stream_file::implementation& impl,
75 : std::filesystem::path const& path,
76 : file_base::flags mode) override
77 : {
78 738 : if (pool_unavailable())
79 2 : return std::make_error_code(std::errc::operation_not_supported);
80 736 : return static_cast<posix_stream_file&>(impl).open_file(path, mode);
81 : }
82 :
83 : /** Check whether file I/O is unavailable on this context.
84 :
85 : The file thread pool completes cross-thread, which the
86 : lockless scheduler of the unsafe tier cannot accept.
87 : */
88 1151 : bool pool_unavailable() const noexcept
89 : {
90 1151 : return sched_->scheduler_locking_disabled();
91 : }
92 :
93 639 : void shutdown() override
94 : {
95 : // See uring_socket_service_base::shutdown(): snapshot under an
96 : // acquired reference, cancel+close without the pool lock held;
97 : // shutdown() sets shutting-down and takes the snapshot in one
98 : // critical section, so a close that drops the last ref deletes
99 : // rather than recycles.
100 639 : std::vector<posix_stream_file*> live;
101 639 : object_pool_.shutdown(
102 639 : [&](posix_stream_file* f)
103 : {
104 4 : acquire(f);
105 4 : live.push_back(f);
106 4 : });
107 643 : for (auto* f : live)
108 : {
109 4 : f->cancel();
110 4 : f->close_file();
111 4 : release(f);
112 : }
113 639 : }
114 :
115 659 : void post(scheduler_op* op)
116 : {
117 659 : sched_->post(op);
118 659 : }
119 :
120 671 : void work_started() noexcept
121 : {
122 671 : sched_->work_started();
123 671 : }
124 :
125 671 : void work_finished() noexcept
126 : {
127 671 : sched_->work_finished();
128 671 : }
129 :
130 : /** Return the thread pool that runs this service's file work.
131 :
132 : The pool's service is created on first use, so this can fail
133 : where a plain accessor could not. Its workers start later, on
134 : the first post, and a thread the system refuses there is
135 : reported by that post rather than thrown here.
136 :
137 : @throws std::bad_alloc If the service cannot be allocated.
138 :
139 : @return The context's shared blocking-I/O pool.
140 :
141 : @see thread_pool_ref::get
142 : */
143 671 : thread_pool& pool()
144 : {
145 671 : return pool_.get();
146 : }
147 :
148 : private:
149 : scheduler* sched_;
150 : thread_pool_ref pool_;
151 : object_pool<posix_stream_file> object_pool_;
152 : };
153 :
154 : // ---------------------------------------------------------------------------
155 : // posix_stream_file inline implementations (require complete service type)
156 : // ---------------------------------------------------------------------------
157 :
158 : inline std::coroutine_handle<>
159 589 : posix_stream_file::read_some(
160 : capy::continuation& cont,
161 : capy::executor_ref ex,
162 : buffer_param param,
163 : std::stop_token token,
164 : std::error_code* ec,
165 : std::size_t* bytes_out)
166 : {
167 589 : auto& op = read_op_;
168 589 : op.reset();
169 589 : op.is_read = true;
170 :
171 : // Closed-object contract outranks the zero-length no-op.
172 589 : if (fd_ < 0)
173 : {
174 6 : *ec = make_error_code(std::errc::bad_file_descriptor);
175 6 : *bytes_out = 0;
176 6 : return dispatch_coro(ex, cont);
177 : }
178 :
179 583 : capy::mutable_buffer bufs[max_buffers];
180 583 : op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
181 :
182 583 : if (op.iovec_count == 0)
183 : {
184 2 : *ec = {};
185 2 : *bytes_out = 0;
186 2 : return dispatch_coro(ex, cont);
187 : }
188 :
189 1162 : for (int i = 0; i < op.iovec_count; ++i)
190 : {
191 581 : op.iovecs[i].iov_base = bufs[i].data();
192 581 : op.iovecs[i].iov_len = bufs[i].size();
193 : }
194 :
195 581 : op.cont = &cont;
196 581 : op.ex = ex;
197 581 : op.ec_out = ec;
198 581 : op.bytes_out = bytes_out;
199 581 : op.start(token);
200 :
201 581 : op.fd = fd_;
202 581 : op.offset = offset_;
203 581 : op.generation = generation_;
204 581 : op.svc = &svc_;
205 :
206 581 : svc_.work_started();
207 :
208 581 : read_pool_op_.file_ = this;
209 581 : read_pool_op_.ref_ = detail::object_ref(this);
210 581 : read_pool_op_.func_ = &posix_stream_file::do_read_work;
211 581 : read_pool_op_.destroy_ = &posix_stream_file::destroy_read_work;
212 581 : if (auto pec = svc_.pool().post(&read_pool_op_))
213 : {
214 : // The pool is shutting down, or the system refused it a thread.
215 : // Nothing of this read went cross-thread, so it answers here
216 : // like the closed-descriptor and zero-length exits above rather
217 : // than through a completion the scheduler has to carry back.
218 6 : read_pool_op_.ref_.reset();
219 6 : op.stop_cb.reset();
220 6 : svc_.work_finished();
221 6 : *ec = pec;
222 6 : *bytes_out = 0;
223 6 : return dispatch_coro(ex, cont);
224 : }
225 575 : return std::noop_coroutine();
226 : }
227 :
228 : inline void
229 571 : posix_stream_file::do_read_work(pool_work_item* w) noexcept
230 : {
231 571 : auto* pw = static_cast<pool_op*>(w);
232 571 : auto* self = pw->file_;
233 571 : auto& op = self->read_op_;
234 :
235 571 : if (!op.cancelled.load(std::memory_order_acquire))
236 : {
237 135 : if (!self->gate_.enter(op.fd))
238 : {
239 : // Closed or replaced since this was queued.
240 MIS 0 : op.request_cancel();
241 : }
242 HIT 135 : else if (op.cancelled.load(std::memory_order_acquire))
243 : {
244 : // Closed and reopened onto the same number since the check
245 : // above: the gate admits the number, but not this op.
246 MIS 0 : self->gate_.leave();
247 : }
248 : else
249 : {
250 : ssize_t n;
251 : do
252 : {
253 HIT 135 : n = file_preadv(
254 135 : op.fd, op.iovecs, op.iovec_count,
255 135 : static_cast<file_off_t>(op.offset));
256 : }
257 135 : while (n < 0 && errno == EINTR);
258 :
259 135 : if (n >= 0)
260 : {
261 128 : op.errn = 0;
262 128 : op.bytes_transferred = static_cast<std::size_t>(n);
263 :
264 : // The file may have been replaced while this ran; its
265 : // position belongs to whatever it now holds.
266 128 : std::lock_guard<std::mutex> lock(self->offset_mutex_);
267 128 : if (self->generation_ == op.generation)
268 122 : self->offset_ += static_cast<std::uint64_t>(n);
269 128 : }
270 : else
271 : {
272 7 : op.errn = errno;
273 7 : op.bytes_transferred = 0;
274 : }
275 135 : self->gate_.leave();
276 : }
277 : }
278 :
279 571 : op.object_ref_ = std::move(pw->ref_);
280 571 : self->svc_.post(&op);
281 571 : }
282 :
283 : inline std::coroutine_handle<>
284 98 : posix_stream_file::write_some(
285 : capy::continuation& cont,
286 : capy::executor_ref ex,
287 : buffer_param param,
288 : std::stop_token token,
289 : std::error_code* ec,
290 : std::size_t* bytes_out)
291 : {
292 98 : auto& op = write_op_;
293 98 : op.reset();
294 98 : op.is_read = false;
295 :
296 : // Closed-object contract outranks the zero-length no-op.
297 98 : if (fd_ < 0)
298 : {
299 6 : *ec = make_error_code(std::errc::bad_file_descriptor);
300 6 : *bytes_out = 0;
301 6 : return dispatch_coro(ex, cont);
302 : }
303 :
304 92 : capy::mutable_buffer bufs[max_buffers];
305 92 : op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
306 :
307 92 : if (op.iovec_count == 0)
308 : {
309 2 : *ec = {};
310 2 : *bytes_out = 0;
311 2 : return dispatch_coro(ex, cont);
312 : }
313 :
314 180 : for (int i = 0; i < op.iovec_count; ++i)
315 : {
316 90 : op.iovecs[i].iov_base = bufs[i].data();
317 90 : op.iovecs[i].iov_len = bufs[i].size();
318 : }
319 :
320 90 : op.cont = &cont;
321 90 : op.ex = ex;
322 90 : op.ec_out = ec;
323 90 : op.bytes_out = bytes_out;
324 90 : op.start(token);
325 :
326 90 : op.fd = fd_;
327 90 : op.offset = offset_;
328 90 : op.generation = generation_;
329 90 : op.svc = &svc_;
330 :
331 90 : svc_.work_started();
332 :
333 90 : write_pool_op_.file_ = this;
334 90 : write_pool_op_.ref_ = detail::object_ref(this);
335 90 : write_pool_op_.func_ = &posix_stream_file::do_write_work;
336 90 : write_pool_op_.destroy_ = &posix_stream_file::destroy_write_work;
337 90 : if (auto pec = svc_.pool().post(&write_pool_op_))
338 : {
339 : // The pool is shutting down, or the system refused it a thread.
340 : // Nothing of this write went cross-thread, so it answers here
341 : // like the closed-descriptor and zero-length exits above rather
342 : // than through a completion the scheduler has to carry back.
343 2 : write_pool_op_.ref_.reset();
344 2 : op.stop_cb.reset();
345 2 : svc_.work_finished();
346 2 : *ec = pec;
347 2 : *bytes_out = 0;
348 2 : return dispatch_coro(ex, cont);
349 : }
350 88 : return std::noop_coroutine();
351 : }
352 :
353 : inline void
354 88 : posix_stream_file::do_write_work(pool_work_item* w) noexcept
355 : {
356 88 : auto* pw = static_cast<pool_op*>(w);
357 88 : auto* self = pw->file_;
358 88 : auto& op = self->write_op_;
359 :
360 88 : if (!op.cancelled.load(std::memory_order_acquire))
361 : {
362 37 : if (!self->gate_.enter(op.fd))
363 : {
364 : // Closed or replaced since this was queued.
365 MIS 0 : op.request_cancel();
366 : }
367 HIT 37 : else if (op.cancelled.load(std::memory_order_acquire))
368 : {
369 : // Closed and reopened onto the same number since the check
370 : // above: the gate admits the number, but not this op.
371 MIS 0 : self->gate_.leave();
372 : }
373 : else
374 : {
375 : ssize_t n;
376 : do
377 : {
378 HIT 37 : n = file_pwritev(
379 37 : op.fd, op.iovecs, op.iovec_count,
380 37 : static_cast<file_off_t>(op.offset));
381 : }
382 37 : while (n < 0 && errno == EINTR);
383 :
384 37 : if (n >= 0)
385 : {
386 30 : op.errn = 0;
387 30 : op.bytes_transferred = static_cast<std::size_t>(n);
388 :
389 : // The file may have been replaced while this ran; its
390 : // position belongs to whatever it now holds.
391 30 : std::lock_guard<std::mutex> lock(self->offset_mutex_);
392 30 : if (self->generation_ == op.generation)
393 25 : self->offset_ += static_cast<std::uint64_t>(n);
394 30 : }
395 : else
396 : {
397 7 : op.errn = errno;
398 7 : op.bytes_transferred = 0;
399 : }
400 37 : self->gate_.leave();
401 : }
402 : }
403 :
404 88 : op.object_ref_ = std::move(pw->ref_);
405 88 : self->svc_.post(&op);
406 88 : }
407 :
408 : inline void
409 4 : posix_stream_file::destroy_read_work(pool_work_item* w) noexcept
410 : {
411 4 : auto* pw = static_cast<pool_op*>(w);
412 4 : auto* self = pw->file_;
413 4 : self->read_op_.object_ref_ = std::move(pw->ref_);
414 4 : self->read_op_.destroy();
415 4 : }
416 :
417 : inline void
418 MIS 0 : posix_stream_file::destroy_write_work(pool_work_item* w) noexcept
419 : {
420 0 : auto* pw = static_cast<pool_op*>(w);
421 0 : auto* self = pw->file_;
422 0 : self->write_op_.object_ref_ = std::move(pw->ref_);
423 0 : self->write_op_.destroy();
424 0 : }
425 :
426 : inline std::error_code
427 HIT 413 : posix_stream_file::assign(native_handle_type handle) noexcept
428 : {
429 : // The public assign() guarantees the object is closed.
430 413 : if (svc_.pool_unavailable())
431 2 : return std::make_error_code(std::errc::operation_not_supported);
432 411 : if (auto ec = validate_file_fd(handle))
433 4 : return ec;
434 :
435 407 : fd_ = handle;
436 407 : offset_ = 0;
437 407 : gate_.open(handle);
438 407 : return {};
439 : }
440 :
441 : inline void
442 759 : posix_stream_file::retire() noexcept
443 : {
444 759 : svc_.object_pool_.recycle(this);
445 759 : }
446 :
447 : inline void
448 663 : posix_stream_file::release_work(posix_stream_file_service& svc) noexcept
449 : {
450 663 : svc.work_finished();
451 663 : }
452 :
453 : } // namespace boost::corosio::detail
454 :
455 : #endif // BOOST_COROSIO_POSIX
456 :
457 : #endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_SERVICE_HPP
|
