From 927316963efc8bcaf29a46ca57248e8148daa338 Mon Sep 17 00:00:00 2001 From: bright Date: Sun, 26 Jul 2026 20:47:28 +0800 Subject: [PATCH] perf: batch SQE submissions into one syscall per poll - io_uring_submit is no longer issued per operation; submitter.submit only registers the prepared SQEs and _poll flushes them all with a single io_uring_enter before waiting for completions. Previously every recv/send/accept paid its own syscall, which negates io_uring's batching advantage over epoll. - Raise DEFAULT_ENTRIES from 16 to 256; the submitter already flushes and retries if the queue fills up mid-batch. - Remove the unused *_nogil C wrappers: CFFI API-mode calls release the GIL around C calls already, so they were dead code. - Poll every 10ms instead of 100ms in the proactor e2e conftest, since operations now only make progress when _poll runs. Co-Authored-By: Claude Fable 5 --- ffi.py | 25 +++----------- tests/e2e/proactor/conftest.py | 2 +- tests/unit/test_submission.py | 60 ++++++++++++++++++++++++++++++++++ uringloop/lib.py | 11 ++++++- uringloop/proactor.py | 41 +++++++++++++++++++++-- 5 files changed, 113 insertions(+), 26 deletions(-) create mode 100644 tests/unit/test_submission.py diff --git a/ffi.py b/ffi.py index ad449b0..1fe5d9b 100644 --- a/ffi.py +++ b/ffi.py @@ -3,6 +3,8 @@ ffibuilder = cffi.FFI() +# NOTE: no Py_BEGIN_ALLOW_THREADS wrappers are needed here; CFFI API-mode +# calls already release the GIL around C function calls. source_code = """ #include #include @@ -13,25 +15,7 @@ #include #include #include - #include #include "liburing.h" - - int io_uring_wait_cqe_nogil(struct io_uring *ring, struct io_uring_cqe **cqe_ptr) { - int res; - Py_BEGIN_ALLOW_THREADS - res = io_uring_wait_cqe(ring, cqe_ptr); - Py_END_ALLOW_THREADS - return res; - } - - int io_uring_wait_cqe_timeout_nogil(struct io_uring *ring, struct io_uring_cqe **cqe_ptr, - struct __kernel_timespec *ts) { - int res; - Py_BEGIN_ALLOW_THREADS - res = io_uring_wait_cqe_timeout(ring, cqe_ptr, ts); - Py_END_ALLOW_THREADS - return res; - } """ @@ -321,6 +305,8 @@ static inline void io_uring_prep_write(struct io_uring_sqe *sqe, int fd, const void *buf, unsigned nbytes, __u64 offset); struct io_uring_sqe *io_uring_get_sqe(struct io_uring *ring); + static inline unsigned io_uring_sq_ready(const struct io_uring *ring); + static inline unsigned io_uring_sq_space_left(const struct io_uring *ring); static inline void io_uring_sqe_set_data64(struct io_uring_sqe *sqe, __u64 data); static inline void io_uring_sqe_set_flags(struct io_uring_sqe *sqe, unsigned flags); @@ -329,9 +315,6 @@ static inline int io_uring_wait_cqe(struct io_uring *ring, struct io_uring_cqe **cqe_ptr); static inline void io_uring_cqe_seen(struct io_uring *ring, struct io_uring_cqe *cqe); int io_uring_wait_cqe_timeout(struct io_uring *ring, struct io_uring_cqe **cqe_ptr, struct __kernel_timespec *ts); - - int io_uring_wait_cqe_nogil(struct io_uring *ring, struct io_uring_cqe **cqe_ptr); - int io_uring_wait_cqe_timeout_nogil(struct io_uring *ring, struct io_uring_cqe **cqe_ptr, struct __kernel_timespec *ts); """) # TODO: remove this trick, i feel it is not a official way. diff --git a/tests/e2e/proactor/conftest.py b/tests/e2e/proactor/conftest.py index 0479a31..248f322 100644 --- a/tests/e2e/proactor/conftest.py +++ b/tests/e2e/proactor/conftest.py @@ -88,7 +88,7 @@ async def init_proactor() -> AsyncGenerator[IoUringProactor, None]: async def _run_proactor_task(): while True: proactor._poll(timeout=0.0) # type: ignore[reportPrivateUsage] - await asyncio.sleep(0.1) + await asyncio.sleep(0.01) task = loop.create_task(_run_proactor_task()) diff --git a/tests/unit/test_submission.py b/tests/unit/test_submission.py new file mode 100644 index 0000000..8e8de56 --- /dev/null +++ b/tests/unit/test_submission.py @@ -0,0 +1,60 @@ +import errno + +import pytest + +import uringloop.proactor +from uringloop.proactor import _ProactorSubmit + + +def test_flush_retries_partial_submissions(monkeypatch): + submitter = _ProactorSubmit(object(), {}) + submitter._pending_submit = True + submissions = iter((1, 1)) + ready = iter((1, 0)) + submit_calls = 0 + + def submit(ring): + nonlocal submit_calls + submit_calls += 1 + return next(submissions) + + monkeypatch.setattr(uringloop.proactor, "io_uring_submit", submit) + monkeypatch.setattr(uringloop.proactor, "io_uring_sq_ready", lambda ring: next(ready)) + + submitter.flush() + + assert submit_calls == 2 + assert not submitter._pending_submit + + +def test_flush_preserves_pending_state_on_error(monkeypatch): + submitter = _ProactorSubmit(object(), {}) + submitter._pending_submit = True + + def submit(ring): + raise OSError(errno.EAGAIN, "try again") + + monkeypatch.setattr(uringloop.proactor, "io_uring_submit", submit) + + with pytest.raises(OSError, match="try again"): + submitter.flush() + + assert submitter._pending_submit + + +def test_linked_submission_reserves_all_slots(monkeypatch): + submitter = _ProactorSubmit(object(), {}) + submitter._pending_submit = True + available = iter((1, 2)) + flushed = False + + def flush(): + nonlocal flushed + flushed = True + + monkeypatch.setattr(uringloop.proactor, "io_uring_sq_space_left", lambda ring: next(available)) + monkeypatch.setattr(submitter, "flush", flush) + + submitter.ensure_capacity(2) + + assert flushed diff --git a/uringloop/lib.py b/uringloop/lib.py index 69a3689..69450ca 100644 --- a/uringloop/lib.py +++ b/uringloop/lib.py @@ -430,10 +430,19 @@ def io_uring_cqe_seen(ring: IoUring, cqe: IoUringCqe) -> None: lib.io_uring_cqe_seen(ring, cqe) -def io_uring_submit(ring: IoUring) -> None: +def io_uring_submit(ring: IoUring) -> int: res = lib.io_uring_submit(ring) if res < 0: raise OSError(-res, os.strerror(-res)) + return res + + +def io_uring_sq_ready(ring: IoUring) -> int: + return lib.io_uring_sq_ready(ring) + + +def io_uring_sq_space_left(ring: IoUring) -> int: + return lib.io_uring_sq_space_left(ring) def io_uring_peek_cqe(ring: IoUring) -> IoUringCqe: diff --git a/uringloop/proactor.py b/uringloop/proactor.py index 82d4587..1e3e603 100644 --- a/uringloop/proactor.py +++ b/uringloop/proactor.py @@ -31,6 +31,8 @@ io_uring_prep_write, io_uring_queue_exit, io_uring_queue_init, + io_uring_sq_ready, + io_uring_sq_space_left, io_uring_sqe_set_data64, io_uring_sqe_set_flags, io_uring_submit, @@ -85,7 +87,7 @@ class PendingCompletion: ProatorCache: TypeAlias = dict[int, PendingCompletion] -DEFAULT_ENTRIES = 16 +DEFAULT_ENTRIES = 256 class _IoUringFuture(futures.Future[Any]): @@ -114,18 +116,26 @@ def __init__(self, ring: IoUring, cache: ProatorCache) -> None: self._iouring = ring self._cache: ProatorCache = cache self._unsubmitted: list[tuple[int, KernelRequest]] = [] + self._pending_submit = False def _get_sqe(self) -> Any: sqe = io_uring_get_sqe(self._iouring) if sqe is None: # submission queue is full: flush the prepared SQEs to the kernel # to free up slots, then retry - io_uring_submit(self._iouring) + self.flush() sqe = io_uring_get_sqe(self._iouring) if sqe is None: raise RuntimeError("io_uring submission queue is full") return sqe + def ensure_capacity(self, count: int) -> None: + """Ensure a group of linked SQEs fits in one submission.""" + if io_uring_sq_space_left(self._iouring) < count: + self.flush() + if io_uring_sq_space_left(self._iouring) < count: + raise RuntimeError(f"io_uring submission queue cannot fit {count} linked entries") + def recv(self, request: RecvRequest, user_data: int, flags: int = 0) -> Self: sqe = self._get_sqe() io_uring_prep_recv(sqe, request) @@ -226,10 +236,29 @@ def poll_add(self, request: PollAddRequest, user_data: int, flags: int = 0) -> S return self def submit(self, op: BaseOperation, fut: _IoUringFuture | None): + """Register the prepared SQEs; the syscall is deferred to flush(). + + Batching all SQEs prepared between two polls into one + io_uring_submit call is what makes io_uring cheaper than epoll: + one io_uring_enter per loop iteration instead of one per operation. + """ for user_data, request in self._unsubmitted: self._cache[user_data] = PendingCompletion(op, fut, request) - io_uring_submit(self._iouring) self._unsubmitted = [] + self._pending_submit = True + + def flush(self): + while self._pending_submit: + try: + submitted = io_uring_submit(self._iouring) + except OSError as exc: + if exc.errno == errno.EINTR: + continue + raise + if io_uring_sq_ready(self._iouring) == 0: + self._pending_submit = False + elif submitted == 0: + raise RuntimeError("io_uring made no progress submitting queued entries") class IoUringProactor: @@ -369,6 +398,8 @@ def connect(self, conn: socket.socket, address: pyAddress) -> futures.Future[Non def sendfile(self, sock: socket.socket, file: BufferedReader, offset: int, count: int) -> futures.Future[int]: """NOTE: edge cases would be handled by event loop""" + # Linked SQEs cannot cross submission boundaries. + self.submitter.ensure_capacity(2) pipe_r, pipe_w = os.pipe() f2p_request = SpliceRequest( @@ -423,6 +454,10 @@ def poll_add(self, file: socket.socket | IOBase | int, poll_mask: int) -> future return fut def _poll(self, timeout: float | None = None): + # push every SQE prepared since the last poll to the kernel in a + # single syscall before waiting for completions + self.submitter.flush() + if timeout is None: while True: try: