From d86b6e89ce4a670aa25c826c4f5b662e6a570646 Mon Sep 17 00:00:00 2001 From: bright Date: Sun, 26 Jul 2026 20:49:45 +0800 Subject: [PATCH 1/2] refactor: collapse duplicated operation classes and submitter methods - Add SingleCqeOperation, which carries user_data/cqe_received handling and the common operate() shape; each concrete operation now only declares its fields, get_file_obj and _result. SendfileOperation keeps its custom multi-CQE logic. ~330 lines of copy-paste removed with identical behavior (all construction sites already used keyword args). - _ProactorSubmit's eleven identical prep methods now delegate to a single _prep helper, keeping the TypedDict-typed signatures. Co-Authored-By: Claude Fable 5 --- uringloop/operation.py | 360 +++++++++-------------------------------- uringloop/proactor.py | 95 +++-------- 2 files changed, 100 insertions(+), 355 deletions(-) diff --git a/uringloop/operation.py b/uringloop/operation.py index e91ed23..1e5aec8 100644 --- a/uringloop/operation.py +++ b/uringloop/operation.py @@ -1,6 +1,6 @@ from abc import ABC, abstractmethod from collections.abc import Buffer -from dataclasses import dataclass +from dataclasses import dataclass, field from io import BufferedReader, IOBase import os import socket @@ -43,371 +43,199 @@ def get_os_error(res: int) -> OSError: @dataclass(slots=True) -class SendOperation(BaseOperation): - sock: socket.socket - buffer: Annotated[Buffer, "readable"] - flags: int - user_data: int - cqe_received: bool = False +class SingleCqeOperation(BaseOperation): + """An operation completed by exactly one CQE. + + Subclasses only define their fields, get_file_obj and _result. + """ + + user_data: int = field(kw_only=True) + cqe_received: bool = field(default=False, kw_only=True) def get_user_data( self, ) -> int | None: return None if self.cqe_received else self.user_data - def get_file_obj(self) -> Any: - return self.sock + def mark_seen(self, user_data: int) -> None: + if user_data != self.user_data: + raise RuntimeError(f"Unknown user_data: {user_data} is not expected.") + self.cqe_received = True + + def all_seen(self) -> bool: + return self.cqe_received def operate(self, cqe: IoUringCqe, fut: "_IoUringFuture"): res: int = cqe.res if res < 0: fut.set_exception(get_os_error(res)) else: - fut.set_result(res) + fut.set_result(self._result(res)) - def mark_seen(self, user_data: int): - if user_data == self.user_data: - self.cqe_received = True - else: - raise RuntimeError(f"Unknown user_data: {user_data} is not expected.") + @abstractmethod + def _result(self, res: int) -> Any: + """Build the future's result from a non-negative CQE res.""" + ... - def all_seen(self) -> bool: - return self.cqe_received + +@dataclass(slots=True) +class SendOperation(SingleCqeOperation): + sock: socket.socket + buffer: Annotated[Buffer, "readable"] + flags: int + + def get_file_obj(self) -> Any: + return self.sock + + def _result(self, res: int) -> int: + return res @dataclass(slots=True) -class WriteOperation(BaseOperation): +class WriteOperation(SingleCqeOperation): file: IOBase buffer: Annotated[Buffer, "readable"] offset: int - user_data: int - cqe_received: bool = False - - def get_user_data( - self, - ) -> int | None: - return None if self.cqe_received else self.user_data def get_file_obj(self) -> Any: return self.file - def operate(self, cqe: IoUringCqe, fut: "_IoUringFuture"): - res: int = cqe.res - if res < 0: - fut.set_exception(get_os_error(res)) - else: - fut.set_result(res) - - def mark_seen(self, user_data: int): - if user_data == self.user_data: - self.cqe_received = True - else: - raise RuntimeError(f"Unknown user_data: {user_data} is not expected.") - - def all_seen(self) -> bool: - return self.cqe_received + def _result(self, res: int) -> int: + return res @dataclass(slots=True) -class RecvOperation(BaseOperation): +class RecvOperation(SingleCqeOperation): sock: socket.socket buffer: Annotated[Buffer, "writable"] flags: int - user_data: int - cqe_received: bool = False - - def get_user_data( - self, - ) -> int | None: - return None if self.cqe_received else self.user_data def get_file_obj(self) -> Any: return self.sock - def operate(self, cqe: IoUringCqe, fut: "_IoUringFuture"): - res: int = cqe.res - if res < 0: - fut.set_exception(get_os_error(res)) - else: - fut.set_result(memoryview(self.buffer)[:res].tobytes()) - - def mark_seen(self, user_data: int): - if self.user_data == user_data: - self.cqe_received = True - else: - raise RuntimeError(f"Unknown user_data: {user_data} is not expected.") - - def all_seen(self) -> bool: - return self.cqe_received + def _result(self, res: int) -> bytes: + return memoryview(self.buffer)[:res].tobytes() @dataclass(slots=True) -class ReadOperation(BaseOperation): +class ReadOperation(SingleCqeOperation): file: IOBase buffer: Annotated[Buffer, "writable"] - user_data: int - cqe_received: bool = False - - def get_user_data( - self, - ) -> int | None: - return None if self.cqe_received else self.user_data def get_file_obj(self) -> Any: return self.file - def operate(self, cqe: IoUringCqe, fut: "_IoUringFuture"): - res: int = cqe.res - if res < 0: - fut.set_exception(get_os_error(res)) - else: - fut.set_result(memoryview(self.buffer)[:res].tobytes()) - - def mark_seen(self, user_data: int): - if self.user_data == user_data: - self.cqe_received = True - else: - raise RuntimeError(f"Unknown user_data: {user_data} is not expected.") - - def all_seen(self) -> bool: - return self.cqe_received + def _result(self, res: int) -> bytes: + return memoryview(self.buffer)[:res].tobytes() @dataclass(slots=True) -class RecvIntoOperation(BaseOperation): +class RecvIntoOperation(SingleCqeOperation): sock: socket.socket buffer: Annotated[Buffer, "writable"] flags: int - user_data: int - cqe_received: bool = False - - def get_user_data( - self, - ) -> int | None: - return None if self.cqe_received else self.user_data def get_file_obj(self) -> Any: return self.sock - def operate(self, cqe: IoUringCqe, fut: "_IoUringFuture"): - res: int = cqe.res - if res < 0: - fut.set_exception(get_os_error(res)) - else: - fut.set_result(res) - - def mark_seen(self, user_data: int): - if self.user_data == user_data: - self.cqe_received = True - else: - raise RuntimeError(f"Unknown user_data: {user_data} is not expected.") - - def all_seen(self) -> bool: - return self.cqe_received + def _result(self, res: int) -> int: + return res @dataclass(slots=True) -class ReadIntoOperation(BaseOperation): +class ReadIntoOperation(SingleCqeOperation): file: IOBase buffer: Annotated[Buffer, "writable"] - user_data: int - cqe_received: bool = False - - def get_user_data( - self, - ) -> int | None: - return None if self.cqe_received else self.user_data def get_file_obj(self) -> Any: return self.file - def operate(self, cqe: IoUringCqe, fut: "_IoUringFuture"): - res: int = cqe.res - if res < 0: - fut.set_exception(get_os_error(res)) - else: - fut.set_result(res) - - def mark_seen(self, user_data: int): - if self.user_data == user_data: - self.cqe_received = True - else: - raise RuntimeError(f"Unknown user_data: {user_data} is not expected.") - - def all_seen(self) -> bool: - return self.cqe_received + def _result(self, res: int) -> int: + return res @dataclass(slots=True) -class RecvFromOperation(BaseOperation): +class RecvFromOperation(SingleCqeOperation): sock: socket.socket buffer: Annotated[Buffer, "writable"] sockaddr: Annotated[Sockaddr, "writable"] flags: int - user_data: int - cqe_received: bool = False - - def get_user_data( - self, - ) -> int | None: - return None if self.cqe_received else self.user_data def get_file_obj(self) -> Any: return self.sock - def operate(self, cqe: IoUringCqe, fut: "_IoUringFuture"): - res: int = cqe.res - if res < 0: - fut.set_exception(get_os_error(res)) - else: - fut.set_result((memoryview(self.buffer)[:res].tobytes(), parse_addr(self.sock.family, self.sockaddr))) - - def mark_seen(self, user_data: int): - if self.user_data == user_data: - self.cqe_received = True - else: - raise RuntimeError(f"Unknown user_data: {user_data} is not expected.") - - def all_seen(self) -> bool: - return self.cqe_received + def _result(self, res: int) -> tuple[bytes, Any]: + return (memoryview(self.buffer)[:res].tobytes(), parse_addr(self.sock.family, self.sockaddr)) @dataclass(slots=True) -class RecvFromIntoOperation(BaseOperation): +class RecvFromIntoOperation(SingleCqeOperation): sock: socket.socket buffer: Annotated[Buffer, "writable"] sockaddr: Annotated[Sockaddr, "writable"] flags: int - user_data: int - cqe_received: bool = False - - def get_user_data( - self, - ) -> int | None: - return None if self.cqe_received else self.user_data def get_file_obj(self) -> Any: return self.sock - def operate(self, cqe: IoUringCqe, fut: "_IoUringFuture"): - res: int = cqe.res - if res < 0: - fut.set_exception(get_os_error(res)) - else: - fut.set_result((res, parse_addr(self.sock.family, self.sockaddr))) - - def mark_seen(self, user_data: int): - if self.user_data == user_data: - self.cqe_received = True - else: - raise RuntimeError(f"Unknown user_data: {user_data} is not expected.") - - def all_seen(self) -> bool: - return self.cqe_received + def _result(self, res: int) -> tuple[int, Any]: + return (res, parse_addr(self.sock.family, self.sockaddr)) @dataclass(slots=True) -class SendToOperation(BaseOperation): +class SendToOperation(SingleCqeOperation): sock: socket.socket buffer: Annotated[Buffer, "readable"] sockaddr: Annotated[Sockaddr, "readable"] | None flags: int - user_data: int - cqe_received: bool = False - - def get_user_data( - self, - ) -> int | None: - return None if self.cqe_received else self.user_data def get_file_obj(self) -> Any: return self.sock - def operate(self, cqe: IoUringCqe, fut: "_IoUringFuture"): - res: int = cqe.res - if res < 0: - fut.set_exception(get_os_error(res)) - else: - fut.set_result(res) - - def mark_seen(self, user_data: int): - if self.user_data == user_data: - self.cqe_received = True - else: - raise RuntimeError(f"Unknown user_data: {user_data} is not expected.") - - def all_seen(self) -> bool: - return self.cqe_received + def _result(self, res: int) -> int: + return res @dataclass(slots=True) -class ConnectOperation(BaseOperation): +class ConnectOperation(SingleCqeOperation): sock: socket.socket sockaddr: Annotated[Sockaddr, "readable"] - user_data: int - cqe_received: bool = False - - def get_user_data( - self, - ) -> int | None: - return None if self.cqe_received else self.user_data def get_file_obj(self) -> Any: return self.sock - def operate(self, cqe: IoUringCqe, fut: "_IoUringFuture"): - res: int = cqe.res - if res < 0: - fut.set_exception(get_os_error(res)) - else: - fut.set_result(None) - - def mark_seen(self, user_data: int): - if self.user_data == user_data: - self.cqe_received = True - else: - raise RuntimeError(f"Unknown user_data: {user_data} is not expected.") - - def all_seen(self) -> bool: - return self.cqe_received + def _result(self, res: int) -> None: + return None @dataclass(slots=True) -class AcceptOperation(BaseOperation): +class AcceptOperation(SingleCqeOperation): sock: socket.socket sockaddr: Annotated[Sockaddr, "writable"] flags: int - user_data: int - cqe_received: bool = False - - def get_user_data( - self, - ) -> int | None: - return None if self.cqe_received else self.user_data def get_file_obj(self) -> Any: return self.sock - def operate(self, cqe: IoUringCqe, fut: "_IoUringFuture"): - res: int = cqe.res - if res < 0: - fut.set_exception(get_os_error(res)) - else: - # socket.fromfd would dup() the fd and leak the original one; - # wrap the accepted fd directly instead (same as socket.accept) - sock = socket.socket(self.sock.family, self.sock.type, self.sock.proto, fileno=res) - if socket.getdefaulttimeout() is None and self.sock.gettimeout(): - sock.setblocking(True) - fut.set_result((sock, parse_addr(self.sock.family, self.sockaddr))) + def _result(self, res: int) -> tuple[socket.socket, Any]: + # socket.fromfd would dup() the fd and leak the original one; + # wrap the accepted fd directly instead (same as socket.accept) + sock = socket.socket(self.sock.family, self.sock.type, self.sock.proto, fileno=res) + if socket.getdefaulttimeout() is None and self.sock.gettimeout(): + sock.setblocking(True) + return (sock, parse_addr(self.sock.family, self.sockaddr)) - def mark_seen(self, user_data: int): - if self.user_data == user_data: - self.cqe_received = True - else: - raise RuntimeError(f"Unknown user_data: {user_data} is not expected.") - def all_seen(self) -> bool: - return self.cqe_received +@dataclass(slots=True) +class PollAddOperation(SingleCqeOperation): + file: IOBase | socket.socket | int # some source is not a obj + poll_mask: int + + def get_file_obj(self) -> Any: + return self.file + + def _result(self, res: int) -> int: + return res @dataclass(slots=True) @@ -461,35 +289,3 @@ def mark_seen(self, user_data: int): def all_seen(self) -> bool: return self.p2s_done - - -@dataclass(slots=True) -class PollAddOperation(BaseOperation): - file: IOBase | socket.socket | int # some source is not a obj - poll_mask: int - user_data: int - cqe_received: bool = False - - def get_user_data( - self, - ) -> int | None: - return None if self.cqe_received else self.user_data - - def get_file_obj(self) -> Any: - return self.file - - def operate(self, cqe: IoUringCqe, fut: "_IoUringFuture"): - res: int = cqe.res - if res < 0: - fut.set_exception(get_os_error(res)) - else: - fut.set_result(res) - - def mark_seen(self, user_data: int): - if self.user_data == user_data: - self.cqe_received = True - else: - raise RuntimeError(f"Unknown user_data: {user_data} is not expected.") - - def all_seen(self) -> bool: - return self.cqe_received diff --git a/uringloop/proactor.py b/uringloop/proactor.py index 1e3e603..7647ae4 100644 --- a/uringloop/proactor.py +++ b/uringloop/proactor.py @@ -1,5 +1,5 @@ from asyncio import events, futures -from collections.abc import Buffer +from collections.abc import Buffer, Callable from dataclasses import dataclass import errno from io import BufferedReader, IOBase @@ -136,104 +136,53 @@ def ensure_capacity(self, count: int) -> None: 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: + def _prep( + self, + prep_fn: Callable[[Any, Any], None], + request: KernelRequest, + user_data: int, + flags: int, + ) -> Self: sqe = self._get_sqe() - io_uring_prep_recv(sqe, request) + prep_fn(sqe, request) io_uring_sqe_set_data64(sqe, user_data) if flags: io_uring_sqe_set_flags(sqe, flags) self._unsubmitted.append((user_data, request)) return self + def recv(self, request: RecvRequest, user_data: int, flags: int = 0) -> Self: + return self._prep(io_uring_prep_recv, request, user_data, flags) + def read(self, request: ReadRequest, user_data: int, flags: int = 0) -> Self: - sqe = self._get_sqe() - io_uring_prep_read(sqe, request) - io_uring_sqe_set_data64(sqe, user_data) - if flags: - io_uring_sqe_set_flags(sqe, flags) - self._unsubmitted.append((user_data, request)) - return self + return self._prep(io_uring_prep_read, request, user_data, flags) def recvfrom(self, request: RecvFromRequest, user_data: int, flags: int = 0) -> Self: - sqe = self._get_sqe() - io_uring_prep_recvfrom(sqe, request) - io_uring_sqe_set_data64(sqe, user_data) - if flags: - io_uring_sqe_set_flags(sqe, flags) - self._unsubmitted.append((user_data, request)) - return self + return self._prep(io_uring_prep_recvfrom, request, user_data, flags) def sendto(self, request: SendToRequest, user_data: int, flags: int = 0) -> Self: - sqe = self._get_sqe() - io_uring_prep_sendto(sqe, request) - io_uring_sqe_set_data64(sqe, user_data) - if flags: - io_uring_sqe_set_flags(sqe, flags) - self._unsubmitted.append((user_data, request)) - return self + return self._prep(io_uring_prep_sendto, request, user_data, flags) def send(self, request: SendRequest, user_data: int, flags: int = 0) -> Self: - sqe = self._get_sqe() - io_uring_prep_send(sqe, request) - io_uring_sqe_set_data64(sqe, user_data) - if flags: - io_uring_sqe_set_flags(sqe, flags) - self._unsubmitted.append((user_data, request)) - return self + return self._prep(io_uring_prep_send, request, user_data, flags) def write(self, request: WriteRequest, user_data: int, flags: int = 0) -> Self: - sqe = self._get_sqe() - io_uring_prep_write(sqe, request) - io_uring_sqe_set_data64(sqe, user_data) - if flags: - io_uring_sqe_set_flags(sqe, flags) - self._unsubmitted.append((user_data, request)) - return self + return self._prep(io_uring_prep_write, request, user_data, flags) def accept(self, request: AcceptRequest, user_data: int, flags: int = 0) -> Self: - sqe = self._get_sqe() - io_uring_prep_accept(sqe, request) - io_uring_sqe_set_data64(sqe, user_data) - if flags: - io_uring_sqe_set_flags(sqe, flags) - self._unsubmitted.append((user_data, request)) - return self + return self._prep(io_uring_prep_accept, request, user_data, flags) def connect(self, request: ConnectRequest, user_data: int, flags: int = 0) -> Self: - sqe = self._get_sqe() - io_uring_prep_connect(sqe, request) - io_uring_sqe_set_data64(sqe, user_data) - if flags: - io_uring_sqe_set_flags(sqe, flags) - self._unsubmitted.append((user_data, request)) - return self + return self._prep(io_uring_prep_connect, request, user_data, flags) def splice(self, request: SpliceRequest, user_data: int, flags: int = 0) -> Self: - sqe = self._get_sqe() - io_uring_prep_splice(sqe, request) - io_uring_sqe_set_data64(sqe, user_data) - if flags: - io_uring_sqe_set_flags(sqe, flags) - self._unsubmitted.append((user_data, request)) - return self + return self._prep(io_uring_prep_splice, request, user_data, flags) def cancel(self, request: Cancel64Request, user_data: int, flags: int = 0) -> Self: - sqe = self._get_sqe() - io_uring_prep_cancel64(sqe, request) - io_uring_sqe_set_data64(sqe, user_data) - if flags: - io_uring_sqe_set_flags(sqe, flags) - self._unsubmitted.append((user_data, request)) - return self + return self._prep(io_uring_prep_cancel64, request, user_data, flags) def poll_add(self, request: PollAddRequest, user_data: int, flags: int = 0) -> Self: - sqe = self._get_sqe() - io_uring_prep_poll_add(sqe, request) - io_uring_sqe_set_data64(sqe, user_data) - if flags: - io_uring_sqe_set_flags(sqe, flags) - self._unsubmitted.append((user_data, request)) - return self + return self._prep(io_uring_prep_poll_add, request, user_data, flags) def submit(self, op: BaseOperation, fut: _IoUringFuture | None): """Register the prepared SQEs; the syscall is deferred to flush(). From 02b083b3d86f4bb67b80ea277feacd51b209f3c4 Mon Sep 17 00:00:00 2001 From: bright Date: Sun, 26 Jul 2026 20:51:26 +0800 Subject: [PATCH 2/2] refactor: consistent IoUring naming, typo fixes, test conftest cleanup - Rename IouringProactorEventLoop(-Policy) and _IouringWritePipeTransport to the IoUring prefix already used by IoUringProactor; the old public names remain importable as deprecated aliases. - Fix ProatorCache -> ProactorCache, test_tcp_conmmunication, 'C extention', and a 'woulbe' comment. - Drop the no-op try/finally in IoUringProactor.select. - Loop-test conftest now overrides pytest-asyncio's event_loop_policy fixture (the documented mechanism) instead of setting the global policy from an async autouse fixture, and drops the deprecated event_loop fixture nothing used. - Proactor-test conftest: use get_running_loop, guard the failure callback against cancelled tasks (task.exception() raises there), and remove the run_until_complete call that is invalid in a running loop. Co-Authored-By: Claude Fable 5 --- README.md | 10 +++++----- tests/e2e/loop/conftest.py | 19 ++++++------------- tests/e2e/loop/test_subprocess.py | 10 +++++----- tests/e2e/loop/test_unix.py | 4 ++-- tests/e2e/proactor/conftest.py | 8 ++++---- tests/e2e/proactor/test_tcp.py | 2 +- uringloop/__init__.py | 16 ++++++++++++++-- uringloop/loop.py | 17 +++++++++++------ uringloop/proactor.py | 15 ++++++--------- 9 files changed, 54 insertions(+), 47 deletions(-) diff --git a/README.md b/README.md index e6d8f91..99f0c0f 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,7 @@ A Python implementation of a liburing-based proactor event loop for asyncio, des ## Goals -- Provide a primarily Python (with CFFI, maybe C extention in future) implementation of a liburing-based event loop +- Provide a primarily Python (with CFFI, maybe C extension in future) implementation of a liburing-based event loop - Maintain full compatibility with standard asyncio APIs - Follow Python standard library implementation patterns @@ -46,7 +46,7 @@ Pass the loop factory to `asyncio.run` to use the io_uring-based event loop: ```python import asyncio -from uringloop import IouringProactorEventLoop +from uringloop import IoUringProactorEventLoop async def handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter): @@ -110,7 +110,7 @@ async def main(): if __name__ == "__main__": - asyncio.run(main(), loop_factory=IouringProactorEventLoop) + asyncio.run(main(), loop_factory=IoUringProactorEventLoop) ``` @@ -131,9 +131,9 @@ An event loop policy is also available for code that still uses the (deprecated ```python import asyncio -from uringloop import IouringProactorEventLoopPolicy +from uringloop import IoUringProactorEventLoopPolicy -asyncio.set_event_loop_policy(IouringProactorEventLoopPolicy()) +asyncio.set_event_loop_policy(IoUringProactorEventLoopPolicy()) asyncio.run(main()) ``` diff --git a/tests/e2e/loop/conftest.py b/tests/e2e/loop/conftest.py index 4eeeaea..6da19b0 100644 --- a/tests/e2e/loop/conftest.py +++ b/tests/e2e/loop/conftest.py @@ -1,24 +1,17 @@ -import asyncio import os import tempfile -from typing import cast import pytest -import pytest_asyncio -from uringloop.loop import IouringProactorEventLoop, IouringProactorEventLoopPolicy +from uringloop.loop import IoUringProactorEventLoopPolicy -@pytest_asyncio.fixture(scope="package", autouse=True) -async def event_loop_policy(): - asyncio.set_event_loop_policy(IouringProactorEventLoopPolicy()) +@pytest.fixture(scope="package") +def event_loop_policy(): + # overriding this fixture is the documented pytest-asyncio way to run + # every test in the package on loops created by this policy + return IoUringProactorEventLoopPolicy() -# TODO: remove pytest_asyncio warning -@pytest.fixture -def event_loop(): - loop = asyncio.get_event_loop() - yield cast(IouringProactorEventLoop, loop) - loop.close() @pytest.fixture def unix_socket_path(): diff --git a/tests/e2e/loop/test_subprocess.py b/tests/e2e/loop/test_subprocess.py index 97d54d6..cbeb085 100644 --- a/tests/e2e/loop/test_subprocess.py +++ b/tests/e2e/loop/test_subprocess.py @@ -3,13 +3,13 @@ import pytest -from uringloop.loop import IouringProactorEventLoop, _IouringWritePipeTransport # type: ignore[reportPrivateUsage] +from uringloop.loop import IoUringProactorEventLoop, _IoUringWritePipeTransport # type: ignore[reportPrivateUsage] @pytest.mark.asyncio async def test_subprocess_basic(): event_loop = asyncio.get_running_loop() - assert isinstance(event_loop, IouringProactorEventLoop) + assert isinstance(event_loop, IoUringProactorEventLoop) # Create a protocol class that tracks completion class SubprocessProtocol(asyncio.SubprocessProtocol): def __init__(self, exit_future: asyncio.Future[Any]): @@ -37,7 +37,7 @@ def process_exited(self): @pytest.mark.asyncio async def test_subprocess_io(): event_loop = asyncio.get_running_loop() - assert isinstance(event_loop, IouringProactorEventLoop) + assert isinstance(event_loop, IoUringProactorEventLoop) class SubprocessProtocol(asyncio.SubprocessProtocol): def __init__(self, exit_future: asyncio.Future[Any]): self.exit_future = exit_future @@ -56,7 +56,7 @@ def process_exited(self): # Write to stdin and close stdin = transport.get_pipe_transport(0) - assert isinstance(stdin, _IouringWritePipeTransport), f"stdin got unexpected type {type(stdin)}" + assert isinstance(stdin, _IoUringWritePipeTransport), f"stdin got unexpected type {type(stdin)}" stdin.write(b"test data\n") stdin.close() @@ -70,7 +70,7 @@ def process_exited(self): @pytest.mark.asyncio async def test_subprocess_error(): event_loop = asyncio.get_running_loop() - assert isinstance(event_loop, IouringProactorEventLoop) + assert isinstance(event_loop, IoUringProactorEventLoop) class SubprocessProtocol(asyncio.SubprocessProtocol): def __init__(self, exit_future: asyncio.Future[Any]): self.exit_future = exit_future diff --git a/tests/e2e/loop/test_unix.py b/tests/e2e/loop/test_unix.py index 5fd52fa..81912e1 100644 --- a/tests/e2e/loop/test_unix.py +++ b/tests/e2e/loop/test_unix.py @@ -4,13 +4,13 @@ import pytest -from uringloop.loop import IouringProactorEventLoop +from uringloop.loop import IoUringProactorEventLoop @pytest.mark.asyncio async def test_unix_connection(unix_socket_path: str): event_loop = asyncio.get_running_loop() - assert isinstance(event_loop, IouringProactorEventLoop) + assert isinstance(event_loop, IoUringProactorEventLoop) # Server protocol class EchoServerProtocol(asyncio.Protocol): def connection_made(self, transport: asyncio.BaseTransport): diff --git a/tests/e2e/proactor/conftest.py b/tests/e2e/proactor/conftest.py index 248f322..ff53844 100644 --- a/tests/e2e/proactor/conftest.py +++ b/tests/e2e/proactor/conftest.py @@ -79,7 +79,7 @@ def server_udp_sock() -> Generator[socket.socket, None, None]: @pytest_asyncio.fixture async def init_proactor() -> AsyncGenerator[IoUringProactor, None]: """Fixture to create and clean up the proactor and its polling task.""" - loop = asyncio.get_event_loop() + loop = asyncio.get_running_loop() proactor = IoUringProactor() proactor.set_loop(loop) @@ -93,14 +93,14 @@ async def _run_proactor_task(): task = loop.create_task(_run_proactor_task()) def stop_all_coro_if_raise_exception(task: asyncio.Task[None]): + if task.cancelled(): + return if task.exception(): # If task failed - # Cancel all other tasks + # Cancel all other tasks so the test fails instead of hanging for t in asyncio.all_tasks(loop): if t != task and not t.done(): t.cancel() - loop.run_until_complete(loop.shutdown_asyncgens()) - # Add failure callback task.add_done_callback(stop_all_coro_if_raise_exception) diff --git a/tests/e2e/proactor/test_tcp.py b/tests/e2e/proactor/test_tcp.py index 10def35..5d8a1d8 100644 --- a/tests/e2e/proactor/test_tcp.py +++ b/tests/e2e/proactor/test_tcp.py @@ -8,7 +8,7 @@ @pytest.mark.asyncio -async def test_tcp_conmmunication( +async def test_tcp_communication( init_proactor: IoUringProactor, client_tcp_sock: socket.socket, server_tcp_sock: socket.socket ) -> None: """Integration test using real socket communication with the echo server.""" diff --git a/uringloop/__init__.py b/uringloop/__init__.py index 6c4e871..66739c8 100644 --- a/uringloop/__init__.py +++ b/uringloop/__init__.py @@ -1,10 +1,22 @@ -__all__ = ["IoUringProactor", "IouringProactorEventLoopPolicy", "IouringProactorEventLoop"] +__all__ = [ + "IoUringProactor", + "IoUringProactorEventLoop", + "IoUringProactorEventLoopPolicy", + # deprecated aliases (0.1.x naming) + "IouringProactorEventLoop", + "IouringProactorEventLoopPolicy", +] import platform import re -from uringloop.loop import IouringProactorEventLoop, IouringProactorEventLoopPolicy +from uringloop.loop import ( + IoUringProactorEventLoop, + IoUringProactorEventLoopPolicy, + IouringProactorEventLoop, + IouringProactorEventLoopPolicy, +) from uringloop.proactor import IoUringProactor diff --git a/uringloop/loop.py b/uringloop/loop.py index 1464612..a7c2cb5 100644 --- a/uringloop/loop.py +++ b/uringloop/loop.py @@ -12,7 +12,7 @@ from uringloop.proactor import IoUringProactor -class _IouringWritePipeTransport(proactor_events._ProactorBaseWritePipeTransport): # type: ignore[reportPrivateUsage] +class _IoUringWritePipeTransport(proactor_events._ProactorBaseWritePipeTransport): # type: ignore[reportPrivateUsage] def __init__(self, *args, **kw): super().__init__(*args, **kw) self._read_fut = cast(IoUringProactor, self._loop._proactor).poll_add(self._sock, POLLERR | POLLHUP) # type: ignore[reportPrivateUsage] @@ -35,8 +35,8 @@ def _pipe_closed(self, fut: futures.Future[int]): self.close() -class IouringProactorEventLoop(proactor_events.BaseProactorEventLoop): - """Linux version of proactor event loop using Iouring.""" +class IoUringProactorEventLoop(proactor_events.BaseProactorEventLoop): + """Linux version of proactor event loop using IoUring.""" def __init__(self, proactor: IoUringProactor | None = None): # BaseEventLoop.__del__ may run when constructing the default proactor @@ -237,7 +237,7 @@ def _child_exited(fut: futures.Future[int]): fut.add_done_callback(_child_exited) def _make_write_pipe_transport(self, sock, protocol, waiter=None, extra=None): - return _IouringWritePipeTransport(self, sock, protocol, waiter, extra) + return _IoUringWritePipeTransport(self, sock, protocol, waiter, extra) # Preserve the child-watcher methods on Python 3.12 and 3.13 by using the Unix @@ -248,5 +248,10 @@ def _make_write_pipe_transport(self, sock, protocol, waiter=None, extra=None): ) -class IouringProactorEventLoopPolicy(_BaseDefaultEventLoopPolicy): - _loop_factory = IouringProactorEventLoop +class IoUringProactorEventLoopPolicy(_BaseDefaultEventLoopPolicy): + _loop_factory = IoUringProactorEventLoop + + +# deprecated aliases kept for backwards compatibility with the 0.1.x naming +IouringProactorEventLoop = IoUringProactorEventLoop +IouringProactorEventLoopPolicy = IoUringProactorEventLoopPolicy diff --git a/uringloop/proactor.py b/uringloop/proactor.py index 7647ae4..592d76f 100644 --- a/uringloop/proactor.py +++ b/uringloop/proactor.py @@ -84,7 +84,7 @@ class PendingCompletion: request: KernelRequest -ProatorCache: TypeAlias = dict[int, PendingCompletion] +ProactorCache: TypeAlias = dict[int, PendingCompletion] DEFAULT_ENTRIES = 256 @@ -112,9 +112,9 @@ def cancel(self, msg: Any | None = None): class _ProactorSubmit: - def __init__(self, ring: IoUring, cache: ProatorCache) -> None: + def __init__(self, ring: IoUring, cache: ProactorCache) -> None: self._iouring = ring - self._cache: ProatorCache = cache + self._cache: ProactorCache = cache self._unsubmitted: list[tuple[int, KernelRequest]] = [] self._pending_submit = False @@ -221,7 +221,7 @@ def __init__(self, entries: int = DEFAULT_ENTRIES, flags: int = 0): io_uring_queue_init(entries, ring, flags) self._iouring = ring - self._cache: ProatorCache = {} + self._cache: ProactorCache = {} self._stopped_serving: WeakSet[Any] = weakref.WeakSet() self.submitter = _ProactorSubmit(self._iouring, self._cache) # unique per-operation key for the SQE user_data field; an object id() @@ -244,10 +244,7 @@ def select(self, timeout: float | None = None): self._poll(timeout) tmp = self._results self._results = [] - try: - return tmp - finally: - tmp = None + return tmp def recv(self, conn: socket.socket | IOBase, nbytes: int, flags: int = 0) -> futures.Future[bytes]: buf = bytearray(nbytes) @@ -481,7 +478,7 @@ def _handle_cqe(self, cqe: IoUringCqe): op.mark_seen(cqe.user_data) # TODO: figure out the correct way to _stopped_serving, may be io_uring_prep_cancel_fd? if op.get_file_obj() in self._stopped_serving: - # the self.cancel_operation woulbe be triggered + # the self.cancel_operation would be triggered # if the user_data is seen, the op.get_user_data would not appeared fut.cancel() else: