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 216 : explicit posix_stream_file_service(capy::execution_context& ctx)
36 648 : : sched_(&get_scheduler(ctx))
37 216 : , pool_(ctx)
38 : {
39 216 : }
40 :
41 432 : ~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 272 : io_object::implementation* construct() override
48 : {
49 272 : auto ptr = std::make_shared<posix_stream_file>(*this);
50 272 : auto* impl = ptr.get();
51 :
52 : {
53 272 : std::lock_guard<std::mutex> lock(mutex_);
54 272 : file_list_.push_back(impl);
55 272 : file_ptrs_[impl] = std::move(ptr);
56 272 : }
57 :
58 272 : return impl;
59 272 : }
60 :
61 270 : void destroy(io_object::implementation* p) override
62 : {
63 270 : auto& impl = static_cast<posix_stream_file&>(*p);
64 270 : impl.cancel();
65 270 : impl.close_file();
66 270 : destroy_impl(impl);
67 270 : }
68 :
69 507 : void close(io_object::handle& h) override
70 : {
71 507 : if (h.get())
72 : {
73 507 : auto& impl = static_cast<posix_stream_file&>(*h.get());
74 507 : impl.cancel();
75 507 : impl.close_file();
76 : }
77 507 : }
78 :
79 255 : 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 255 : if (sched_->scheduler_locking_disabled())
87 2 : return std::make_error_code(std::errc::operation_not_supported);
88 253 : return static_cast<posix_stream_file&>(impl).open_file(path, mode);
89 : }
90 :
91 216 : void shutdown() override
92 : {
93 216 : std::lock_guard<std::mutex> lock(mutex_);
94 218 : 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 216 : file_ptrs_.clear();
101 216 : }
102 :
103 270 : void destroy_impl(posix_stream_file& impl)
104 : {
105 270 : std::lock_guard<std::mutex> lock(mutex_);
106 270 : file_list_.remove(&impl);
107 270 : file_ptrs_.erase(&impl);
108 270 : }
109 :
110 217 : void post(scheduler_op* op)
111 : {
112 217 : sched_->post(op);
113 217 : }
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 225 : thread_pool& pool()
139 : {
140 225 : 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 : // ---------------------------------------------------------------------------
153 : // posix_stream_file inline implementations (require complete service type)
154 : // ---------------------------------------------------------------------------
155 :
156 : inline std::coroutine_handle<>
157 152 : posix_stream_file::read_some(
158 : std::coroutine_handle<> h,
159 : capy::executor_ref ex,
160 : buffer_param param,
161 : std::stop_token token,
162 : std::error_code* ec,
163 : std::size_t* bytes_out)
164 : {
165 152 : auto& op = read_op_;
166 152 : op.reset();
167 152 : op.is_read = true;
168 :
169 : // Closed-object contract outranks the zero-length no-op.
170 152 : if (fd_ < 0)
171 : {
172 6 : *ec = make_error_code(std::errc::bad_file_descriptor);
173 6 : *bytes_out = 0;
174 6 : op.cont.h = h;
175 6 : return dispatch_coro(ex, op.cont);
176 : }
177 :
178 146 : capy::mutable_buffer bufs[max_buffers];
179 146 : op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
180 :
181 146 : if (op.iovec_count == 0)
182 : {
183 2 : *ec = {};
184 2 : *bytes_out = 0;
185 2 : op.cont.h = h;
186 2 : return dispatch_coro(ex, op.cont);
187 : }
188 :
189 288 : for (int i = 0; i < op.iovec_count; ++i)
190 : {
191 144 : op.iovecs[i].iov_base = bufs[i].data();
192 144 : op.iovecs[i].iov_len = bufs[i].size();
193 : }
194 :
195 144 : op.h = h;
196 144 : op.ex = ex;
197 144 : op.ec_out = ec;
198 144 : op.bytes_out = bytes_out;
199 144 : op.start(token);
200 :
201 144 : op.ex.on_work_started();
202 :
203 144 : read_pool_op_.file_ = this;
204 144 : read_pool_op_.ref_ = this->shared_from_this();
205 144 : read_pool_op_.func_ = &posix_stream_file::do_read_work;
206 144 : if (auto pec = svc_.pool().post(&read_pool_op_))
207 : {
208 : // The pool is shutting down, or the system refused it a thread.
209 : // Nothing of this read went cross-thread, so it answers here
210 : // like the closed-descriptor and zero-length exits above rather
211 : // than through a completion the scheduler has to carry back.
212 6 : read_pool_op_.ref_.reset();
213 6 : op.stop_cb.reset();
214 6 : op.ex.on_work_finished();
215 6 : *ec = pec;
216 6 : *bytes_out = 0;
217 6 : op.cont.h = h;
218 6 : return dispatch_coro(ex, op.cont);
219 : }
220 138 : return std::noop_coroutine();
221 : }
222 :
223 : inline void
224 138 : posix_stream_file::do_read_work(pool_work_item* w) noexcept
225 : {
226 138 : auto* pw = static_cast<pool_op*>(w);
227 138 : auto* self = pw->file_;
228 138 : auto& op = self->read_op_;
229 :
230 138 : if (!op.cancelled.load(std::memory_order_acquire))
231 : {
232 : ssize_t n;
233 : do
234 : {
235 240 : n = ::preadv(
236 120 : self->fd_, op.iovecs, op.iovec_count,
237 120 : static_cast<off_t>(self->offset_));
238 : }
239 120 : while (n < 0 && errno == EINTR);
240 :
241 120 : if (n >= 0)
242 : {
243 113 : op.errn = 0;
244 113 : op.bytes_transferred = static_cast<std::size_t>(n);
245 113 : self->offset_ += static_cast<std::uint64_t>(n);
246 : }
247 : else
248 : {
249 7 : op.errn = errno;
250 7 : op.bytes_transferred = 0;
251 : }
252 : }
253 :
254 138 : op.impl_ptr = std::move(pw->ref_);
255 138 : self->svc_.post(&op);
256 138 : }
257 :
258 : inline std::coroutine_handle<>
259 89 : posix_stream_file::write_some(
260 : std::coroutine_handle<> h,
261 : capy::executor_ref ex,
262 : buffer_param param,
263 : std::stop_token token,
264 : std::error_code* ec,
265 : std::size_t* bytes_out)
266 : {
267 89 : auto& op = write_op_;
268 89 : op.reset();
269 89 : op.is_read = false;
270 :
271 : // Closed-object contract outranks the zero-length no-op.
272 89 : if (fd_ < 0)
273 : {
274 6 : *ec = make_error_code(std::errc::bad_file_descriptor);
275 6 : *bytes_out = 0;
276 6 : op.cont.h = h;
277 6 : return dispatch_coro(ex, op.cont);
278 : }
279 :
280 83 : capy::mutable_buffer bufs[max_buffers];
281 83 : op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
282 :
283 83 : if (op.iovec_count == 0)
284 : {
285 2 : *ec = {};
286 2 : *bytes_out = 0;
287 2 : op.cont.h = h;
288 2 : return dispatch_coro(ex, op.cont);
289 : }
290 :
291 162 : for (int i = 0; i < op.iovec_count; ++i)
292 : {
293 81 : op.iovecs[i].iov_base = bufs[i].data();
294 81 : op.iovecs[i].iov_len = bufs[i].size();
295 : }
296 :
297 81 : op.h = h;
298 81 : op.ex = ex;
299 81 : op.ec_out = ec;
300 81 : op.bytes_out = bytes_out;
301 81 : op.start(token);
302 :
303 81 : op.ex.on_work_started();
304 :
305 81 : write_pool_op_.file_ = this;
306 81 : write_pool_op_.ref_ = this->shared_from_this();
307 81 : write_pool_op_.func_ = &posix_stream_file::do_write_work;
308 81 : if (auto pec = svc_.pool().post(&write_pool_op_))
309 : {
310 : // The pool is shutting down, or the system refused it a thread.
311 : // Nothing of this write went cross-thread, so it answers here
312 : // like the closed-descriptor and zero-length exits above rather
313 : // than through a completion the scheduler has to carry back.
314 2 : write_pool_op_.ref_.reset();
315 2 : op.stop_cb.reset();
316 2 : op.ex.on_work_finished();
317 2 : *ec = pec;
318 2 : *bytes_out = 0;
319 2 : op.cont.h = h;
320 2 : return dispatch_coro(ex, op.cont);
321 : }
322 79 : return std::noop_coroutine();
323 : }
324 :
325 : inline void
326 79 : posix_stream_file::do_write_work(pool_work_item* w) noexcept
327 : {
328 79 : auto* pw = static_cast<pool_op*>(w);
329 79 : auto* self = pw->file_;
330 79 : auto& op = self->write_op_;
331 :
332 79 : if (!op.cancelled.load(std::memory_order_acquire))
333 : {
334 : ssize_t n;
335 : do
336 : {
337 58 : n = ::pwritev(
338 29 : self->fd_, op.iovecs, op.iovec_count,
339 29 : static_cast<off_t>(self->offset_));
340 : }
341 29 : while (n < 0 && errno == EINTR);
342 :
343 29 : if (n >= 0)
344 : {
345 22 : op.errn = 0;
346 22 : op.bytes_transferred = static_cast<std::size_t>(n);
347 22 : self->offset_ += static_cast<std::uint64_t>(n);
348 : }
349 : else
350 : {
351 7 : op.errn = errno;
352 7 : op.bytes_transferred = 0;
353 : }
354 : }
355 :
356 79 : op.impl_ptr = std::move(pw->ref_);
357 79 : self->svc_.post(&op);
358 79 : }
359 :
360 : } // namespace boost::corosio::detail
361 :
362 : #endif // BOOST_COROSIO_POSIX
363 :
364 : #endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_SERVICE_HPP
|
