30713547314

Committed 01 Aug 2026 06:53PM UTC coverage: 86.197%. First build

Build # 30713547314

Build Type

Pull #68

github

Committed by

web-flow

Commit Message

Merge cf65dbe26 into 962e4bb2d

Pull Request Pull Request #68: Rewrite native backends from C to Rust (thread_aio, linux_aio, linux_uring)

Coverage Stats

11 of 14 new or added lines in 3 files covered. (78.57%)

306 of 355 relevant lines covered (86.2%)

8.02 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.67

/caio/python_aio.py

import os
from collections import defaultdict
from collections.abc import Callable
from enum import IntEnum, unique
from io import BytesIO
from multiprocessing.pool import ThreadPool
from threading import Lock, RLock
from types import MappingProxyType
from typing import Any

from .abstract import AbstractContext, AbstractOperation

fdsync = getattr(os, "fdatasync", os.fsync)
NATIVE_PREAD_PWRITE = hasattr(os, "pread") and hasattr(os, "pwrite")


@unique
class OpCode(IntEnum):
    READ = 0
    WRITE = 1
    FSYNC = 2
    FDSYNC = 3
    NOOP = -1


class Context(AbstractContext):
    """
    python aio context implementation
    """

    MAX_POOL_SIZE = 128

    def __init__(self, max_requests: int = 32, pool_size: int = 8):
        assert pool_size < self.MAX_POOL_SIZE

        self.__max_requests = max_requests
        self.pool = ThreadPool(pool_size)
        self._in_progress = 0
        self._closed = False
        self._closed_lock = Lock()

        if not NATIVE_PREAD_PWRITE:
            self._locks_cleaner = RLock()       # type: ignore
            self._locks = defaultdict(RLock)    # type: ignore

    @property
    def max_requests(self) -> int:
        return self.__max_requests

    def _execute(self, operation: "Operation"):
        handler = self._OP_MAP[operation.opcode]

        def on_error(exc):
            self._in_progress -= 1
            operation.exception = exc
            operation.written = 0
            operation.callback(None)

        def on_success(result):
            self._in_progress -= 1
            operation.written = result
            operation.callback(result)

        if self._in_progress > self.__max_requests:
            raise RuntimeError(
                "Maximum simultaneous requests have been reached",
            )

        self._in_progress += 1

        self.pool.apply_async(
            handler, args=(self, operation),
            callback=on_success,
            error_callback=on_error,
        )

    if NATIVE_PREAD_PWRITE:
        def __pread(self, fd, size, offset):
            return os.pread(fd, size, offset)

        def __pwrite(self, fd, bytes, offset):
            return os.pwrite(fd, bytes, offset)
    else:
        def __pread(self, fd, size, offset):
            with self._locks[fd]:
                os.lseek(fd, 0, os.SEEK_SET)
                os.lseek(fd, offset, os.SEEK_SET)
                return os.read(fd, size)

        def __pwrite(self, fd, bytes, offset):
            with self._locks[fd]:
                os.lseek(fd, 0, os.SEEK_SET)
                os.lseek(fd, offset, os.SEEK_SET)
                return os.write(fd, bytes)

    def _handle_read(self, operation: "Operation"):
        return operation.buffer.write(
            self.__pread(
                operation.fileno,
                operation.nbytes,
                operation.offset,
            ),
        )

    def _handle_write(self, operation: "Operation"):
        return self.__pwrite(
            operation.fileno, operation.buffer.getvalue(), operation.offset,
        )

    def _handle_fsync(self, operation: "Operation"):
        return os.fsync(operation.fileno)

    def _handle_fdsync(self, operation: "Operation"):
        return fdsync(operation.fileno)

    def _handle_noop(self, operation: "Operation"):
        return

    def submit(self, *aio_operations) -> int:
        operations = []

        for operation in aio_operations:
            if not isinstance(operation, Operation):
                raise ValueError(f"Invalid Operation {operation!r}")  # noqa: TRY004 (public API, keep ValueError)

            operations.append(operation)

        count = 0
        for operation in operations:
            self._execute(operation)
            count += 1

        return count

    def cancel(self, *aio_operations) -> int:
        """
        Cancels multiple Operations. Returns

         Operation.cancel(aio_op1, aio_op2, aio_opN, ...) -> int

        (Always returns zero, this method exists for compatibility reasons)
        """
        return 0

    def close(self):
        if self._closed:
            return

        with self._closed_lock:
            self.pool.close()
            self._closed = True

    def __del__(self):
        if self.pool.close():
            self.close()

    _OP_MAP = MappingProxyType({
        OpCode.READ: _handle_read,
        OpCode.WRITE: _handle_write,
        OpCode.FSYNC: _handle_fsync,
        OpCode.FDSYNC: _handle_fdsync,
        OpCode.NOOP: _handle_noop,
    })


# noinspection PyPropertyDefinition
class Operation(AbstractOperation):
    """
    python aio operation implementation
    """
    def __init__(
        self,
        fd: int,
        nbytes: int | None,
        offset: int | None,
        opcode: OpCode,
        payload: bytes | None = None,
        priority: int | None = None,
    ):
        self.callback: Callable[[int], Any] | None = None
        self.buffer = BytesIO()

        if opcode == OpCode.WRITE and payload:
            self.buffer = BytesIO(payload)

        self.opcode = opcode
        self.__fileno = fd
        self.__offset = offset or 0
        self.__opcode = opcode
        self.__nbytes = nbytes or 0
        self.__priority = priority or 0
        self.exception = None
        self.written = 0

    @classmethod
    def read(
        cls, nbytes: int, fd: int, offset: int, priority=0,
    ) -> "Operation":
        """
        Creates a new instance of Operation on read mode.
        """
        return cls(fd, nbytes, offset, opcode=OpCode.READ, priority=priority)

    @classmethod
    def write(
        cls, payload_bytes: bytes, fd: int, offset: int, priority=0,
    ) -> "Operation":
        """
        Creates a new instance of AIOOperation on write mode.
        """
        return cls(
            fd,
            len(payload_bytes),
            offset,
            payload=payload_bytes,
            opcode=OpCode.WRITE,
            priority=priority,
        )

    @classmethod
    def fsync(cls, fd: int, priority=0) -> "Operation":

        """
        Creates a new instance of AIOOperation on fsync mode.
        """
        return cls(fd, None, None, opcode=OpCode.FSYNC, priority=priority)

    @classmethod
    def fdsync(cls, fd: int, priority=0) -> "Operation":

        """
        Creates a new instance of AIOOperation on fdsync mode.
        """
        return cls(fd, None, None, opcode=OpCode.FDSYNC, priority=priority)

    def get_value(self) -> bytes | int:
        """
        Method returns a bytes value of AIOOperation's result or None.
        """
        if self.exception:
            raise self.exception

        if self.opcode == OpCode.WRITE:
            return self.written

        if self.buffer is None:
            return

        return self.buffer.getvalue()

    @property
    def fileno(self) -> int:
        return self.__fileno

    @property
    def offset(self) -> int:
        return self.__offset

    @property
    def payload(self) -> memoryview | None:
        return self.buffer.getbuffer()

    @property
    def nbytes(self) -> int:
        return self.__nbytes

    def set_callback(self, callback: Callable[[int], Any]) -> bool:
        self.callback = callback
        return True

1	import os	10✔
2	from collections import defaultdict	10✔
3	from collections.abc import Callable	10✔
4	from enum import IntEnum, unique	10✔
5	from io import BytesIO	10✔
6	from multiprocessing.pool import ThreadPool	10✔
7	from threading import Lock, RLock	10✔
8	from types import MappingProxyType	10✔
9	from typing import Any	10✔
10
11	from .abstract import AbstractContext, AbstractOperation	10✔
12
13	fdsync = getattr(os, "fdatasync", os.fsync)	10✔
14	NATIVE_PREAD_PWRITE = hasattr(os, "pread") and hasattr(os, "pwrite")	10✔
15
16
17	@unique	10✔
18	class OpCode(IntEnum):	10✔
19	READ = 0	10✔
20	WRITE = 1	10✔
21	FSYNC = 2	10✔
22	FDSYNC = 3	10✔
23	NOOP = -1	10✔
24
25
26	class Context(AbstractContext):	10✔
27	"""
28	python aio context implementation
29	"""
30
31	MAX_POOL_SIZE = 128	10✔
32
33	def __init__(self, max_requests: int = 32, pool_size: int = 8):	10✔
34	assert pool_size < self.MAX_POOL_SIZE	10✔
35
36	self.__max_requests = max_requests	10✔
37	self.pool = ThreadPool(pool_size)	10✔
38	self._in_progress = 0	10✔
39	self._closed = False	10✔
40	self._closed_lock = Lock()	10✔
41
42	if not NATIVE_PREAD_PWRITE:	10✔
43	self._locks_cleaner = RLock() # type: ignore	×
44	self._locks = defaultdict(RLock) # type: ignore	×
45
46	@property	10✔
47	def max_requests(self) -> int:	10✔
48	return self.__max_requests	×
49
50	def _execute(self, operation: "Operation"):	10✔
51	handler = self._OP_MAP[operation.opcode]	10✔
52
53	def on_error(exc):	10✔
54	self._in_progress -= 1	10✔
55	operation.exception = exc	10✔
56	operation.written = 0	10✔
57	operation.callback(None)	10✔
58
59	def on_success(result):	10✔
60	self._in_progress -= 1	10✔
61	operation.written = result	10✔
62	operation.callback(result)	10✔
63
64	if self._in_progress > self.__max_requests:	10✔
65	raise RuntimeError(	10✔
66	"Maximum simultaneous requests have been reached",
67	)
68
69	self._in_progress += 1	10✔
70
71	self.pool.apply_async(	10✔
72	handler, args=(self, operation),
73	callback=on_success,
74	error_callback=on_error,
75	)
76
77	if NATIVE_PREAD_PWRITE:	10✔
78	def __pread(self, fd, size, offset):	10✔
79	return os.pread(fd, size, offset)	10✔
80
81	def __pwrite(self, fd, bytes, offset):	10✔
82	return os.pwrite(fd, bytes, offset)	10✔
83	else:
84	def __pread(self, fd, size, offset):	×
85	with self._locks[fd]:	×
86	os.lseek(fd, 0, os.SEEK_SET)	×
87	os.lseek(fd, offset, os.SEEK_SET)	×
88	return os.read(fd, size)	×
89
90	def __pwrite(self, fd, bytes, offset):	×
91	with self._locks[fd]:	×
92	os.lseek(fd, 0, os.SEEK_SET)	×
93	os.lseek(fd, offset, os.SEEK_SET)	×
94	return os.write(fd, bytes)	×
95
96	def _handle_read(self, operation: "Operation"):	10✔
97	return operation.buffer.write(	10✔
98	self.__pread(
99	operation.fileno,
100	operation.nbytes,
101	operation.offset,
102	),
103	)
104
105	def _handle_write(self, operation: "Operation"):	10✔
106	return self.__pwrite(	10✔
107	operation.fileno, operation.buffer.getvalue(), operation.offset,
108	)
109
110	def _handle_fsync(self, operation: "Operation"):	10✔
111	return os.fsync(operation.fileno)	10✔
112
113	def _handle_fdsync(self, operation: "Operation"):	10✔
114	return fdsync(operation.fileno)	10✔
115
116	def _handle_noop(self, operation: "Operation"):	10✔
117	return	×
118
119	def submit(self, *aio_operations) -> int:	10✔
120	operations = []	10✔
121
122	for operation in aio_operations:	10✔
123	if not isinstance(operation, Operation):	10✔
NEW 124	raise ValueError(f"Invalid Operation {operation!r}") # noqa: TRY004 (public API, keep ValueError)	×
125
126	operations.append(operation)	10✔
127
128	count = 0	10✔
129	for operation in operations:	10✔
130	self._execute(operation)	10✔
131	count += 1	10✔
132
133	return count	10✔
134
135	def cancel(self, *aio_operations) -> int:	10✔
136	"""
137	Cancels multiple Operations. Returns
138
139	Operation.cancel(aio_op1, aio_op2, aio_opN, ...) -> int
140
141	(Always returns zero, this method exists for compatibility reasons)
142	"""
143	return 0	10✔
144
145	def close(self):	10✔
146	if self._closed:	10✔
147	return	×
148
149	with self._closed_lock:	10✔
150	self.pool.close()	10✔
151	self._closed = True	10✔
152
153	def __del__(self):	10✔
154	if self.pool.close():	10✔
155	self.close()	×
156
157	_OP_MAP = MappingProxyType({	10✔
158	OpCode.READ: _handle_read,
159	OpCode.WRITE: _handle_write,
160	OpCode.FSYNC: _handle_fsync,
161	OpCode.FDSYNC: _handle_fdsync,
162	OpCode.NOOP: _handle_noop,
163	})
164
165
166	# noinspection PyPropertyDefinition
167	class Operation(AbstractOperation):	10✔
168	"""
169	python aio operation implementation
170	"""
171	def __init__(	10✔
172	self,
173	fd: int,
174	nbytes: int \| None,
175	offset: int \| None,
176	opcode: OpCode,
177	payload: bytes \| None = None,
178	priority: int \| None = None,
179	):
180	self.callback: Callable[[int], Any] \| None = None	10✔
181	self.buffer = BytesIO()	10✔
182
183	if opcode == OpCode.WRITE and payload:	10✔
184	self.buffer = BytesIO(payload)	10✔
185
186	self.opcode = opcode	10✔
187	self.__fileno = fd	10✔
188	self.__offset = offset or 0	10✔
189	self.__opcode = opcode	10✔
190	self.__nbytes = nbytes or 0	10✔
191	self.__priority = priority or 0	10✔
192	self.exception = None	10✔
193	self.written = 0	10✔
194
195	@classmethod	10✔
196	def read(	10✔
197	cls, nbytes: int, fd: int, offset: int, priority=0,
198	) -> "Operation":
199	"""
200	Creates a new instance of Operation on read mode.
201	"""
202	return cls(fd, nbytes, offset, opcode=OpCode.READ, priority=priority)	10✔
203
204	@classmethod	10✔
205	def write(	10✔
206	cls, payload_bytes: bytes, fd: int, offset: int, priority=0,
207	) -> "Operation":
208	"""
209	Creates a new instance of AIOOperation on write mode.
210	"""
211	return cls(	10✔
212	fd,
213	len(payload_bytes),
214	offset,
215	payload=payload_bytes,
216	opcode=OpCode.WRITE,
217	priority=priority,
218	)
219
220	@classmethod	10✔
221	def fsync(cls, fd: int, priority=0) -> "Operation":	10✔
222
223	"""
224	Creates a new instance of AIOOperation on fsync mode.
225	"""
226	return cls(fd, None, None, opcode=OpCode.FSYNC, priority=priority)	10✔
227
228	@classmethod	10✔
229	def fdsync(cls, fd: int, priority=0) -> "Operation":	10✔
230
231	"""
232	Creates a new instance of AIOOperation on fdsync mode.
233	"""
234	return cls(fd, None, None, opcode=OpCode.FDSYNC, priority=priority)	10✔
235
236	def get_value(self) -> bytes \| int:	10✔
237	"""
238	Method returns a bytes value of AIOOperation's result or None.
239	"""
240	if self.exception:	10✔
241	raise self.exception	10✔
242
243	if self.opcode == OpCode.WRITE:	10✔
244	return self.written	10✔
245
246	if self.buffer is None:	10✔
247	return	×
248
249	return self.buffer.getvalue()	10✔
250
251	@property	10✔
252	def fileno(self) -> int:	10✔
253	return self.__fileno	10✔
254
255	@property	10✔
256	def offset(self) -> int:	10✔
257	return self.__offset	10✔
258
259	@property	10✔
260	def payload(self) -> memoryview \| None:	10✔
261	return self.buffer.getbuffer()	10✔
262
263	@property	10✔
264	def nbytes(self) -> int:	10✔
265	return self.__nbytes	10✔
266
267	def set_callback(self, callback: Callable[[int], Any]) -> bool:	10✔
268	self.callback = callback	10✔
269	return True	10✔

mosquito / caio / 30713547314

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous