TLA Line data Source code
1 : //
2 : // Copyright (c) 2026 Steve Gerbino
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_REACTOR_REACTOR_BASIC_SOCKET_HPP
11 : #define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_BASIC_SOCKET_HPP
12 :
13 : #include <boost/corosio/detail/intrusive.hpp>
14 : #include <boost/corosio/detail/native_handle.hpp>
15 : #include <boost/corosio/endpoint.hpp>
16 : #include <boost/corosio/native/detail/native_socket_base.hpp>
17 : #include <boost/corosio/native/detail/reactor/reactor_op_base.hpp>
18 : #include <boost/corosio/native/detail/make_err.hpp>
19 : #include <boost/corosio/native/detail/endpoint_convert.hpp>
20 :
21 : #include <memory>
22 : #include <mutex>
23 : #include <utility>
24 :
25 : #include <errno.h>
26 : #include <netinet/in.h>
27 : #include <sys/socket.h>
28 : #include <unistd.h>
29 :
30 : namespace boost::corosio::detail {
31 :
32 : /** CRTP base for reactor-backed socket implementations.
33 :
34 : Extracts the shared data members, virtual overrides, and
35 : cancel/close/register logic that is identical across TCP
36 : (reactor_stream_socket) and UDP (reactor_datagram_socket).
37 :
38 : Derived classes provide CRTP callbacks that enumerate their
39 : specific op slots so cancel/close can iterate them generically.
40 :
41 : @tparam Derived The concrete socket type (CRTP).
42 : @tparam ImplBase The public vtable base (tcp_socket::implementation
43 : or udp_socket::implementation).
44 : @tparam Service The backend's service type.
45 : @tparam DescState The backend's descriptor_state type.
46 : @tparam Endpoint The endpoint type (endpoint or local_endpoint).
47 : */
48 : template<
49 : class Derived,
50 : class ImplBase,
51 : class Service,
52 : class DescState,
53 : class Endpoint = endpoint>
54 : class reactor_basic_socket
55 : : public native_socket_base<Derived, ImplBase, Endpoint>
56 : , public intrusive_list<Derived>::node
57 : {
58 : friend Derived;
59 :
60 : template<class, class, class, class, class, class, class, class, class>
61 : friend class reactor_stream_socket;
62 :
63 : template<class, class, class, class, class, class, class, class, class, class, class>
64 : friend class reactor_datagram_socket;
65 :
66 HIT 21975 : explicit reactor_basic_socket(Service& svc) noexcept : svc_(svc) {}
67 :
68 : protected:
69 : // fd_ / local_endpoint_ and the synchronous accessors (native_handle,
70 : // is_open, set_option/get_option, set_socket/set_local_endpoint, do_bind)
71 : // live in native_socket_base — the readiness/completion-agnostic base
72 : // shared with io_uring's sockets. The using-declarations make the
73 : // inherited members visible to this template's own unqualified
74 : // references below (two-phase lookup).
75 : using native_socket_base<Derived, ImplBase, Endpoint>::fd_;
76 : using native_socket_base<Derived, ImplBase, Endpoint>::local_endpoint_;
77 :
78 : Service& svc_;
79 :
80 : public:
81 : /// Per-descriptor state for persistent reactor registration.
82 : DescState desc_state_;
83 :
84 21975 : ~reactor_basic_socket() override = default;
85 :
86 : /** Assign the fd, initialize descriptor state, and register with
87 : the reactor.
88 :
89 : @param fd The descriptor to adopt.
90 :
91 : @return The error if the reactor rejects the descriptor, in
92 : which case the implementation is left closed and the caller
93 : retains ownership of @a fd; otherwise a default constructed
94 : error code.
95 : */
96 7648 : std::error_code init_and_register(int fd) noexcept
97 : {
98 7648 : fd_ = fd;
99 7648 : desc_state_.fd = fd;
100 : {
101 7648 : std::lock_guard lock(desc_state_.mutex);
102 7648 : desc_state_.read_op = nullptr;
103 7648 : desc_state_.write_op = nullptr;
104 7648 : desc_state_.connect_op = nullptr;
105 7648 : }
106 7648 : if (auto ec = svc_.scheduler().register_descriptor(fd, &desc_state_))
107 : {
108 : // Undo the partial state so a failed adopt is
109 : // indistinguishable from a closed implementation.
110 1 : fd_ = -1;
111 1 : desc_state_.fd = -1;
112 1 : desc_state_.registered_events = 0;
113 1 : return ec;
114 : }
115 7647 : return {};
116 : }
117 :
118 : /** Register an op with the reactor.
119 :
120 : Handles cached edge events and deferred cancellation.
121 : Called on the EAGAIN/EINPROGRESS path when speculative
122 : I/O failed.
123 : */
124 : template<class Op>
125 : void register_op(
126 : Op& op,
127 : reactor_op_base*& desc_slot,
128 : bool& ready_flag,
129 : bool& cancel_flag,
130 : bool is_write_direction = false) noexcept;
131 :
132 : /** Cancel a single pending operation.
133 :
134 : Claims the operation from its descriptor_state slot under
135 : the mutex and posts it to the scheduler as cancelled.
136 : Derived must implement:
137 : op_to_desc_slot(Op&) -> reactor_op_base**
138 : op_to_cancel_flag(Op&) -> bool*
139 : */
140 : template<class Op>
141 : void cancel_single_op(Op& op) noexcept;
142 :
143 : /** Cancel all pending operations.
144 :
145 : Invoked by the derived class's cancel() override.
146 : Derived must implement:
147 : for_each_op(auto fn)
148 : for_each_desc_entry(auto fn)
149 : */
150 : void do_cancel() noexcept;
151 :
152 : /** Close the socket and cancel pending operations.
153 :
154 : Invoked by the derived class's close_socket(). The
155 : derived class may add backend-specific cleanup after
156 : calling this method.
157 : Derived must implement:
158 : for_each_op(auto fn)
159 : for_each_desc_entry(auto fn)
160 : */
161 : void do_close_socket() noexcept;
162 :
163 : /** Release the socket without closing the fd.
164 :
165 : Like do_close_socket() but does not call ::close().
166 : Returns the fd so the caller can take ownership.
167 : */
168 : native_handle_type do_release_socket() noexcept;
169 : };
170 :
171 : template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
172 : template<class Op>
173 : void
174 8151 : reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::register_op(
175 : Op& op,
176 : reactor_op_base*& desc_slot,
177 : bool& ready_flag,
178 : bool& cancel_flag,
179 : bool is_write_direction) noexcept
180 : {
181 8151 : svc_.work_started();
182 :
183 8151 : std::lock_guard lock(desc_state_.mutex);
184 8151 : bool io_done = false;
185 8151 : if (ready_flag)
186 : {
187 268 : ready_flag = false;
188 268 : op.perform_io();
189 268 : io_done = (op.errn != EAGAIN && op.errn != EWOULDBLOCK);
190 268 : if (!io_done)
191 268 : op.errn = 0;
192 : }
193 :
194 8151 : if (cancel_flag)
195 : {
196 MIS 0 : cancel_flag = false;
197 0 : op.cancelled.store(true, std::memory_order_relaxed);
198 : }
199 :
200 HIT 8151 : if (io_done || op.cancelled.load(std::memory_order_acquire))
201 : {
202 26 : svc_.post(&op);
203 26 : svc_.work_finished();
204 : }
205 : else
206 : {
207 8125 : desc_slot = &op;
208 :
209 : // Select must rebuild its fd_sets when a write-direction op
210 : // is parked, so select() watches for writability. Compiled
211 : // away to nothing for epoll and kqueue.
212 : if constexpr (requires { Service::needs_write_notification; })
213 : {
214 : if constexpr (Service::needs_write_notification)
215 : {
216 3841 : if (is_write_direction)
217 3392 : svc_.scheduler().notify_reactor();
218 : }
219 : }
220 : }
221 8151 : }
222 :
223 : template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
224 : template<class Op>
225 : void
226 278 : reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::cancel_single_op(
227 : Op& op) noexcept
228 : {
229 278 : auto self = this->weak_from_this().lock();
230 278 : if (!self)
231 MIS 0 : return;
232 :
233 HIT 278 : op.request_cancel();
234 :
235 278 : auto* d = static_cast<Derived*>(this);
236 278 : reactor_op_base** desc_op_ptr = d->op_to_desc_slot(op);
237 :
238 278 : if (desc_op_ptr)
239 : {
240 278 : reactor_op_base* claimed = nullptr;
241 : {
242 278 : std::lock_guard lock(desc_state_.mutex);
243 278 : if (*desc_op_ptr == &op)
244 234 : claimed = std::exchange(*desc_op_ptr, nullptr);
245 : // Not in the slot: request_cancel() above already set
246 : // op.cancelled, which register_op consults before parking
247 : // and the completion decode consults on delivery. Latching
248 : // a descriptor flag here instead would outlive this op and
249 : // cancel the next wait in the same direction.
250 278 : }
251 278 : if (claimed)
252 : {
253 234 : op.impl_ptr = self;
254 234 : svc_.post(&op);
255 234 : svc_.work_finished();
256 : }
257 : }
258 278 : }
259 :
260 : template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
261 : void
262 233 : reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
263 : do_cancel() noexcept
264 : {
265 233 : auto self = this->weak_from_this().lock();
266 233 : if (!self)
267 MIS 0 : return;
268 :
269 HIT 233 : auto* d = static_cast<Derived*>(this);
270 :
271 1665 : d->for_each_op([](auto& op) { op.request_cancel(); });
272 :
273 : // Claim ops under a single lock acquisition
274 : struct claimed_entry
275 : {
276 : reactor_op_base* op = nullptr;
277 : reactor_op_base* base = nullptr;
278 : };
279 : // Max 8 ops: conn, rd, wr, wait_rd, wait_wr, wait_er, recv_rd, send_wr
280 233 : claimed_entry claimed[8];
281 233 : int count = 0;
282 :
283 : {
284 233 : std::lock_guard lock(desc_state_.mutex);
285 3097 : d->for_each_desc_entry([&](auto& op, reactor_op_base*& desc_slot) {
286 1432 : if (desc_slot == &op)
287 : {
288 137 : claimed[count].op = std::exchange(desc_slot, nullptr);
289 137 : claimed[count].base = &op;
290 137 : ++count;
291 : }
292 : });
293 233 : }
294 :
295 370 : for (int i = 0; i < count; ++i)
296 : {
297 137 : claimed[i].base->impl_ptr = self;
298 137 : svc_.post(claimed[i].base);
299 137 : svc_.work_finished();
300 : }
301 233 : }
302 :
303 : template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
304 : void
305 66252 : reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
306 : do_close_socket() noexcept
307 : {
308 66252 : auto self = this->weak_from_this().lock();
309 66252 : if (self)
310 : {
311 66252 : auto* d = static_cast<Derived*>(this);
312 :
313 467236 : d->for_each_op([](auto& op) { op.request_cancel(); });
314 :
315 : struct claimed_entry
316 : {
317 : reactor_op_base* base = nullptr;
318 : };
319 66252 : claimed_entry claimed[8];
320 66252 : int count = 0;
321 :
322 : {
323 66252 : std::lock_guard lock(desc_state_.mutex);
324 66252 : d->for_each_desc_entry(
325 801968 : [&](auto& /*op*/, reactor_op_base*& desc_slot) {
326 400984 : auto* c = std::exchange(desc_slot, nullptr);
327 400984 : if (c)
328 : {
329 28 : claimed[count].base = c;
330 28 : ++count;
331 : }
332 : });
333 66252 : desc_state_.read_ready = false;
334 66252 : desc_state_.write_ready = false;
335 66252 : desc_state_.read_cancel_pending = false;
336 66252 : desc_state_.write_cancel_pending = false;
337 66252 : desc_state_.connect_cancel_pending = false;
338 66252 : desc_state_.wait_read_cancel_pending = false;
339 66252 : desc_state_.wait_write_cancel_pending = false;
340 66252 : desc_state_.wait_error_cancel_pending = false;
341 :
342 66252 : if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
343 694 : desc_state_.impl_ref_ = self;
344 66252 : }
345 :
346 66280 : for (int i = 0; i < count; ++i)
347 : {
348 28 : claimed[i].base->impl_ptr = self;
349 28 : svc_.post(claimed[i].base);
350 28 : svc_.work_finished();
351 : }
352 : }
353 :
354 66252 : if (fd_ >= 0)
355 : {
356 14662 : if (desc_state_.registered_events != 0)
357 14662 : svc_.scheduler().deregister_descriptor(fd_);
358 14662 : ::close(fd_);
359 14662 : fd_ = -1;
360 : }
361 :
362 66252 : desc_state_.fd = -1;
363 66252 : desc_state_.registered_events = 0;
364 :
365 66252 : local_endpoint_ = Endpoint{};
366 66252 : }
367 :
368 : template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
369 : native_handle_type
370 10 : reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
371 : do_release_socket() noexcept
372 : {
373 : // Cancel pending ops (same as do_close_socket)
374 10 : auto self = this->weak_from_this().lock();
375 10 : if (self)
376 : {
377 10 : auto* d = static_cast<Derived*>(this);
378 :
379 78 : d->for_each_op([](auto& op) { op.request_cancel(); });
380 :
381 : struct claimed_entry
382 : {
383 : reactor_op_base* base = nullptr;
384 : };
385 10 : claimed_entry claimed[8];
386 10 : int count = 0;
387 :
388 : {
389 10 : std::lock_guard lock(desc_state_.mutex);
390 10 : d->for_each_desc_entry(
391 136 : [&](auto& /*op*/, reactor_op_base*& desc_slot) {
392 68 : auto* c = std::exchange(desc_slot, nullptr);
393 68 : if (c)
394 : {
395 6 : claimed[count].base = c;
396 6 : ++count;
397 : }
398 : });
399 10 : desc_state_.read_ready = false;
400 10 : desc_state_.write_ready = false;
401 10 : desc_state_.read_cancel_pending = false;
402 10 : desc_state_.write_cancel_pending = false;
403 10 : desc_state_.connect_cancel_pending = false;
404 10 : desc_state_.wait_read_cancel_pending = false;
405 10 : desc_state_.wait_write_cancel_pending = false;
406 10 : desc_state_.wait_error_cancel_pending = false;
407 :
408 10 : if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
409 3 : desc_state_.impl_ref_ = self;
410 10 : }
411 :
412 16 : for (int i = 0; i < count; ++i)
413 : {
414 6 : claimed[i].base->impl_ptr = self;
415 6 : svc_.post(claimed[i].base);
416 6 : svc_.work_finished();
417 : }
418 : }
419 :
420 10 : native_handle_type released = fd_;
421 :
422 10 : if (fd_ >= 0)
423 : {
424 10 : if (desc_state_.registered_events != 0)
425 10 : svc_.scheduler().deregister_descriptor(fd_);
426 : // Do NOT close -- caller takes ownership
427 10 : fd_ = -1;
428 : }
429 :
430 10 : desc_state_.fd = -1;
431 10 : desc_state_.registered_events = 0;
432 :
433 10 : local_endpoint_ = Endpoint{};
434 :
435 20 : return released;
436 10 : }
437 :
438 : } // namespace boost::corosio::detail
439 :
440 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_BASIC_SOCKET_HPP
|
