Use IORING_POLL_ADD_MULTI for the wakeup eventfd poll.

Instead of registering a new one-shot poll_add on the wakeup eventfd
before every blocking wait, register a multi-shot poll once and let the
kernel keep it alive across fires (IORING_POLL_ADD_MULTI, kernel 5.13+).

This removes one SQE submission per select() call on supporting kernels.
On older kernels the #ifdef falls back to the one-shot-per-wait path.

The wakeup CQE is now identified by a sentinel (address of wakeup_fd)
rather than NULL so select_process_completions can drain the eventfd
and detect when the multi-shot poll needs re-registration
(IORING_CQE_F_MORE absent).

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: samuel-williams-shopify

ext/io/event/selector/uring.c

@@ -40,6 +40,12 @@ struct IO_Event_Selector_URing
 	// This allows IORING_SETUP_SINGLE_ISSUER: only the owner thread ever submits SQEs.
 	int wakeup_fd;
 
+	// Whether a multi-shot poll on wakeup_fd is currently registered in the ring.
+	// With IORING_POLL_ADD_MULTI the poll persists across fires so we only need to
+	// register it once (re-registering only when the kernel cancels it).
+	// Without multi-shot support this stays 0 and we register before every wait.
+	int wakeup_registered;
+	
 	struct timespec idle_duration;
 
 	struct IO_Event_Array completions;
@@ -108,6 +114,7 @@ void close_internal(struct IO_Event_Selector_URing *selector)
 	if (selector->wakeup_fd >= 0) {
 		close(selector->wakeup_fd);
 		selector->wakeup_fd = -1;
+		selector->wakeup_registered = 0;
 	}
 
 	if (selector->ring.ring_fd >= 0) {
@@ -230,6 +237,7 @@ VALUE IO_Event_Selector_URing_allocate(VALUE self) {
 	selector->pending = 0;
 	selector->blocked = 0;
 	selector->wakeup_fd = -1;
+	selector->wakeup_registered = 0;
 
 	IO_Event_List_initialize(&selector->free_list);
 
@@ -1104,24 +1112,29 @@ static
 int select_internal_without_gvl(struct select_arguments *arguments) {
 	struct IO_Event_Selector_URing *selector = arguments->selector;
 
-	// Register a one-shot poll on the wakeup eventfd before releasing the GVL.
-	// This allows wakeup() to signal us by writing to the fd from any thread
-	// without touching the ring's SQ (required for IORING_SETUP_SINGLE_ISSUER).
-	struct io_uring_sqe *sqe = io_get_sqe(selector);
-	io_uring_prep_poll_add(sqe, selector->wakeup_fd, POLLIN);
-	io_uring_sqe_set_data(sqe, NULL);
-	selector->pending += 1;
+	// Register a poll on the wakeup eventfd before releasing the GVL so that
+	// wakeup() can signal us with a plain write() from any thread.
+	// With IORING_POLL_ADD_MULTI the poll persists across fires and we only
+	// register it once; on older kernels we fall back to one-shot per wait.
+	if (!selector->wakeup_registered) {
+		struct io_uring_sqe *sqe = io_get_sqe(selector);
+		io_uring_prep_poll_add(sqe, selector->wakeup_fd, POLLIN);
+#ifdef IORING_POLL_ADD_MULTI
+		sqe->len = IORING_POLL_ADD_MULTI;
+		selector->wakeup_registered = 1;
+#endif
+		// Use the address of wakeup_fd as a unique sentinel so completions
+		// can be identified without allocating a full completion struct.
+		io_uring_sqe_set_data(sqe, &selector->wakeup_fd);
+		selector->pending += 1;
+	}
 
 	io_uring_submit_flush(selector);
 
 	selector->blocked = 1;
 	rb_thread_call_without_gvl(select_internal, (void *)arguments, RUBY_UBF_IO, 0);
 	selector->blocked = 0;
 
-	// Drain the wakeup eventfd so the next poll_add doesn't fire immediately.
-	uint64_t value;
-	while (read(selector->wakeup_fd, &value, sizeof(value)) > 0) {}
-	
 	if (arguments->result == -ETIME) {
 		arguments->result = 0;
 	} else if (arguments->result == -EINTR) {
@@ -1159,6 +1172,20 @@ unsigned select_process_completions(struct IO_Event_Selector_URing *selector) {
 			continue;
 		}
 
+		// Wakeup eventfd poll completion — drain the fd and check if the
+		// multi-shot poll is still live (IORING_CQE_F_MORE set).
+		if (cqe->user_data == (uint64_t)(uintptr_t)&selector->wakeup_fd) {
+			uint64_t value;
+			while (read(selector->wakeup_fd, &value, sizeof(value)) > 0) {}
+#ifdef IORING_POLL_ADD_MULTI
+			if (!(cqe->flags & IORING_CQE_F_MORE)) {
+				selector->wakeup_registered = 0;
+			}
+#endif
+			io_uring_cq_advance(ring, 1);
+			continue;
+		}
+		
 		struct IO_Event_Selector_URing_Completion *completion = (void*)cqe->user_data;
 		struct IO_Event_Selector_URing_Waiting *waiting = completion->waiting;