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_RANDOM_ACCESS_FILE_SERVICE_HPP
11 : #define BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_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_random_access_file.hpp>
18 : #include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp>
19 : #include <boost/corosio/detail/random_access_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 <limits>
25 : #include <vector>
26 :
27 : namespace boost::corosio::detail {
28 :
29 : /** Random-access file service for POSIX backends. */
30 : class BOOST_COROSIO_DECL posix_random_access_file_service final
31 : : public random_access_file_service
32 : {
33 : friend class posix_random_access_file;
34 :
35 : public:
36 HIT 189 : explicit posix_random_access_file_service(capy::execution_context& ctx)
37 567 : : sched_(&get_scheduler(ctx))
38 189 : , pool_(ctx)
39 : {
40 189 : }
41 :
42 378 : ~posix_random_access_file_service() override = default;
43 :
44 : posix_random_access_file_service(posix_random_access_file_service const&) =
45 : delete;
46 : posix_random_access_file_service&
47 : operator=(posix_random_access_file_service const&) = delete;
48 :
49 244 : io_object::implementation* construct() override
50 : {
51 244 : return object_pool_.acquire(*this);
52 : }
53 :
54 244 : void destroy(io_object::implementation* p) override
55 : {
56 244 : auto& impl = static_cast<posix_random_access_file&>(*p);
57 244 : impl.cancel();
58 244 : impl.close_file();
59 244 : release(&impl);
60 244 : }
61 :
62 455 : void close(io_object::handle& h) override
63 : {
64 455 : if (h.get())
65 : {
66 455 : auto& impl = static_cast<posix_random_access_file&>(*h.get());
67 455 : impl.cancel();
68 455 : impl.close_file();
69 : }
70 455 : }
71 :
72 224 : std::error_code open_file(
73 : random_access_file::implementation& impl,
74 : std::filesystem::path const& path,
75 : file_base::flags mode) override
76 : {
77 224 : if (pool_unavailable())
78 MIS 0 : return std::make_error_code(std::errc::operation_not_supported);
79 HIT 224 : return static_cast<posix_random_access_file&>(impl).open_file(
80 224 : 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 233 : bool pool_unavailable() const noexcept
89 : {
90 233 : return sched_->scheduler_locking_disabled();
91 : }
92 :
93 189 : 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 189 : std::vector<posix_random_access_file*> live;
101 189 : object_pool_.shutdown(
102 189 : [&](posix_random_access_file* f)
103 : {
104 6 : acquire(f);
105 6 : live.push_back(f);
106 6 : });
107 195 : for (auto* f : live)
108 : {
109 6 : f->cancel();
110 6 : f->close_file();
111 6 : release(f);
112 : }
113 189 : }
114 :
115 505 : void post(scheduler_op* op)
116 : {
117 505 : sched_->post(op);
118 505 : }
119 :
120 515 : void work_started() noexcept
121 : {
122 515 : sched_->work_started();
123 515 : }
124 :
125 515 : void work_finished() noexcept
126 : {
127 515 : sched_->work_finished();
128 515 : }
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 515 : thread_pool& pool()
144 : {
145 515 : return pool_.get();
146 : }
147 :
148 : private:
149 : scheduler* sched_;
150 : thread_pool_ref pool_;
151 : object_pool<posix_random_access_file> object_pool_;
152 : };
153 :
154 : // ---------------------------------------------------------------------------
155 : // posix_random_access_file inline implementations (require complete service)
156 : // ---------------------------------------------------------------------------
157 :
158 : inline std::coroutine_handle<>
159 391 : posix_random_access_file::read_some_at(
160 : std::uint64_t offset,
161 : capy::continuation& cont,
162 : capy::executor_ref ex,
163 : buffer_param param,
164 : std::stop_token token,
165 : std::error_code* ec,
166 : std::size_t* bytes_out)
167 : {
168 : // Closed-object contract outranks the zero-length no-op.
169 391 : if (fd_ < 0)
170 : {
171 4 : *ec = make_error_code(std::errc::bad_file_descriptor);
172 4 : *bytes_out = 0;
173 4 : return cont.h;
174 : }
175 :
176 387 : capy::mutable_buffer bufs[max_buffers];
177 387 : auto count = param.copy_to(bufs, max_buffers);
178 :
179 387 : if (count == 0)
180 : {
181 2 : *ec = {};
182 2 : *bytes_out = 0;
183 2 : return cont.h;
184 : }
185 :
186 385 : auto* op = acquire_op();
187 385 : op->is_read = true;
188 385 : op->offset = offset;
189 385 : op->fd = fd_;
190 :
191 385 : op->iovec_count = static_cast<int>(count);
192 770 : for (int i = 0; i < op->iovec_count; ++i)
193 : {
194 385 : op->iovecs[i].iov_base = bufs[i].data();
195 385 : op->iovecs[i].iov_len = bufs[i].size();
196 : }
197 :
198 385 : op->cont = &cont;
199 385 : op->ex = ex;
200 385 : op->ec_out = ec;
201 385 : op->bytes_out = bytes_out;
202 385 : op->file_ = this;
203 385 : op->object_ref_ = detail::object_ref(this);
204 385 : op->start(token);
205 :
206 385 : svc_.work_started();
207 :
208 : {
209 385 : std::lock_guard<std::mutex> lock(ops_mutex_);
210 385 : outstanding_ops_.push_back(op);
211 385 : }
212 :
213 385 : static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
214 385 : static_cast<pool_work_item*>(op)->destroy_ = &raf_op::destroy_work;
215 385 : if (auto pec = svc_.pool().post(static_cast<pool_work_item*>(op)))
216 : {
217 : // The pool is shutting down, or the system refused it a thread.
218 : // Nothing of this read went cross-thread, so it answers here
219 : // like the closed-descriptor and zero-length exits above rather
220 : // than through a completion the scheduler has to carry back.
221 : // The op never reached the queue: it unlinks, unwinds the
222 : // work count and recycles, and the coroutine resumes below.
223 2 : op->discard();
224 2 : *ec = pec;
225 2 : *bytes_out = 0;
226 2 : return cont.h;
227 : }
228 383 : return std::noop_coroutine();
229 : }
230 :
231 : inline std::coroutine_handle<>
232 134 : posix_random_access_file::write_some_at(
233 : std::uint64_t offset,
234 : capy::continuation& cont,
235 : capy::executor_ref ex,
236 : buffer_param param,
237 : std::stop_token token,
238 : std::error_code* ec,
239 : std::size_t* bytes_out)
240 : {
241 : // Closed-object contract outranks the zero-length no-op.
242 134 : if (fd_ < 0)
243 : {
244 2 : *ec = make_error_code(std::errc::bad_file_descriptor);
245 2 : *bytes_out = 0;
246 2 : return cont.h;
247 : }
248 :
249 132 : capy::mutable_buffer bufs[max_buffers];
250 132 : auto count = param.copy_to(bufs, max_buffers);
251 :
252 132 : if (count == 0)
253 : {
254 2 : *ec = {};
255 2 : *bytes_out = 0;
256 2 : return cont.h;
257 : }
258 :
259 130 : auto* op = acquire_op();
260 130 : op->is_read = false;
261 130 : op->offset = offset;
262 130 : op->fd = fd_;
263 :
264 130 : op->iovec_count = static_cast<int>(count);
265 260 : for (int i = 0; i < op->iovec_count; ++i)
266 : {
267 130 : op->iovecs[i].iov_base = bufs[i].data();
268 130 : op->iovecs[i].iov_len = bufs[i].size();
269 : }
270 :
271 130 : op->cont = &cont;
272 130 : op->ex = ex;
273 130 : op->ec_out = ec;
274 130 : op->bytes_out = bytes_out;
275 130 : op->file_ = this;
276 130 : op->object_ref_ = detail::object_ref(this);
277 130 : op->start(token);
278 :
279 130 : svc_.work_started();
280 :
281 : {
282 130 : std::lock_guard<std::mutex> lock(ops_mutex_);
283 130 : outstanding_ops_.push_back(op);
284 130 : }
285 :
286 130 : static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
287 130 : static_cast<pool_work_item*>(op)->destroy_ = &raf_op::destroy_work;
288 130 : if (auto pec = svc_.pool().post(static_cast<pool_work_item*>(op)))
289 : {
290 : // The pool is shutting down, or the system refused it a thread.
291 : // Nothing of this write went cross-thread, so it answers here
292 : // like the closed-descriptor and zero-length exits above rather
293 : // than through a completion the scheduler has to carry back.
294 : // The op never reached the queue: it unlinks, unwinds the
295 : // work count and recycles, and the coroutine resumes below.
296 2 : op->discard();
297 2 : *ec = pec;
298 2 : *bytes_out = 0;
299 2 : return cont.h;
300 : }
301 128 : return std::noop_coroutine();
302 : }
303 :
304 : // -- raf_op thread-pool work function --
305 :
306 : inline void
307 505 : posix_random_access_file::raf_op::do_work(pool_work_item* w) noexcept
308 : {
309 505 : auto* op = static_cast<raf_op*>(w);
310 505 : auto* self = op->file_;
311 :
312 505 : if (op->cancelled.load(std::memory_order_acquire))
313 : {
314 68 : op->errn = ECANCELED;
315 68 : op->bytes_transferred = 0;
316 : }
317 437 : else if (
318 874 : op->offset >
319 437 : static_cast<std::uint64_t>(std::numeric_limits<file_off_t>::max()))
320 : {
321 2 : op->errn = EOVERFLOW;
322 2 : op->bytes_transferred = 0;
323 : }
324 435 : else if (!self->gate_.enter(op->fd))
325 : {
326 : // Closed or replaced since this was queued.
327 MIS 0 : op->errn = ECANCELED;
328 0 : op->bytes_transferred = 0;
329 : }
330 HIT 435 : else if (op->cancelled.load(std::memory_order_acquire))
331 : {
332 : // Closed and reopened onto the same number since the check
333 : // above: the gate admits the number, but not this op.
334 MIS 0 : self->gate_.leave();
335 0 : op->errn = ECANCELED;
336 0 : op->bytes_transferred = 0;
337 : }
338 : else
339 : {
340 : ssize_t n;
341 HIT 435 : if (op->is_read)
342 : {
343 : do
344 : {
345 323 : n = file_preadv(
346 323 : op->fd, op->iovecs, op->iovec_count,
347 323 : static_cast<file_off_t>(op->offset));
348 : }
349 323 : while (n < 0 && errno == EINTR);
350 : }
351 : else
352 : {
353 : do
354 : {
355 112 : n = file_pwritev(
356 112 : op->fd, op->iovecs, op->iovec_count,
357 112 : static_cast<file_off_t>(op->offset));
358 : }
359 112 : while (n < 0 && errno == EINTR);
360 : }
361 :
362 435 : if (n >= 0)
363 : {
364 421 : op->errn = 0;
365 421 : op->bytes_transferred = static_cast<std::size_t>(n);
366 : }
367 : else
368 : {
369 14 : op->errn = errno;
370 14 : op->bytes_transferred = 0;
371 : }
372 435 : self->gate_.leave();
373 : }
374 :
375 505 : self->svc_.post(static_cast<scheduler_op*>(op));
376 505 : }
377 :
378 : inline std::error_code
379 9 : posix_random_access_file::assign(native_handle_type handle) noexcept
380 : {
381 : // The public assign() guarantees the object is closed.
382 9 : if (svc_.pool_unavailable())
383 2 : return std::make_error_code(std::errc::operation_not_supported);
384 7 : if (auto ec = validate_file_fd(handle))
385 4 : return ec;
386 :
387 3 : fd_ = handle;
388 3 : gate_.open(handle);
389 3 : return {};
390 : }
391 :
392 : inline void
393 244 : posix_random_access_file::retire() noexcept
394 : {
395 244 : svc_.object_pool_.recycle(this);
396 244 : }
397 :
398 : inline void
399 515 : posix_random_access_file::release_work(
400 : posix_random_access_file_service& svc) noexcept
401 : {
402 515 : svc.work_finished();
403 515 : }
404 :
405 : } // namespace boost::corosio::detail
406 :
407 : #endif // BOOST_COROSIO_POSIX
408 :
409 : #endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_FILE_SERVICE_HPP
|
