Merge branch 'fixes'

Set of fixes, and a test case for one of the issues.

* fixes:
  liburing.h: avoid OOL round trip in io_uring_peek_cqe() on empty CQ
  liburing.h: drop acquire ordering on CQ head load in __io_uring_peek_cqe()
  src/setup: explicitly request 2MB huge pages
  src/arch: use sysconf() for page size on aarch64/riscv64 libc builds
  src/setup: account SQ array in rings_size()
  src/sanitize: check secondary user pointers as well
  src/sanitize: initialize the handler table at compile time
  src/queue: io_uring_wait_cqes_min_timeout() should -EINVAL if unsupported
  liburing.h: don't return stale CQE pointer with timeout error
  src/arch: use prlimit64 for the rlimit syscall wrappers
  src/setup: account SQ array in the rings region in io_uring_alloc_huge()
  src/setup: restore ring fd and flags init in io_uring_queue_mmap()
  test/cq-peek-batch-mixed: add peek-batch test for mixed CQE rings
  src/queue: handle mixed CQEs in io_uring_peek_batch_cqe()
  src/register: fix mmap leak in io_uring_resize_rings()

Author: axboe

src/arch/aarch64/lib.h

@@ -6,6 +6,18 @@
 #include <elf.h>
 #include "../../syscall.h"
 
+#ifndef CONFIG_NOLIBC
+#include <unistd.h>
+
+static inline long __get_page_size(void)
+{
+	long ret = sysconf(_SC_PAGESIZE);
+
+	if (ret < 0)
+		ret = 4096;
+	return ret;
+}
+#else
 static inline long __get_page_size(void)
 {
 	Elf64_Off buf[2];
@@ -32,6 +44,7 @@ static inline long __get_page_size(void)
 	__sys_close(fd);
 	return ret;
 }
