30894786869

Committed 04 Aug 2026 09:05AM UTC coverage: 89.278% (-3.0%) from 92.291%

Build # 30894786869

Build Type

Pull #72

github

Committed by

web-flow

Commit Message

Merge 6f4875622 into 6d03e42c8

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

Coverage Stats

9 of 21 new or added lines in 1 file covered. (42.86%)

90 existing lines in 5 files now uncovered.

408 of 457 relevant lines covered (89.28%)

9.77 hits per line

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

92.57

/caio/python_aio.py

import operator
import os
import sys
import threading
from collections import defaultdict
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 .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 = RLock()       # type: ignore
            self._locks = defaultdict(RLock)    # type: ignore

    @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 __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"):
        # 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	12✔
2	import os	12✔
3	import sys	12✔
4	import threading	12✔
5	from collections import defaultdict	12✔
6	from collections.abc import Callable	12✔
7	from enum import IntEnum, unique	12✔
8	from multiprocessing.pool import ThreadPool	12✔
9	from threading import Lock, RLock	12✔
10	from types import MappingProxyType	12✔
11	from typing import Any	12✔
12
13	from .abstract import AbstractContext, AbstractOperation	12✔
14
15	fdsync = getattr(os, "fdatasync", os.fsync)	12✔
16	NATIVE_PREAD_PWRITE = hasattr(os, "pread") and hasattr(os, "pwrite")	12✔
17
18
19	@unique	12✔
20	class OpCode(IntEnum):	12✔
21	READ = 0	12✔
22	WRITE = 1	12✔
23	FSYNC = 2	12✔
24	FDSYNC = 3	12✔
25	NOOP = -1	12✔
26
27
28	@unique	12✔
29	class ContextState(IntEnum):	12✔
30	OPEN = 0	12✔
31	CLOSING = 1	12✔
32	CLOSED = 2	12✔
33
34
35	class Context(AbstractContext):	12✔
36	"""
37	python aio context implementation
38	"""
39
40	MAX_POOL_SIZE = 128	12✔
41
42	def __init__(self, max_requests: int = 32, pool_size: int = 8):	12✔
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	12✔
48
49	if not (0 < pool_size < self.MAX_POOL_SIZE):	12✔
50	raise ValueError(	12✔
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:	12✔
55	raise ValueError(	12✔
56	f"max_requests must be a positive integer, got {max_requests}",
57	)
58
59	self.__max_requests = max_requests	12✔
60	self.pool = ThreadPool(pool_size)	12✔
61	self._in_progress = 0	12✔
62	self._lock = Lock()	12✔
63	self._state = ContextState.OPEN	12✔
64
65	if not NATIVE_PREAD_PWRITE:	12✔
NEW UNCOV 66	self._locks_cleaner = RLock() # type: ignore	×
NEW UNCOV 67	self._locks = defaultdict(RLock) # type: ignore	×
68
69	@property	12✔
70	def max_requests(self) -> int:	12✔
71	return self.__max_requests	×
72
73	@staticmethod	12✔
74	def _invoke_callback(operation: "Operation", value):	12✔
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:	12✔
84	callback = operation.callback	12✔
85	if callback is None:	12✔
86	return	12✔
87
88	try:	12✔
89	callback(value)	12✔
90	except BaseException: # noqa: BLE001 (must isolate any callback exception, including SystemExit/KeyboardInterrupt)	12✔
91	threading.excepthook(	12✔
92	threading.ExceptHookArgs(
93	(*sys.exc_info(), threading.current_thread()),
94	),
95	)
96
97	def _release_slot(self):	12✔
98	with self._lock:	12✔
99	self._in_progress -= 1	12✔
100
101	def _rollback_claim(self, operation: "Operation"):	12✔
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:	12✔
110	operation.in_progress = False	12✔
111	self._in_progress -= 1	12✔
112
113	def _execute(self, operation: "Operation") -> bool:	12✔
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]	12✔
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):	12✔
127	self._release_slot()	12✔
128	operation.exception = exc	12✔
129	operation.written = 0	12✔
130	self._invoke_callback(operation, None)	12✔
131
132	def on_success(result):	12✔
133	self._release_slot()	12✔
134	operation.written = result	12✔
135	self._invoke_callback(operation, result)	12✔
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:	12✔
142	if operation.in_progress:	12✔
143	return False	12✔
144
145	if self._state != ContextState.OPEN:	12✔
146	raise RuntimeError("Context is closed")	12✔
147
148	if self._in_progress >= self.__max_requests:	12✔
149	raise RuntimeError(	12✔
150	"Maximum simultaneous requests have been reached",
151	)
152
153	operation.in_progress = True	12✔
154	self._in_progress += 1	12✔
155
156	try:	12✔
157	self.pool.apply_async(	12✔
158	handler, args=(self, operation),
159	callback=on_success,
160	error_callback=on_error,
161	)
162	except BaseException:	12✔
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)	12✔
168	raise	12✔
169
170	return True	12✔
171
172	if NATIVE_PREAD_PWRITE:	12✔
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 UNCOV 179	def __pread(self, fd, size, offset):	×
NEW UNCOV 180	with self._locks[fd]:	×
NEW UNCOV 181	os.lseek(fd, 0, os.SEEK_SET)	×
NEW UNCOV 182	os.lseek(fd, offset, os.SEEK_SET)	×
NEW UNCOV 183	return os.read(fd, size)	×
184
NEW UNCOV 185	def __pwrite(self, fd, bytes, offset):	×
NEW UNCOV 186	with self._locks[fd]:	×
NEW UNCOV 187	os.lseek(fd, 0, os.SEEK_SET)	×
NEW UNCOV 188	os.lseek(fd, offset, os.SEEK_SET)	×
NEW UNCOV 189	return os.write(fd, bytes)	×
190
191	def _handle_read(self, operation: "Operation"):	12✔
192	# Stored directly, not copied through a BytesIO - pread() already
193	# returns exactly the bytes object get_value()/payload need to hand
194	# back, and buffering it through BytesIO.write() just to unwrap it
195	# again later cost a full extra copy per read for no benefit.
196	data = self.__pread(	12✔
197	operation.fileno, operation.nbytes, operation.offset,
198	)
199	operation.buffer = data	12✔
200	return len(data)	12✔
201
202	def _handle_write(self, operation: "Operation"):	12✔
203	# operation.buffer is the caller's own payload bytes, unwrapped -
204	# no BytesIO round-trip needed to hand it to pwrite() either.
205	return self.__pwrite(	12✔
206	operation.fileno, operation.buffer, operation.offset,
207	)
208
209	def _handle_fsync(self, operation: "Operation"):	12✔
210	return os.fsync(operation.fileno)	12✔
211
212	def _handle_fdsync(self, operation: "Operation"):	12✔
213	return fdsync(operation.fileno)	12✔
214
215	def _handle_noop(self, operation: "Operation"):	12✔
216	return	2✔
217
218	def submit(self, *aio_operations) -> int:	12✔
219	for operation in aio_operations:	12✔
220	if not isinstance(operation, Operation):	12✔
221	raise ValueError(f"Invalid Operation {operation!r}") # noqa: TRY004 (pre-existing public exception type, not changing it here)	×
222
223	count = 0	12✔
224	for operation in aio_operations:	12✔
225	if self._execute(operation):	12✔
226	count += 1	12✔
227
228	return count	12✔
229
230	def cancel(self, *aio_operations) -> int:	12✔
231	"""
232	Cancels multiple Operations. Returns
233
234	Operation.cancel(aio_op1, aio_op2, aio_opN, ...) -> int
235
236	(Always returns zero, this method exists for compatibility reasons)
237	"""
238	return 0	12✔
239
240	def close(self):	12✔
241	if self._state != ContextState.OPEN:	12✔
242	return	12✔
243	with self._lock:	12✔
244	if self._state != ContextState.OPEN:	12✔
245	return	×
246	self._state = ContextState.CLOSING	12✔
247	self.pool.close()	12✔
248	self._state = ContextState.CLOSED	12✔
249
250	def __del__(self):	12✔
251	self.close()	12✔
252
253	_OP_MAP = MappingProxyType({	12✔
254	OpCode.READ: _handle_read,
255	OpCode.WRITE: _handle_write,
256	OpCode.FSYNC: _handle_fsync,
257	OpCode.FDSYNC: _handle_fdsync,
258	OpCode.NOOP: _handle_noop,
259	})
260
261
262	# noinspection PyPropertyDefinition
263	class Operation(AbstractOperation):	12✔
264	"""
265	python aio operation implementation
266	"""
267	def __init__(	12✔
268	self,
269	fd: int,
270	nbytes: int \| None,
271	offset: int \| None,
272	opcode: OpCode,
273	payload: bytes \| None = None,
274	priority: int \| None = None,
275	):
276	# Validated eagerly, at construction time - matching the other 3
277	# backends, which reject a non-int-like fd/nbytes/offset/priority
278	# (or a non-bytes write payload) synchronously via
279	# PyArg_ParseTupleAndKeywords/PyBytes_Check, rather than storing it
280	# and failing later inside a worker thread. operator.index() is the
281	# same __index__-based coercion PyArg_ParseTupleAndKeywords' "I"/"K"
282	# format codes use internally, so this accepts exactly what the C
283	# constructors accept (plain ints, numpy-style int-likes, ...) and
284	# rejects exactly what they reject (str, float, ...).
285	fd = operator.index(fd)	12✔
286	if nbytes is not None:	12✔
287	nbytes = operator.index(nbytes)	12✔
288	if offset is not None:	12✔
289	offset = operator.index(offset)	12✔
290	if priority is not None:	12✔
291	priority = operator.index(priority)	12✔
292
293	# Plain bytes, not a BytesIO wrapper - for a write this is the
294	# caller's own payload, handed to pwrite() as-is; for a read it
295	# starts empty and _handle_read() replaces it with pread()'s
296	# result directly. Either way there's nothing to unwrap later, so
297	# get_value()/payload return it with no extra copy - and it's
298	# never None, so callers don't need to check.
299	if opcode == OpCode.WRITE:	12✔
300	if not isinstance(payload, bytes):	12✔
301	raise ValueError(f"payload_bytes must be bytes, got {payload!r}")	12✔
302	buffer = payload	12✔
303	else:
304	buffer = b""	12✔
305
306	self.callback: Callable[[int], Any] \| None = None	12✔
307	self.in_progress = False	12✔
308	self._lock = Lock()	12✔
309	self.buffer: bytes = buffer	12✔
310
311	self.opcode = opcode	12✔
312	self.__fileno = fd	12✔
313	self.__offset = offset or 0	12✔
314	self.__opcode = opcode	12✔
315	self.__nbytes = nbytes or 0	12✔
316	self.__priority = priority or 0	12✔
317	self.exception = None	12✔
318	self.written = 0	12✔
319
320	@classmethod	12✔
321	def read(	12✔
322	cls, nbytes: int, fd: int, offset: int, priority=0,
323	) -> "Operation":
324	"""
325	Creates a new instance of Operation on read mode.
326	"""
327	return cls(fd, nbytes, offset, opcode=OpCode.READ, priority=priority)	12✔
328
329	@classmethod	12✔
330	def write(	12✔
331	cls, payload_bytes: bytes, fd: int, offset: int, priority=0,
332	) -> "Operation":
333	"""
334	Creates a new instance of AIOOperation on write mode.
335	"""
336	return cls(	12✔
337	fd,
338	len(payload_bytes),
339	offset,
340	payload=payload_bytes,
341	opcode=OpCode.WRITE,
342	priority=priority,
343	)
344
345	@classmethod	12✔
346	def fsync(cls, fd: int, priority=0) -> "Operation":	12✔
347
348	"""
349	Creates a new instance of AIOOperation on fsync mode.
350	"""
351	return cls(fd, None, None, opcode=OpCode.FSYNC, priority=priority)	12✔
352
353	@classmethod	12✔
354	def fdsync(cls, fd: int, priority=0) -> "Operation":	12✔
355
356	"""
357	Creates a new instance of AIOOperation on fdsync mode.
358	"""
359	return cls(fd, None, None, opcode=OpCode.FDSYNC, priority=priority)	12✔
360
361	def get_value(self) -> bytes \| int \| None:	12✔
362	"""
363	Method returns a bytes value of AIOOperation's result or None.
364	"""
365	if self.exception:	12✔
366	raise self.exception	12✔
367
368	if self.opcode == OpCode.WRITE:	12✔
369	return self.written	12✔
370
371	if self.opcode in (OpCode.FSYNC, OpCode.FDSYNC):	12✔
372	return None	12✔
373
374	return self.buffer	12✔
375
376	@property	12✔
377	def fileno(self) -> int:	12✔
378	return self.__fileno	12✔
379
380	@property	12✔
381	def offset(self) -> int:	12✔
382	return self.__offset	12✔
383
384	@property	12✔
385	def payload(self) -> memoryview \| None:	12✔
386	return memoryview(self.buffer)	12✔
387
388	@property	12✔
389	def nbytes(self) -> int:	12✔
390	return self.__nbytes	12✔
391
392	def set_callback(self, callback: Callable[[int], Any]) -> bool:	12✔
393	if not callable(callback):	12✔
394	raise ValueError(f"callback must be callable, got {callback!r}") # noqa: TRY004 (pre-existing public exception type, not changing it here)	12✔
395	with self._lock:	12✔
396	self.callback = callback	12✔
397	return True	12✔

mosquito / caio / 30894786869

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