include/boost/corosio/native/detail/epoll/epoll_socket_service.hpp

81.0% Lines (336/415) 93.3% List of functions (28/30)

[ ] prev/next file n p prev/next uncovered line

f(x) Functions (30)

Function Calls Lines Branches Blocks

Line	Hits	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_EPOLL_EPOLL_SOCKET_SERVICE_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SOCKET_SERVICE_HPP
12
13		#include <boost/corosio/detail/platform.hpp>
14
15		#if BOOST_COROSIO_HAS_EPOLL
16
17		#include <boost/corosio/detail/config.hpp>
18		#include <boost/capy/ex/execution_context.hpp>
19		#include <boost/corosio/detail/socket_service.hpp>
20
21		#include <boost/corosio/native/detail/epoll/epoll_socket.hpp>
22		#include <boost/corosio/native/detail/epoll/epoll_scheduler.hpp>
23
24		#include <boost/corosio/native/detail/endpoint_convert.hpp>
25		#include <boost/corosio/native/detail/make_err.hpp>
26		#include <boost/corosio/detail/dispatch_coro.hpp>
27		#include <boost/corosio/detail/except.hpp>
28		#include <boost/capy/buffers.hpp>
29
30		#include <coroutine>
31		#include <mutex>
32		#include <unordered_map>
33		#include <utility>
34
35		#include <errno.h>
36		#include <netinet/in.h>
37		#include <netinet/tcp.h>
38		#include <sys/epoll.h>
39		#include <sys/socket.h>
40		#include <unistd.h>
41
42		/*
43		epoll Socket Implementation
44		===========================
45
46		Each I/O operation follows the same pattern:
47		1. Try the syscall immediately (non-blocking socket)
48		2. If it succeeds or fails with a real error, post to completion queue
49		3. If EAGAIN/EWOULDBLOCK, register with epoll and wait
50
51		This "try first" approach avoids unnecessary epoll round-trips for
52		operations that can complete immediately (common for small reads/writes
53		on fast local connections).
54
55		One-Shot Registration
56		---------------------
57		We use one-shot epoll registration: each operation registers, waits for
58		one event, then unregisters. This simplifies the state machine since we
59		don't need to track whether an fd is currently registered or handle
60		re-arming. The tradeoff is slightly more epoll_ctl calls, but the
61		simplicity is worth it.
62
63		Cancellation
64		------------
65		See op.hpp for the completion/cancellation race handling via the
66		`registered` atomic. cancel() must complete pending operations (post
67		them with cancelled flag) so coroutines waiting on them can resume.
68		close_socket() calls cancel() first to ensure this.
69
70		Impl Lifetime with shared_ptr
71		-----------------------------
72		Socket impls use enable_shared_from_this. The service owns impls via
73		shared_ptr maps (socket_ptrs_) keyed by raw pointer for O(1) lookup and
74		removal. When a user calls close(), we call cancel() which posts pending
75		ops to the scheduler.
76
77		CRITICAL: The posted ops must keep the impl alive until they complete.
78		Otherwise the scheduler would process a freed op (use-after-free). The
79		cancel() method captures shared_from_this() into op.impl_ptr before
80		posting. When the op completes, impl_ptr is cleared, allowing the impl
81		to be destroyed if no other references exist.
82
83		Service Ownership
84		-----------------
85		epoll_socket_service owns all socket impls. destroy_impl() removes the
86		shared_ptr from the map, but the impl may survive if ops still hold
87		impl_ptr refs. shutdown() closes all sockets and clears the map; any
88		in-flight ops will complete and release their refs.
89		*/
90
91		namespace boost::corosio::detail {
92
93		/** State for epoll socket service. */
94		class epoll_socket_state
95		{
96		public:
97	244x	explicit epoll_socket_state(epoll_scheduler& sched) noexcept : sched_(sched)
98		{
99	244x	}
100
101		epoll_scheduler& sched_;
102		std::mutex mutex_;
103		intrusive_list<epoll_socket> socket_list_;
104		std::unordered_map<epoll_socket*, std::shared_ptr<epoll_socket>>
105		socket_ptrs_;
106		};
107
108		/** epoll socket service implementation.
109
110		Inherits from socket_service to enable runtime polymorphism.
111		Uses key_type = socket_service for service lookup.
112		*/
113		class BOOST_COROSIO_DECL epoll_socket_service final : public socket_service
114		{
115		public:
116		explicit epoll_socket_service(capy::execution_context& ctx);
117		~epoll_socket_service() override;
118
119		epoll_socket_service(epoll_socket_service const&) = delete;
120		epoll_socket_service& operator=(epoll_socket_service const&) = delete;
121
122		void shutdown() override;
123
124		io_object::implementation* construct() override;
125		void destroy(io_object::implementation*) override;
126		void close(io_object::handle&) override;
127		std::error_code open_socket(
128		tcp_socket::implementation& impl,
129		int family,
130		int type,
131		int protocol) override;
132
133	342658x	epoll_scheduler& scheduler() const noexcept
134		{
135	342658x	return state_->sched_;
136		}
137		void post(epoll_op* op);
138		void work_started() noexcept;
139		void work_finished() noexcept;
140
141		private:
142		std::unique_ptr<epoll_socket_state> state_;
143		};
144
145		//--------------------------------------------------------------------------
146		//
147		// Implementation
148		//
149		//--------------------------------------------------------------------------
150
151		// Register an op with the reactor, handling cached edge events.
152		// Called under the EAGAIN/EINPROGRESS path when speculative I/O failed.
153		inline void
154	5025x	epoll_socket::register_op(
155		epoll_op& op,
156		epoll_op*& desc_slot,
157		bool& ready_flag,
158		bool& cancel_flag) noexcept
159		{
160	5025x	svc_.work_started();
161
162	5025x	std::lock_guard lock(desc_state_.mutex);
163	5025x	bool io_done = false;
164	5025x	if (ready_flag)
165		{
166	142x	ready_flag = false;
167	142x	op.perform_io();
168	142x	io_done = (op.errn != EAGAIN && op.errn != EWOULDBLOCK);
169	142x	if (!io_done)
170	142x	op.errn = 0;
171		}
172
173	5025x	if (cancel_flag)
174		{
175	95x	cancel_flag = false;
176	95x	op.cancelled.store(true, std::memory_order_relaxed);
177		}
178
179	5025x	if (io_done \|\| op.cancelled.load(std::memory_order_acquire))
180		{
181	95x	svc_.post(&op);
182	95x	svc_.work_finished();
183		}
184		else
185		{
186	4930x	desc_slot = &op;
187		}
188	5025x	}
189
190		inline void
191	104x	epoll_op::canceller::operator()() const noexcept
192		{
193	104x	op->cancel();
194	104x	}
195
196		inline void
197	✗	epoll_connect_op::cancel() noexcept
198		{
199	✗	if (socket_impl_)
200	✗	socket_impl_->cancel_single_op(*this);
201		else
202	✗	request_cancel();
203	✗	}
204
205		inline void
206	98x	epoll_read_op::cancel() noexcept
207		{
208	98x	if (socket_impl_)
209	98x	socket_impl_->cancel_single_op(*this);
210		else
211	✗	request_cancel();
212	98x	}
213
214		inline void
215	✗	epoll_write_op::cancel() noexcept
216		{
217	✗	if (socket_impl_)
218	✗	socket_impl_->cancel_single_op(*this);
219		else
220	✗	request_cancel();
221	✗	}
222
223		inline void
224	53193x	epoll_op::operator()()
225		{
226	53193x	stop_cb.reset();
227
228	53193x	socket_impl_->svc_.scheduler().reset_inline_budget();
229
230	53193x	if (cancelled.load(std::memory_order_acquire))
231	205x	*ec_out = capy::error::canceled;
232	52988x	else if (errn != 0)
233	✗	*ec_out = make_err(errn);
234	52988x	else if (is_read_operation() && bytes_transferred == 0)
235	✗	*ec_out = capy::error::eof;
236		else
237	52988x	*ec_out = {};
238
239	53193x	*bytes_out = bytes_transferred;
240
241		// Move to stack before resuming coroutine. The coroutine might close
242		// the socket, releasing the last wrapper ref. If impl_ptr were the
243		// last ref and we destroyed it while still in operator(), we'd have
244		// use-after-free. Moving to local ensures destruction happens at
245		// function exit, after all member accesses are complete.
246	53193x	capy::executor_ref saved_ex(ex);
247	53193x	std::coroutine_handle<> saved_h(h);
248	53193x	auto prevent_premature_destruction = std::move(impl_ptr);
249	53193x	dispatch_coro(saved_ex, saved_h).resume();
250	53193x	}
251
252		inline void
253	4824x	epoll_connect_op::operator()()
254		{
255	4824x	stop_cb.reset();
256
257	4824x	socket_impl_->svc_.scheduler().reset_inline_budget();
258
259	4824x	bool success = (errn == 0 && !cancelled.load(std::memory_order_acquire));
260
261		// Cache endpoints on successful connect
262	4824x	if (success && socket_impl_)
263		{
264	4822x	endpoint local_ep;
265	4822x	sockaddr_storage local_storage{};
266	4822x	socklen_t local_len = sizeof(local_storage);
267	4822x	if (::getsockname(
268	4822x	fd, reinterpret_cast<sockaddr*>(&local_storage), &local_len) ==
269		0)
270	4822x	local_ep = from_sockaddr(local_storage);
271	4822x	static_cast<epoll_socket*>(socket_impl_)
272	4822x	->set_endpoints(local_ep, target_endpoint);
273		}
274
275	4824x	if (cancelled.load(std::memory_order_acquire))
276	✗	*ec_out = capy::error::canceled;
277	4824x	else if (errn != 0)
278	2x	*ec_out = make_err(errn);
279		else
280	4822x	*ec_out = {};
281
282		// Move to stack before resuming. See epoll_op::operator()() for rationale.
283	4824x	capy::executor_ref saved_ex(ex);
284	4824x	std::coroutine_handle<> saved_h(h);
285	4824x	auto prevent_premature_destruction = std::move(impl_ptr);
286	4824x	dispatch_coro(saved_ex, saved_h).resume();
287	4824x	}
288
289	14529x	inline epoll_socket::epoll_socket(epoll_socket_service& svc) noexcept
290	14529x	: svc_(svc)
291		{
292	14529x	}
293
294	14529x	inline epoll_socket::~epoll_socket() = default;
295
296		inline std::coroutine_handle<>
297	4824x	epoll_socket::connect(
298		std::coroutine_handle<> h,
299		capy::executor_ref ex,
300		endpoint ep,
301		std::stop_token token,
302		std::error_code* ec)
303		{
304	4824x	auto& op = conn_;
305
306	4824x	sockaddr_storage storage{};
307		socklen_t addrlen =
308	4824x	detail::to_sockaddr(ep, detail::socket_family(fd_), storage);
309	4824x	int result = ::connect(fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
310
311	4824x	if (result == 0)
312		{
313	✗	sockaddr_storage local_storage{};
314	✗	socklen_t local_len = sizeof(local_storage);
315	✗	if (::getsockname(
316	✗	fd_, reinterpret_cast<sockaddr*>(&local_storage), &local_len) ==
317		0)
318	✗	local_endpoint_ = detail::from_sockaddr(local_storage);
319	✗	remote_endpoint_ = ep;
320		}
321
322	4824x	if (result == 0 \|\| errno != EINPROGRESS)
323		{
324	✗	int err = (result < 0) ? errno : 0;
325	✗	if (svc_.scheduler().try_consume_inline_budget())
326		{
327	✗	*ec = err ? make_err(err) : std::error_code{};
328	✗	return dispatch_coro(ex, h);
329		}
330	✗	op.reset();
331	✗	op.h = h;
332	✗	op.ex = ex;
333	✗	op.ec_out = ec;
334	✗	op.fd = fd_;
335	✗	op.target_endpoint = ep;
336	✗	op.start(token, this);
337	✗	op.impl_ptr = shared_from_this();
338	✗	op.complete(err, 0);
339	✗	svc_.post(&op);
340	✗	return std::noop_coroutine();
341		}
342
343		// EINPROGRESS — register with reactor
344	4824x	op.reset();
345	4824x	op.h = h;
346	4824x	op.ex = ex;
347	4824x	op.ec_out = ec;
348	4824x	op.fd = fd_;
349	4824x	op.target_endpoint = ep;
350	4824x	op.start(token, this);
351	4824x	op.impl_ptr = shared_from_this();
352
353	4824x	register_op(
354	4824x	op, desc_state_.connect_op, desc_state_.write_ready,
355	4824x	desc_state_.connect_cancel_pending);
356	4824x	return std::noop_coroutine();
357		}
358
359		inline std::coroutine_handle<>
360	132861x	epoll_socket::read_some(
361		std::coroutine_handle<> h,
362		capy::executor_ref ex,
363		buffer_param param,
364		std::stop_token token,
365		std::error_code* ec,
366		std::size_t* bytes_out)
367		{
368	132861x	auto& op = rd_;
369	132861x	op.reset();
370
371	132861x	capy::mutable_buffer bufs[epoll_read_op::max_buffers];
372	132861x	op.iovec_count =
373	132861x	static_cast<int>(param.copy_to(bufs, epoll_read_op::max_buffers));
374
375	132861x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
376		{
377	1x	op.empty_buffer_read = true;
378	1x	op.h = h;
379	1x	op.ex = ex;
380	1x	op.ec_out = ec;
381	1x	op.bytes_out = bytes_out;
382	1x	op.start(token, this);
383	1x	op.impl_ptr = shared_from_this();
384	1x	op.complete(0, 0);
385	1x	svc_.post(&op);
386	1x	return std::noop_coroutine();
387		}
388
389	265720x	for (int i = 0; i < op.iovec_count; ++i)
390		{
391	132860x	op.iovecs[i].iov_base = bufs[i].data();
392	132860x	op.iovecs[i].iov_len = bufs[i].size();
393		}
394
395		// Speculative read
396		ssize_t n;
397		do
398		{
399	132860x	n = ::readv(fd_, op.iovecs, op.iovec_count);
400		}
401	132860x	while (n < 0 && errno == EINTR);
402
403	132860x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
404		{
405	132659x	int err = (n < 0) ? errno : 0;
406	132659x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
407
408	132659x	if (svc_.scheduler().try_consume_inline_budget())
409		{
410	106173x	if (err)
411	✗	*ec = make_err(err);
412	106173x	else if (n == 0)
413	5x	*ec = capy::error::eof;
414		else
415	106168x	*ec = {};
416	106173x	*bytes_out = bytes;
417	106173x	return dispatch_coro(ex, h);
418		}
419	26486x	op.h = h;
420	26486x	op.ex = ex;
421	26486x	op.ec_out = ec;
422	26486x	op.bytes_out = bytes_out;
423	26486x	op.start(token, this);
424	26486x	op.impl_ptr = shared_from_this();
425	26486x	op.complete(err, bytes);
426	26486x	svc_.post(&op);
427	26486x	return std::noop_coroutine();
428		}
429
430		// EAGAIN — register with reactor
431	201x	op.h = h;
432	201x	op.ex = ex;
433	201x	op.ec_out = ec;
434	201x	op.bytes_out = bytes_out;
435	201x	op.fd = fd_;
436	201x	op.start(token, this);
437	201x	op.impl_ptr = shared_from_this();
438
439	201x	register_op(
440	201x	op, desc_state_.read_op, desc_state_.read_ready,
441	201x	desc_state_.read_cancel_pending);
442	201x	return std::noop_coroutine();
443		}
444
445		inline std::coroutine_handle<>
446	132661x	epoll_socket::write_some(
447		std::coroutine_handle<> h,
448		capy::executor_ref ex,
449		buffer_param param,
450		std::stop_token token,
451		std::error_code* ec,
452		std::size_t* bytes_out)
453		{
454	132661x	auto& op = wr_;
455	132661x	op.reset();
456
457	132661x	capy::mutable_buffer bufs[epoll_write_op::max_buffers];
458	132661x	op.iovec_count =
459	132661x	static_cast<int>(param.copy_to(bufs, epoll_write_op::max_buffers));
460
461	132661x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
462		{
463	1x	op.h = h;
464	1x	op.ex = ex;
465	1x	op.ec_out = ec;
466	1x	op.bytes_out = bytes_out;
467	1x	op.start(token, this);
468	1x	op.impl_ptr = shared_from_this();
469	1x	op.complete(0, 0);
470	1x	svc_.post(&op);
471	1x	return std::noop_coroutine();
472		}
473
474	265320x	for (int i = 0; i < op.iovec_count; ++i)
475		{
476	132660x	op.iovecs[i].iov_base = bufs[i].data();
477	132660x	op.iovecs[i].iov_len = bufs[i].size();
478		}
479
480		// Speculative write
481	132660x	msghdr msg{};
482	132660x	msg.msg_iov = op.iovecs;
483	132660x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
484
485		ssize_t n;
486		do
487		{
488	132660x	n = ::sendmsg(fd_, &msg, MSG_NOSIGNAL);
489		}
490	132660x	while (n < 0 && errno == EINTR);
491
492	132660x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
493		{
494	132660x	int err = (n < 0) ? errno : 0;
495	132660x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
496
497	132660x	if (svc_.scheduler().try_consume_inline_budget())
498		{
499	106156x	*ec = err ? make_err(err) : std::error_code{};
500	106156x	*bytes_out = bytes;
501	106156x	return dispatch_coro(ex, h);
502		}
503	26504x	op.h = h;
504	26504x	op.ex = ex;
505	26504x	op.ec_out = ec;
506	26504x	op.bytes_out = bytes_out;
507	26504x	op.start(token, this);
508	26504x	op.impl_ptr = shared_from_this();
509	26504x	op.complete(err, bytes);
510	26504x	svc_.post(&op);
511	26504x	return std::noop_coroutine();
512		}
513
514		// EAGAIN — register with reactor
515	✗	op.h = h;
516	✗	op.ex = ex;
517	✗	op.ec_out = ec;
518	✗	op.bytes_out = bytes_out;
519	✗	op.fd = fd_;
520	✗	op.start(token, this);
521	✗	op.impl_ptr = shared_from_this();
522
523	✗	register_op(
524	✗	op, desc_state_.write_op, desc_state_.write_ready,
525	✗	desc_state_.write_cancel_pending);
526	✗	return std::noop_coroutine();
527		}
528
529		inline std::error_code
530	3x	epoll_socket::shutdown(tcp_socket::shutdown_type what) noexcept
531		{
532		int how;
533	3x	switch (what)
534		{
535	1x	case tcp_socket::shutdown_receive:
536	1x	how = SHUT_RD;
537	1x	break;
538	1x	case tcp_socket::shutdown_send:
539	1x	how = SHUT_WR;
540	1x	break;
541	1x	case tcp_socket::shutdown_both:
542	1x	how = SHUT_RDWR;
543	1x	break;
544	✗	default:
545	✗	return make_err(EINVAL);
546		}
547	3x	if (::shutdown(fd_, how) != 0)
548	✗	return make_err(errno);
549	3x	return {};
550		}
551
552		inline std::error_code
553	32x	epoll_socket::set_option(
554		int level, int optname, void const* data, std::size_t size) noexcept
555		{
556	32x	if (::setsockopt(fd_, level, optname, data, static_cast<socklen_t>(size)) !=
557		0)
558	✗	return make_err(errno);
559	32x	return {};
560		}
561
562		inline std::error_code
563	31x	epoll_socket::get_option(
564		int level, int optname, void* data, std::size_t* size) const noexcept
565		{
566	31x	socklen_t len = static_cast<socklen_t>(*size);
567	31x	if (::getsockopt(fd_, level, optname, data, &len) != 0)
568	✗	return make_err(errno);
569	31x	*size = static_cast<std::size_t>(len);
570	31x	return {};
571		}
572
573		inline void
574	187x	epoll_socket::cancel() noexcept
575		{
576	187x	auto self = weak_from_this().lock();
577	187x	if (!self)
578	✗	return;
579
580	187x	conn_.request_cancel();
581	187x	rd_.request_cancel();
582	187x	wr_.request_cancel();
583
584	187x	epoll_op* conn_claimed = nullptr;
585	187x	epoll_op* rd_claimed = nullptr;
586	187x	epoll_op* wr_claimed = nullptr;
587		{
588	187x	std::lock_guard lock(desc_state_.mutex);
589	187x	if (desc_state_.connect_op == &conn_)
590	✗	conn_claimed = std::exchange(desc_state_.connect_op, nullptr);
591		else
592	187x	desc_state_.connect_cancel_pending = true;
593	187x	if (desc_state_.read_op == &rd_)
594	3x	rd_claimed = std::exchange(desc_state_.read_op, nullptr);
595		else
596	184x	desc_state_.read_cancel_pending = true;
597	187x	if (desc_state_.write_op == &wr_)
598	✗	wr_claimed = std::exchange(desc_state_.write_op, nullptr);
599		else
600	187x	desc_state_.write_cancel_pending = true;
601	187x	}
602
603	187x	if (conn_claimed)
604		{
605	✗	conn_.impl_ptr = self;
606	✗	svc_.post(&conn_);
607	✗	svc_.work_finished();
608		}
609	187x	if (rd_claimed)
610		{
611	3x	rd_.impl_ptr = self;
612	3x	svc_.post(&rd_);
613	3x	svc_.work_finished();
614		}
615	187x	if (wr_claimed)
616		{
617	✗	wr_.impl_ptr = self;
618	✗	svc_.post(&wr_);
619	✗	svc_.work_finished();
620		}
621	187x	}
622
623		inline void
624	98x	epoll_socket::cancel_single_op(epoll_op& op) noexcept
625		{
626	98x	auto self = weak_from_this().lock();
627	98x	if (!self)
628	✗	return;
629
630	98x	op.request_cancel();
631
632	98x	epoll_op** desc_op_ptr = nullptr;
633	98x	if (&op == &conn_)
634	✗	desc_op_ptr = &desc_state_.connect_op;
635	98x	else if (&op == &rd_)
636	98x	desc_op_ptr = &desc_state_.read_op;
637	✗	else if (&op == &wr_)
638	✗	desc_op_ptr = &desc_state_.write_op;
639
640	98x	if (desc_op_ptr)
641		{
642	98x	epoll_op* claimed = nullptr;
643		{
644	98x	std::lock_guard lock(desc_state_.mutex);
645	98x	if (*desc_op_ptr == &op)
646	98x	claimed = std::exchange(*desc_op_ptr, nullptr);
647	✗	else if (&op == &conn_)
648	✗	desc_state_.connect_cancel_pending = true;
649	✗	else if (&op == &rd_)
650	✗	desc_state_.read_cancel_pending = true;
651	✗	else if (&op == &wr_)
652	✗	desc_state_.write_cancel_pending = true;
653	98x	}
654	98x	if (claimed)
655		{
656	98x	op.impl_ptr = self;
657	98x	svc_.post(&op);
658	98x	svc_.work_finished();
659		}
660		}
661	98x	}
662
663		inline void
664	43558x	epoll_socket::close_socket() noexcept
665		{
666	43558x	auto self = weak_from_this().lock();
667	43558x	if (self)
668		{
669	43558x	conn_.request_cancel();
670	43558x	rd_.request_cancel();
671	43558x	wr_.request_cancel();
672
673	43558x	epoll_op* conn_claimed = nullptr;
674	43558x	epoll_op* rd_claimed = nullptr;
675	43558x	epoll_op* wr_claimed = nullptr;
676		{
677	43558x	std::lock_guard lock(desc_state_.mutex);
678	43558x	conn_claimed = std::exchange(desc_state_.connect_op, nullptr);
679	43558x	rd_claimed = std::exchange(desc_state_.read_op, nullptr);
680	43558x	wr_claimed = std::exchange(desc_state_.write_op, nullptr);
681	43558x	desc_state_.read_ready = false;
682	43558x	desc_state_.write_ready = false;
683	43558x	desc_state_.read_cancel_pending = false;
684	43558x	desc_state_.write_cancel_pending = false;
685	43558x	desc_state_.connect_cancel_pending = false;
686	43558x	}
687
688	43558x	if (conn_claimed)
689		{
690	✗	conn_.impl_ptr = self;
691	✗	svc_.post(&conn_);
692	✗	svc_.work_finished();
693		}
694	43558x	if (rd_claimed)
695		{
696	1x	rd_.impl_ptr = self;
697	1x	svc_.post(&rd_);
698	1x	svc_.work_finished();
699		}
700	43558x	if (wr_claimed)
701		{
702	✗	wr_.impl_ptr = self;
703	✗	svc_.post(&wr_);
704	✗	svc_.work_finished();
705		}
706
707	43558x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
708	97x	desc_state_.impl_ref_ = self;
709		}
710
711	43558x	if (fd_ >= 0)
712		{
713	9661x	if (desc_state_.registered_events != 0)
714	9661x	svc_.scheduler().deregister_descriptor(fd_);
715	9661x	::close(fd_);
716	9661x	fd_ = -1;
717		}
718
719	43558x	desc_state_.fd = -1;
720	43558x	desc_state_.registered_events = 0;
721
722	43558x	local_endpoint_ = endpoint{};
723	43558x	remote_endpoint_ = endpoint{};
724	43558x	}
725
726	244x	inline epoll_socket_service::epoll_socket_service(capy::execution_context& ctx)
727	244x	: state_(
728		std::make_unique<epoll_socket_state>(
729	244x	ctx.use_service<epoll_scheduler>()))
730		{
731	244x	}
732
733	488x	inline epoll_socket_service::~epoll_socket_service() {}
734
735		inline void
736	244x	epoll_socket_service::shutdown()
737		{
738	244x	std::lock_guard lock(state_->mutex_);
739
740	244x	while (auto* impl = state_->socket_list_.pop_front())
741	✗	impl->close_socket();
742
743		// Don't clear socket_ptrs_ here. The scheduler shuts down after us and
744		// drains completed_ops_, calling destroy() on each queued op. If we
745		// released our shared_ptrs now, an epoll_op::destroy() could free the
746		// last ref to an impl whose embedded descriptor_state is still linked
747		// in the queue — use-after-free on the next pop(). Letting ~state_
748		// release the ptrs (during service destruction, after scheduler
749		// shutdown) keeps every impl alive until all ops have been drained.
750	244x	}
751
752		inline io_object::implementation*
753	14529x	epoll_socket_service::construct()
754		{
755	14529x	auto impl = std::make_shared<epoll_socket>(*this);
756	14529x	auto* raw = impl.get();
757
758		{
759	14529x	std::lock_guard lock(state_->mutex_);
760	14529x	state_->socket_list_.push_back(raw);
761	14529x	state_->socket_ptrs_.emplace(raw, std::move(impl));
762	14529x	}
763
764	14529x	return raw;
765	14529x	}
766
767		inline void
768	14529x	epoll_socket_service::destroy(io_object::implementation* impl)
769		{
770	14529x	auto* epoll_impl = static_cast<epoll_socket*>(impl);
771	14529x	epoll_impl->close_socket();
772	14529x	std::lock_guard lock(state_->mutex_);
773	14529x	state_->socket_list_.remove(epoll_impl);
774	14529x	state_->socket_ptrs_.erase(epoll_impl);
775	14529x	}
776
777		inline std::error_code
778	4839x	epoll_socket_service::open_socket(
779		tcp_socket::implementation& impl, int family, int type, int protocol)
780		{
781	4839x	auto* epoll_impl = static_cast<epoll_socket*>(&impl);
782	4839x	epoll_impl->close_socket();
783
784	4839x	int fd = ::socket(family, type \| SOCK_NONBLOCK \| SOCK_CLOEXEC, protocol);
785	4839x	if (fd < 0)
786	✗	return make_err(errno);
787
788	4839x	if (family == AF_INET6)
789		{
790	5x	int one = 1;
791	5x	::setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &one, sizeof(one));
792		}
793
794	4839x	epoll_impl->fd_ = fd;
795
796		// Register fd with epoll (edge-triggered mode)
797	4839x	epoll_impl->desc_state_.fd = fd;
798		{
799	4839x	std::lock_guard lock(epoll_impl->desc_state_.mutex);
800	4839x	epoll_impl->desc_state_.read_op = nullptr;
801	4839x	epoll_impl->desc_state_.write_op = nullptr;
802	4839x	epoll_impl->desc_state_.connect_op = nullptr;
803	4839x	}
804	4839x	scheduler().register_descriptor(fd, &epoll_impl->desc_state_);
805
806	4839x	return {};
807		}
808
809		inline void
810	24190x	epoll_socket_service::close(io_object::handle& h)
811		{
812	24190x	static_cast<epoll_socket*>(h.get())->close_socket();
813	24190x	}
814
815		inline void
816	53189x	epoll_socket_service::post(epoll_op* op)
817		{
818	53189x	state_->sched_.post(op);
819	53189x	}
820
821		inline void
822	5025x	epoll_socket_service::work_started() noexcept
823		{
824	5025x	state_->sched_.work_started();
825	5025x	}
826
827		inline void
828	197x	epoll_socket_service::work_finished() noexcept
829		{
830	197x	state_->sched_.work_finished();
831	197x	}
832
833		} // namespace boost::corosio::detail
834
835		#endif // BOOST_COROSIO_HAS_EPOLL
836
837		#endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SOCKET_SERVICE_HPP
838