Report IOR_CQE_F_SOCK_NONEMPTY on accept completions

README.md

@@ -23,7 +23,10 @@ The goal is to provide maximum performance on platforms with native async I/O su
   `IOR_CQE_F_MORE`, until cancelled - `IORING_ACCEPT_MULTISHOT` on io_uring,
   an edge-triggered watch of the listener on the thread pool's poller, which
   accepts until `EAGAIN` at every edge, a few AcceptEx kept outstanding on
-  Windows
+  Windows; an accept's completion carries `IOR_CQE_F_SOCK_NONEMPTY` when
+  another connection is queued behind it (io_uring from Linux 6.10, a
+  zero-timeout poll elsewhere), and on io_uring a receive's when more data
+  waits
 - Timer/timeout operations, relative or absolute on the monotonic, boot-time
   or wall clock (`IOR_TIMEOUT_ABS`, `IOR_TIMEOUT_BOOTTIME`,
   `IOR_TIMEOUT_REALTIME`); the timespec is copied at submit on every backend

src/ior.h

@@ -280,6 +280,18 @@ typedef struct ior_timespec {
  *  accept (ior_prep_accept_multishot) one per connection, and stays armed.
  *  A completion without this bit is the operation's last. */
 #define IOR_CQE_F_MORE (1U << 1)
+/** The socket has more waiting after this completion: another connection
+ *  queued on the listener behind an accept (ior_prep_accept(),
+ *  ior_prep_accept_multishot()), more data to read behind a receive
+ *  (ior_prep_recv()). A hint, never a promise: it is set where the backend
+ *  can tell at no great cost, and a completion without it says nothing.
+ *  Accept: io_uring from Linux 6.10, the thread backend (a zero-timeout
+ *  poll of the listener after each accept), IOCP for a one-shot accept
+ *  (WSAPoll, likewise); a multishot accept's connections on IOCP never
+ *  carry it, its outstanding AcceptEx taking what is queued. Receive:
+ *  io_uring only (IORING_CQE_F_SOCK_NONEMPTY, Linux 5.19). Lets a caller
+ *  accept or read again at once instead of waiting for -EAGAIN. */
+#define IOR_CQE_F_SOCK_NONEMPTY (1U << 2)
 /** @} */
 
 /** Asynchronous I/O backend implementation. */
@@ -775,6 +787,9 @@ void ior_prep_recv(
  * else puts it in non-blocking mode: accepted without the flag, it is a
  * blocking descriptor the backend will take at its word.
  *
+ * IOR_CQE_F_SOCK_NONEMPTY among the completion's flags says another
+ * connection is queued already (a hint; see the flag).
+ *
  * @param ctx      I/O context.
  * @param sqe      Entry from ior_get_sqe().
  * @param fd       Listening socket.
@@ -803,7 +818,9 @@ void ior_prep_accept(ior_ctx *ctx, ior_sqe *sqe, ior_fd_t fd, struct sockaddr *a
  *
  * Connections accepted before the last completion still arrive, with
  * IOR_CQE_F_MORE, ahead of it: reap until the completion without the flag,
- * and close what you do not want. On io_uring and the thread backend the
+ * and close what you do not want; one may carry IOR_CQE_F_SOCK_NONEMPTY,
+ * another connection being queued behind it (a hint; see the flag). On
+ * io_uring and the thread backend the
  * operation also ends on its own when a connection finds the completion
  * queue full: that connection is its last completion, with the socket as
  * res (>= 0) and no IOR_CQE_F_MORE, as io_uring ends a multishot accept.

src/ior_iocp.c

@@ -126,6 +126,7 @@ typedef struct ior_iocp_op {
 	uint32_t poll_mask;
 	bool poll_multi;
 	bool cqe_more;
+	bool cqe_nonempty; // IOR_CQE_F_SOCK_NONEMPTY: a one-shot accept found more queued
 	bool poll_held;
 	_Atomic uint32_t poll_rearm;
 	struct ior_iocp_op *poll_parent;
@@ -748,6 +749,7 @@ static ior_iocp_op *alloc_op(ior_ctx_iocp *ctx)
 	op->poll_mask = 0;
 	op->poll_multi = false;
 	op->cqe_more = false;
+	op->cqe_nonempty = false;
 	op->poll_held = false;
 	atomic_store(&op->poll_rearm, 0);
 	op->poll_parent = NULL;
@@ -1664,6 +1666,15 @@ static void finish_socket_op(ior_ctx_iocp *ctx, ior_iocp_op *op)
 				}
 				*op->sa_len = (socklen_t) ra_len;
 			}
+			if (op->error_code == ERROR_SUCCESS && !op->poll_parent) {
+				// IOR_CQE_F_SOCK_NONEMPTY for a one-shot accept: the listener
+				// reads ready while a connection is queued (a poll with no
+				// wait; an error counts as nothing queued, the hint being
+				// optional). Not for a multishot's child: its siblings take
+				// what is queued as it arrives.
+				WSAPOLLFD pfd = { .fd = ls, .events = POLLRDNORM };
+				op->cqe_nonempty = WSAPoll(&pfd, 1, 0) > 0 && (pfd.revents & POLLRDNORM);
+			}
 		}
 		if (op->error_code != ERROR_SUCCESS) {
 			closesocket(op->accept_sock);
@@ -2593,7 +2604,8 @@ static int issue_poll(ior_ctx_iocp *ctx, ior_iocp_op *op)
 static void op_to_cqe(ior_iocp_op *op)
 {
 	op->cqe.iocp.user_data = op->user_data;
-	op->cqe.iocp.flags = op->cqe_more ? IOR_CQE_F_MORE : 0;
+	op->cqe.iocp.flags = (op->cqe_more ? IOR_CQE_F_MORE : 0)
+			| (op->cqe_nonempty ? IOR_CQE_F_SOCK_NONEMPTY : 0);
 
 	if (op->submit_res) {
 		op->cqe.iocp.res = op->submit_res;

src/ior_threads_pool.c

@@ -593,6 +593,22 @@ static int ior_threads_pool_accept_multi(const ior_sqe *sqe)
 	return sqe->threads.opcode == IOR_OP_ACCEPT && (sqe->threads.ioprio & IOR_ACCEPT_MULTISHOT);
 }
 
+/*
+ * IOR_CQE_F_SOCK_NONEMPTY for an accept's completion: another connection is
+ * queued if the listener still reads ready. One poll(2) with no wait per
+ * accepted connection, which io_uring gets for free from the protocol
+ * (Linux 6.10); an error counts as nothing queued, the hint being optional.
+ */
+static uint32_t ior_threads_pool_accept_nonempty(int fd)
+{
+	struct pollfd pfd = { .fd = fd, .events = POLLIN };
+	int pret;
+	do {
+		pret = poll(&pfd, 1, 0);
+	} while (pret < 0 && errno == EINTR);
+	return pret > 0 && (pfd.revents & POLLIN) ? IOR_CQE_F_SOCK_NONEMPTY : 0;
+}
+
 /*
  * An edge on a multishot accept's listener, on the poller thread (and once
  * on the worker, before the op is handed over): accept until the backlog is
@@ -634,7 +650,7 @@ static int ior_threads_pool_accept_edge(ior_threads_pool *pool, ior_work *w)
 		memset(&cqe, 0, sizeof(cqe));
 		cqe.threads.user_data = sqe->threads.user_data;
 		cqe.threads.res = nfd;
-		cqe.threads.flags = IOR_CQE_F_MORE;
+		cqe.threads.flags = IOR_CQE_F_MORE | ior_threads_pool_accept_nonempty(sqe->threads.fd);
 		if (ior_threads_pool_post(pool, NULL, &cqe) < 0) {
 			w->accept_last = nfd;
 			return 1;
@@ -2493,6 +2509,9 @@ static void ior_threads_pool_process_single_sqe(
 			cqe->threads.res = ior_threads_pool_accept(sqe->threads.fd,
 					(struct sockaddr *) (uintptr_t) sqe->threads.addr,
 					(socklen_t *) (uintptr_t) sqe->threads.off, sqe->threads.rw_flags);
+			if (cqe->threads.res >= 0) {
+				cqe->threads.flags = ior_threads_pool_accept_nonempty(sqe->threads.fd);
+			}
 			IOR_LOG_TRACE("accept end: res=%d", cqe->threads.res);
 			break;
 		}

src/ior_uring.c

@@ -1678,6 +1678,9 @@ static void ior_uring_backend_prep_poll_add(ior_sqe *sqe, ior_fd_t fd, uint32_t
 
 /* The kernel's own flag marks the edge completions of a multishot poll. */
 _Static_assert(IOR_CQE_F_MORE == IORING_CQE_F_MORE, "IOR_CQE_F_MORE must match");
+/* The kernel's hint on accept (6.10) and recv (5.19) completions, passed through. */
+_Static_assert(IOR_CQE_F_SOCK_NONEMPTY == IORING_CQE_F_SOCK_NONEMPTY,
+		"IOR_CQE_F_SOCK_NONEMPTY must match");
 _Static_assert(IOR_POLL_ADD_MULTI == IORING_POLL_ADD_MULTI, "IOR_POLL_ADD_MULTI must match");
 
 static void ior_uring_backend_prep_poll_multishot(ior_sqe *sqe, ior_fd_t fd, uint32_t poll_mask)

tests/test_accept_connect.c

@@ -15,6 +15,9 @@
 #ifndef _WIN32
 #include <netinet/in.h>
 #endif
+#ifdef __linux__
+#include <sys/utsname.h>
+#endif
 
 #define TAG_ACCEPT ((void *) 0x1)
 #define TAG_CONNECT ((void *) 0x2)
@@ -492,6 +495,63 @@ static void reap_cancelled(ior_ctx *ctx, int32_t cancel_res)
 	}
 }
 
+/*
+ * Whether IOR_CQE_F_SOCK_NONEMPTY is exact on this backend for a one-shot
+ * accept (multi == 0) or a multishot's connections: the thread backend and
+ * IOCP probe the listener, IOCP only for a one-shot accept, and io_uring
+ * reports it from Linux 6.10.
+ */
+static int nonempty_exact(ior_ctx *ctx, int multi)
+{
+	ior_backend_type backend = ior_get_backend_type(ctx);
+	if (backend == IOR_BACKEND_THREADS) {
+		return 1;
+	}
+	if (backend == IOR_BACKEND_IOCP) {
+		return !multi;
+	}
+#ifdef __linux__
+	struct utsname u;
+	int major = 0, minor = 0;
+	if (uname(&u) == 0 && sscanf(u.release, "%d.%d", &major, &minor) == 2) {
+		return major > 6 || (major == 6 && minor >= 10);
+	}
+#endif
+	return 0;
+}
+
+/*
+ * IOR_CQE_F_SOCK_NONEMPTY on a one-shot accept: set with another connection
+ * queued behind the accepted one, clear on the last.
+ */
+static void test_accept_sock_nonempty(void **state)
+{
+	ac_state *s = (ac_state *) *state;
+	ior_fd_t c1 = connect_client(s);
+	ior_fd_t c2 = connect_client(s);
+	int exact = nonempty_exact(s->ctx, 0);
+
+	for (int i = 0; i < 2; i++) {
+		submit_accept(s, NULL, NULL, 0);
+		assert_true(ior_submit(s->ctx) >= 0);
+		cqe_rec r;
+		if (reap_one(s->ctx, &r, 3000) != 0) {
+			fail_msg("accept %d missing", i);
+		}
+		assert_ptr_equal(r.tag, TAG_ACCEPT);
+		ior_fd_t a = accepted_fd(r.res);
+		if (exact) {
+			assert_int_equal(
+					r.flags & IOR_CQE_F_SOCK_NONEMPTY, i == 0 ? IOR_CQE_F_SOCK_NONEMPTY : 0);
+		} else if (i == 1) {
+			assert_false(r.flags & IOR_CQE_F_SOCK_NONEMPTY);
+		}
+		test_close_fd(a);
+	}
+	test_close_fd(c1);
+	test_close_fd(c2);
+}
+
 /*
  * Connections arriving one by one: each is one edge completion, the
  * accepted socket works for I/O both ways, and a cancel ends the operation
@@ -562,8 +622,22 @@ static void test_accept_multishot_backlog(void **state)
 		clients[i] = connect_client(s);
 	}
 	submit_maccept(s, 0, 0);
+	int exact = nonempty_exact(s->ctx, 1);
 	for (int i = 0; i < 5; i++) {
-		accepted[i] = reap_maccept(s->ctx);
+		cqe_rec r;
+		if (reap_one(s->ctx, &r, 3000) != 0) {
+			fail_msg("connection %d not reported", i);
+		}
+		assert_ptr_equal(r.tag, TAG_MACCEPT);
+		assert_true(r.flags & IOR_CQE_F_MORE);
+		accepted[i] = accepted_fd(r.res);
+		// Another is queued behind each but the last.
+		if (exact) {
+			assert_int_equal(
+					r.flags & IOR_CQE_F_SOCK_NONEMPTY, i < 4 ? IOR_CQE_F_SOCK_NONEMPTY : 0);
+		} else if (i == 4) {
+			assert_false(r.flags & IOR_CQE_F_SOCK_NONEMPTY);
+		}
 	}
 	assert_silent(s->ctx, 50);
 
@@ -955,6 +1029,7 @@ int main(void)
 #ifndef _WIN32
 		cmocka_unit_test_setup_teardown(test_accept_flags, setup_ac, teardown_ac),
 #endif
+		cmocka_unit_test_setup_teardown(test_accept_sock_nonempty, setup_ac, teardown_ac),
 		cmocka_unit_test_setup_teardown(test_accept_multishot_stream, setup_ac, teardown_ac),
 		cmocka_unit_test_setup_teardown(test_accept_multishot_backlog, setup_ac, teardown_ac),
 #ifndef _WIN32