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 1225 : posix_random_access_file_service(
34 : capy::execution_context& ctx, scheduler& sched)
35 2450 : : sched_(&sched)
36 1225 : , pool_(get_or_create_pool(ctx))
37 : {
38 1225 : }
39 :
40 2450 : ~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 76 : io_object::implementation* construct() override
48 : {
49 76 : auto ptr = std::make_shared<posix_random_access_file>(*this);
50 76 : auto* impl = ptr.get();
51 :
52 : {
53 76 : std::lock_guard<std::mutex> lock(mutex_);
54 76 : file_list_.push_back(impl);
55 76 : file_ptrs_[impl] = std::move(ptr);
56 76 : }
57 :
58 76 : return impl;
59 76 : }
60 :
61 76 : void destroy(io_object::implementation* p) override
62 : {
63 76 : auto& impl = static_cast<posix_random_access_file&>(*p);
64 76 : impl.cancel();
65 76 : impl.close_file();
66 76 : destroy_impl(impl);
67 76 : }
68 :
69 136 : void close(io_object::handle& h) override
70 : {
71 136 : if (h.get())
72 : {
73 136 : auto& impl = static_cast<posix_random_access_file&>(*h.get());
74 136 : impl.cancel();
75 136 : impl.close_file();
76 : }
77 136 : }
78 :
79 64 : 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 64 : if (sched_->scheduler_locking_disabled())
87 MIS 0 : return std::make_error_code(std::errc::operation_not_supported);
88 HIT 64 : return static_cast<posix_random_access_file&>(impl).open_file(
89 64 : path, mode);
90 : }
91 :
92 1225 : void shutdown() override
93 : {
94 1225 : std::lock_guard<std::mutex> lock(mutex_);
95 1225 : 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 1225 : file_ptrs_.clear();
102 1225 : }
103 :
104 76 : void destroy_impl(posix_random_access_file& impl)
105 : {
106 76 : std::lock_guard<std::mutex> lock(mutex_);
107 76 : file_list_.remove(&impl);
108 76 : file_ptrs_.erase(&impl);
109 76 : }
110 :
111 294 : void post(scheduler_op* op)
112 : {
113 294 : sched_->post(op);
114 294 : }
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 294 : thread_pool& pool() noexcept
127 : {
128 294 : return pool_;
129 : }
130 :
131 : private:
132 1225 : static thread_pool& get_or_create_pool(capy::execution_context& ctx)
133 : {
134 1225 : auto* p = ctx.find_service<thread_pool>();
135 1225 : if (p)
136 1225 : 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 1225 : get_random_access_file_service(capy::execution_context& ctx, scheduler& sched)
153 : {
154 1225 : 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 272 : 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 272 : capy::mutable_buffer bufs[max_buffers];
172 272 : auto count = param.copy_to(bufs, max_buffers);
173 :
174 272 : if (count == 0)
175 : {
176 2 : *ec = {};
177 2 : *bytes_out = 0;
178 2 : return h;
179 : }
180 :
181 270 : auto* op = new raf_op();
182 270 : op->is_read = true;
183 270 : op->offset = offset;
184 :
185 270 : op->iovec_count = static_cast<int>(count);
186 540 : for (int i = 0; i < op->iovec_count; ++i)
187 : {
188 270 : op->iovecs[i].iov_base = bufs[i].data();
189 270 : op->iovecs[i].iov_len = bufs[i].size();
190 : }
191 :
192 270 : op->h = h;
193 270 : op->ex = ex;
194 270 : op->ec_out = ec;
195 270 : op->bytes_out = bytes_out;
196 270 : op->file_ = this;
197 270 : op->file_ref = this->shared_from_this();
198 270 : op->start(token);
199 :
200 270 : op->ex.on_work_started();
201 :
202 : {
203 270 : std::lock_guard<std::mutex> lock(ops_mutex_);
204 270 : outstanding_ops_.push_back(op);
205 270 : }
206 :
207 270 : static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
208 270 : if (!svc_.pool().post(static_cast<pool_work_item*>(op)))
209 : {
210 MIS 0 : op->cancelled.store(true, std::memory_order_release);
211 0 : svc_.post(static_cast<scheduler_op*>(op));
212 : }
213 HIT 270 : return std::noop_coroutine();
214 : }
215 :
216 : inline std::coroutine_handle<>
217 26 : posix_random_access_file::write_some_at(
218 : std::uint64_t offset,
219 : std::coroutine_handle<> h,
220 : capy::executor_ref ex,
221 : buffer_param param,
222 : std::stop_token token,
223 : std::error_code* ec,
224 : std::size_t* bytes_out)
225 : {
226 26 : capy::mutable_buffer bufs[max_buffers];
227 26 : auto count = param.copy_to(bufs, max_buffers);
228 :
229 26 : if (count == 0)
230 : {
231 2 : *ec = {};
232 2 : *bytes_out = 0;
233 2 : return h;
234 : }
235 :
236 24 : auto* op = new raf_op();
237 24 : op->is_read = false;
238 24 : op->offset = offset;
239 :
240 24 : op->iovec_count = static_cast<int>(count);
241 48 : for (int i = 0; i < op->iovec_count; ++i)
242 : {
243 24 : op->iovecs[i].iov_base = bufs[i].data();
244 24 : op->iovecs[i].iov_len = bufs[i].size();
245 : }
246 :
247 24 : op->h = h;
248 24 : op->ex = ex;
249 24 : op->ec_out = ec;
250 24 : op->bytes_out = bytes_out;
251 24 : op->file_ = this;
252 24 : op->file_ref = this->shared_from_this();
253 24 : op->start(token);
254 :
255 24 : op->ex.on_work_started();
256 :
257 : {
258 24 : std::lock_guard<std::mutex> lock(ops_mutex_);
259 24 : outstanding_ops_.push_back(op);
260 24 : }
261 :
262 24 : static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
263 24 : if (!svc_.pool().post(static_cast<pool_work_item*>(op)))
264 : {
265 MIS 0 : op->cancelled.store(true, std::memory_order_release);
266 0 : svc_.post(static_cast<scheduler_op*>(op));
267 : }
268 HIT 24 : return std::noop_coroutine();
269 : }
270 :
271 : // -- raf_op thread-pool work function --
272 :
273 : inline void
274 294 : posix_random_access_file::raf_op::do_work(pool_work_item* w) noexcept
275 : {
276 294 : auto* op = static_cast<raf_op*>(w);
277 294 : auto* self = op->file_;
278 :
279 294 : if (op->cancelled.load(std::memory_order_acquire))
280 : {
281 2 : op->errn = ECANCELED;
282 2 : op->bytes_transferred = 0;
283 : }
284 584 : else if (op->offset >
285 292 : static_cast<std::uint64_t>(std::numeric_limits<off_t>::max()))
286 : {
287 MIS 0 : op->errn = EOVERFLOW;
288 0 : op->bytes_transferred = 0;
289 : }
290 : else
291 : {
292 : ssize_t n;
293 HIT 292 : if (op->is_read)
294 : {
295 : do
296 : {
297 536 : n = ::preadv(self->fd_, op->iovecs, op->iovec_count,
298 268 : static_cast<off_t>(op->offset));
299 : }
300 268 : while (n < 0 && errno == EINTR);
301 : }
302 : else
303 : {
304 : do
305 : {
306 48 : n = ::pwritev(self->fd_, op->iovecs, op->iovec_count,
307 24 : static_cast<off_t>(op->offset));
308 : }
309 24 : while (n < 0 && errno == EINTR);
310 : }
311 :
312 292 : if (n >= 0)
313 : {
314 292 : op->errn = 0;
315 292 : op->bytes_transferred = static_cast<std::size_t>(n);
316 : }
317 : else
318 : {
319 MIS 0 : op->errn = errno;
320 0 : op->bytes_transferred = 0;
321 : }
322 : }
323 :
324 HIT 294 : self->svc_.post(static_cast<scheduler_op*>(op));
325 294 : }
326 :
327 : } // namespace boost::corosio::detail
328 :
329 : #endif // BOOST_COROSIO_POSIX
330 :
331 : #endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_FILE_SERVICE_HPP
|
