30897283339

Committed 04 Aug 2026 09:39AM UTC coverage: 92.657% (+0.4%) from 92.291%

Build # 30897283339

Build Type

Pull #72

github

Committed by

web-flow

Commit Message

Merge 3b92a60eb into 6d03e42c8

Pull Request Pull Request #72: Make all four backends safe under free-threaded CPython

Coverage Stats

10 of 20 new or added lines in 1 file covered. (50.0%)

86 existing lines in 5 files now uncovered.

429 of 463 relevant lines covered (92.66%)

12.61 hits per line

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

98.56

/caio/python_aio.py

import operator
import os
import sys
import threading
from collections.abc import Callable
from enum import IntEnum, unique
from multiprocessing.pool import ThreadPool
from threading import Lock, RLock
from types import MappingProxyType
from typing import Any
from weakref import WeakValueDictionary

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


@unique
class ContextState(IntEnum):
    OPEN = 0
    CLOSING = 1
    CLOSED = 2


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

    MAX_POOL_SIZE = 128

    def __init__(self, max_requests: int = 32, pool_size: int = 8):
        # Set before any validation that can raise: __del__ runs even on a
        # partially-constructed object, and close() below relies on _state
        # existing and being non-OPEN to safely no-op before ever touching
        # _lock/pool, which aren't created until validation passes.
        self._state = ContextState.CLOSED

        if not (0 < pool_size < self.MAX_POOL_SIZE):
            raise ValueError(
                f"pool_size must be between 1 and {self.MAX_POOL_SIZE - 1}, "
                f"got {pool_size}",
            )
        if max_requests <= 0:
            raise ValueError(
                f"max_requests must be a positive integer, got {max_requests}",
            )

        self.__max_requests = max_requests
        self.pool = ThreadPool(pool_size)
        self._in_progress = 0
        self._lock = Lock()
        self._state = ContextState.OPEN

        if not NATIVE_PREAD_PWRITE:
            self._locks_cleaner = Lock()
            self._locks: WeakValueDictionary = WeakValueDictionary()

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

    @staticmethod
    def _invoke_callback(operation: "Operation", value):
        """
        Calls ``operation``'s callback with ``value``, if one was set via
        ``set_callback()``. A missing callback is a normal, expected case.
        A raising callback must not escape - this runs on ThreadPool's
        single internal result-handler thread, and an uncaught exception
        there kills that thread, silently stalling every future
        result/callback for the rest of this Context's lifetime.
        """
        with operation._lock:
            callback = operation.callback
        if callback is None:
            return

        try:
            callback(value)
        except BaseException:  # noqa: BLE001 (must isolate any callback exception, including SystemExit/KeyboardInterrupt)
            threading.excepthook(
                threading.ExceptHookArgs(
                    (*sys.exc_info(), threading.current_thread()),
                ),
            )

    def _release_slot(self):
        with self._lock:
            self._in_progress -= 1

    def _rollback_claim(self, operation: "Operation"):
        """
        Undoes a claim that was never actually scheduled (e.g. a
        concurrent close() tore down the pool between the capacity
        reservation and apply_async()) - unlike genuine completion, this
        must reset operation.in_progress, since the operation never
        actually ran and must stay retryable.
        """
        with operation._lock, self._lock:
            operation.in_progress = False
            self._in_progress -= 1

    def _execute(self, operation: "Operation") -> bool:
        """
        Returns True if actually scheduled, False if skipped because
        ``operation`` was already in progress - matches the other three
        backends, which all silently skip (rather than raise on, or
        dispatch twice) a resubmit of an Operation still in flight.
        """
        handler = self._OP_MAP[operation.opcode]

        # operation.in_progress is deliberately NOT reset on genuine
        # completion below - one-shot forever once actually scheduled,
        # matching all three native backends: a completed Operation must
        # not be resubmittable, only a fresh one constructed for a retry.
        def on_error(exc):
            self._release_slot()
            operation.exception = exc
            operation.written = 0
            self._invoke_callback(operation, None)

        def on_success(result):
            self._release_slot()
            operation.written = result
            self._invoke_callback(operation, result)

        # operation.in_progress is the Operation's own lock, not this
        # Context's - two different Contexts submitting the same Operation
        # only ever share the Operation, never a Context, so a per-Context
        # lock alone can't stop them both from claiming it.
        with operation._lock, self._lock:
            if operation.in_progress:
                return False

            if self._state != ContextState.OPEN:
                raise RuntimeError("Context is closed")

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

            operation.in_progress = True
            self._in_progress += 1

        try:
            self.pool.apply_async(
                handler, args=(self, operation),
                callback=on_success,
                error_callback=on_error,
            )
        except BaseException:
            # Scheduling itself failed (e.g. a concurrent close() already
            # tore down the pool) - the slot reserved above was never
            # actually claimed by a real job, so it must be given back
            # instead of permanently inflating _in_progress.
            self._rollback_claim(operation)
            raise

        return True

    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 __fd_lock(self, fd):
            # Plain get-or-create under _locks_cleaner - two threads racing
            # on the same fd's first access must get back the same RLock,
            # or their lseek()+read()/write() pairs can interleave. The
            # WeakValueDictionary itself only keeps an fd's entry alive
            # while some __pread()/__pwrite() call still holds a strong ref
            # to it (the `with` block below) - otherwise this Context would
            # accumulate one RLock per fd it's ever touched, forever.
            with self._locks_cleaner:
                lock = self._locks.get(fd)
                if lock is None:
                    lock = self._locks[fd] = RLock()
                return lock

        def __pread(self, fd, size, offset):
            with self.__fd_lock(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.__fd_lock(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"):
        # Stored directly, not copied through a BytesIO - pread() already
        # returns exactly the bytes object get_value()/payload need to hand
        # back, and buffering it through BytesIO.write() just to unwrap it
        # again later cost a full extra copy per read for no benefit.
        data = self.__pread(
            operation.fileno, operation.nbytes, operation.offset,
        )
        operation.buffer = data
        return len(data)

    def _handle_write(self, operation: "Operation"):
        # operation.buffer is the caller's own payload bytes, unwrapped -
        # no BytesIO round-trip needed to hand it to pwrite() either.
        return self.__pwrite(
            operation.fileno, operation.buffer, 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:
        for operation in aio_operations:
            if not isinstance(operation, Operation):
                raise ValueError(f"Invalid Operation {operation!r}")  # noqa: TRY004 (pre-existing public exception type, not changing it here)

        count = 0
        for operation in aio_operations:
            if 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._state != ContextState.OPEN:
            return
        with self._lock:
            if self._state != ContextState.OPEN:
                return
            self._state = ContextState.CLOSING
            self.pool.close()
            self._state = ContextState.CLOSED

    def __del__(self):
        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,
    ):
        # Validated eagerly, at construction time - matching the other 3
        # backends, which reject a non-int-like fd/nbytes/offset/priority
        # (or a non-bytes write payload) synchronously via
        # PyArg_ParseTupleAndKeywords/PyBytes_Check, rather than storing it
        # and failing later inside a worker thread. operator.index() is the
        # same __index__-based coercion PyArg_ParseTupleAndKeywords' "I"/"K"
        # format codes use internally, so this accepts exactly what the C
        # constructors accept (plain ints, numpy-style int-likes, ...) and
        # rejects exactly what they reject (str, float, ...).
        fd = operator.index(fd)
        if nbytes is not None:
            nbytes = operator.index(nbytes)
        if offset is not None:
            offset = operator.index(offset)
        if priority is not None:
            priority = operator.index(priority)

        # Plain bytes, not a BytesIO wrapper - for a write this is the
        # caller's own payload, handed to pwrite() as-is; for a read it
        # starts empty and _handle_read() replaces it with pread()'s
        # result directly. Either way there's nothing to unwrap later, so
        # get_value()/payload return it with no extra copy - and it's
        # never None, so callers don't need to check.
        if opcode == OpCode.WRITE:
            if not isinstance(payload, bytes):
                raise ValueError(f"payload_bytes must be bytes, got {payload!r}")
            buffer = payload
        else:
            buffer = b""

        self.callback: Callable[[int], Any] | None = None
        self.in_progress = False
        self._lock = Lock()
        self.buffer: bytes = buffer

        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 | None:
        """
        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.opcode in (OpCode.FSYNC, OpCode.FDSYNC):
            return None

        return self.buffer

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

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

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

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

    def set_callback(self, callback: Callable[[int], Any]) -> bool:
        if not callable(callback):
            raise ValueError(f"callback must be callable, got {callback!r}")  # noqa: TRY004 (pre-existing public exception type, not changing it here)
        with self._lock:
            self.callback = callback
        return True

1	import operator	16✔
2	import os	16✔
3	import sys	16✔
4	import threading	16✔
5	from collections.abc import Callable	16✔
6	from enum import IntEnum, unique	16✔
7	from multiprocessing.pool import ThreadPool	16✔
8	from threading import Lock, RLock	16✔
9	from types import MappingProxyType	16✔
10	from typing import Any	16✔
11	from weakref import WeakValueDictionary	16✔
12
13	from .abstract import AbstractContext, AbstractOperation	16✔
14
15	fdsync = getattr(os, "fdatasync", os.fsync)	16✔
16	NATIVE_PREAD_PWRITE = hasattr(os, "pread") and hasattr(os, "pwrite")	16✔
17
18
19	@unique	16✔
20	class OpCode(IntEnum):	16✔
21	READ = 0	16✔
22	WRITE = 1	16✔
23	FSYNC = 2	16✔
24	FDSYNC = 3	16✔
25	NOOP = -1	16✔
26
27
28	@unique	16✔
29	class ContextState(IntEnum):	16✔
30	OPEN = 0	16✔
31	CLOSING = 1	16✔
32	CLOSED = 2	16✔
33
34
35	class Context(AbstractContext):	16✔
36	"""
37	python aio context implementation
38	"""
39
40	MAX_POOL_SIZE = 128	16✔
41
42	def __init__(self, max_requests: int = 32, pool_size: int = 8):	16✔
43	# Set before any validation that can raise: __del__ runs even on a
44	# partially-constructed object, and close() below relies on _state
45	# existing and being non-OPEN to safely no-op before ever touching
46	# _lock/pool, which aren't created until validation passes.
47	self._state = ContextState.CLOSED	16✔
48
49	if not (0 < pool_size < self.MAX_POOL_SIZE):	16✔
50	raise ValueError(	16✔
51	f"pool_size must be between 1 and {self.MAX_POOL_SIZE - 1}, "
52	f"got {pool_size}",
53	)
54	if max_requests <= 0:	16✔
55	raise ValueError(	16✔
56	f"max_requests must be a positive integer, got {max_requests}",
57	)
58
59	self.__max_requests = max_requests	16✔
60	self.pool = ThreadPool(pool_size)	16✔
61	self._in_progress = 0	16✔
62	self._lock = Lock()	16✔
63	self._state = ContextState.OPEN	16✔
64
65	if not NATIVE_PREAD_PWRITE:	16✔
NEW 66	self._locks_cleaner = Lock()	4✔
NEW 67	self._locks: WeakValueDictionary = WeakValueDictionary()	4✔
68
69	@property	16✔
70	def max_requests(self) -> int:	16✔
71	return self.__max_requests	×
72
73	@staticmethod	16✔
74	def _invoke_callback(operation: "Operation", value):	16✔
75	"""
76	Calls ``operation``'s callback with ``value``, if one was set via
77	``set_callback()``. A missing callback is a normal, expected case.
78	A raising callback must not escape - this runs on ThreadPool's
79	single internal result-handler thread, and an uncaught exception
80	there kills that thread, silently stalling every future
81	result/callback for the rest of this Context's lifetime.
82	"""
83	with operation._lock:	16✔
84	callback = operation.callback	16✔
85	if callback is None:	16✔
86	return	16✔
87
88	try:	16✔
89	callback(value)	16✔
90	except BaseException: # noqa: BLE001 (must isolate any callback exception, including SystemExit/KeyboardInterrupt)	16✔
91	threading.excepthook(	16✔
92	threading.ExceptHookArgs(
93	(*sys.exc_info(), threading.current_thread()),
94	),
95	)
96
97	def _release_slot(self):	16✔
98	with self._lock:	16✔
99	self._in_progress -= 1	16✔
100
101	def _rollback_claim(self, operation: "Operation"):	16✔
102	"""
103	Undoes a claim that was never actually scheduled (e.g. a
104	concurrent close() tore down the pool between the capacity
105	reservation and apply_async()) - unlike genuine completion, this
106	must reset operation.in_progress, since the operation never
107	actually ran and must stay retryable.
108	"""
109	with operation._lock, self._lock:	16✔
110	operation.in_progress = False	16✔
111	self._in_progress -= 1	16✔
112
113	def _execute(self, operation: "Operation") -> bool:	16✔
114	"""
115	Returns True if actually scheduled, False if skipped because
116	``operation`` was already in progress - matches the other three
117	backends, which all silently skip (rather than raise on, or
118	dispatch twice) a resubmit of an Operation still in flight.
119	"""
120	handler = self._OP_MAP[operation.opcode]	16✔
121
122	# operation.in_progress is deliberately NOT reset on genuine
123	# completion below - one-shot forever once actually scheduled,
124	# matching all three native backends: a completed Operation must
125	# not be resubmittable, only a fresh one constructed for a retry.
126	def on_error(exc):	16✔
127	self._release_slot()	16✔
128	operation.exception = exc	16✔
129	operation.written = 0	16✔
130	self._invoke_callback(operation, None)	16✔
131
132	def on_success(result):	16✔
133	self._release_slot()	16✔
134	operation.written = result	16✔
135	self._invoke_callback(operation, result)	16✔
136
137	# operation.in_progress is the Operation's own lock, not this
138	# Context's - two different Contexts submitting the same Operation
139	# only ever share the Operation, never a Context, so a per-Context
140	# lock alone can't stop them both from claiming it.
141	with operation._lock, self._lock:	16✔
142	if operation.in_progress:	16✔
143	return False	16✔
144
145	if self._state != ContextState.OPEN:	16✔
146	raise RuntimeError("Context is closed")	16✔
147
148	if self._in_progress >= self.__max_requests:	16✔
149	raise RuntimeError(	16✔
150	"Maximum simultaneous requests have been reached",
151	)
152
153	operation.in_progress = True	16✔
154	self._in_progress += 1	16✔
155
156	try:	16✔
157	self.pool.apply_async(	16✔
158	handler, args=(self, operation),
159	callback=on_success,
160	error_callback=on_error,
161	)
162	except BaseException:	16✔
163	# Scheduling itself failed (e.g. a concurrent close() already
164	# tore down the pool) - the slot reserved above was never
165	# actually claimed by a real job, so it must be given back
166	# instead of permanently inflating _in_progress.
167	self._rollback_claim(operation)	16✔
168	raise	16✔
169
170	return True	16✔
171
172	if NATIVE_PREAD_PWRITE:	16✔
173	def __pread(self, fd, size, offset):	12✔
174	return os.pread(fd, size, offset)	12✔
175
176	def __pwrite(self, fd, bytes, offset):	12✔
177	return os.pwrite(fd, bytes, offset)	12✔
178	else:
NEW 179	def __fd_lock(self, fd):	4✔
180	# Plain get-or-create under _locks_cleaner - two threads racing
181	# on the same fd's first access must get back the same RLock,
182	# or their lseek()+read()/write() pairs can interleave. The
183	# WeakValueDictionary itself only keeps an fd's entry alive
184	# while some __pread()/__pwrite() call still holds a strong ref
185	# to it (the `with` block below) - otherwise this Context would
186	# accumulate one RLock per fd it's ever touched, forever.
NEW 187	with self._locks_cleaner:	4✔
NEW 188	lock = self._locks.get(fd)	4✔
NEW 189	if lock is None:	4✔
NEW 190	lock = self._locks[fd] = RLock()	4✔
NEW 191	return lock	4✔
192
UNCOV 193	def __pread(self, fd, size, offset):	4✔
NEW 194	with self.__fd_lock(fd):	4✔
UNCOV 195	os.lseek(fd, 0, os.SEEK_SET)	4✔
UNCOV 196	os.lseek(fd, offset, os.SEEK_SET)	4✔
UNCOV 197	return os.read(fd, size)	4✔
198
UNCOV 199	def __pwrite(self, fd, bytes, offset):	4✔
NEW 200	with self.__fd_lock(fd):	4✔
UNCOV 201	os.lseek(fd, 0, os.SEEK_SET)	4✔
UNCOV 202	os.lseek(fd, offset, os.SEEK_SET)	4✔
UNCOV 203	return os.write(fd, bytes)	4✔
204
205	def _handle_read(self, operation: "Operation"):	16✔
206	# Stored directly, not copied through a BytesIO - pread() already
207	# returns exactly the bytes object get_value()/payload need to hand
208	# back, and buffering it through BytesIO.write() just to unwrap it
209	# again later cost a full extra copy per read for no benefit.
210	data = self.__pread(	16✔
211	operation.fileno, operation.nbytes, operation.offset,
212	)
213	operation.buffer = data	16✔
214	return len(data)	16✔
215
216	def _handle_write(self, operation: "Operation"):	16✔
217	# operation.buffer is the caller's own payload bytes, unwrapped -
218	# no BytesIO round-trip needed to hand it to pwrite() either.
219	return self.__pwrite(	16✔
220	operation.fileno, operation.buffer, operation.offset,
221	)
222
223	def _handle_fsync(self, operation: "Operation"):	16✔
224	return os.fsync(operation.fileno)	16✔
225
226	def _handle_fdsync(self, operation: "Operation"):	16✔
227	return fdsync(operation.fileno)	16✔
228
229	def _handle_noop(self, operation: "Operation"):	16✔
230	return	2✔
231
232	def submit(self, *aio_operations) -> int:	16✔
233	for operation in aio_operations:	16✔
234	if not isinstance(operation, Operation):	16✔
235	raise ValueError(f"Invalid Operation {operation!r}") # noqa: TRY004 (pre-existing public exception type, not changing it here)	×
236
237	count = 0	16✔
238	for operation in aio_operations:	16✔
239	if self._execute(operation):	16✔
240	count += 1	16✔
241
242	return count	16✔
243
244	def cancel(self, *aio_operations) -> int:	16✔
245	"""
246	Cancels multiple Operations. Returns
247
248	Operation.cancel(aio_op1, aio_op2, aio_opN, ...) -> int
249
250	(Always returns zero, this method exists for compatibility reasons)
251	"""
252	return 0	16✔
253
254	def close(self):	16✔
255	if self._state != ContextState.OPEN:	16✔
256	return	16✔
257	with self._lock:	16✔
258	if self._state != ContextState.OPEN:	16✔
259	return	×
260	self._state = ContextState.CLOSING	16✔
261	self.pool.close()	16✔
262	self._state = ContextState.CLOSED	16✔
263
264	def __del__(self):	16✔
265	self.close()	16✔
266
267	_OP_MAP = MappingProxyType({	16✔
268	OpCode.READ: _handle_read,
269	OpCode.WRITE: _handle_write,
270	OpCode.FSYNC: _handle_fsync,
271	OpCode.FDSYNC: _handle_fdsync,
272	OpCode.NOOP: _handle_noop,
273	})
274
275
276	# noinspection PyPropertyDefinition
277	class Operation(AbstractOperation):	16✔
278	"""
279	python aio operation implementation
280	"""
281	def __init__(	16✔
282	self,
283	fd: int,
284	nbytes: int \| None,
285	offset: int \| None,
286	opcode: OpCode,
287	payload: bytes \| None = None,
288	priority: int \| None = None,
289	):
290	# Validated eagerly, at construction time - matching the other 3
291	# backends, which reject a non-int-like fd/nbytes/offset/priority
292	# (or a non-bytes write payload) synchronously via
293	# PyArg_ParseTupleAndKeywords/PyBytes_Check, rather than storing it
294	# and failing later inside a worker thread. operator.index() is the
295	# same __index__-based coercion PyArg_ParseTupleAndKeywords' "I"/"K"
296	# format codes use internally, so this accepts exactly what the C
297	# constructors accept (plain ints, numpy-style int-likes, ...) and
298	# rejects exactly what they reject (str, float, ...).
299	fd = operator.index(fd)	16✔
300	if nbytes is not None:	16✔
301	nbytes = operator.index(nbytes)	16✔
302	if offset is not None:	16✔
303	offset = operator.index(offset)	16✔
304	if priority is not None:	16✔
305	priority = operator.index(priority)	16✔
306
307	# Plain bytes, not a BytesIO wrapper - for a write this is the
308	# caller's own payload, handed to pwrite() as-is; for a read it
309	# starts empty and _handle_read() replaces it with pread()'s
310	# result directly. Either way there's nothing to unwrap later, so
311	# get_value()/payload return it with no extra copy - and it's
312	# never None, so callers don't need to check.
313	if opcode == OpCode.WRITE:	16✔
314	if not isinstance(payload, bytes):	16✔
315	raise ValueError(f"payload_bytes must be bytes, got {payload!r}")	16✔
316	buffer = payload	16✔
317	else:
318	buffer = b""	16✔
319
320	self.callback: Callable[[int], Any] \| None = None	16✔
321	self.in_progress = False	16✔
322	self._lock = Lock()	16✔
323	self.buffer: bytes = buffer	16✔
324
325	self.opcode = opcode	16✔
326	self.__fileno = fd	16✔
327	self.__offset = offset or 0	16✔
328	self.__opcode = opcode	16✔
329	self.__nbytes = nbytes or 0	16✔
330	self.__priority = priority or 0	16✔
331	self.exception = None	16✔
332	self.written = 0	16✔
333
334	@classmethod	16✔
335	def read(	16✔
336	cls, nbytes: int, fd: int, offset: int, priority=0,
337	) -> "Operation":
338	"""
339	Creates a new instance of Operation on read mode.
340	"""
341	return cls(fd, nbytes, offset, opcode=OpCode.READ, priority=priority)	16✔
342
343	@classmethod	16✔
344	def write(	16✔
345	cls, payload_bytes: bytes, fd: int, offset: int, priority=0,
346	) -> "Operation":
347	"""
348	Creates a new instance of AIOOperation on write mode.
349	"""
350	return cls(	16✔
351	fd,
352	len(payload_bytes),
353	offset,
354	payload=payload_bytes,
355	opcode=OpCode.WRITE,
356	priority=priority,
357	)
358
359	@classmethod	16✔
360	def fsync(cls, fd: int, priority=0) -> "Operation":	16✔
361
362	"""
363	Creates a new instance of AIOOperation on fsync mode.
364	"""
365	return cls(fd, None, None, opcode=OpCode.FSYNC, priority=priority)	16✔
366
367	@classmethod	16✔
368	def fdsync(cls, fd: int, priority=0) -> "Operation":	16✔
369
370	"""
371	Creates a new instance of AIOOperation on fdsync mode.
372	"""
373	return cls(fd, None, None, opcode=OpCode.FDSYNC, priority=priority)	16✔
374
375	def get_value(self) -> bytes \| int \| None:	16✔
376	"""
377	Method returns a bytes value of AIOOperation's result or None.
378	"""
379	if self.exception:	16✔
380	raise self.exception	16✔
381
382	if self.opcode == OpCode.WRITE:	16✔
383	return self.written	16✔
384
385	if self.opcode in (OpCode.FSYNC, OpCode.FDSYNC):	16✔
386	return None	16✔
387
388	return self.buffer	16✔
389
390	@property	16✔
391	def fileno(self) -> int:	16✔
392	return self.__fileno	16✔
393
394	@property	16✔
395	def offset(self) -> int:	16✔
396	return self.__offset	16✔
397
398	@property	16✔
399	def payload(self) -> memoryview \| None:	16✔
400	return memoryview(self.buffer)	16✔
401
402	@property	16✔
403	def nbytes(self) -> int:	16✔
404	return self.__nbytes	16✔
405
406	def set_callback(self, callback: Callable[[int], Any]) -> bool:	16✔
407	if not callable(callback):	16✔
408	raise ValueError(f"callback must be callable, got {callback!r}") # noqa: TRY004 (pre-existing public exception type, not changing it here)	16✔
409	with self._lock:	16✔
410	self.callback = callback	16✔
411	return True	16✔

mosquito / caio / 30897283339

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