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