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/thread_pool.hpp>
21 :
22 : #include <mutex>
23 : #include <unordered_map>
24 :
25 : namespace boost::corosio::detail {
26 :
27 : /** Stream file service for POSIX backends.
28 :
29 : Owns all posix_stream_file instances. Thread lifecycle is
30 : managed by the thread_pool service (shared with resolver).
31 : */
32 : class BOOST_COROSIO_DECL posix_stream_file_service final
33 : : public file_service
34 : {
35 : public:
36 HIT 1603 : posix_stream_file_service(
37 : capy::execution_context& ctx, scheduler& sched)
38 3206 : : sched_(&sched)
39 1603 : , pool_(get_or_create_pool(ctx))
40 : {
41 1603 : }
42 :
43 3206 : ~posix_stream_file_service() override = default;
44 :
45 : posix_stream_file_service(posix_stream_file_service const&) = delete;
46 : posix_stream_file_service& operator=(posix_stream_file_service const&) = delete;
47 :
48 95 : io_object::implementation* construct() override
49 : {
50 95 : auto ptr = std::make_shared<posix_stream_file>(*this);
51 95 : auto* impl = ptr.get();
52 :
53 : {
54 95 : std::lock_guard<std::mutex> lock(mutex_);
55 95 : file_list_.push_back(impl);
56 95 : file_ptrs_[impl] = std::move(ptr);
57 95 : }
58 :
59 95 : return impl;
60 95 : }
61 :
62 95 : void destroy(io_object::implementation* p) override
63 : {
64 95 : auto& impl = static_cast<posix_stream_file&>(*p);
65 95 : impl.cancel();
66 95 : impl.close_file();
67 95 : destroy_impl(impl);
68 95 : }
69 :
70 168 : void close(io_object::handle& h) override
71 : {
72 168 : if (h.get())
73 : {
74 168 : auto& impl = static_cast<posix_stream_file&>(*h.get());
75 168 : impl.cancel();
76 168 : impl.close_file();
77 : }
78 168 : }
79 :
80 75 : std::error_code open_file(
81 : stream_file::implementation& impl,
82 : std::filesystem::path const& path,
83 : file_base::flags mode) override
84 : {
85 : // Unavailable in the unsafe tier: the file thread pool completes
86 : // cross-thread, which the lockless scheduler cannot accept.
87 75 : if (sched_->scheduler_locking_disabled())
88 2 : return std::make_error_code(std::errc::operation_not_supported);
89 73 : return static_cast<posix_stream_file&>(impl).open_file(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 95 : void destroy_impl(posix_stream_file& impl)
105 : {
106 95 : std::lock_guard<std::mutex> lock(mutex_);
107 95 : file_list_.remove(&impl);
108 95 : file_ptrs_.erase(&impl);
109 95 : }
110 :
111 39 : void post(scheduler_op* op)
112 : {
113 39 : sched_->post(op);
114 39 : }
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 39 : thread_pool& pool() noexcept
127 : {
128 39 : 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_stream_file> file_list_;
144 : std::unordered_map<posix_stream_file*, std::shared_ptr<posix_stream_file>>
145 : file_ptrs_;
146 : };
147 :
148 : /** Get or create the stream file service for the given context. */
149 : inline posix_stream_file_service&
150 HIT 1603 : get_stream_file_service(capy::execution_context& ctx, scheduler& sched)
151 : {
152 1603 : return ctx.make_service<posix_stream_file_service>(sched);
153 : }
154 :
155 : // ---------------------------------------------------------------------------
156 : // posix_stream_file inline implementations (require complete service type)
157 : // ---------------------------------------------------------------------------
158 :
159 : inline std::coroutine_handle<>
160 28 : posix_stream_file::read_some(
161 : std::coroutine_handle<> h,
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 28 : auto& op = read_op_;
169 28 : op.reset();
170 28 : op.is_read = true;
171 :
172 : // Closed-object contract outranks the zero-length no-op.
173 28 : if (fd_ < 0)
174 : {
175 6 : *ec = make_error_code(std::errc::bad_file_descriptor);
176 6 : *bytes_out = 0;
177 6 : op.cont.h = h;
178 6 : return dispatch_coro(ex, op.cont);
179 : }
180 :
181 22 : capy::mutable_buffer bufs[max_buffers];
182 22 : op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
183 :
184 22 : if (op.iovec_count == 0)
185 : {
186 2 : *ec = {};
187 2 : *bytes_out = 0;
188 2 : op.cont.h = h;
189 2 : return dispatch_coro(ex, op.cont);
190 : }
191 :
192 40 : for (int i = 0; i < op.iovec_count; ++i)
193 : {
194 20 : op.iovecs[i].iov_base = bufs[i].data();
195 20 : op.iovecs[i].iov_len = bufs[i].size();
196 : }
197 :
198 20 : op.h = h;
199 20 : op.ex = ex;
200 20 : op.ec_out = ec;
201 20 : op.bytes_out = bytes_out;
202 20 : op.start(token);
203 :
204 20 : op.ex.on_work_started();
205 :
206 20 : read_pool_op_.file_ = this;
207 20 : read_pool_op_.ref_ = this->shared_from_this();
208 20 : read_pool_op_.func_ = &posix_stream_file::do_read_work;
209 20 : if (!svc_.pool().post(&read_pool_op_))
210 : {
211 MIS 0 : op.impl_ref = std::move(read_pool_op_.ref_);
212 0 : op.cancelled.store(true, std::memory_order_release);
213 0 : svc_.post(&read_op_);
214 : }
215 HIT 20 : return std::noop_coroutine();
216 : }
217 :
218 : inline void
219 20 : posix_stream_file::do_read_work(pool_work_item* w) noexcept
220 : {
221 20 : auto* pw = static_cast<pool_op*>(w);
222 20 : auto* self = pw->file_;
223 20 : auto& op = self->read_op_;
224 :
225 20 : if (!op.cancelled.load(std::memory_order_acquire))
226 : {
227 : ssize_t n;
228 : do
229 : {
230 36 : n = ::preadv(self->fd_, op.iovecs, op.iovec_count,
231 18 : static_cast<off_t>(self->offset_));
232 : }
233 18 : while (n < 0 && errno == EINTR);
234 :
235 18 : if (n >= 0)
236 : {
237 16 : op.errn = 0;
238 16 : op.bytes_transferred = static_cast<std::size_t>(n);
239 16 : self->offset_ += static_cast<std::uint64_t>(n);
240 : }
241 : else
242 : {
243 2 : op.errn = errno;
244 2 : op.bytes_transferred = 0;
245 : }
246 : }
247 :
248 20 : op.impl_ref = std::move(pw->ref_);
249 20 : self->svc_.post(&op);
250 20 : }
251 :
252 : inline std::coroutine_handle<>
253 27 : posix_stream_file::write_some(
254 : std::coroutine_handle<> h,
255 : capy::executor_ref ex,
256 : buffer_param param,
257 : std::stop_token token,
258 : std::error_code* ec,
259 : std::size_t* bytes_out)
260 : {
261 27 : auto& op = write_op_;
262 27 : op.reset();
263 27 : op.is_read = false;
264 :
265 : // Closed-object contract outranks the zero-length no-op.
266 27 : if (fd_ < 0)
267 : {
268 6 : *ec = make_error_code(std::errc::bad_file_descriptor);
269 6 : *bytes_out = 0;
270 6 : op.cont.h = h;
271 6 : return dispatch_coro(ex, op.cont);
272 : }
273 :
274 21 : capy::mutable_buffer bufs[max_buffers];
275 21 : op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
276 :
277 21 : if (op.iovec_count == 0)
278 : {
279 2 : *ec = {};
280 2 : *bytes_out = 0;
281 2 : op.cont.h = h;
282 2 : return dispatch_coro(ex, op.cont);
283 : }
284 :
285 38 : for (int i = 0; i < op.iovec_count; ++i)
286 : {
287 19 : op.iovecs[i].iov_base = bufs[i].data();
288 19 : op.iovecs[i].iov_len = bufs[i].size();
289 : }
290 :
291 19 : op.h = h;
292 19 : op.ex = ex;
293 19 : op.ec_out = ec;
294 19 : op.bytes_out = bytes_out;
295 19 : op.start(token);
296 :
297 19 : op.ex.on_work_started();
298 :
299 19 : write_pool_op_.file_ = this;
300 19 : write_pool_op_.ref_ = this->shared_from_this();
301 19 : write_pool_op_.func_ = &posix_stream_file::do_write_work;
302 19 : if (!svc_.pool().post(&write_pool_op_))
303 : {
304 MIS 0 : op.impl_ref = std::move(write_pool_op_.ref_);
305 0 : op.cancelled.store(true, std::memory_order_release);
306 0 : svc_.post(&write_op_);
307 : }
308 HIT 19 : return std::noop_coroutine();
309 : }
310 :
311 : inline void
312 19 : posix_stream_file::do_write_work(pool_work_item* w) noexcept
313 : {
314 19 : auto* pw = static_cast<pool_op*>(w);
315 19 : auto* self = pw->file_;
316 19 : auto& op = self->write_op_;
317 :
318 19 : if (!op.cancelled.load(std::memory_order_acquire))
319 : {
320 : ssize_t n;
321 : do
322 : {
323 38 : n = ::pwritev(self->fd_, op.iovecs, op.iovec_count,
324 19 : static_cast<off_t>(self->offset_));
325 : }
326 19 : while (n < 0 && errno == EINTR);
327 :
328 19 : if (n >= 0)
329 : {
330 17 : op.errn = 0;
331 17 : op.bytes_transferred = static_cast<std::size_t>(n);
332 17 : self->offset_ += static_cast<std::uint64_t>(n);
333 : }
334 : else
335 : {
336 2 : op.errn = errno;
337 2 : op.bytes_transferred = 0;
338 : }
339 : }
340 :
341 19 : op.impl_ref = std::move(pw->ref_);
342 19 : self->svc_.post(&op);
343 19 : }
344 :
345 : } // namespace boost::corosio::detail
346 :
347 : #endif // BOOST_COROSIO_POSIX
348 :
349 : #endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_SERVICE_HPP
|
