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/thread_pool.hpp>
21 :
22 : #include <limits>
23 : #include <mutex>
24 : #include <unordered_map>
25 :
26 : namespace boost::corosio::detail {
27 :
28 : /** Random-access file service for POSIX backends. */
29 : class BOOST_COROSIO_DECL posix_random_access_file_service final
30 : : public random_access_file_service
31 : {
32 : public:
33 HIT 2106 : posix_random_access_file_service(
34 : capy::execution_context& ctx, scheduler& sched)
35 4212 : : sched_(&sched)
36 2106 : , pool_(ctx)
37 : {
38 2106 : }
39 :
40 4212 : ~posix_random_access_file_service() override = default;
41 :
42 : posix_random_access_file_service(posix_random_access_file_service const&) =
43 : delete;
44 : posix_random_access_file_service&
45 : operator=(posix_random_access_file_service const&) = delete;
46 :
47 113 : io_object::implementation* construct() override
48 : {
49 113 : auto ptr = std::make_shared<posix_random_access_file>(*this);
50 113 : auto* impl = ptr.get();
51 :
52 : {
53 113 : std::lock_guard<std::mutex> lock(mutex_);
54 113 : file_list_.push_back(impl);
55 113 : file_ptrs_[impl] = std::move(ptr);
56 113 : }
57 :
58 113 : return impl;
59 113 : }
60 :
61 111 : void destroy(io_object::implementation* p) override
62 : {
63 111 : auto& impl = static_cast<posix_random_access_file&>(*p);
64 111 : impl.cancel();
65 111 : impl.close_file();
66 111 : destroy_impl(impl);
67 111 : }
68 :
69 203 : void close(io_object::handle& h) override
70 : {
71 203 : if (h.get())
72 : {
73 203 : auto& impl = static_cast<posix_random_access_file&>(*h.get());
74 203 : impl.cancel();
75 203 : impl.close_file();
76 : }
77 203 : }
78 :
79 99 : std::error_code open_file(
80 : random_access_file::implementation& impl,
81 : std::filesystem::path const& path,
82 : file_base::flags mode) override
83 : {
84 : // Unavailable in the unsafe tier: the file thread pool completes
85 : // cross-thread, which the lockless scheduler cannot accept.
86 99 : if (sched_->scheduler_locking_disabled())
87 MIS 0 : return std::make_error_code(std::errc::operation_not_supported);
88 HIT 99 : return static_cast<posix_random_access_file&>(impl).open_file(
89 99 : path, mode);
90 : }
91 :
92 2106 : void shutdown() override
93 : {
94 2106 : std::lock_guard<std::mutex> lock(mutex_);
95 2108 : for (auto* impl = file_list_.pop_front(); impl != nullptr;
96 2 : impl = file_list_.pop_front())
97 : {
98 2 : impl->cancel();
99 2 : impl->close_file();
100 : }
101 2106 : file_ptrs_.clear();
102 2106 : }
103 :
104 111 : void destroy_impl(posix_random_access_file& impl)
105 : {
106 111 : std::lock_guard<std::mutex> lock(mutex_);
107 111 : file_list_.remove(&impl);
108 111 : file_ptrs_.erase(&impl);
109 111 : }
110 :
111 332 : void post(scheduler_op* op)
112 : {
113 332 : sched_->post(op);
114 332 : }
115 :
116 : void work_started() noexcept
117 : {
118 : sched_->work_started();
119 : }
120 :
121 : void work_finished() noexcept
122 : {
123 : sched_->work_finished();
124 : }
125 :
126 : /** Return the thread pool that runs this service's file work.
127 :
128 : The pool's service is created on first use, so this can fail
129 : where a plain accessor could not. Its workers start later, on
130 : the first post, and a thread the system refuses there is
131 : reported by that post rather than thrown here.
132 :
133 : @throws std::bad_alloc If the service cannot be allocated.
134 :
135 : @return The context's shared blocking-I/O pool.
136 :
137 : @see thread_pool_ref::get
138 : */
139 336 : thread_pool& pool()
140 : {
141 336 : return pool_.get();
142 : }
143 :
144 : private:
145 : scheduler* sched_;
146 : thread_pool_ref pool_;
147 : std::mutex mutex_;
148 : intrusive_list<posix_random_access_file> file_list_;
149 : std::unordered_map<
150 : posix_random_access_file*,
151 : std::shared_ptr<posix_random_access_file>>
152 : file_ptrs_;
153 : };
154 :
155 : /** Get or create the random-access file service for the given context. */
156 : inline posix_random_access_file_service&
157 2106 : get_random_access_file_service(capy::execution_context& ctx, scheduler& sched)
158 : {
159 2106 : return ctx.make_service<posix_random_access_file_service>(sched);
160 : }
161 :
162 : // ---------------------------------------------------------------------------
163 : // posix_random_access_file inline implementations (require complete service)
164 : // ---------------------------------------------------------------------------
165 :
166 : inline std::coroutine_handle<>
167 299 : posix_random_access_file::read_some_at(
168 : std::uint64_t offset,
169 : std::coroutine_handle<> h,
170 : capy::executor_ref ex,
171 : buffer_param param,
172 : std::stop_token token,
173 : std::error_code* ec,
174 : std::size_t* bytes_out)
175 : {
176 : // Closed-object contract outranks the zero-length no-op.
177 299 : if (fd_ < 0)
178 : {
179 4 : *ec = make_error_code(std::errc::bad_file_descriptor);
180 4 : *bytes_out = 0;
181 4 : return h;
182 : }
183 :
184 295 : capy::mutable_buffer bufs[max_buffers];
185 295 : auto count = param.copy_to(bufs, max_buffers);
186 :
187 295 : if (count == 0)
188 : {
189 2 : *ec = {};
190 2 : *bytes_out = 0;
191 2 : return h;
192 : }
193 :
194 293 : auto* op = new raf_op();
195 293 : op->is_read = true;
196 293 : op->offset = offset;
197 :
198 293 : op->iovec_count = static_cast<int>(count);
199 586 : for (int i = 0; i < op->iovec_count; ++i)
200 : {
201 293 : op->iovecs[i].iov_base = bufs[i].data();
202 293 : op->iovecs[i].iov_len = bufs[i].size();
203 : }
204 :
205 293 : op->h = h;
206 293 : op->ex = ex;
207 293 : op->ec_out = ec;
208 293 : op->bytes_out = bytes_out;
209 293 : op->file_ = this;
210 293 : op->impl_ptr = this->shared_from_this();
211 293 : op->start(token);
212 :
213 293 : op->ex.on_work_started();
214 :
215 : {
216 293 : std::lock_guard<std::mutex> lock(ops_mutex_);
217 293 : outstanding_ops_.push_back(op);
218 293 : }
219 :
220 293 : static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
221 293 : if (auto pec = svc_.pool().post(static_cast<pool_work_item*>(op)))
222 : {
223 : // The pool is shutting down, or the system refused it a thread.
224 : // Nothing of this read went cross-thread, so it answers here
225 : // like the closed-descriptor and zero-length exits above rather
226 : // than through a completion the scheduler has to carry back.
227 : // destroy() is the discard the op never reaching the queue
228 : // needs: it unlinks, unwinds the work count and frees.
229 2 : op->destroy();
230 2 : *ec = pec;
231 2 : *bytes_out = 0;
232 2 : return h;
233 : }
234 291 : return std::noop_coroutine();
235 : }
236 :
237 : inline std::coroutine_handle<>
238 47 : posix_random_access_file::write_some_at(
239 : std::uint64_t offset,
240 : std::coroutine_handle<> h,
241 : capy::executor_ref ex,
242 : buffer_param param,
243 : std::stop_token token,
244 : std::error_code* ec,
245 : std::size_t* bytes_out)
246 : {
247 : // Closed-object contract outranks the zero-length no-op.
248 47 : if (fd_ < 0)
249 : {
250 2 : *ec = make_error_code(std::errc::bad_file_descriptor);
251 2 : *bytes_out = 0;
252 2 : return h;
253 : }
254 :
255 45 : capy::mutable_buffer bufs[max_buffers];
256 45 : auto count = param.copy_to(bufs, max_buffers);
257 :
258 45 : if (count == 0)
259 : {
260 2 : *ec = {};
261 2 : *bytes_out = 0;
262 2 : return h;
263 : }
264 :
265 43 : auto* op = new raf_op();
266 43 : op->is_read = false;
267 43 : op->offset = offset;
268 :
269 43 : op->iovec_count = static_cast<int>(count);
270 86 : for (int i = 0; i < op->iovec_count; ++i)
271 : {
272 43 : op->iovecs[i].iov_base = bufs[i].data();
273 43 : op->iovecs[i].iov_len = bufs[i].size();
274 : }
275 :
276 43 : op->h = h;
277 43 : op->ex = ex;
278 43 : op->ec_out = ec;
279 43 : op->bytes_out = bytes_out;
280 43 : op->file_ = this;
281 43 : op->impl_ptr = this->shared_from_this();
282 43 : op->start(token);
283 :
284 43 : op->ex.on_work_started();
285 :
286 : {
287 43 : std::lock_guard<std::mutex> lock(ops_mutex_);
288 43 : outstanding_ops_.push_back(op);
289 43 : }
290 :
291 43 : static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
292 43 : if (auto pec = svc_.pool().post(static_cast<pool_work_item*>(op)))
293 : {
294 : // The pool is shutting down, or the system refused it a thread.
295 : // Nothing of this write went cross-thread, so it answers here
296 : // like the closed-descriptor and zero-length exits above rather
297 : // than through a completion the scheduler has to carry back.
298 : // destroy() is the discard the op never reaching the queue
299 : // needs: it unlinks, unwinds the work count and frees.
300 2 : op->destroy();
301 2 : *ec = pec;
302 2 : *bytes_out = 0;
303 2 : return h;
304 : }
305 41 : return std::noop_coroutine();
306 : }
307 :
308 : // -- raf_op thread-pool work function --
309 :
310 : inline void
311 332 : posix_random_access_file::raf_op::do_work(pool_work_item* w) noexcept
312 : {
313 332 : auto* op = static_cast<raf_op*>(w);
314 332 : auto* self = op->file_;
315 :
316 332 : if (op->cancelled.load(std::memory_order_acquire))
317 : {
318 12 : op->errn = ECANCELED;
319 12 : op->bytes_transferred = 0;
320 : }
321 320 : else if (
322 640 : op->offset >
323 320 : static_cast<std::uint64_t>(std::numeric_limits<off_t>::max()))
324 : {
325 2 : op->errn = EOVERFLOW;
326 2 : op->bytes_transferred = 0;
327 : }
328 : else
329 : {
330 : ssize_t n;
331 318 : if (op->is_read)
332 : {
333 : do
334 : {
335 562 : n = ::preadv(
336 281 : self->fd_, op->iovecs, op->iovec_count,
337 281 : static_cast<off_t>(op->offset));
338 : }
339 281 : while (n < 0 && errno == EINTR);
340 : }
341 : else
342 : {
343 : do
344 : {
345 74 : n = ::pwritev(
346 37 : self->fd_, op->iovecs, op->iovec_count,
347 37 : static_cast<off_t>(op->offset));
348 : }
349 37 : while (n < 0 && errno == EINTR);
350 : }
351 :
352 318 : if (n >= 0)
353 : {
354 304 : op->errn = 0;
355 304 : op->bytes_transferred = static_cast<std::size_t>(n);
356 : }
357 : else
358 : {
359 14 : op->errn = errno;
360 14 : op->bytes_transferred = 0;
361 : }
362 : }
363 :
364 332 : self->svc_.post(static_cast<scheduler_op*>(op));
365 332 : }
366 :
367 : } // namespace boost::corosio::detail
368 :
369 : #endif // BOOST_COROSIO_POSIX
370 :
371 : #endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_FILE_SERVICE_HPP
|
