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