30855072398

Committed 03 Aug 2026 09:32PM UTC coverage: 91.304% (+2.6%) from 88.669%

Build # 30855072398

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #69 from mosquito/c-extension-fixes

Fix segfault, hang, and ~30 real bugs across all 4 I/O backends

Coverage Stats

94 of 106 new or added lines in 5 files covered. (88.68%)

32 existing lines in 2 files now uncovered.

378 of 414 relevant lines covered (91.3%)

11.44 hits per line

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

97.99

/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.
        """
        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 self._lock:
            self._in_progress -= 1
            operation.in_progress = False

    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 checked and set under the same lock as
        # the capacity check/reservation - otherwise two concurrent
        # submits of the very same Operation (or the same object appearing
        # twice in one submit(op, op) call) could both see it unset and
        # both dispatch, running the I/O twice against one result object.
        with 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",
                )

            self._in_progress += 1
            operation.in_progress = True

        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.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)
        self.callback = callback
        return True

1	import operator	14✔
2	import os	14✔
3	import sys	14✔
4	import threading	14✔
5	from collections import defaultdict	14✔
6	from collections.abc import Callable	14✔
7	from enum import IntEnum, unique	14✔
8	from multiprocessing.pool import ThreadPool	14✔
9	from threading import Lock, RLock	14✔
10	from types import MappingProxyType	14✔
11	from typing import Any	14✔
12
13	from .abstract import AbstractContext, AbstractOperation	14✔
14
15	fdsync = getattr(os, "fdatasync", os.fsync)	14✔
16	NATIVE_PREAD_PWRITE = hasattr(os, "pread") and hasattr(os, "pwrite")	14✔
17
18
19	@unique	14✔
20	class OpCode(IntEnum):	14✔
21	READ = 0	14✔
22	WRITE = 1	14✔
23	FSYNC = 2	14✔
24	FDSYNC = 3	14✔
25	NOOP = -1	14✔
26
27
28	@unique	14✔
29	class ContextState(IntEnum):	14✔
30	OPEN = 0	14✔
31	CLOSING = 1	14✔
32	CLOSED = 2	14✔
33
34
35	class Context(AbstractContext):	14✔
36	"""
37	python aio context implementation
38	"""
39
40	MAX_POOL_SIZE = 128	14✔
41
42	def __init__(self, max_requests: int = 32, pool_size: int = 8):	14✔
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	14✔
48
49	if not (0 < pool_size < self.MAX_POOL_SIZE):	14✔
50	raise ValueError(	14✔
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:	14✔
55	raise ValueError(	14✔
56	f"max_requests must be a positive integer, got {max_requests}",
57	)
58
59	self.__max_requests = max_requests	14✔
60	self.pool = ThreadPool(pool_size)	14✔
61	self._in_progress = 0	14✔
62	self._lock = Lock()	14✔
63	self._state = ContextState.OPEN	14✔
64
65	if not NATIVE_PREAD_PWRITE:	14✔
66	self._locks_cleaner = RLock() # type: ignore	4✔
67	self._locks = defaultdict(RLock) # type: ignore	4✔
68
69	@property	14✔
70	def max_requests(self) -> int:	14✔
71	return self.__max_requests	×
72
73	@staticmethod	14✔
74	def _invoke_callback(operation: "Operation", value):	14✔
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	callback = operation.callback	14✔
84	if callback is None:	14✔
85	return	14✔
86
87	try:	14✔
88	callback(value)	14✔
89	except BaseException: # noqa: BLE001 (must isolate any callback exception, including SystemExit/KeyboardInterrupt)	14✔
90	threading.excepthook(	14✔
91	threading.ExceptHookArgs(
92	(*sys.exc_info(), threading.current_thread()),
93	),
94	)
95
96	def _release_slot(self):	14✔
97	with self._lock:	14✔
98	self._in_progress -= 1	14✔
99
100	def _rollback_claim(self, operation: "Operation"):	14✔
101	"""
102	Undoes a claim that was never actually scheduled (e.g. a
103	concurrent close() tore down the pool between the capacity
104	reservation and apply_async()) - unlike genuine completion, this
105	must reset operation.in_progress, since the operation never
106	actually ran and must stay retryable.
107	"""
108	with self._lock:	14✔
109	self._in_progress -= 1	14✔
110	operation.in_progress = False	14✔
111
112	def _execute(self, operation: "Operation") -> bool:	14✔
113	"""
114	Returns True if actually scheduled, False if skipped because
115	``operation`` was already in progress - matches the other three
116	backends, which all silently skip (rather than raise on, or
117	dispatch twice) a resubmit of an Operation still in flight.
118	"""
119	handler = self._OP_MAP[operation.opcode]	14✔
120
121	# operation.in_progress is deliberately NOT reset on genuine
122	# completion below - one-shot forever once actually scheduled,
123	# matching all three native backends: a completed Operation must
124	# not be resubmittable, only a fresh one constructed for a retry.
125	def on_error(exc):	14✔
126	self._release_slot()	14✔
127	operation.exception = exc	14✔
128	operation.written = 0	14✔
129	self._invoke_callback(operation, None)	14✔
130
131	def on_success(result):	14✔
132	self._release_slot()	14✔
133	operation.written = result	14✔
134	self._invoke_callback(operation, result)	14✔
135
136	# operation.in_progress is checked and set under the same lock as
137	# the capacity check/reservation - otherwise two concurrent
138	# submits of the very same Operation (or the same object appearing
139	# twice in one submit(op, op) call) could both see it unset and
140	# both dispatch, running the I/O twice against one result object.
141	with self._lock:	14✔
142	if operation.in_progress:	14✔
143	return False	14✔
144
145	if self._state != ContextState.OPEN:	14✔
146	raise RuntimeError("Context is closed")	14✔
147
148	if self._in_progress >= self.__max_requests:	14✔
149	raise RuntimeError(	14✔
150	"Maximum simultaneous requests have been reached",
151	)
152
153	self._in_progress += 1	14✔
154	operation.in_progress = True	14✔
155
156	try:	14✔
157	self.pool.apply_async(	14✔
158	handler, args=(self, operation),
159	callback=on_success,
160	error_callback=on_error,
161	)
162	except BaseException:	14✔
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)	14✔
168	raise	14✔
169
170	return True	14✔
171
172	if NATIVE_PREAD_PWRITE:	14✔
173	def __pread(self, fd, size, offset):	10✔
174	return os.pread(fd, size, offset)	10✔
175
176	def __pwrite(self, fd, bytes, offset):	10✔
177	return os.pwrite(fd, bytes, offset)	10✔
178	else:
179	def __pread(self, fd, size, offset):	4✔
180	with self._locks[fd]:	4✔
181	os.lseek(fd, 0, os.SEEK_SET)	4✔
182	os.lseek(fd, offset, os.SEEK_SET)	4✔
183	return os.read(fd, size)	4✔
184
185	def __pwrite(self, fd, bytes, offset):	4✔
186	with self._locks[fd]:	4✔
187	os.lseek(fd, 0, os.SEEK_SET)	4✔
188	os.lseek(fd, offset, os.SEEK_SET)	4✔
189	return os.write(fd, bytes)	4✔
190
191	def _handle_read(self, operation: "Operation"):	14✔
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(	14✔
197	operation.fileno, operation.nbytes, operation.offset,
198	)
199	operation.buffer = data	14✔
200	return len(data)	14✔
201
202	def _handle_write(self, operation: "Operation"):	14✔
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(	14✔
206	operation.fileno, operation.buffer, operation.offset,
207	)
208
209	def _handle_fsync(self, operation: "Operation"):	14✔
210	return os.fsync(operation.fileno)	14✔
211
212	def _handle_fdsync(self, operation: "Operation"):	14✔
213	return fdsync(operation.fileno)	14✔
214
215	def _handle_noop(self, operation: "Operation"):	14✔
216	return	×
217
218	def submit(self, *aio_operations) -> int:	14✔
219	for operation in aio_operations:	14✔
220	if not isinstance(operation, Operation):	14✔
NEW 221	raise ValueError(f"Invalid Operation {operation!r}") # noqa: TRY004 (pre-existing public exception type, not changing it here)	×
222
223	count = 0	14✔
224	for operation in aio_operations:	14✔
225	if self._execute(operation):	14✔
226	count += 1	14✔
227
228	return count	14✔
229
230	def cancel(self, *aio_operations) -> int:	14✔
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	14✔
239
240	def close(self):	14✔
241	if self._state != ContextState.OPEN:	14✔
242	return	14✔
243	with self._lock:	14✔
244	if self._state != ContextState.OPEN:	14✔
NEW 245	return	×
246	self._state = ContextState.CLOSING	14✔
247	self.pool.close()	14✔
248	self._state = ContextState.CLOSED	14✔
249
250	def __del__(self):	14✔
251	self.close()	14✔
252
253	_OP_MAP = MappingProxyType({	14✔
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):	14✔
264	"""
265	python aio operation implementation
266	"""
267	def __init__(	14✔
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)	14✔
286	if nbytes is not None:	14✔
287	nbytes = operator.index(nbytes)	14✔
288	if offset is not None:	14✔
289	offset = operator.index(offset)	14✔
290	if priority is not None:	14✔
291	priority = operator.index(priority)	14✔
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:	14✔
300	if not isinstance(payload, bytes):	14✔
301	raise ValueError(f"payload_bytes must be bytes, got {payload!r}")	14✔
302	buffer = payload	14✔
303	else:
304	buffer = b""	14✔
305
306	self.callback: Callable[[int], Any] \| None = None	14✔
307	self.in_progress = False	14✔
308	self.buffer: bytes = buffer	14✔
309
310	self.opcode = opcode	14✔
311	self.__fileno = fd	14✔
312	self.__offset = offset or 0	14✔
313	self.__opcode = opcode	14✔
314	self.__nbytes = nbytes or 0	14✔
315	self.__priority = priority or 0	14✔
316	self.exception = None	14✔
317	self.written = 0	14✔
318
319	@classmethod	14✔
320	def read(	14✔
321	cls, nbytes: int, fd: int, offset: int, priority=0,
322	) -> "Operation":
323	"""
324	Creates a new instance of Operation on read mode.
325	"""
326	return cls(fd, nbytes, offset, opcode=OpCode.READ, priority=priority)	14✔
327
328	@classmethod	14✔
329	def write(	14✔
330	cls, payload_bytes: bytes, fd: int, offset: int, priority=0,
331	) -> "Operation":
332	"""
333	Creates a new instance of AIOOperation on write mode.
334	"""
335	return cls(	14✔
336	fd,
337	len(payload_bytes),
338	offset,
339	payload=payload_bytes,
340	opcode=OpCode.WRITE,
341	priority=priority,
342	)
343
344	@classmethod	14✔
345	def fsync(cls, fd: int, priority=0) -> "Operation":	14✔
346
347	"""
348	Creates a new instance of AIOOperation on fsync mode.
349	"""
350	return cls(fd, None, None, opcode=OpCode.FSYNC, priority=priority)	14✔
351
352	@classmethod	14✔
353	def fdsync(cls, fd: int, priority=0) -> "Operation":	14✔
354
355	"""
356	Creates a new instance of AIOOperation on fdsync mode.
357	"""
358	return cls(fd, None, None, opcode=OpCode.FDSYNC, priority=priority)	14✔
359
360	def get_value(self) -> bytes \| int \| None:	14✔
361	"""
362	Method returns a bytes value of AIOOperation's result or None.
363	"""
364	if self.exception:	14✔
365	raise self.exception	14✔
366
367	if self.opcode == OpCode.WRITE:	14✔
368	return self.written	14✔
369
370	if self.opcode in (OpCode.FSYNC, OpCode.FDSYNC):	14✔
371	return None	14✔
372
373	return self.buffer	14✔
374
375	@property	14✔
376	def fileno(self) -> int:	14✔
377	return self.__fileno	14✔
378
379	@property	14✔
380	def offset(self) -> int:	14✔
381	return self.__offset	14✔
382
383	@property	14✔
384	def payload(self) -> memoryview \| None:	14✔
385	return memoryview(self.buffer)	14✔
386
387	@property	14✔
388	def nbytes(self) -> int:	14✔
389	return self.__nbytes	14✔
390
391	def set_callback(self, callback: Callable[[int], Any]) -> bool:	14✔
392	if not callable(callback):	14✔
393	raise ValueError(f"callback must be callable, got {callback!r}") # noqa: TRY004 (pre-existing public exception type, not changing it here)	14✔
394	self.callback = callback	14✔
395	return True	14✔

mosquito / caio / 30855072398

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