+#endif
 
 static inline long get_page_size(void)
 {

src/arch/riscv64/lib.h

@@ -7,6 +7,18 @@
 #include <sys/auxv.h>
 #include "../../syscall.h"
 
+#ifndef CONFIG_NOLIBC
+#include <unistd.h>
+
+static inline long __get_page_size(void)
+{
+	long ret = sysconf(_SC_PAGESIZE);
+
+	if (ret < 0)
+		ret = 4096;
+	return ret;
+}
+#else
 static inline long __get_page_size(void)
 {
 	Elf64_Off buf[2];
@@ -33,6 +45,7 @@ static inline long __get_page_size(void)
 	__sys_close(fd);
 	return ret;
 }
+#endif
 
 static inline long get_page_size(void)
 {

src/arch/syscall-defs.h

@@ -48,12 +48,12 @@ static inline int __sys_madvise(void *addr, size_t length, int advice)
 
 static inline int __sys_getrlimit(int resource, struct rlimit *rlim)
 {
-	return (int) __do_syscall2(__NR_getrlimit, resource, rlim);
+	return (int) __do_syscall4(__NR_prlimit64, 0, resource, NULL, rlim);
 }
 
 static inline int __sys_setrlimit(int resource, const struct rlimit *rlim)
 {
-	return (int) __do_syscall2(__NR_setrlimit, resource, rlim);
+	return (int) __do_syscall4(__NR_prlimit64, 0, resource, rlim, NULL);
 }
 
 static inline int __sys_close(int fd)

src/include/liburing.h

@@ -1881,20 +1881,30 @@ IOURINGINLINE int __io_uring_peek_cqe(struct io_uring *ring,
 	do {
 		unsigned tail = io_uring_smp_load_acquire(ring->cq.ktail);
 
-		/**
-		 * A load_acquire on the head prevents reordering with the
-		 * cqe load below, ensuring that we see the correct cq entry.
+		/*
+		 * The acquire ordering on the tail load pairs with the kernel
+		 * side publishing CQEs, and guarantees the contents of any
+		 * entry in [head, tail). The CQ head is only ever written by
+		 * the application, so a plain load is sufficient.
 		 */
-		unsigned head = io_uring_smp_load_acquire(ring->cq.khead);
+		unsigned head = *ring->cq.khead;
 
 		cqe = NULL;
 		available = tail - head;
 		if (!available)
 			break;
 
 		cqe = &ring->cq.cqes[(head & mask) << shift];
-		if (!io_uring_skip_cqe(ring, cqe, &err))
+		if (!io_uring_skip_cqe(ring, cqe, &err)) {
+			/*
+			 * If an error was set, the CQE was an internal
+			 * timeout and has already been consumed - don't
+			 * return a pointer to it.
+			 */
+			if (err)
+				cqe = NULL;
 			break;
+		}
 		cqe = NULL;
 	} while (1);
 
@@ -1912,8 +1922,20 @@ IOURINGINLINE int io_uring_peek_cqe(struct io_uring *ring,
 				    struct io_uring_cqe **cqe_ptr)
 	LIBURING_NOEXCEPT
 {
-	if (!__io_uring_peek_cqe(ring, cqe_ptr, NULL) && *cqe_ptr)
-		return 0;
+	if (!__io_uring_peek_cqe(ring, cqe_ptr, NULL)) {
+		if (*cqe_ptr)
+			return 0;
+		/*
+		 * If the CQ is empty and there's nothing the kernel could
+		 * flush to it (no IOPOLL completions to reap, no overflown
+		 * CQEs, no pending task work), avoid the round trip into
+		 * the full get_cqe machinery.
+		 */
+		if (!(ring->flags & IORING_SETUP_IOPOLL) &&
+		    !(IO_URING_READ_ONCE(*ring->sq.kflags) &
+		      (IORING_SQ_CQ_OVERFLOW | IORING_SQ_TASKRUN)))
+			return -EAGAIN;
+	}
 
 	return io_uring_wait_cqe_nr(ring, cqe_ptr, 0);
 }

src/queue.c

@@ -169,24 +169,55 @@ static inline bool io_uring_peek_batch_cqe_(struct io_uring *ring,
 					    unsigned *count)
 {
 	unsigned ready = io_uring_cq_ready(ring);
-	unsigned shift;
 	unsigned head;
 	unsigned mask;
 	unsigned last;
+	unsigned nr;
 
 	if (!ready)
 		return false;
 
-	shift = io_uring_cqe_shift(ring);
 	head = *ring->cq.khead;
 	mask = ring->cq.ring_mask;
-	if (ready < *count)
-		*count = ready;
-	last = head + *count;
-	for (;head != last; head++)
-		*(cqes++) = &ring->cq.cqes[(head & mask) << shift];
+	if (!(ring->flags & IORING_SETUP_CQE_MIXED)) {
+		unsigned shift = io_uring_cqe_shift(ring);
 
-	return true;
+		if (ready < *count)
+			*count = ready;
+		last = head + *count;
+		for (;head != last; head++)
+			*(cqes++) = &ring->cq.cqes[(head & mask) << shift];
+
+		return true;
+	}
+
+	/*
+	 * For mixed CQE rings, CQEs take up one or two slots, and the kernel
+	 * may post skip entries to pad out the ring at wrap time. Only return
+	 * pointers to real CQEs, with *count denoting the number of CQEs.
+	 */
+	last = head + ready;
+	nr = 0;
+	while (head != last && nr < *count) {
+		struct io_uring_cqe *cqe = &ring->cq.cqes[head & mask];
+
+		if (cqe->flags & IORING_CQE_F_SKIP) {
+			/*
+			 * A skip entry can only be consumed if it's at the
+			 * current CQ head, stop the batch otherwise. It'll
+			 * be at the head for the next peek.
+			 */
+			if (nr)
+				break;
+			io_uring_cq_advance(ring, 1);
+			head++;
+			continue;
+		}
+		head += io_uring_cqe_nr(cqe);
+		cqes[nr++] = cqe;
+	}
+	*count = nr;
+	return nr != 0;
 }
 
 /*
@@ -344,6 +375,8 @@ int io_uring_wait_cqes_min_timeout(struct io_uring *ring,
 				   struct __kernel_timespec *ts,
 				   unsigned int min_wait_usec, sigset_t *sigmask)
 {
+	if (!(ring->features & IORING_FEAT_MIN_TIMEOUT))
+		return -EINVAL;
 	return io_uring_wait_cqes_new(ring, cqe_ptr, wait_nr, ts, min_wait_usec,
 					sigmask);
 }

src/register.c

@@ -482,26 +482,8 @@ int io_uring_resize_rings(struct io_uring *ring, struct io_uring_params *p)
 	__sys_munmap(ring->sq.sqes, ring->sq.sqes_sz);
 	io_uring_unmap_rings(&ring->sq, &ring->cq);
 
-
-	memset(&ring->sq, 0, sizeof(ring->sq));
-	memset(&ring->cq, 0, sizeof(ring->cq));
-	ret = io_uring_mmap(ring->ring_fd, p, &ring->sq, &ring->cq);
-	if (ret) {
-		/*
-		 * The kernel has already committed to the new ring layout,
-		 * but we failed to mmap it. io_uring_mmap() cleaned up its
-		 * own partial mappings but may leave stale or error-encoded
-		 * pointers in sq/cq (e.g. ring_ptr = (void *)-ENOMEM).
-		 * Re-zero to scrub those so io_uring_queue_exit() won't
-		 * pass dangling pointers to munmap.  The ring is
-		 * unrecoverable — destroy it.
-		 */
-		memset(&ring->sq, 0, sizeof(ring->sq));
-		memset(&ring->cq, 0, sizeof(ring->cq));
-		io_uring_queue_exit(ring);
-		goto out;
-	}
-
+	ring->sq = sq;
+	ring->cq = cq;
 	ring->sq.sqe_head = sq_head;
 	ring->sq.sqe_tail = sq_tail;