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