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 1603 : posix_random_access_file_service(
34 : capy::execution_context& ctx, scheduler& sched)
35 3206 : : sched_(&sched)
36 1603 : , pool_(get_or_create_pool(ctx))
37 : {
38 1603 : }
39 :
40 3206 : ~posix_random_access_file_service() override = default;
41 :
42 : posix_random_access_file_service(
43 : posix_random_access_file_service const&) = delete;
44 : posix_random_access_file_service& operator=(
45 : posix_random_access_file_service const&) = delete;
46 :
47 96 : io_object::implementation* construct() override
48 : {
49 96 : auto ptr = std::make_shared<posix_random_access_file>(*this);
50 96 : auto* impl = ptr.get();
51 :
52 : {
53 96 : std::lock_guard<std::mutex> lock(mutex_);
54 96 : file_list_.push_back(impl);
55 96 : file_ptrs_[impl] = std::move(ptr);
56 96 : }
57 :
58 96 : return impl;
59 96 : }
60 :
61 96 : void destroy(io_object::implementation* p) override
62 : {
63 96 : auto& impl = static_cast<posix_random_access_file&>(*p);
64 96 : impl.cancel();
65 96 : impl.close_file();
66 96 : destroy_impl(impl);
67 96 : }
68 :
69 173 : void close(io_object::handle& h) override
70 : {
71 173 : if (h.get())
72 : {
73 173 : auto& impl = static_cast<posix_random_access_file&>(*h.get());
74 173 : impl.cancel();
75 173 : impl.close_file();
76 : }
77 173 : }
78 :
79 77 : 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 77 : if (sched_->scheduler_locking_disabled())
87 MIS 0 : return std::make_error_code(std::errc::operation_not_supported);
88 HIT 77 : return static_cast<posix_random_access_file&>(impl).open_file(
89 77 : path, mode);
90 : }
91 :
92 1603 : void shutdown() override
93 : {
94 1603 : std::lock_guard<std::mutex> lock(mutex_);
95 1603 : for (auto* impl = file_list_.pop_front(); impl != nullptr;
96 MIS 0 : impl = file_list_.pop_front())
97 : {
98 0 : impl->cancel();
99 0 : impl->close_file();
100 : }
101 HIT 1603 : file_ptrs_.clear();
102 1603 : }
103 :
104 96 : void destroy_impl(posix_random_access_file& impl)
105 : {
106 96 : std::lock_guard<std::mutex> lock(mutex_);
107 96 : file_list_.remove(&impl);
108 96 : file_ptrs_.erase(&impl);
109 96 : }
110 :
111 302 : void post(scheduler_op* op)
112 : {
113 302 : sched_->post(op);
114 302 : }
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 302 : thread_pool& pool() noexcept
127 : {
128 302 : return pool_;
129 : }
130 :
131 : private:
132 1603 : static thread_pool& get_or_create_pool(capy::execution_context& ctx)
133 : {
134 1603 : auto* p = ctx.find_service<thread_pool>();
135 1603 : if (p)
136 1603 : return *p;
137 MIS 0 : return ctx.make_service<thread_pool>();
138 : }
139 :
140 : scheduler* sched_;
141 : thread_pool& pool_;
142 : std::mutex mutex_;
143 : intrusive_list<posix_random_access_file> file_list_;
144 : std::unordered_map<
145 : posix_random_access_file*,
146 : std::shared_ptr<posix_random_access_file>>
147 : file_ptrs_;
148 : };
149 :
150 : /** Get or create the random-access file service for the given context. */
151 : inline posix_random_access_file_service&
152 HIT 1603 : get_random_access_file_service(capy::execution_context& ctx, scheduler& sched)
153 : {
154 1603 : return ctx.make_service<posix_random_access_file_service>(sched);
155 : }
156 :
157 : // ---------------------------------------------------------------------------
158 : // posix_random_access_file inline implementations (require complete service)
159 : // ---------------------------------------------------------------------------
160 :
161 : inline std::coroutine_handle<>
162 281 : posix_random_access_file::read_some_at(
163 : std::uint64_t offset,
164 : std::coroutine_handle<> h,
165 : capy::executor_ref ex,
166 : buffer_param param,
167 : std::stop_token token,
168 : std::error_code* ec,
169 : std::size_t* bytes_out)
170 : {
171 : // Closed-object contract outranks the zero-length no-op.
172 281 : if (fd_ < 0)
173 : {
174 4 : *ec = make_error_code(std::errc::bad_file_descriptor);
175 4 : *bytes_out = 0;
176 4 : return h;
177 : }
178 :
179 277 : capy::mutable_buffer bufs[max_buffers];
180 277 : auto count = param.copy_to(bufs, max_buffers);
181 :
182 277 : if (count == 0)
183 : {
184 2 : *ec = {};
185 2 : *bytes_out = 0;
186 2 : return h;
187 : }
188 :
189 275 : auto* op = new raf_op();
190 275 : op->is_read = true;
191 275 : op->offset = offset;
192 :
193 275 : op->iovec_count = static_cast<int>(count);
194 550 : for (int i = 0; i < op->iovec_count; ++i)
195 : {
196 275 : op->iovecs[i].iov_base = bufs[i].data();
197 275 : op->iovecs[i].iov_len = bufs[i].size();
198 : }
199 :
200 275 : op->h = h;
201 275 : op->ex = ex;
202 275 : op->ec_out = ec;
203 275 : op->bytes_out = bytes_out;
204 275 : op->file_ = this;
205 275 : op->file_ref = this->shared_from_this();
206 275 : op->start(token);
207 :
208 275 : op->ex.on_work_started();
209 :
210 : {
211 275 : std::lock_guard<std::mutex> lock(ops_mutex_);
212 275 : outstanding_ops_.push_back(op);
213 275 : }
214 :
215 275 : static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
216 275 : if (!svc_.pool().post(static_cast<pool_work_item*>(op)))
217 : {
218 MIS 0 : op->cancelled.store(true, std::memory_order_release);
219 0 : svc_.post(static_cast<scheduler_op*>(op));
220 : }
221 HIT 275 : return std::noop_coroutine();
222 : }
223 :
224 : inline std::coroutine_handle<>
225 31 : posix_random_access_file::write_some_at(
226 : std::uint64_t offset,
227 : std::coroutine_handle<> h,
228 : capy::executor_ref ex,
229 : buffer_param param,
230 : std::stop_token token,
231 : std::error_code* ec,
232 : std::size_t* bytes_out)
233 : {
234 : // Closed-object contract outranks the zero-length no-op.
235 31 : if (fd_ < 0)
236 : {
237 2 : *ec = make_error_code(std::errc::bad_file_descriptor);
238 2 : *bytes_out = 0;
239 2 : return h;
240 : }
241 :
242 29 : capy::mutable_buffer bufs[max_buffers];
243 29 : auto count = param.copy_to(bufs, max_buffers);
244 :
245 29 : if (count == 0)
246 : {
247 2 : *ec = {};
248 2 : *bytes_out = 0;
249 2 : return h;
250 : }
251 :
252 27 : auto* op = new raf_op();
253 27 : op->is_read = false;
254 27 : op->offset = offset;
255 :
256 27 : op->iovec_count = static_cast<int>(count);
257 54 : for (int i = 0; i < op->iovec_count; ++i)
258 : {
259 27 : op->iovecs[i].iov_base = bufs[i].data();
260 27 : op->iovecs[i].iov_len = bufs[i].size();
261 : }
262 :
263 27 : op->h = h;
264 27 : op->ex = ex;
265 27 : op->ec_out = ec;
266 27 : op->bytes_out = bytes_out;
267 27 : op->file_ = this;
268 27 : op->file_ref = this->shared_from_this();
269 27 : op->start(token);
270 :
271 27 : op->ex.on_work_started();
272 :
273 : {
274 27 : std::lock_guard<std::mutex> lock(ops_mutex_);
275 27 : outstanding_ops_.push_back(op);
276 27 : }
277 :
278 27 : static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
279 27 : if (!svc_.pool().post(static_cast<pool_work_item*>(op)))
280 : {
281 MIS 0 : op->cancelled.store(true, std::memory_order_release);
282 0 : svc_.post(static_cast<scheduler_op*>(op));
283 : }
284 HIT 27 : return std::noop_coroutine();
285 : }
286 :
287 : // -- raf_op thread-pool work function --
288 :
289 : inline void
290 302 : posix_random_access_file::raf_op::do_work(pool_work_item* w) noexcept
291 : {
292 302 : auto* op = static_cast<raf_op*>(w);
293 302 : auto* self = op->file_;
294 :
295 302 : if (op->cancelled.load(std::memory_order_acquire))
296 : {
297 2 : op->errn = ECANCELED;
298 2 : op->bytes_transferred = 0;
299 : }
300 600 : else if (op->offset >
301 300 : static_cast<std::uint64_t>(std::numeric_limits<off_t>::max()))
302 : {
303 2 : op->errn = EOVERFLOW;
304 2 : op->bytes_transferred = 0;
305 : }
306 : else
307 : {
308 : ssize_t n;
309 298 : if (op->is_read)
310 : {
311 : do
312 : {
313 542 : n = ::preadv(self->fd_, op->iovecs, op->iovec_count,
314 271 : static_cast<off_t>(op->offset));
315 : }
316 271 : while (n < 0 && errno == EINTR);
317 : }
318 : else
319 : {
320 : do
321 : {
322 54 : n = ::pwritev(self->fd_, op->iovecs, op->iovec_count,
323 27 : static_cast<off_t>(op->offset));
324 : }
325 27 : while (n < 0 && errno == EINTR);
326 : }
327 :
328 298 : if (n >= 0)
329 : {
330 294 : op->errn = 0;
331 294 : op->bytes_transferred = static_cast<std::size_t>(n);
332 : }
333 : else
334 : {
335 4 : op->errn = errno;
336 4 : op->bytes_transferred = 0;
337 : }
338 : }
339 :
340 302 : self->svc_.post(static_cast<scheduler_op*>(op));
341 302 : }
342 :
343 : } // namespace boost::corosio::detail
344 :
345 : #endif // BOOST_COROSIO_POSIX
346 :
347 : #endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_FILE_SERVICE_HPP
|
