From 85dbe6b055af27a48d5a5a57c605a3304e415f6f Mon Sep 17 00:00:00 2001 From: Dmitry Orlov Date: Tue, 4 Aug 2026 09:37:54 +0200 Subject: [PATCH 1/8] Try free-threaded CPython support for the python_aio backend None of the three C extensions declare Py_mod_gil support, so importing any of them under a free-threaded interpreter silently re-enables the GIL for the whole process - not attempting that here. Two CI checks: a dedicated free-threaded-python-only job builds normally under a regular interpreter, then runs the suite through the matching free-threaded one against that same install - the compiled extensions' SOABI doesn't match, so they fail to import exactly like on any unsupported platform, leaving python_aio genuinely running with no GIL. Separately, the existing tests matrix gets 3.13t/3.14t entries that build fresh under a free-threaded interpreter directly, to see what happens when the C extensions do compile there (they do; the GIL just comes back). Adds a regression test asserting the GIL actually stays off whenever python_aio is the only backend available, plus a concurrency stress test hammering python_aio's own locking with many concurrent read/write ops across a real multi-worker ThreadPool. --- .github/workflows/ci.yml | 37 +++++++++++++++ pyproject.toml | 1 + tests/test_free_threading.py | 90 ++++++++++++++++++++++++++++++++++++ 3 files changed, 128 insertions(+) create mode 100644 tests/test_free_threading.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f74d6fd..660b617 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -44,6 +44,15 @@ jobs: os: ubuntu-latest - python: "3.14" os: ubuntu-latest + # Fresh build directly under a free-threaded interpreter - none of + # the C extensions declare Py_mod_gil support yet, so importing + # one (if it compiles here at all) re-enables the GIL for the + # whole process. See free-threaded-python-only below for the + # actual no-GIL check. + - python: "3.13t" + os: ubuntu-latest + - python: "3.14t" + os: ubuntu-latest - python: "3.10" os: windows-latest - python: "3.11" @@ -81,6 +90,34 @@ jobs: env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + # Builds normally under a regular interpreter, then runs the suite through + # a free-threaded one of the same X.Y against that same install. The + # compiled extensions' SOABI won't match the free-threaded interpreter, so + # they simply fail to import (same ImportError path as any unsupported + # platform) - only python_aio ends up available, and unlike the `tests` + # matrix's own free-threaded entries above, nothing gets the chance to + # re-enable the GIL first. + free-threaded-python-only: + needs: lint + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python: ["3.13", "3.14"] + steps: + - uses: actions/checkout@v4 + - uses: astral-sh/setup-uv@v5 + with: + python-version: "${{ matrix.python }}" + - run: uv sync --extra develop + - run: uv python install "${{ matrix.python }}t" + - name: Run tests under the free-threaded interpreter against this install + run: | + FREE_THREADED_PYTHON="$(uv python find "${{ matrix.python }}t")" + PYTHONPATH="$(pwd)/.venv/lib/python${{ matrix.python }}/site-packages" "$FREE_THREADED_PYTHON" -m pytest -sv tests + env: + FORCE_COLOR: 1 + finish: needs: - tests diff --git a/pyproject.toml b/pyproject.toml index 8e2f163..6bd0613 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,6 +27,7 @@ classifiers = [ "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: Implementation :: CPython", + "Programming Language :: Python :: Free Threading :: 1 - Unstable", ] [project.urls] diff --git a/tests/test_free_threading.py b/tests/test_free_threading.py new file mode 100644 index 0000000..8a23901 --- /dev/null +++ b/tests/test_free_threading.py @@ -0,0 +1,90 @@ +import os +import sys +import sysconfig +import threading + +import pytest + +import caio +from caio import python_aio + + +def test_gil_stays_disabled_when_only_python_aio_available(): + """caio ships no Py_mod_gil declaration on any C extension, so importing + thread_aio/linux_aio/linux_uring under a free-threaded interpreter + silently re-enables the GIL (CPython's own safety fallback). When none + of them are importable here (e.g. built against a regular interpreter, + then run under a free-threaded one - different SOABI, so they simply + don't match), nothing should have flipped the GIL back on.""" + if not sysconfig.get_config_var("Py_GIL_DISABLED"): + pytest.skip("not a free-threaded build") + if any((caio.thread_aio, caio.linux_aio, caio.linux_uring)): + pytest.skip("a C extension is available here too - GIL re-enable is expected") + assert sys._is_gil_enabled() is False + + +def test_high_concurrency_stress_no_data_races(tmp_path): + """Hammers python_aio's own bookkeeping (the _lock-protected + _in_progress counter and per-operation in_progress flag) with many + concurrent writes then reads dispatched across a real multi-worker + ThreadPool. Under a free-threaded interpreter these workers can run + truly in parallel instead of being serialized by the GIL, so a race in + that bookkeeping - two workers claiming the same slot, a result + crossing over to the wrong Operation - would show up here as a lost + write or a chunk read back with the wrong content, not just a + thread-timing fluke.""" + count = 200 + chunk = 4096 + path = tmp_path / "stress.bin" + path.write_bytes(b"\x00" * (count * chunk)) + fd = os.open(str(path), os.O_RDWR) + try: + ctx = python_aio.Context(max_requests=count, pool_size=32) + expected = [bytes([i % 256]) * chunk for i in range(count)] + + written = [None] * count + lock = threading.Lock() + remaining = [count] + done = threading.Event() + + def make_write_cb(i): + def cb(n): + with lock: + written[i] = n + remaining[0] -= 1 + if remaining[0] == 0: + done.set() + return cb + + for i in range(count): + op = python_aio.Operation.write(expected[i], fd, i * chunk) + op.set_callback(make_write_cb(i)) + assert ctx.submit(op) == 1 + + assert done.wait(10), f"only {count - remaining[0]}/{count} writes completed" + assert written == [chunk] * count + + read_back = [None] * count + remaining = [count] + done = threading.Event() + + def make_read_cb(i, op): + def cb(_n): + with lock: + read_back[i] = op.get_value() + remaining[0] -= 1 + if remaining[0] == 0: + done.set() + return cb + + for i in range(count): + op = python_aio.Operation.read(chunk, fd, i * chunk) + op.set_callback(make_read_cb(i, op)) + assert ctx.submit(op) == 1 + + assert done.wait(10), f"only {count - remaining[0]}/{count} reads completed" + for i, (got, want) in enumerate(zip(read_back, expected)): + assert got == want, f"chunk {i} mismatch" + finally: + os.close(fd) + ctx.close() From f8b5ecbd5dfbde4296fa70eebdbd93323978289c Mon Sep 17 00:00:00 2001 From: Dmitry Orlov Date: Tue, 4 Aug 2026 09:42:06 +0200 Subject: [PATCH 2/8] Fix python_aio: cross-Context race on the same Operation Context._execute() serialized the check-and-set of operation.in_progress under its own lock, which only protects submissions through that one Context - two different Contexts submitting the same Operation shared no lock and could both observe it unclaimed. Reproduced reliably under a free-threaded interpreter (~0.5% of submissions double-dispatched). Moves the claim itself onto the Operation via its own lock (_claim()/_unclaim()), independent of which Context is dispatching it. --- caio/python_aio.py | 35 ++++++++++++------ tests/test_free_threading.py | 69 ++++++++++++++++++++++++++++++++++++ 2 files changed, 94 insertions(+), 10 deletions(-) diff --git a/caio/python_aio.py b/caio/python_aio.py index de42f00..a129eb7 100644 --- a/caio/python_aio.py +++ b/caio/python_aio.py @@ -107,7 +107,7 @@ def _rollback_claim(self, operation: "Operation"): """ with self._lock: self._in_progress -= 1 - operation.in_progress = False + operation._unclaim() def _execute(self, operation: "Operation") -> bool: """ @@ -133,25 +133,25 @@ def on_success(result): 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 + # The claim itself is synchronized by 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 here can't stop them both from claiming it. + if not operation._claim(): + return False + with self._lock: if self._state != ContextState.OPEN: + operation._unclaim() raise RuntimeError("Context is closed") if self._in_progress >= self.__max_requests: + operation._unclaim() raise RuntimeError( "Maximum simultaneous requests have been reached", ) self._in_progress += 1 - operation.in_progress = True try: self.pool.apply_async( @@ -305,6 +305,7 @@ def __init__( self.callback: Callable[[int], Any] | None = None self.in_progress = False + self._lock = Lock() self.buffer: bytes = buffer self.opcode = opcode @@ -316,6 +317,20 @@ def __init__( self.exception = None self.written = 0 + def _claim(self) -> bool: + """Atomically claims this Operation for execution - False if some + Context (this one or another) already claimed it first.""" + with self._lock: + if self.in_progress: + return False + self.in_progress = True + return True + + def _unclaim(self) -> None: + """Reverts a claim that never actually got scheduled.""" + with self._lock: + self.in_progress = False + @classmethod def read( cls, nbytes: int, fd: int, offset: int, priority=0, diff --git a/tests/test_free_threading.py b/tests/test_free_threading.py index 8a23901..9c7b85b 100644 --- a/tests/test_free_threading.py +++ b/tests/test_free_threading.py @@ -88,3 +88,72 @@ def cb(_n): finally: os.close(fd) ctx.close() + + +def test_same_operation_is_claimed_once_across_contexts(): + """An Operation's one-shot claim must be synchronized by the Operation. + + Context._execute() currently protects ``operation.in_progress`` with the + Context's lock. That only serializes submissions through the *same* + Context: two Contexts use two unrelated locks and can both observe False + before either stores True when the GIL is disabled. + + Synchronizing each pair of submissions makes the race frequent enough to + be a useful regression test without depending on filesystem timing. + """ + if getattr(sys, "_is_gil_enabled", lambda: True)(): + pytest.skip("requires a free-threaded interpreter with the GIL disabled") + + iterations = 20_000 + contexts = ( + python_aio.Context(max_requests=iterations * 2), + python_aio.Context(max_requests=iterations * 2), + ) + start = threading.Barrier(3, timeout=30) + finished = threading.Barrier(3, timeout=30) + current = [None] + submitted = [0, 0] + errors = [] + + def submitter(index, context): + try: + for _ in range(iterations): + start.wait() + submitted[index] += context.submit(current[0]) + finished.wait() + except Exception as exc: # noqa: BLE001 (surface child-thread failures) + errors.append(exc) + start.abort() + finished.abort() + + threads = [ + threading.Thread(target=submitter, args=(index, context)) + for index, context in enumerate(contexts) + ] + + try: + for thread in threads: + thread.start() + + for _ in range(iterations): + current[0] = python_aio.Operation( + 0, None, None, python_aio.OpCode.NOOP, + ) + start.wait() + finished.wait() + + for thread in threads: + thread.join() + + assert not errors + assert sum(submitted) == iterations, ( + f"{sum(submitted) - iterations} Operations were submitted twice" + ) + finally: + start.abort() + finished.abort() + for thread in threads: + thread.join(timeout=5) + for context in contexts: + context.close() + context.pool.join() From 5a18340c91ce24ee9cd56c7a723c309473cdf262 Mon Sep 17 00:00:00 2001 From: Dmitry Orlov Date: Tue, 4 Aug 2026 11:04:57 +0200 Subject: [PATCH 3/8] Make all four backends safe under free-threaded CPython Fixes #64 Every backend's submit() (and linux_uring's cancel()) claimed an Operation via a plain check-then-set of in_progress - safe only because the GIL serialized it. Under a free-threaded interpreter, two Contexts racing to submit() the same Operation could both see it unclaimed and both dispatch it: reproduced as a stable segfault in thread_aio (two workers running the same Operation concurrently), silent double- delivery in linux_aio/linux_uring, and the same double-submit in python_aio. Fixed with a CAS on in_progress in the three C backends, and a proper per-Operation lock (instead of the Context's own, which only serializes submissions through that one Context) in python_aio. linux_uring's own submit()/cancel()/uring_drain_cq() had a second, separate race: each reads the ring's head/tail once, does a batch of work sized off that snapshot, and only commits a new head/tail at the end. Two concurrent callers (typically concurrent process_events() drainers) could claim the same CQE range and each fire every callback in it - occasionally corrupting the ring enough to livelock the whole process. Fixed with a critical section (Py_BEGIN_CRITICAL_SECTION, compiles to nothing under the GIL) around each ring bookkeeping section, always released before invoking a Python callback so a reentrant call from inside one can't deadlock on itself. linux_aio/linux_uring's callback field also wasn't safe to read concurrently with set_callback() replacing it - both now go through the same per-object critical section instead of a bare pointer read/write, with the old callback properly released after. Adds tests/test_free_threading.py: a GIL-stays-disabled regression check, and stress tests (import-time GIL check, concurrent submit/claim races, concurrent process_events() drainers, concurrent result readers) across every backend, gated on actually running under a free-threaded interpreter with the GIL disabled. CI gets two free-threaded ubuntu jobs (3.13t/3.14t) that force PYTHON_GIL=0 so the suite exercises real no-GIL concurrency, plus a free-threaded-python-only job proving python_aio alone stays genuinely no-GIL without any override. The release workflow, Makefile, and make-wheels.sh now also build cp313t/cp314t wheels alongside the existing versions (Windows ships pure-Python only, already version-agnostic). None of the three C extensions declare Py_mod_gil support yet, so a real free-threaded install still needs PYTHON_GIL=0 to keep the GIL off after importing one - a possible follow-up once this is battle-tested. --- caio/linux_aio.c | 100 +++-- caio/linux_uring.c | 106 ++++- caio/python_aio.py | 41 +- caio/thread_aio.c | 105 +++-- tests/test_free_threading.py | 772 ++++++++++++++++++++++++++++++++++- 5 files changed, 1011 insertions(+), 113 deletions(-) diff --git a/caio/linux_aio.c b/caio/linux_aio.c index 4e8d1d3..5f7911b 100644 --- a/caio/linux_aio.c +++ b/caio/linux_aio.c @@ -11,6 +11,33 @@ #define PY_SSIZE_T_CLEAN #include #include + +#if PY_VERSION_HEX >= 0x030D0000 && defined(Py_GIL_DISABLED) +#define CAIO_BEGIN_CRITICAL_SECTION(object) \ + PyCriticalSection caio_critical_section; \ + PyCriticalSection_Begin( \ + &caio_critical_section, (PyObject *)(object) \ + ) +#define CAIO_END_CRITICAL_SECTION() \ + PyCriticalSection_End(&caio_critical_section) +#else +#define CAIO_BEGIN_CRITICAL_SECTION(object) +#define CAIO_END_CRITICAL_SECTION() +#endif + +#define CAIO_ATOMIC_LOAD(value) \ + __atomic_load_n(&(value), __ATOMIC_ACQUIRE) +#define CAIO_ATOMIC_STORE(value, new_value) \ + __atomic_store_n(&(value), (new_value), __ATOMIC_RELEASE) +#define CAIO_ATOMIC_LOAD_STORE(value, new_value) \ + __atomic_exchange_n(&(value), (new_value), __ATOMIC_ACQ_REL) + +#if PY_VERSION_HEX >= 0x030D0000 && defined(Py_GIL_DISABLED) +#define CAIO_DECLARE_FREE_THREADED(module) \ + PyUnstable_Module_SetGIL((module), Py_MOD_GIL_NOT_USED) +#else +#define CAIO_DECLARE_FREE_THREADED(module) 0 +#endif #include @@ -152,6 +179,15 @@ static PyTypeObject* AIOOperationTypeP = NULL; static PyTypeObject* AIOContextTypeP = NULL; +static PyObject *AIOOperation_callback_ref(AIOOperation *self) { + PyObject *callback; + CAIO_BEGIN_CRITICAL_SECTION(self); + callback = Py_XNewRef(self->callback); + CAIO_END_CRITICAL_SECTION(); + return callback; +} + + static void AIOContext_dealloc(AIOContext *self) { if (self->weakreflist != NULL) @@ -276,21 +312,16 @@ static PyObject* AIOContext_submit(AIOContext *self, PyObject *args) { return NULL; } - /* Skip anything already in_progress (this exact Operation submitted - * earlier and not yet completed) - otherwise the kernel would get - * handed the very same embedded iocb struct a second time, and two - * completions would eventually deliver for one Python object. Every - * claimed op's context is reset first (Py_XDECREF, not just an - * overwrite) since a previously-completed submission never clears it - - * without this, a normal resubmit after completion silently leaks the - * old context reference forever. */ + /* Atomic exchange, not check-then-set: two Contexts racing on the same Operation + * must not both hand its iocb to the kernel. Context reset via + * Py_XDECREF, not overwrite - a previously-completed submission never + * clears it. */ Py_ssize_t to_submit = 0; for (Py_ssize_t i = 0; i < nr; i++) { AIOOperation *op = (AIOOperation *) PyTuple_GET_ITEM(args, i); - if (op->in_progress) - continue; - op->in_progress = 1; + if (CAIO_ATOMIC_LOAD_STORE(op->in_progress, 1)) continue; + Py_XDECREF(op->context); op->context = self; Py_INCREF(self); @@ -317,7 +348,7 @@ static PyObject* AIOContext_submit(AIOContext *self, PyObject *args) { * each op is exactly as retryable as it was before this call. */ for (Py_ssize_t i = 0; i < to_submit; i++) { AIOOperation *op = claimed[i]; - op->in_progress = 0; + CAIO_ATOMIC_STORE(op->in_progress, 0); Py_CLEAR(op->context); Py_DECREF(op); } @@ -333,7 +364,7 @@ static PyObject* AIOContext_submit(AIOContext *self, PyObject *args) { * back the same way instead of leaking its claim forever. */ for (Py_ssize_t i = result; i < to_submit; i++) { AIOOperation *op = claimed[i]; - op->in_progress = 0; + CAIO_ATOMIC_STORE(op->in_progress, 0); Py_CLEAR(op->context); Py_DECREF(op); } @@ -391,15 +422,18 @@ static PyObject* AIOContext_cancel(AIOContext *self, PyObject *args, PyObject *k * the (paused) Rust rewrite; a cancelled Operation is still terminal, * not retryable, only a fresh one constructed for a retry. */ Py_CLEAR(op->context); - op->done = 1; + CAIO_ATOMIC_STORE(op->done, 1); - if (op->callback != NULL) { - PyObject *rv = PyObject_CallFunction(op->callback, "K", ev.res); + PyObject *callback = AIOOperation_callback_ref(op); + if (callback != NULL) { + PyObject *rv = PyObject_CallFunction(callback, "K", ev.res); if (rv == NULL) { + Py_DECREF(callback); Py_DECREF(op); return NULL; } Py_DECREF(rv); + Py_DECREF(callback); } Py_DECREF(op); @@ -524,22 +558,24 @@ static PyObject* AIOContext_process_events( op = (AIOOperation*)(uintptr_t) ev->data; - Py_CLEAR(op->context); - op->done = 1; - if (ev->res >= 0) { op->iocb.aio_nbytes = ev->res; } else { op->error = -ev->res; } - if (op->callback != NULL) { - PyObject *rv = PyObject_CallFunction(op->callback, "K", ev->res); + Py_CLEAR(op->context); + CAIO_ATOMIC_STORE(op->done, 1); + + PyObject *callback = AIOOperation_callback_ref(op); + if (callback != NULL) { + PyObject *rv = PyObject_CallFunction(callback, "K", ev->res); if (rv == NULL) { - PyErr_WriteUnraisable(op->callback); + PyErr_WriteUnraisable(callback); } else { Py_DECREF(rv); } + Py_DECREF(callback); } Py_DECREF(op); @@ -990,7 +1026,10 @@ PyDoc_STRVAR(AIOOperation_get_value_docstring, static PyObject* AIOOperation_get_value( AIOOperation *self, PyObject *args, PyObject *kwds ) { - if (self->in_progress && !self->done) { + if ( + CAIO_ATOMIC_LOAD(self->in_progress) && + !CAIO_ATOMIC_LOAD(self->done) + ) { PyErr_SetString( PyExc_RuntimeError, "get_value() is not available while the operation is in flight" @@ -1044,6 +1083,7 @@ static PyObject* AIOOperation_set_callback( static char *kwlist[] = {"callback", NULL}; PyObject* callback; + PyObject* old_callback; int argIsOk = PyArg_ParseTupleAndKeywords( args, kwds, "O", kwlist, @@ -1062,13 +1102,20 @@ static PyObject* AIOOperation_set_callback( } Py_INCREF(callback); + CAIO_BEGIN_CRITICAL_SECTION(self); + old_callback = self->callback; self->callback = callback; + CAIO_END_CRITICAL_SECTION(); + Py_XDECREF(old_callback); Py_RETURN_TRUE; } static PyObject *AIOOperation_payload_getter(AIOOperation *self, void *closure) { - if (self->in_progress && !self->done) { + if ( + CAIO_ATOMIC_LOAD(self->in_progress) && + !CAIO_ATOMIC_LOAD(self->done) + ) { PyErr_SetString( PyExc_RuntimeError, "payload is not available while the operation is in flight" @@ -1251,6 +1298,10 @@ PyMODINIT_FUNC PyInit_linux_aio(void) { m = PyModule_Create(&linux_aio_module); if (m == NULL) return NULL; + if (CAIO_DECLARE_FREE_THREADED(m) < 0) { + Py_DECREF(m); + return NULL; + } if (PyType_Ready(AIOContextTypeP) < 0) return NULL; @@ -1274,4 +1325,3 @@ PyMODINIT_FUNC PyInit_linux_aio(void) { return m; } - diff --git a/caio/linux_uring.c b/caio/linux_uring.c index f50dbd3..876a951 100644 --- a/caio/linux_uring.c +++ b/caio/linux_uring.c @@ -22,6 +22,33 @@ #include #include +#if PY_VERSION_HEX >= 0x030D0000 && defined(Py_GIL_DISABLED) +#define CAIO_BEGIN_CRITICAL_SECTION(object) \ + PyCriticalSection caio_critical_section; \ + PyCriticalSection_Begin( \ + &caio_critical_section, (PyObject *)(object) \ + ) +#define CAIO_END_CRITICAL_SECTION() \ + PyCriticalSection_End(&caio_critical_section) +#else +#define CAIO_BEGIN_CRITICAL_SECTION(object) +#define CAIO_END_CRITICAL_SECTION() +#endif + +#define CAIO_ATOMIC_LOAD(value) \ + __atomic_load_n(&(value), __ATOMIC_ACQUIRE) +#define CAIO_ATOMIC_STORE(value, new_value) \ + __atomic_store_n(&(value), (new_value), __ATOMIC_RELEASE) +#define CAIO_ATOMIC_LOAD_STORE(value, new_value) \ + __atomic_exchange_n(&(value), (new_value), __ATOMIC_ACQ_REL) + +#if PY_VERSION_HEX >= 0x030D0000 && defined(Py_GIL_DISABLED) +#define CAIO_DECLARE_FREE_THREADED(module) \ + PyUnstable_Module_SetGIL((module), Py_MOD_GIL_NOT_USED) +#else +#define CAIO_DECLARE_FREE_THREADED(module) 0 +#endif + /* ---- syscall wrappers ---- */ static inline int io_uring_setup(uint32_t entries, struct io_uring_params *p) { return (int) syscall(__NR_io_uring_setup, entries, p); @@ -120,6 +147,15 @@ static int AIOOperation_clear(AIOOperation *self) { } +static PyObject *AIOOperation_callback_ref(AIOOperation *self) { + PyObject *callback; + CAIO_BEGIN_CRITICAL_SECTION(self); + callback = Py_XNewRef(self->callback); + CAIO_END_CRITICAL_SECTION(); + return callback; +} + + static void AIOOperation_dealloc(AIOOperation *self) { PyObject_GC_UnTrack(self); @@ -343,7 +379,10 @@ PyDoc_STRVAR(AIOOperation_get_value_docstring, static PyObject *AIOOperation_get_value( AIOOperation *self, PyObject *args, PyObject *kwds ) { - if (self->in_progress && !self->done) { + if ( + CAIO_ATOMIC_LOAD(self->in_progress) && + !CAIO_ATOMIC_LOAD(self->done) + ) { PyErr_SetString( PyExc_RuntimeError, "get_value() is not available while the operation is in flight" @@ -389,21 +428,29 @@ PyDoc_STRVAR(AIOOperation_set_callback_docstring, static PyObject *AIOOperation_set_callback( AIOOperation *self, PyObject *callback ) { + PyObject *old_callback; + if (!PyCallable_Check(callback)) { PyErr_Format(PyExc_ValueError, "object %r is not callable", callback); return NULL; } Py_INCREF(callback); - Py_XDECREF(self->callback); + CAIO_BEGIN_CRITICAL_SECTION(self); + old_callback = self->callback; self->callback = callback; + CAIO_END_CRITICAL_SECTION(); + Py_XDECREF(old_callback); Py_RETURN_TRUE; } static PyObject *AIOOperation_payload_getter(AIOOperation *self, void *closure) { - if (self->in_progress && !self->done) { + if ( + CAIO_ATOMIC_LOAD(self->in_progress) && + !CAIO_ATOMIC_LOAD(self->done) + ) { PyErr_SetString( PyExc_RuntimeError, "payload is not available while the operation is in flight" @@ -809,6 +856,8 @@ static PyObject *AIOContext_repr(AIOContext *self) { * callbacks afterward removes the race entirely. */ static int uring_drain_cq(AIOContext *self, uint32_t max) { + CAIO_BEGIN_CRITICAL_SECTION(self); /* released before any callback runs */ + uint32_t head = __atomic_load_n(self->cq_head, __ATOMIC_RELAXED); uint32_t tail = __atomic_load_n(self->cq_tail, __ATOMIC_ACQUIRE); uint32_t mask = *self->cq_ring_mask; @@ -824,6 +873,7 @@ static int uring_drain_cq(AIOContext *self, uint32_t max) { if (ops == NULL || results == NULL) { PyMem_Free(ops); PyMem_Free(results); + CAIO_END_CRITICAL_SECTION(); PyErr_NoMemory(); return -1; } @@ -845,14 +895,14 @@ static int uring_drain_cq(AIOContext *self, uint32_t max) { * done touching anything for this op, so there's no more reason * to keep the Context alive on its behalf. */ AIOOperation *op = (AIOOperation *)(uintptr_t) cqe->user_data; - Py_CLEAR(op->context); - op->done = 1; op->result = cqe->res; if (cqe->res < 0) { op->error = -cqe->res; } else if (op->opcode == URING_READ) { op->buf_size = cqe->res; } + Py_CLEAR(op->context); + CAIO_ATOMIC_STORE(op->done, 1); ops[count] = op; results[count] = cqe->res; @@ -862,23 +912,26 @@ static int uring_drain_cq(AIOContext *self, uint32_t max) { /* Ring state fully committed - reentrant callers now see this whole * batch as already consumed, before a single callback has run. */ __atomic_store_n(self->cq_head, head, __ATOMIC_RELEASE); + CAIO_END_CRITICAL_SECTION(); for (uint32_t i = 0; i < count; i++) { AIOOperation *op = ops[i]; + PyObject *callback = AIOOperation_callback_ref(op); - if (op->callback != NULL) { + if (callback != NULL) { PyObject *arg = PyLong_FromLong((long) results[i]); if (arg == NULL) { - PyErr_WriteUnraisable(op->callback); + PyErr_WriteUnraisable(callback); } else { - PyObject *rv = PyObject_CallOneArg(op->callback, arg); + PyObject *rv = PyObject_CallOneArg(callback, arg); Py_DECREF(arg); if (rv == NULL) { - PyErr_WriteUnraisable(op->callback); + PyErr_WriteUnraisable(callback); } else { Py_DECREF(rv); } } + Py_DECREF(callback); } Py_DECREF(op); @@ -917,6 +970,8 @@ static PyObject *AIOContext_submit(AIOContext *self, PyObject *args) { } } + CAIO_BEGIN_CRITICAL_SECTION(self); /* serializes tail/SQE writes */ + uint32_t tail = __atomic_load_n(self->sq_tail, __ATOMIC_RELAXED); uint32_t head = __atomic_load_n(self->sq_head, __ATOMIC_ACQUIRE); uint32_t mask = *self->sq_ring_mask; @@ -926,19 +981,19 @@ static PyObject *AIOContext_submit(AIOContext *self, PyObject *args) { for (Py_ssize_t i = 0; i < nr; i++) { AIOOperation *op = (AIOOperation *) PyTuple_GET_ITEM(args, i); - if (op->in_progress) - continue; + /* Atomic exchange, not check-then-set: two Contexts racing on the same + * Operation must not both stage an SQE for it. Claimed up front + * so the check-and-set is one atomic step - the exit paths below + * that can still skip a freshly-claimed op undo the claim. */ + if (CAIO_ATOMIC_LOAD_STORE(op->in_progress, 1)) continue; if ((tail - head) >= capacity) { - /* Commit whatever WAS successfully staged earlier in this same - * call before returning - those ops already have in_progress=1 - * and their own Py_INCREF applied, and their SQEs are already - * written into the ring buffer; without this they'd be - * invisible to the kernel forever (sq_tail never advanced past - * them) despite looking submitted from Python's side - stuck - * in_progress permanently, no completion ever able to arrive - * to clear it, and their reference leaked for good. */ + /* Not staged - give the claim back. Commit whatever WAS + * staged earlier in this call first, or it'd be invisible to + * the kernel forever (sq_tail never advanced past it). */ + CAIO_ATOMIC_STORE(op->in_progress, 0); __atomic_store_n(self->sq_tail, tail, __ATOMIC_RELEASE); + CAIO_END_CRITICAL_SECTION(); PyErr_SetString(PyExc_OverflowError, "io_uring SQ ring full"); return NULL; } @@ -970,6 +1025,9 @@ static PyObject *AIOContext_submit(AIOContext *self, PyObject *args) { sqe->fsync_flags = IORING_FSYNC_DATASYNC; break; default: + /* Unrecognized opcode: give the claim back, this op was + * never staged. */ + CAIO_ATOMIC_STORE(op->in_progress, 0); continue; } @@ -977,7 +1035,6 @@ static PyObject *AIOContext_submit(AIOContext *self, PyObject *args) { self->sq_array[index] = index; tail++; - op->in_progress = 1; Py_INCREF(op); /* Held only while genuinely in flight (cleared on completion in @@ -993,6 +1050,7 @@ static PyObject *AIOContext_submit(AIOContext *self, PyObject *args) { } __atomic_store_n(self->sq_tail, tail, __ATOMIC_RELEASE); + CAIO_END_CRITICAL_SECTION(); /* * Do NOT call io_uring_enter here. The Python asyncio layer batches @@ -1089,9 +1147,12 @@ static PyObject *AIOContext_cancel( args, kwds, "O!", kwlist, &AIOOperationType, &op)) return NULL; + CAIO_BEGIN_CRITICAL_SECTION(self); + uint32_t tail = __atomic_load_n(self->sq_tail, __ATOMIC_RELAXED); uint32_t head = __atomic_load_n(self->sq_head, __ATOMIC_ACQUIRE); if ((tail - head) >= *self->sq_ring_entries) { + CAIO_END_CRITICAL_SECTION(); PyErr_SetString(PyExc_OverflowError, "io_uring SQ ring full"); return NULL; } @@ -1106,6 +1167,7 @@ static PyObject *AIOContext_cancel( if (!self->no_sqarray) self->sq_array[index] = index; __atomic_store_n(self->sq_tail, tail + 1, __ATOMIC_RELEASE); + CAIO_END_CRITICAL_SECTION(); io_uring_enter(self->uring_fd, 1, 0, 0, NULL); @@ -1431,6 +1493,10 @@ PyMODINIT_FUNC PyInit_linux_uring(void) { PyObject *m = PyModule_Create(&linux_uring_module); if (m == NULL) return NULL; + if (CAIO_DECLARE_FREE_THREADED(m) < 0) { + Py_DECREF(m); + return NULL; + } if (PyModule_AddObject(m, "SQPOLL_ALLOWED", PyBool_FromLong(sqpoll_allowed)) < 0) { Py_DECREF(m); diff --git a/caio/python_aio.py b/caio/python_aio.py index a129eb7..9eaef3b 100644 --- a/caio/python_aio.py +++ b/caio/python_aio.py @@ -80,7 +80,8 @@ def _invoke_callback(operation: "Operation", value): there kills that thread, silently stalling every future result/callback for the rest of this Context's lifetime. """ - callback = operation.callback + with operation._lock: + callback = operation.callback if callback is None: return @@ -105,9 +106,9 @@ def _rollback_claim(self, operation: "Operation"): must reset operation.in_progress, since the operation never actually ran and must stay retryable. """ - with self._lock: + with operation._lock, self._lock: + operation.in_progress = False self._in_progress -= 1 - operation._unclaim() def _execute(self, operation: "Operation") -> bool: """ @@ -133,24 +134,23 @@ def on_success(result): operation.written = result self._invoke_callback(operation, result) - # The claim itself is synchronized by 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 here can't stop them both from claiming it. - if not operation._claim(): - return False + # 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 - with self._lock: if self._state != ContextState.OPEN: - operation._unclaim() raise RuntimeError("Context is closed") if self._in_progress >= self.__max_requests: - operation._unclaim() raise RuntimeError( "Maximum simultaneous requests have been reached", ) + operation.in_progress = True self._in_progress += 1 try: @@ -317,20 +317,6 @@ def __init__( self.exception = None self.written = 0 - def _claim(self) -> bool: - """Atomically claims this Operation for execution - False if some - Context (this one or another) already claimed it first.""" - with self._lock: - if self.in_progress: - return False - self.in_progress = True - return True - - def _unclaim(self) -> None: - """Reverts a claim that never actually got scheduled.""" - with self._lock: - self.in_progress = False - @classmethod def read( cls, nbytes: int, fd: int, offset: int, priority=0, @@ -406,5 +392,6 @@ def nbytes(self) -> int: 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 + with self._lock: + self.callback = callback return True diff --git a/caio/thread_aio.c b/caio/thread_aio.c index 8e0e209..49948ff 100644 --- a/caio/thread_aio.c +++ b/caio/thread_aio.c @@ -6,6 +6,33 @@ #include #include +#if PY_VERSION_HEX >= 0x030D0000 && defined(Py_GIL_DISABLED) +#define CAIO_BEGIN_CRITICAL_SECTION(object) \ + PyCriticalSection caio_critical_section; \ + PyCriticalSection_Begin( \ + &caio_critical_section, (PyObject *)(object) \ + ) +#define CAIO_END_CRITICAL_SECTION() \ + PyCriticalSection_End(&caio_critical_section) +#else +#define CAIO_BEGIN_CRITICAL_SECTION(object) +#define CAIO_END_CRITICAL_SECTION() +#endif + +#define CAIO_ATOMIC_LOAD(value) \ + __atomic_load_n(&(value), __ATOMIC_ACQUIRE) +#define CAIO_ATOMIC_STORE(value, new_value) \ + __atomic_store_n(&(value), (new_value), __ATOMIC_RELEASE) +#define CAIO_ATOMIC_LOAD_STORE(value, new_value) \ + __atomic_exchange_n(&(value), (new_value), __ATOMIC_ACQ_REL) + +#if PY_VERSION_HEX >= 0x030D0000 && defined(Py_GIL_DISABLED) +#define CAIO_DECLARE_FREE_THREADED(module) \ + PyUnstable_Module_SetGIL((module), Py_MOD_GIL_NOT_USED) +#else +#define CAIO_DECLARE_FREE_THREADED(module) 0 +#endif + #include "src/threadpool/threadpool.h" @@ -60,6 +87,15 @@ enum THAIO_OP_CODE { }; +static PyObject *AIOOperation_callback_ref(AIOOperation *self) { + PyObject *callback; + CAIO_BEGIN_CRITICAL_SECTION(self); + callback = Py_XNewRef(self->callback); + CAIO_END_CRITICAL_SECTION(); + return callback; +} + + static void AIOContext_dealloc(AIOContext *self) { if (self->weakreflist != NULL) @@ -219,22 +255,25 @@ void worker(void *arg) { op->buf_size = result; } - /* Release store, paired with payload/get_value()'s acquire load of - * done - the plain writes to result/error/buf_size above happen on - * this thread without holding the GIL, so without a real memory - * barrier a concurrent GIL-holding reader on another thread has no - * guarantee of ever observing them (or of observing them in this - * order), regardless of in_progress. */ - __atomic_store_n(&op->done, 1, __ATOMIC_RELEASE); - state = PyGILState_Ensure(); - if (op->callback != NULL) { - PyObject_CallFunction(op->callback, "i", result); + if (op->opcode == THAIO_WRITE) { + Py_CLEAR(op->py_buffer); } - if (op->opcode == THAIO_WRITE) { - Py_DECREF(op->py_buffer); - op->py_buffer = NULL; + /* Publish completion only after every result field and Python-owned + * buffer transition is complete. payload/get_value() acquire-load done, + * so a reader that observes true must also observe all writes above. */ + CAIO_ATOMIC_STORE(op->done, 1); + + PyObject *callback = AIOOperation_callback_ref(op); + if (callback != NULL) { + PyObject *rv = PyObject_CallFunction(callback, "i", result); + if (rv == NULL) { + PyErr_WriteUnraisable(callback); + } else { + Py_DECREF(rv); + } + Py_DECREF(callback); } Py_DECREF(ctx); @@ -340,28 +379,17 @@ static PyObject* AIOContext_submit( int result = 0; for (i=0; i < nr; i++) { - if (ops[i]->in_progress) continue; - - // Claim the op (mark in_progress, set ctx, take references) only - // right before actually handing it to the pool - previously this - // was done for every argument up front, in the first loop above, - // even for ops this call was about to skip as already in_progress. - // That silently overwrote an in-flight op's ctx pointer with no - // matching incref, leaking the old Context's reference and leaving - // the original worker()'s eventual Py_DECREF(ctx) to decrement the - // wrong (new) Context instead - a real use-after-free/over-decref - // risk, not just a leak. A threadpool_add() failure below must - // also leave this op exactly as retryable as before this call, - // not permanently stuck in_progress=1 with no worker ever assigned - // to clear it. - ops[i]->in_progress = 1; + // Atomic exchange, not check-then-set: two Contexts racing on the same + // Operation must not both dispatch it to a worker. + if (CAIO_ATOMIC_LOAD_STORE(ops[i]->in_progress, 1)) continue; + ops[i]->ctx = (void*) self; Py_INCREF(ops[i]); Py_INCREF(self); result = threadpool_add(self->pool, worker, (void*) ops[i], 0); if (process_pool_error(result) < 0) { - ops[i]->in_progress = 0; + CAIO_ATOMIC_STORE(ops[i]->in_progress, 0); ops[i]->ctx = NULL; Py_DECREF(ops[i]); Py_DECREF(self); @@ -790,7 +818,10 @@ PyDoc_STRVAR(AIOOperation_get_value_docstring, static PyObject* AIOOperation_get_value( AIOOperation *self, PyObject *args, PyObject *kwds ) { - if (self->in_progress && !__atomic_load_n(&self->done, __ATOMIC_ACQUIRE)) { + if ( + CAIO_ATOMIC_LOAD(self->in_progress) && + !CAIO_ATOMIC_LOAD(self->done) + ) { PyErr_SetString( PyExc_RuntimeError, "get_value() is not available while the operation is in flight" @@ -844,6 +875,7 @@ static PyObject* AIOOperation_set_callback( static char *kwlist[] = {"callback", NULL}; PyObject* callback; + PyObject* old_callback; int argIsOk = PyArg_ParseTupleAndKeywords( args, kwds, "O", kwlist, @@ -862,14 +894,21 @@ static PyObject* AIOOperation_set_callback( } Py_INCREF(callback); + CAIO_BEGIN_CRITICAL_SECTION(self); + old_callback = self->callback; self->callback = callback; + CAIO_END_CRITICAL_SECTION(); + Py_XDECREF(old_callback); Py_RETURN_TRUE; } static PyObject *AIOOperation_payload_getter(AIOOperation *self, void *closure) { - if (self->in_progress && !__atomic_load_n(&self->done, __ATOMIC_ACQUIRE)) { + if ( + CAIO_ATOMIC_LOAD(self->in_progress) && + !CAIO_ATOMIC_LOAD(self->done) + ) { PyErr_SetString( PyExc_RuntimeError, "payload is not available while the operation is in flight" @@ -1009,6 +1048,10 @@ PyMODINIT_FUNC PyInit_thread_aio(void) { m = PyModule_Create(&thread_aio_module); if (m == NULL) return NULL; + if (CAIO_DECLARE_FREE_THREADED(m) < 0) { + Py_DECREF(m); + return NULL; + } if (PyType_Ready(&AIOContextType) < 0) return NULL; diff --git a/tests/test_free_threading.py b/tests/test_free_threading.py index 9c7b85b..272ae8d 100644 --- a/tests/test_free_threading.py +++ b/tests/test_free_threading.py @@ -1,7 +1,11 @@ +import gc import os +import select import sys import sysconfig import threading +import time +import weakref import pytest @@ -9,18 +13,14 @@ from caio import python_aio -def test_gil_stays_disabled_when_only_python_aio_available(): - """caio ships no Py_mod_gil declaration on any C extension, so importing - thread_aio/linux_aio/linux_uring under a free-threaded interpreter - silently re-enables the GIL (CPython's own safety fallback). When none - of them are importable here (e.g. built against a regular interpreter, - then run under a free-threaded one - different SOABI, so they simply - don't match), nothing should have flipped the GIL back on.""" +def test_importing_caio_does_not_enable_gil(): + """Every importable C backend must declare and support free threading.""" if not sysconfig.get_config_var("Py_GIL_DISABLED"): pytest.skip("not a free-threaded build") - if any((caio.thread_aio, caio.linux_aio, caio.linux_uring)): - pytest.skip("a C extension is available here too - GIL re-enable is expected") - assert sys._is_gil_enabled() is False + assert sys._is_gil_enabled() is False, ( + "importing caio enabled the GIL; at least one C extension has not " + "declared free-threading support" + ) def test_high_concurrency_stress_no_data_races(tmp_path): @@ -157,3 +157,755 @@ def submitter(index, context): for context in contexts: context.close() context.pool.join() + + +def _require_gil_disabled(): + if getattr(sys, "_is_gil_enabled", lambda: True)(): + pytest.skip("requires a free-threaded interpreter with the GIL disabled") + + +def _pump_contexts(contexts, predicate, timeout=10): + deadline = time.monotonic() + timeout + while not predicate(): + if time.monotonic() >= deadline: + raise TimeoutError("contexts did not complete in time") + for context in contexts: + if hasattr(context, "flush"): + context.flush() + for context in contexts: + if hasattr(context, "process_events"): + context.process_events() + time.sleep(0.001) + + +def _close_contexts(contexts): + for context in contexts: + close = getattr(context, "close", None) + if close is not None: + close() + for context in contexts: + pool = getattr(context, "pool", None) + if pool is not None: + pool.join() + + +@pytest.fixture( + params=[2, 4, 8, 16, 32], + ids=lambda value: f"submitters={value}", +) +def ft_submitter_count(request): + return request.param + + +@pytest.fixture(params=[1, 3, 17, 64], ids=lambda value: f"batch={value}") +def ft_operations_per_submit(request): + return request.param + + +@pytest.fixture(params=[1, 8], ids=lambda value: f"submit-rounds={value}") +def ft_submit_rounds(request): + return request.param + + +@pytest.fixture(params=[2, 4, 8], ids=lambda value: f"claimants={value}") +def ft_claimant_count(request): + return request.param + + +@pytest.fixture( + params=[257, 4_096, 20_000], + ids=lambda value: f"claims={value}", +) +def ft_claim_operation_count(request): + return request.param + + +@pytest.fixture( + params=[63, 256, 4_096], + ids=lambda value: f"operations={value}", +) +def ft_operation_count(request): + return request.param + + +@pytest.fixture( + params=[2, 4, 8, 16], + ids=lambda value: f"drainers={value}", +) +def ft_drainer_count(request): + return request.param + + +@pytest.fixture(params=[1, 3, 8, 32], ids=lambda value: f"rounds={value}") +def ft_drain_rounds(request): + return request.param + + +@pytest.fixture(params=[2, 8, 16], ids=lambda value: f"readers={value}") +def ft_reader_count(request): + return request.param + + +@pytest.fixture(params=[1, 257, 2_000], ids=lambda value: f"reads={value}") +def ft_reads_per_thread(request): + return request.param + + +@pytest.fixture(params=[2, 8, 16], ids=lambda value: f"observers={value}") +def ft_observer_count(request): + return request.param + + +@pytest.fixture( + params=[257, 2_000], + ids=lambda value: f"completions={value}", +) +def ft_completion_count(request): + return request.param + + +@pytest.fixture(params=[2, 8, 32], ids=lambda value: f"setters={value}") +def ft_callback_setter_count(request): + return request.param + + +@pytest.fixture(params=[1, 257, 2_000], ids=lambda value: f"sets={value}") +def ft_callback_sets_per_thread(request): + return request.param + + +@pytest.fixture(params=[2, 8, 32], ids=lambda value: f"cancellers={value}") +def ft_canceller_count(request): + return request.param + + +@pytest.fixture(params=[1, 8, 64], ids=lambda value: f"cancel-rounds={value}") +def ft_cancel_rounds(request): + return request.param + + +def test_same_operation_is_claimed_once_across_contexts_all_backends( + tmp_path, + backend, + ft_claimant_count, + ft_claim_operation_count, +): + """Every backend must synchronize a one-shot claim on the Operation. + + This is deliberately separate from the more aggressive python_aio test + above: it only uses the shared public API, so future free-threading C + backends run exactly the same cross-Context race. + """ + _require_gil_disabled() + + iterations = ft_claim_operation_count + claim_window = min(iterations, 512) + path = tmp_path / "cross-context-claim.bin" + fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o600) + contexts = tuple( + backend.Context(max_requests=claim_window) + for _ in range(ft_claimant_count) + ) + start = threading.Barrier(ft_claimant_count + 1, timeout=30) + finished = threading.Barrier(ft_claimant_count + 1, timeout=30) + current = [None] + submitted = [0] * ft_claimant_count + callback_count = [0] + callback_lock = threading.Lock() + errors = [] + errors_lock = threading.Lock() + + def on_done(_result): + with callback_lock: + callback_count[0] += 1 + + def submitter(index, context): + try: + for _ in range(iterations): + start.wait() + submitted[index] += context.submit(current[0]) + finished.wait() + except Exception as exc: # noqa: BLE001 (surface child-thread failures) + with errors_lock: + errors.append(exc) + start.abort() + finished.abort() + + threads = [ + threading.Thread(target=submitter, args=(index, context)) + for index, context in enumerate(contexts) + ] + + try: + for thread in threads: + thread.start() + + for iteration in range(iterations): + operation = backend.Operation.write(b"x", fd, 0) + operation.set_callback(on_done) + current[0] = operation + start.wait() + finished.wait() + if (iteration + 1) % claim_window == 0: + expected_callbacks = sum(submitted) + + def window_finished(expected=expected_callbacks): + with callback_lock: + return callback_count[0] >= expected + + _pump_contexts( + contexts, + window_finished, + timeout=20, + ) + + for thread in threads: + thread.join() + + assert not errors + accepted = sum(submitted) + + def all_callbacks_finished(): + with callback_lock: + return callback_count[0] >= accepted + + _pump_contexts( + contexts, + all_callbacks_finished, + timeout=20, + ) + assert accepted == iterations, ( + f"{accepted - iterations} Operations were submitted twice" + ) + assert callback_count[0] == iterations + finally: + start.abort() + finished.abort() + for thread in threads: + thread.join(timeout=5) + _close_contexts(contexts) + os.close(fd) + + +def test_concurrent_submits_to_one_context_all_backends( + tmp_path, + backend, + ft_submitter_count, + ft_operations_per_submit, + ft_submit_rounds, +): + """Distinct Operations submitted concurrently must not corrupt Context.""" + _require_gil_disabled() + + worker_count = ft_submitter_count + operations_per_worker = ft_operations_per_submit + submit_rounds = ft_submit_rounds + operations_per_worker_total = operations_per_worker * submit_rounds + total = worker_count * operations_per_worker_total + path = tmp_path / "one-context-submit.bin" + fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o600) + context = backend.Context(max_requests=total) + submit_start = threading.Barrier(worker_count, timeout=30) + operations = [] + callback_results = [None] * total + callback_lock = threading.Lock() + submitted = [0] * worker_count + errors = [] + errors_lock = threading.Lock() + + def make_callback(index, operation): + def callback(result): + with callback_lock: + callback_results[index] = (result, operation.get_value()) + return callback + + for index in range(total): + payload = index.to_bytes(4, "little") + operation = backend.Operation.write(payload, fd, index * 4) + operation.set_callback(make_callback(index, operation)) + operations.append(operation) + + def submitter(worker_index): + first = worker_index * operations_per_worker_total + try: + accepted = 0 + for round_index in range(submit_rounds): + batch_first = first + round_index * operations_per_worker + batch_last = batch_first + operations_per_worker + # Every round stages a batch after the same barrier. Multiple + # rounds vary scheduling and repeatedly collide inside the + # Context's SQ-tail read/write window. + submit_start.wait() + accepted += context.submit( + *operations[batch_first:batch_last], + ) + submitted[worker_index] = accepted + except Exception as exc: # noqa: BLE001 (surface child-thread failures) + with errors_lock: + errors.append(exc) + submit_start.abort() + + threads = [ + threading.Thread(target=submitter, args=(index,)) + for index in range(worker_count) + ] + + try: + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=30) + + assert all(not thread.is_alive() for thread in threads) + assert not errors + assert sum(submitted) == total + + def all_callbacks_finished(): + with callback_lock: + return all( + result is not None + for result in callback_results + ) + + _pump_contexts( + (context,), + all_callbacks_finished, + timeout=20, + ) + assert callback_results == [(4, 4)] * total + + os.lseek(fd, 0, os.SEEK_SET) + expected = b"".join(index.to_bytes(4, "little") for index in range(total)) + assert os.read(fd, len(expected)) == expected + finally: + submit_start.abort() + for thread in threads: + thread.join(timeout=5) + _close_contexts((context,)) + os.close(fd) + + +def test_concurrent_process_events_delivers_each_completion_once( + tmp_path, + polling_backend, + ft_operation_count, + ft_drainer_count, + ft_drain_rounds, +): + """Concurrent drainers must never consume or callback one event twice.""" + _require_gil_disabled() + + operation_count = ft_operation_count + drainer_count = ft_drainer_count + drain_rounds = ft_drain_rounds + path = tmp_path / "concurrent-process-events.bin" + fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o600) + context = polling_backend.Context(max_requests=operation_count) + callback_counts = [0] * operation_count + callback_lock = threading.Lock() + drain_round = threading.Barrier(drainer_count, timeout=30) + errors = [] + errors_lock = threading.Lock() + + def make_callback(index): + def callback(_result): + with callback_lock: + callback_counts[index] += 1 + return callback + + operations = [] + for index in range(operation_count): + operation = polling_backend.Operation.write(b"x", fd, index) + operation.set_callback(make_callback(index)) + operations.append(operation) + + assert context.submit(*operations) == operation_count + is_sqpoll = bool(getattr(context, "sqpoll", False)) + if hasattr(context, "flush"): + context.flush() + if is_sqpoll: + with callback_lock: + completed_inline = sum(callback_counts) + if completed_inline < operation_count: + # flush() above wakes a sleeping SQPOLL thread and may drain + # already-ready CQEs. If work remains, wait until eventfd says + # the kernel has populated more of the CQ, but leave that CQ + # untouched for the synchronized drainers below. + readable, _, _ = select.select([context.fileno], [], [], 10) + assert readable, "SQPOLL produced no completion notification" + time.sleep(0.01) + + def drain(): + try: + for _ in range(drain_rounds): + # Force all drainers to enter every round together. Without + # this barrier a fast first thread can consume the whole CQ + # before the others even start, accidentally serializing the + # test and hiding a duplicate cq_head read/commit. + drain_round.wait() + context.process_events( + max_requests=operation_count, + min_requests=0, + timeout=0, + ) + except Exception as exc: # noqa: BLE001 (surface child-thread failures) + with errors_lock: + errors.append(exc) + drain_round.abort() + + threads = [threading.Thread(target=drain) for _ in range(drainer_count)] + + try: + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=30) + + assert all(not thread.is_alive() for thread in threads) + assert not errors + + def all_callbacks_finished(): + with callback_lock: + return sum(callback_counts) >= operation_count + + # Fixed synchronized rounds should normally drain everything. Finish + # any genuinely late kernel completions serially so the assertion + # below diagnoses duplicate delivery, not storage latency. + _pump_contexts((context,), all_callbacks_finished, timeout=20) + assert callback_counts == [1] * operation_count + finally: + drain_round.abort() + for thread in threads: + thread.join(timeout=5) + os.close(fd) + + +def test_completed_operation_supports_concurrent_readers( + tmp_path, + backend, + ft_reader_count, + ft_reads_per_thread, +): + """C backends must safely return owned result references to all threads.""" + _require_gil_disabled() + + payload = bytes(range(256)) * 16 + path = tmp_path / "concurrent-result-readers.bin" + path.write_bytes(payload) + fd = os.open(path, os.O_RDONLY) + context = backend.Context(max_requests=8) + operation = backend.Operation.read(len(payload), fd, 0) + completed = threading.Event() + operation.set_callback(lambda _result: completed.set()) + assert context.submit(operation) == 1 + _pump_contexts((context,), completed.is_set) + assert operation.get_value() == payload + + reader_count = ft_reader_count + reads_per_thread = ft_reads_per_thread + start = threading.Barrier(reader_count + 1, timeout=30) + errors = [] + errors_lock = threading.Lock() + + def read_result(): + try: + start.wait() + for _ in range(reads_per_thread): + if operation.get_value() != payload: + raise AssertionError("concurrent get_value() returned wrong data") + except Exception as exc: # noqa: BLE001 (surface child-thread failures) + with errors_lock: + errors.append(exc) + start.abort() + + threads = [ + threading.Thread(target=read_result) + for _ in range(reader_count) + ] + + try: + for thread in threads: + thread.start() + start.wait() + for thread in threads: + thread.join(timeout=30) + + assert all(not thread.is_alive() for thread in threads) + assert not errors + finally: + start.abort() + for thread in threads: + thread.join(timeout=5) + _close_contexts((context,)) + os.close(fd) + + +def test_payload_access_is_safe_while_worker_publishes_completion( + tmp_path, + pooled_backend, + ft_observer_count, + ft_completion_count, +): + """Publishing ``done`` must not expose a buffer being cleared concurrently. + + In thread_aio the worker used to release-store ``done = 1`` before + acquiring a Python thread state and clearing a completed write's + ``py_buffer``. A free-running reader could therefore pass the in-flight + check and Py_INCREF the same pointer while the worker Py_DECREFed it. + """ + _require_gil_disabled() + + operation_count = ft_completion_count + observer_count = ft_observer_count + path = tmp_path / "payload-completion-race.bin" + fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o600) + context = pooled_backend.Context( + max_requests=operation_count + 1, + pool_size=1, + ) + + blocker_started = threading.Event() + blocker_release = threading.Event() + blocker = pooled_backend.Operation.fsync(fd) + blocker.set_callback( + lambda _result: ( + blocker_started.set(), + blocker_release.wait(30), + ), + ) + assert context.submit(blocker) == 1 + assert blocker_started.wait(10), "worker did not enter blocker callback" + + completed_count = [0] + completed_lock = threading.Lock() + all_completed = threading.Event() + operations = [] + + def on_done(_result): + # thread_aio publishes done before invoking this callback and clears + # the write buffer immediately after it returns. Yielding here makes + # observers repeatedly enter payload/get_value in that exact state + # and then overlap the buffer's ownership transition. + time.sleep(0.001) + with completed_lock: + completed_count[0] += 1 + if completed_count[0] == operation_count: + all_completed.set() + + for index in range(operation_count): + operation = pooled_backend.Operation.write(b"x", fd, index) + operation.set_callback(on_done) + operations.append(operation) + + assert context.submit(*operations) == operation_count + + observer_start = threading.Barrier(observer_count + 1, timeout=30) + errors = [] + errors_lock = threading.Lock() + + def observe(): + try: + observer_start.wait() + while not all_completed.is_set(): + for operation in operations: + try: + payload = operation.payload + if payload is not None: + bytes(payload) + except RuntimeError: + pass + + try: + operation.get_value() + except RuntimeError: + pass + except Exception as exc: # noqa: BLE001 (surface child-thread failures) + with errors_lock: + errors.append(exc) + observer_start.abort() + + observers = [ + threading.Thread(target=observe) + for _ in range(observer_count) + ] + + try: + for observer in observers: + observer.start() + observer_start.wait() + blocker_release.set() + assert all_completed.wait(30), "worker did not finish queued operations" + for observer in observers: + observer.join(timeout=10) + + assert all(not observer.is_alive() for observer in observers) + assert not errors + assert completed_count[0] == operation_count + finally: + blocker_release.set() + observer_start.abort() + for observer in observers: + observer.join(timeout=5) + _close_contexts((context,)) + os.close(fd) + + +def test_concurrent_callback_replacement_releases_every_old_callback( + backend, + ft_callback_setter_count, + ft_callback_sets_per_thread, +): + """set_callback must atomically replace and release its owned reference.""" + _require_gil_disabled() + + setter_count = ft_callback_setter_count + sets_per_thread = ft_callback_sets_per_thread + operation = backend.Operation.fsync(0) + start = threading.Barrier(setter_count + 1, timeout=30) + errors = [] + errors_lock = threading.Lock() + + class Callback: + def __call__(self, _result): + return None + + callbacks = [ + Callback() + for _ in range(setter_count * sets_per_thread) + ] + callback_refs = [weakref.ref(callback) for callback in callbacks] + + def replace_callbacks(thread_index): + try: + start.wait() + first = thread_index * sets_per_thread + for index in range(first, first + sets_per_thread): + assert operation.set_callback(callbacks[index]) is True + except Exception as exc: # noqa: BLE001 (surface child-thread failures) + with errors_lock: + errors.append(exc) + start.abort() + + threads = [ + threading.Thread(target=replace_callbacks, args=(index,)) + for index in range(setter_count) + ] + + try: + for thread in threads: + thread.start() + start.wait() + for thread in threads: + thread.join(timeout=30) + + assert all(not thread.is_alive() for thread in threads) + assert not errors + + # Move the Operation off every callback in the contested set. Each + # set_callback owns exactly one reference and must release the old + # one, even when several threads replace the same slot concurrently. + operation.set_callback(Callback()) + callbacks.clear() + gc.collect() + leaked = sum(ref() is not None for ref in callback_refs) + assert leaked == 0, f"{leaked} replaced callbacks are still referenced" + finally: + start.abort() + for thread in threads: + thread.join(timeout=5) + + +def test_concurrent_uring_cancel_requests_do_not_corrupt_submission_ring( + ft_canceller_count, + ft_cancel_rounds, +): + """cancel and submit share io_uring's SQ tail and must share its lock.""" + _require_gil_disabled() + if caio.linux_uring is None: + pytest.skip("linux_uring backend is unavailable") + + canceller_count = ft_canceller_count + cancel_rounds = ft_cancel_rounds + operation_count = canceller_count * cancel_rounds + context = caio.linux_uring.Context( + max_requests=operation_count * 2, + sqpoll=False, + ) + read_fd, write_fd = os.pipe() + callback_counts = [0] * operation_count + callback_lock = threading.Lock() + + def make_callback(index): + def callback(_result): + with callback_lock: + callback_counts[index] += 1 + return callback + + targets = [ + caio.linux_uring.Operation.read( + 1, + read_fd, + (1 << 64) - 1, + ) + for _ in range(operation_count) + ] + for index, operation in enumerate(targets): + operation.set_callback(make_callback(index)) + assert context.submit(*targets) == operation_count + context.flush() + + start_round = threading.Barrier(canceller_count, timeout=30) + errors = [] + errors_lock = threading.Lock() + cancelled = [0] * canceller_count + + def cancel(thread_index): + try: + for round_index in range(cancel_rounds): + start_round.wait() + cancelled[thread_index] += context.cancel( + targets[round_index * canceller_count + thread_index], + ) + except Exception as exc: # noqa: BLE001 (surface child-thread failures) + with errors_lock: + errors.append(exc) + start_round.abort() + + threads = [ + threading.Thread(target=cancel, args=(index,)) + for index in range(canceller_count) + ] + + try: + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=30) + + assert all(not thread.is_alive() for thread in threads) + assert not errors + assert cancelled == [0] * canceller_count + + def all_targets_completed(): + with callback_lock: + return sum(callback_counts) >= operation_count + + _pump_contexts((context,), all_targets_completed, timeout=10) + assert callback_counts == [1] * operation_count + finally: + start_round.abort() + for thread in threads: + thread.join(timeout=5) + # Release any read whose cancel SQE was lost so Context teardown + # cannot leave the kernel holding pointers to live Operations. + try: + os.write(write_fd, b"x" * operation_count) + _pump_contexts( + (context,), + lambda: sum(callback_counts) >= operation_count, + timeout=5, + ) + except (BrokenPipeError, TimeoutError): + pass + os.close(write_fd) + os.close(read_fd) From 6f48756222d7123b520f0c4d53388285b7a1621a Mon Sep 17 00:00:00 2001 From: Dmitry Orlov Date: Tue, 4 Aug 2026 11:05:03 +0200 Subject: [PATCH 4/8] Test and ship free-threaded builds in CI and releases CI's 3.13t/3.14t jobs now force PYTHON_GIL=0 so they actually exercise no-GIL concurrency for every backend, not just build-and-import. The release workflow, Makefile, and make-wheels.sh build cp313t/cp314t wheels alongside the existing versions on every platform that builds C extensions at all (Windows ships pure-Python only, already version-agnostic). macOS switches from actions/setup-python to uv for interpreter provisioning, since it's the one already confirmed to handle free-threaded version strings correctly. --- .github/workflows/ci.yml | 13 ++++++++----- .github/workflows/publish.yml | 18 ++++++++---------- Makefile | 12 +++++++++++- scripts/make-wheels.sh | 2 ++ 4 files changed, 29 insertions(+), 16 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 660b617..2f1d0fb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -44,15 +44,17 @@ jobs: os: ubuntu-latest - python: "3.14" os: ubuntu-latest - # Fresh build directly under a free-threaded interpreter - none of - # the C extensions declare Py_mod_gil support yet, so importing - # one (if it compiles here at all) re-enables the GIL for the - # whole process. See free-threaded-python-only below for the - # actual no-GIL check. + # Fresh build under a free-threaded interpreter. None of the C + # extensions declare Py_mod_gil yet, so importing one re-enables + # the GIL by default - PYTHON_GIL=0 below keeps it off anyway so + # the suite actually exercises real no-GIL concurrency (their + # thread-safety is what tests/test_free_threading.py checks). - python: "3.13t" os: ubuntu-latest + free_threaded: true - python: "3.14t" os: ubuntu-latest + free_threaded: true - python: "3.10" os: windows-latest - python: "3.11" @@ -84,6 +86,7 @@ jobs: COVERALLS_PARALLEL: 'true' COVERALLS_SERVICE_NAME: github GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + PYTHON_GIL: ${{ matrix.free_threaded && '0' || '' }} - name: Report coverage run: uv run coveralls continue-on-error: true diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index aff50be..40cac43 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -35,7 +35,7 @@ jobs: contents: write strategy: matrix: - python: [cp310-cp310, cp311-cp311, cp312-cp312, cp313-cp313, cp314-cp314] + python: [cp310-cp310, cp311-cp311, cp312-cp312, cp313-cp313, cp314-cp314, cp313-cp313t, cp314-cp314t] steps: - uses: actions/checkout@v4 - name: Set version from release tag @@ -61,7 +61,7 @@ jobs: contents: write strategy: matrix: - python: [cp310-cp310, cp311-cp311, cp312-cp312, cp313-cp313, cp314-cp314] + python: [cp310-cp310, cp311-cp311, cp312-cp312, cp313-cp313, cp314-cp314, cp313-cp313t, cp314-cp314t] steps: - uses: actions/checkout@v4 - name: Set version from release tag @@ -87,7 +87,7 @@ jobs: contents: write strategy: matrix: - python: [cp310-cp310, cp311-cp311, cp312-cp312, cp313-cp313, cp314-cp314] + python: [cp310-cp310, cp311-cp311, cp312-cp312, cp313-cp313, cp314-cp314, cp313-cp313t, cp314-cp314t] steps: - uses: actions/checkout@v4 - name: Set version from release tag @@ -124,7 +124,7 @@ jobs: contents: write strategy: matrix: - python: [cp310-cp310, cp311-cp311, cp312-cp312, cp313-cp313, cp314-cp314] + python: [cp310-cp310, cp311-cp311, cp312-cp312, cp313-cp313, cp314-cp314, cp313-cp313t, cp314-cp314t] steps: - uses: actions/checkout@v4 - uses: astral-sh/setup-uv@v5 @@ -154,17 +154,15 @@ jobs: contents: write strategy: matrix: - python: ["3.10", "3.11", "3.12", "3.13", "3.14"] + python: ["3.10", "3.11", "3.12", "3.13", "3.14", "3.13t", "3.14t"] steps: - uses: actions/checkout@v4 - uses: astral-sh/setup-uv@v5 - - name: Set version from release tag - run: uv version --frozen "${GITHUB_REF_NAME#v}" - - uses: actions/setup-python@v5 with: python-version: "${{ matrix.python }}" - - run: pip install build - - run: python -m build --wheel + - name: Set version from release tag + run: uv version --frozen "${GITHUB_REF_NAME#v}" + - run: uv build --wheel - uses: actions/upload-artifact@v4 with: name: wheel-macos-${{ matrix.python }} diff --git a/Makefile b/Makefile index cb9a126..2a2e05a 100644 --- a/Makefile +++ b/Makefile @@ -28,12 +28,22 @@ sdist: python3.14 -m venv $@ $@/bin/python -m pip install -U pip setuptools build wheel -mac_wheel: .venvs/3.10 .venvs/3.11 .venvs/3.12 .venvs/3.13 .venvs/3.14 +.venvs/3.13t: .venvs + python3.13t -m venv $@ + $@/bin/python -m pip install -U pip setuptools build wheel + +.venvs/3.14t: .venvs + python3.14t -m venv $@ + $@/bin/python -m pip install -U pip setuptools build wheel + +mac_wheel: .venvs/3.10 .venvs/3.11 .venvs/3.12 .venvs/3.13 .venvs/3.14 .venvs/3.13t .venvs/3.14t .venvs/3.10/bin/python -m build .venvs/3.11/bin/python -m build .venvs/3.12/bin/python -m build .venvs/3.13/bin/python -m build .venvs/3.14/bin/python -m build + .venvs/3.13t/bin/python -m build + .venvs/3.14t/bin/python -m build linux_wheel: docker run -it --rm \ diff --git a/scripts/make-wheels.sh b/scripts/make-wheels.sh index c79469b..5294887 100644 --- a/scripts/make-wheels.sh +++ b/scripts/make-wheels.sh @@ -14,6 +14,8 @@ build_wheel cp311-cp311 build_wheel cp312-cp312 build_wheel cp313-cp313 build_wheel cp314-cp314 +build_wheel cp313-cp313t +build_wheel cp314-cp314t cd dist From f696c0372070b8f4bd94ad0c925806066af71784 Mon Sep 17 00:00:00 2001 From: Dmitry Orlov Date: Tue, 4 Aug 2026 11:20:10 +0200 Subject: [PATCH 5/8] Fix Windows test failure and a real lock-identity race in python_aio test_high_concurrency_stress_no_data_races failed on Windows CI at a constant "chunk 10" - the only chunk whose entire payload is the raw byte 0x0A ('\n'). os.open() defaults to text mode on Windows without os.O_BINARY, so every '\n' silently became '\r\n' on write and back on read - not a race at all, just a missing flag. Fixed in the test. While investigating, found and fixed a real (if harder to trigger) bug along the way: python_aio's Windows-only pread/pwrite fallback locked per-fd via `self._locks[fd]` on a `defaultdict(RLock)` - two threads racing on the same fd's first access could each create and store their own RLock instance, ending up serialized against two different objects instead of each other. Replaced with an explicit get-or-create under the (previously declared but unused) _locks_cleaner lock, and switched the dict to a WeakValueDictionary so a long-lived Context doesn't accumulate one RLock per fd forever. Verified directly on the Windows box this originally failed on: reproduced pre-fix, clean across 5 repeated runs post-fix. --- caio/python_aio.py | 24 +++++++++++++++++++----- tests/test_free_threading.py | 16 ++++++++++------ 2 files changed, 29 insertions(+), 11 deletions(-) diff --git a/caio/python_aio.py b/caio/python_aio.py index 9eaef3b..3b3bb1f 100644 --- a/caio/python_aio.py +++ b/caio/python_aio.py @@ -2,13 +2,13 @@ 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 weakref import WeakValueDictionary from .abstract import AbstractContext, AbstractOperation @@ -63,8 +63,8 @@ def __init__(self, max_requests: int = 32, pool_size: int = 8): self._state = ContextState.OPEN if not NATIVE_PREAD_PWRITE: - self._locks_cleaner = RLock() # type: ignore - self._locks = defaultdict(RLock) # type: ignore + self._locks_cleaner = Lock() + self._locks: WeakValueDictionary = WeakValueDictionary() @property def max_requests(self) -> int: @@ -176,14 +176,28 @@ def __pread(self, 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._locks[fd]: + 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._locks[fd]: + with self.__fd_lock(fd): os.lseek(fd, 0, os.SEEK_SET) os.lseek(fd, offset, os.SEEK_SET) return os.write(fd, bytes) diff --git a/tests/test_free_threading.py b/tests/test_free_threading.py index 272ae8d..7278747 100644 --- a/tests/test_free_threading.py +++ b/tests/test_free_threading.py @@ -12,6 +12,10 @@ import caio from caio import python_aio +# os.open() defaults to text mode on Windows without this - \n <-> \r\n +# translation would corrupt any payload containing a raw \n byte. +_O_BINARY = getattr(os, "O_BINARY", 0) + def test_importing_caio_does_not_enable_gil(): """Every importable C backend must declare and support free threading.""" @@ -37,7 +41,7 @@ def test_high_concurrency_stress_no_data_races(tmp_path): chunk = 4096 path = tmp_path / "stress.bin" path.write_bytes(b"\x00" * (count * chunk)) - fd = os.open(str(path), os.O_RDWR) + fd = os.open(str(path), os.O_RDWR | _O_BINARY) try: ctx = python_aio.Context(max_requests=count, pool_size=32) expected = [bytes([i % 256]) * chunk for i in range(count)] @@ -301,7 +305,7 @@ def test_same_operation_is_claimed_once_across_contexts_all_backends( iterations = ft_claim_operation_count claim_window = min(iterations, 512) path = tmp_path / "cross-context-claim.bin" - fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o600) + fd = os.open(path, os.O_RDWR | os.O_CREAT | _O_BINARY, 0o600) contexts = tuple( backend.Context(max_requests=claim_window) for _ in range(ft_claimant_count) @@ -403,7 +407,7 @@ def test_concurrent_submits_to_one_context_all_backends( operations_per_worker_total = operations_per_worker * submit_rounds total = worker_count * operations_per_worker_total path = tmp_path / "one-context-submit.bin" - fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o600) + fd = os.open(path, os.O_RDWR | os.O_CREAT | _O_BINARY, 0o600) context = backend.Context(max_requests=total) submit_start = threading.Barrier(worker_count, timeout=30) operations = [] @@ -499,7 +503,7 @@ def test_concurrent_process_events_delivers_each_completion_once( drainer_count = ft_drainer_count drain_rounds = ft_drain_rounds path = tmp_path / "concurrent-process-events.bin" - fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o600) + fd = os.open(path, os.O_RDWR | os.O_CREAT | _O_BINARY, 0o600) context = polling_backend.Context(max_requests=operation_count) callback_counts = [0] * operation_count callback_lock = threading.Lock() @@ -592,7 +596,7 @@ def test_completed_operation_supports_concurrent_readers( payload = bytes(range(256)) * 16 path = tmp_path / "concurrent-result-readers.bin" path.write_bytes(payload) - fd = os.open(path, os.O_RDONLY) + fd = os.open(path, os.O_RDONLY | _O_BINARY) context = backend.Context(max_requests=8) operation = backend.Operation.read(len(payload), fd, 0) completed = threading.Event() @@ -658,7 +662,7 @@ def test_payload_access_is_safe_while_worker_publishes_completion( operation_count = ft_completion_count observer_count = ft_observer_count path = tmp_path / "payload-completion-race.bin" - fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o600) + fd = os.open(path, os.O_RDWR | os.O_CREAT | _O_BINARY, 0o600) context = pooled_backend.Context( max_requests=operation_count + 1, pool_size=1, From fa9d465c91519a38fa7f19cc91f2ef42d4a15be4 Mon Sep 17 00:00:00 2001 From: Dmitry Orlov Date: Tue, 4 Aug 2026 11:34:30 +0200 Subject: [PATCH 6/8] Add GIL vs free-threaded benchmark comparison plot_results.py now loads two runs (bench_all.csv.gz / bench_all_nogil.csv.gz) and stacks the GIL/no-GIL charts for each figure, so the actual throughput and latency impact of running under a genuinely free-threaded interpreter is visible directly alongside the normal GIL baseline. Charts moved under benchmark/results/ alongside the raw data they're generated from. --- .gitattributes | 3 +- benchmark/chunk_sweep_latency_rand.png | 3 - benchmark/chunk_sweep_latency_seq.png | 3 - benchmark/chunk_sweep_throughput_rand.png | 3 - benchmark/chunk_sweep_throughput_seq.png | 3 - benchmark/concurrency_sweep_latency_rand.png | 3 - benchmark/concurrency_sweep_latency_seq.png | 3 - .../concurrency_sweep_throughput_rand.png | 3 - .../concurrency_sweep_throughput_seq.png | 3 - benchmark/latency_histograms.png | 3 - benchmark/plot_results.py | 279 ++++++++++++++---- benchmark/results/bench_all.csv.gz | 4 +- benchmark/results/bench_all_nogil.csv.gz | 3 + .../results/chunk_sweep_latency_rand.png | 3 + benchmark/results/chunk_sweep_latency_seq.png | 3 + .../results/chunk_sweep_throughput_rand.png | 3 + .../results/chunk_sweep_throughput_seq.png | 3 + .../concurrency_sweep_latency_rand.png | 3 + .../results/concurrency_sweep_latency_seq.png | 3 + .../concurrency_sweep_throughput_rand.png | 3 + .../concurrency_sweep_throughput_seq.png | 3 + benchmark/results/latency_histograms.png | 3 + benchmark/uv.lock | 5 +- 23 files changed, 256 insertions(+), 92 deletions(-) delete mode 100644 benchmark/chunk_sweep_latency_rand.png delete mode 100644 benchmark/chunk_sweep_latency_seq.png delete mode 100644 benchmark/chunk_sweep_throughput_rand.png delete mode 100644 benchmark/chunk_sweep_throughput_seq.png delete mode 100644 benchmark/concurrency_sweep_latency_rand.png delete mode 100644 benchmark/concurrency_sweep_latency_seq.png delete mode 100644 benchmark/concurrency_sweep_throughput_rand.png delete mode 100644 benchmark/concurrency_sweep_throughput_seq.png delete mode 100644 benchmark/latency_histograms.png create mode 100644 benchmark/results/bench_all_nogil.csv.gz create mode 100644 benchmark/results/chunk_sweep_latency_rand.png create mode 100644 benchmark/results/chunk_sweep_latency_seq.png create mode 100644 benchmark/results/chunk_sweep_throughput_rand.png create mode 100644 benchmark/results/chunk_sweep_throughput_seq.png create mode 100644 benchmark/results/concurrency_sweep_latency_rand.png create mode 100644 benchmark/results/concurrency_sweep_latency_seq.png create mode 100644 benchmark/results/concurrency_sweep_throughput_rand.png create mode 100644 benchmark/results/concurrency_sweep_throughput_seq.png create mode 100644 benchmark/results/latency_histograms.png diff --git a/.gitattributes b/.gitattributes index 95acbf9..9577472 100644 --- a/.gitattributes +++ b/.gitattributes @@ -1,2 +1,3 @@ benchmark/*.png filter=lfs diff=lfs merge=lfs -text -benchmark/results/bench_all.csv.gz filter=lfs diff=lfs merge=lfs -text +benchmark/results/*.png filter=lfs diff=lfs merge=lfs -text +benchmark/results/*.csv.gz filter=lfs diff=lfs merge=lfs -text diff --git a/benchmark/chunk_sweep_latency_rand.png b/benchmark/chunk_sweep_latency_rand.png deleted file mode 100644 index a7899ee..0000000 --- a/benchmark/chunk_sweep_latency_rand.png +++ /dev/null @@ -1,3 +0,0 @@ -version https://git-lfs.github.com/spec/v1 -oid sha256:6750b9152b457231248b1531afa8043eb6ef095182840f9fc783e7891d74ec4e -size 297775 diff --git a/benchmark/chunk_sweep_latency_seq.png b/benchmark/chunk_sweep_latency_seq.png deleted file mode 100644 index 9b893d7..0000000 --- a/benchmark/chunk_sweep_latency_seq.png +++ /dev/null @@ -1,3 +0,0 @@ -version https://git-lfs.github.com/spec/v1 -oid sha256:f860d1b86fdd9d2860a44832056d454c111041f3f19395377bf72c365ac893f9 -size 294864 diff --git a/benchmark/chunk_sweep_throughput_rand.png b/benchmark/chunk_sweep_throughput_rand.png deleted file mode 100644 index 7b095d5..0000000 --- a/benchmark/chunk_sweep_throughput_rand.png +++ /dev/null @@ -1,3 +0,0 @@ -version https://git-lfs.github.com/spec/v1 -oid sha256:b378e33d1510b9fa620b7f03c477a79be8de5fb4ce1a39966b449ae86b64945e -size 148823 diff --git a/benchmark/chunk_sweep_throughput_seq.png b/benchmark/chunk_sweep_throughput_seq.png deleted file mode 100644 index 29800f6..0000000 --- a/benchmark/chunk_sweep_throughput_seq.png +++ /dev/null @@ -1,3 +0,0 @@ -version https://git-lfs.github.com/spec/v1 -oid sha256:3c8636aa9446cca65c1bce9dc02d2e76b88f035a681c1eac1d64511be67e5371 -size 149952 diff --git a/benchmark/concurrency_sweep_latency_rand.png b/benchmark/concurrency_sweep_latency_rand.png deleted file mode 100644 index 8f5c145..0000000 --- a/benchmark/concurrency_sweep_latency_rand.png +++ /dev/null @@ -1,3 +0,0 @@ -version https://git-lfs.github.com/spec/v1 -oid sha256:4d985e01dd20b692382e164e597097b0cca5ccf73aeff49dd618c97c08d17d75 -size 350573 diff --git a/benchmark/concurrency_sweep_latency_seq.png b/benchmark/concurrency_sweep_latency_seq.png deleted file mode 100644 index 09d62fb..0000000 --- a/benchmark/concurrency_sweep_latency_seq.png +++ /dev/null @@ -1,3 +0,0 @@ -version https://git-lfs.github.com/spec/v1 -oid sha256:1617d9604cbf92fbf451d6a06531a64e71199cb83e333d2101f11396c17e821e -size 345652 diff --git a/benchmark/concurrency_sweep_throughput_rand.png b/benchmark/concurrency_sweep_throughput_rand.png deleted file mode 100644 index 058f7f5..0000000 --- a/benchmark/concurrency_sweep_throughput_rand.png +++ /dev/null @@ -1,3 +0,0 @@ -version https://git-lfs.github.com/spec/v1 -oid sha256:7dccef412d17b0d5f06b51524a0cc82375e9fb0475da7ba26e75ad3a7ee9ccf8 -size 166242 diff --git a/benchmark/concurrency_sweep_throughput_seq.png b/benchmark/concurrency_sweep_throughput_seq.png deleted file mode 100644 index bafcafa..0000000 --- a/benchmark/concurrency_sweep_throughput_seq.png +++ /dev/null @@ -1,3 +0,0 @@ -version https://git-lfs.github.com/spec/v1 -oid sha256:046b41fb77df09f37b57a4959781d51c2051b93633d0b65ccbfebeac20a9fb36 -size 165142 diff --git a/benchmark/latency_histograms.png b/benchmark/latency_histograms.png deleted file mode 100644 index bc22447..0000000 --- a/benchmark/latency_histograms.png +++ /dev/null @@ -1,3 +0,0 @@ -version https://git-lfs.github.com/spec/v1 -oid sha256:c65fe6aed023eaa54599cf432b500a1a8690cdb4686f64c97c7494b9197efd5e -size 126937 diff --git a/benchmark/plot_results.py b/benchmark/plot_results.py index c88bd38..cfa8103 100644 --- a/benchmark/plot_results.py +++ b/benchmark/plot_results.py @@ -1,6 +1,5 @@ -#!/usr/bin/env python3 """ -Plot caio benchmark results from bench_all.csv. +Plot and compare caio benchmark results with and without the GIL. Produces figures saved to CAIO_RESULTS dir (once per access pattern: rand/seq): concurrency_sweep_throughput_{rand,seq}.png @@ -10,24 +9,42 @@ latency_histograms.png (rand only) Usage: - CAIO_RESULTS=/tmp/results PLOT_RESULTS=/tmp/results \ + CAIO_RESULTS_GIL=/tmp/results-gil \ + CAIO_RESULTS_NOGIL=/tmp/results-nogil \ + PLOT_RESULTS=/tmp/results \ uv run --with matplotlib --with numpy python plot_results.py """ import csv +import gzip import os import pathlib from collections import defaultdict -from typing import Any, DefaultDict, Dict, List +from typing import Any import matplotlib + matplotlib.use("Agg") import matplotlib.pyplot as plt -import matplotlib.ticker as ticker import numpy as np - -PLOT_RESULTS_DIR = pathlib.Path(os.environ.get("PLOT_RESULTS", ".")) -RESULTS_DIR = pathlib.Path(os.environ.get("CAIO_RESULTS", "/tmp/results")) -CSV_PATH = RESULTS_DIR / "bench_all.csv" +from matplotlib import ticker +from PIL import Image + +SCRIPT_DIR = pathlib.Path(__file__).parent +RESULTS_DIR = SCRIPT_DIR / "results" +PLOT_RESULTS_DIR = pathlib.Path(os.environ.get("PLOT_RESULTS", RESULTS_DIR)) +CSV_GIL_PATH = ( + pathlib.Path(os.environ["CAIO_RESULTS_GIL"]) / "bench_all.csv" + if "CAIO_RESULTS_GIL" in os.environ + else RESULTS_DIR / "bench_all.csv.gz" +) +CSV_NOGIL_PATH = ( + pathlib.Path(os.environ["CAIO_RESULTS_NOGIL"]) / "bench_all.csv" + if "CAIO_RESULTS_NOGIL" in os.environ + else RESULTS_DIR / "bench_all_nogil.csv.gz" +) + +OUTPUT_SUFFIX = "" +MODE_LABEL = "" BACKENDS = [ "linux_uring", @@ -43,33 +60,36 @@ # ── data loading ────────────────────────────────────────────────────────────── -Row = Dict[str, str] +Row = dict[str, str] -def load_csv() -> List[Row]: - with open(CSV_PATH) as f: - return list(csv.DictReader(f)) +def load_csv(path: pathlib.Path) -> list[Row]: + if path.suffix == ".gz": + with gzip.open(path, mode="rt", newline="") as fp: + return list(csv.DictReader(fp)) + with path.open(mode="r", newline="") as fp: + return list(csv.DictReader(fp)) -def pct(values: List[float], p: float) -> float: +def pct(values: list[float], p: float) -> float: s = sorted(values) return s[min(int(len(s) * p), len(s) - 1)] -CellData = Dict[str, Any] # keys: lats (List[float]), wall_s (float), n_ops (int) +CellData = dict[str, Any] # keys: lats (List[float]), wall_s (float), n_ops (int) def group( - rows: List[Row], + rows: list[Row], sweep: str, op: str, -) -> DefaultDict[str, DefaultDict[int, CellData]]: +) -> defaultdict[str, defaultdict[int, CellData]]: """Returns {backend: {pivot_value: {lats: [...], wall_s: float, n_ops: int}}}. `sweep` is the full sweep tag, e.g. 'conc_sweep_rand' or 'chunk_sweep_seq'. The pivot key is concurrency for conc sweeps, chunk_bytes for chunk sweeps. """ is_conc = sweep.startswith("conc_sweep") - result: DefaultDict[str, DefaultDict[int, CellData]] = defaultdict( + result: defaultdict[str, defaultdict[int, CellData]] = defaultdict( lambda: defaultdict(lambda: {"lats": [], "wall_s": 0.0, "n_ops": 0}), ) for r in rows: @@ -101,6 +121,71 @@ def apply_style(ax, xlabel: str, ylabel: str, title: str, ax.spines["right"].set_visible(False) +def _mode_title(title: str) -> str: + if not MODE_LABEL: + return title + return f"{title} — {MODE_LABEL}" + + +def _output_path(name: str) -> pathlib.Path: + stem, suffix = os.path.splitext(name) + return PLOT_RESULTS_DIR / f"{stem}{OUTPUT_SUFFIX}{suffix}" + + +def _percentile_footer( + ax, + rows: list[Row], + sweep: str, + backend: str, +) -> None: + """Compact aggregate latency summary below every subplot.""" + lines = [] + for op in ("read", "write"): + values = [ + float(row["latency_us"]) / 1000 + for row in rows + if row["sweep"] == sweep + and row["backend"] == backend + and row["op"] == op + and row.get("latency_us") + ] + if not values: + continue + stats = " ".join( + f"p{int(quantile * 100)} {pct(values, quantile):.3f}" + for quantile in (0.25, 0.50, 0.95, 0.99) + ) + lines.append(f"{op}: {stats} ms") + + if lines: + ax.text( + 0.5, + -0.30, + "\n".join(lines), + transform=ax.transAxes, + ha="center", + va="top", + fontsize=6.2, + color="#555555", + linespacing=1.35, + ) + + +def _combine_mode_rows( + gil_path: pathlib.Path, + nogil_path: pathlib.Path, + output_path: pathlib.Path, +) -> None: + """Stack GIL above no-GIL while preserving each row's exact layout.""" + with Image.open(gil_path) as gil_image, Image.open(nogil_path) as nogil_image: + width = max(gil_image.width, nogil_image.width) + height = gil_image.height + nogil_image.height + combined = Image.new("RGB", (width, height), "white") + combined.paste(gil_image.convert("RGB"), (0, 0)) + combined.paste(nogil_image.convert("RGB"), (0, gil_image.height)) + combined.save(output_path) + + def fmt_chunk(b: int) -> str: if b >= 1024 * 1024: return f"{b // (1024*1024)}M" @@ -122,13 +207,13 @@ def _op_legend(ax): ax.legend(handles=elems, fontsize=7.5, framealpha=0.7) -def _chunk_xticks(ax, chunks: List[int]): +def _chunk_xticks(ax, chunks: list[int]): """Set numeric x-positions with human-readable tick labels.""" ax.set_xticks(range(len(chunks))) ax.set_xticklabels([fmt_chunk(c) for c in chunks]) -def _conc_xticks(ax, values: List[int]): +def _conc_xticks(ax, values: list[int]): """Keep logarithmic concurrency labels readable in wide backend grids.""" ticks = values if len(ticks) > 6: @@ -139,7 +224,7 @@ def _conc_xticks(ax, values: List[int]): ax.tick_params(axis="x", labelsize=8) -def _available(rows: List[Row]) -> List[str]: +def _available(rows: list[Row]) -> list[str]: """Backends that actually appear in the CSV data.""" present = {r["backend"] for r in rows} return [b for b in BACKENDS if b in present] @@ -147,14 +232,19 @@ def _available(rows: List[Row]) -> List[str]: # ── figure 1: concurrency sweep — throughput ────────────────────────────────── -def plot_conc_throughput(rows: List[Row], access: str = "rand"): +def plot_conc_throughput(rows: list[Row], access: str = "rand"): backends = _available(rows) n = len(backends) if not n: return fig, axes = plt.subplots(1, n, figsize=(4.5 * n, 5), sharey=True) - fig.suptitle(f"Throughput vs Concurrency (chunk=16 KB, {access})", - fontsize=13, fontweight="bold") + fig.suptitle( + _mode_title( + f"Throughput vs Concurrency (chunk=16 KB, {access})", + ), + fontsize=13, + fontweight="bold", + ) sweep = f"conc_sweep_{access}" if n == 1: axes = [axes] @@ -186,24 +276,28 @@ def plot_conc_throughput(rows: List[Row], access: str = "rand"): ax.xaxis.set_major_formatter(ticker.ScalarFormatter()) _conc_xticks(ax, xs_all) ax.legend(fontsize=8, framealpha=0.7) + _percentile_footer(ax, rows, sweep, backend) - fig.tight_layout() - out = PLOT_RESULTS_DIR / f"concurrency_sweep_throughput_{access}.png" - fig.savefig(out, dpi=150) + fig.tight_layout(rect=(0, 0.07, 1, 0.95), h_pad=3) + out = _output_path(f"concurrency_sweep_throughput_{access}.png") + fig.savefig(out, dpi=150, bbox_inches="tight") print(f"saved {out}") plt.close(fig) # ── figure 2: concurrency sweep — latency ──────────────────────────────────── -def plot_conc_latency(rows: List[Row], access: str = "rand"): +def plot_conc_latency(rows: list[Row], access: str = "rand"): backends = _available(rows) n = len(backends) if not n: return fig, axes = plt.subplots(1, n, figsize=(4.5 * n, 5), sharey=True) - fig.suptitle(f"Latency vs Concurrency (chunk=16 KB, {access})", - fontsize=13, fontweight="bold") + fig.suptitle( + _mode_title(f"Latency vs Concurrency (chunk=16 KB, {access})"), + fontsize=13, + fontweight="bold", + ) sweep = f"conc_sweep_{access}" if n == 1: axes = [axes] @@ -239,24 +333,30 @@ def plot_conc_latency(rows: List[Row], access: str = "rand"): if pivots: _conc_xticks(ax, pivots) _op_legend(ax) + _percentile_footer(ax, rows, sweep, backend) - fig.tight_layout() - out = PLOT_RESULTS_DIR / f"concurrency_sweep_latency_{access}.png" - fig.savefig(out, dpi=150) + fig.tight_layout(rect=(0, 0.07, 1, 0.95), h_pad=3) + out = _output_path(f"concurrency_sweep_latency_{access}.png") + fig.savefig(out, dpi=150, bbox_inches="tight") print(f"saved {out}") plt.close(fig) # ── figure 3: chunk sweep — throughput (MB/s) ───────────────────────────────── -def plot_chunk_throughput(rows: List[Row], access: str = "rand"): +def plot_chunk_throughput(rows: list[Row], access: str = "rand"): backends = _available(rows) n = len(backends) if not n: return fig, axes = plt.subplots(1, n, figsize=(4.5 * n, 5), sharey=True) - fig.suptitle(f"Throughput vs Chunk Size (concurrency=64, {access})", - fontsize=13, fontweight="bold") + fig.suptitle( + _mode_title( + f"Throughput vs Chunk Size (concurrency=64, {access})", + ), + fontsize=13, + fontweight="bold", + ) sweep = f"chunk_sweep_{access}" if n == 1: axes = [axes] @@ -288,24 +388,28 @@ def plot_chunk_throughput(rows: List[Row], access: str = "rand"): apply_style(ax, "Chunk size", "MB/s", backend) _chunk_xticks(ax, chunks) ax.legend(fontsize=8, framealpha=0.7) + _percentile_footer(ax, rows, sweep, backend) - fig.tight_layout() - out = PLOT_RESULTS_DIR / f"chunk_sweep_throughput_{access}.png" - fig.savefig(out, dpi=150) + fig.tight_layout(rect=(0, 0.07, 1, 0.95), h_pad=3) + out = _output_path(f"chunk_sweep_throughput_{access}.png") + fig.savefig(out, dpi=150, bbox_inches="tight") print(f"saved {out}") plt.close(fig) # ── figure 4: chunk sweep — latency ────────────────────────────────────────── -def plot_chunk_latency(rows: List[Row], access: str = "rand"): +def plot_chunk_latency(rows: list[Row], access: str = "rand"): backends = _available(rows) n = len(backends) if not n: return fig, axes = plt.subplots(1, n, figsize=(4.5 * n, 5), sharey=True) - fig.suptitle(f"Latency vs Chunk Size (concurrency=64, {access})", - fontsize=13, fontweight="bold") + fig.suptitle( + _mode_title(f"Latency vs Chunk Size (concurrency=64, {access})"), + fontsize=13, + fontweight="bold", + ) sweep = f"chunk_sweep_{access}" if n == 1: axes = [axes] @@ -338,10 +442,11 @@ def plot_chunk_latency(rows: List[Row], access: str = "rand"): apply_style(ax, "Chunk size", "Latency (ms)", backend, yscale="log") _chunk_xticks(ax, chunks) _op_legend(ax) + _percentile_footer(ax, rows, sweep, backend) - fig.tight_layout() - out = PLOT_RESULTS_DIR / f"chunk_sweep_latency_{access}.png" - fig.savefig(out, dpi=150) + fig.tight_layout(rect=(0, 0.07, 1, 0.95), h_pad=3) + out = _output_path(f"chunk_sweep_latency_{access}.png") + fig.savefig(out, dpi=150, bbox_inches="tight") print(f"saved {out}") plt.close(fig) @@ -351,7 +456,7 @@ def plot_chunk_latency(rows: List[Row], access: str = "rand"): _HIST_COLORS = ["#3498db", "#e74c3c", "#2ecc71", "#f39c12", "#9b59b6"] -def plot_histograms(rows: List[Row]): +def plot_histograms(rows: list[Row]): """Per-backend histogram of read latency at concurrency=64, chunk=16K.""" backends = _available(rows) if not backends: @@ -363,7 +468,10 @@ def plot_histograms(rows: List[Row]): fig, axes = plt.subplots(nrows, ncols, figsize=(6.5 * ncols, 4.5 * nrows), squeeze=False) fig.suptitle( - f"Read latency distribution (concurrency={target_conc}, chunk=16 KB)", + _mode_title( + f"Read latency distribution " + f"(concurrency={target_conc}, chunk=16 KB)", + ), fontsize=13, fontweight="bold", ) @@ -382,9 +490,10 @@ def plot_histograms(rows: List[Row]): ax.set_visible(False) continue - p50 = pct(lats, 0.50) - p95 = pct(lats, 0.95) - p99 = pct(lats, 0.99) + p25 = pct(lats, 0.25) + p50 = pct(lats, 0.50) + p95 = pct(lats, 0.95) + p99 = pct(lats, 0.99) clip = pct(lats, 0.999) ax.hist( @@ -405,27 +514,37 @@ def plot_histograms(rows: List[Row]): ax.grid(True, linestyle="--", alpha=0.4) ax.spines["top"].set_visible(False) ax.spines["right"].set_visible(False) + ax.text( + 0.5, + -0.24, + ( + f"read: p25 {p25:.3f} p50 {p50:.3f} " + f"p95 {p95:.3f} p99 {p99:.3f} ms" + ), + transform=ax.transAxes, + ha="center", + va="top", + fontsize=6.5, + color="#555555", + ) # Hide unused subplot cells for idx in range(len(backends), nrows * ncols): axes[idx // ncols][idx % ncols].set_visible(False) - fig.tight_layout() - out = PLOT_RESULTS_DIR / "latency_histograms.png" - fig.savefig(out, dpi=150) + fig.tight_layout(rect=(0, 0.07, 1, 0.95), h_pad=3) + out = _output_path("latency_histograms.png") + fig.savefig(out, dpi=150, bbox_inches="tight") print(f"saved {out}") plt.close(fig) # ── main ────────────────────────────────────────────────────────────────────── -def main(): - if not CSV_PATH.exists(): - raise SystemExit(f"CSV not found: {CSV_PATH}\nRun bench_runner.py first.") - - rows = load_csv() - print(f"loaded {len(rows):,} rows from {CSV_PATH}") - +def _render_mode(rows: list[Row], label: str, suffix: str) -> None: + global MODE_LABEL, OUTPUT_SUFFIX + MODE_LABEL = label + OUTPUT_SUFFIX = suffix for access in ("rand", "seq"): plot_conc_throughput(rows, access) plot_conc_latency(rows, access) @@ -433,6 +552,46 @@ def main(): plot_chunk_latency(rows, access) plot_histograms(rows) + +def main(): + missing = [ + path for path in (CSV_GIL_PATH, CSV_NOGIL_PATH) + if not path.exists() + ] + if missing: + paths = "\n".join(str(path) for path in missing) + raise SystemExit(f"Benchmark CSV not found:\n{paths}") + + PLOT_RESULTS_DIR.mkdir(parents=True, exist_ok=True) + gil_rows = load_csv(CSV_GIL_PATH) + nogil_rows = load_csv(CSV_NOGIL_PATH) + print(f"loaded {len(gil_rows):,} GIL rows from {CSV_GIL_PATH}") + print(f"loaded {len(nogil_rows):,} no-GIL rows from {CSV_NOGIL_PATH}") + + _render_mode(gil_rows, "GIL", "_gil") + _render_mode(nogil_rows, "free-threaded / no GIL", "_nogil") + + names = [ + f"{prefix}_{access}.png" + for access in ("rand", "seq") + for prefix in ( + "concurrency_sweep_throughput", + "concurrency_sweep_latency", + "chunk_sweep_throughput", + "chunk_sweep_latency", + ) + ] + ["latency_histograms.png"] + + for name in names: + stem, suffix = os.path.splitext(name) + gil_path = PLOT_RESULTS_DIR / f"{stem}_gil{suffix}" + nogil_path = PLOT_RESULTS_DIR / f"{stem}_nogil{suffix}" + output_path = PLOT_RESULTS_DIR / name + _combine_mode_rows(gil_path, nogil_path, output_path) + gil_path.unlink() + nogil_path.unlink() + print(f"combined {output_path}") + print(f"\nAll plots saved to {PLOT_RESULTS_DIR.resolve()}") diff --git a/benchmark/results/bench_all.csv.gz b/benchmark/results/bench_all.csv.gz index 4fad60f..eeecaf8 100644 --- a/benchmark/results/bench_all.csv.gz +++ b/benchmark/results/bench_all.csv.gz @@ -1,3 +1,3 @@ version https://git-lfs.github.com/spec/v1 -oid sha256:38da57086d60b8d2b9e73ad56f9d47b1a7a548ee448de26448454603f79eb260 -size 836960 +oid sha256:7659a5411926479d07cae526e0746c2636b4a8a5b3989fce986f95fe402ccb78 +size 886698 diff --git a/benchmark/results/bench_all_nogil.csv.gz b/benchmark/results/bench_all_nogil.csv.gz new file mode 100644 index 0000000..c14f740 --- /dev/null +++ b/benchmark/results/bench_all_nogil.csv.gz @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:4d01e61d6f5479f545cc3c61aa28dba94fe0189a11964bbfdb89ff55c4b7442d +size 858860 diff --git a/benchmark/results/chunk_sweep_latency_rand.png b/benchmark/results/chunk_sweep_latency_rand.png new file mode 100644 index 0000000..045ac0e --- /dev/null +++ b/benchmark/results/chunk_sweep_latency_rand.png @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:b33ccbc3adaa2d0d7b97c060b22add3897e532e7ede2ec4467d752c91ba7f6ec +size 448718 diff --git a/benchmark/results/chunk_sweep_latency_seq.png b/benchmark/results/chunk_sweep_latency_seq.png new file mode 100644 index 0000000..9432123 --- /dev/null +++ b/benchmark/results/chunk_sweep_latency_seq.png @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:060062635b0a698aa39230402f611325fdef3c86997a13915b15a32354bb105a +size 443957 diff --git a/benchmark/results/chunk_sweep_throughput_rand.png b/benchmark/results/chunk_sweep_throughput_rand.png new file mode 100644 index 0000000..5cbf34e --- /dev/null +++ b/benchmark/results/chunk_sweep_throughput_rand.png @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:d047c271d3be2ab593f3c0915a30547d50278a1209b81e9f371f89697b5f7b3a +size 238270 diff --git a/benchmark/results/chunk_sweep_throughput_seq.png b/benchmark/results/chunk_sweep_throughput_seq.png new file mode 100644 index 0000000..567aa57 --- /dev/null +++ b/benchmark/results/chunk_sweep_throughput_seq.png @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:b80d399806a7e74f3356da70a026e333ccea49804a3ffa1a1ae931a946aa0e22 +size 241990 diff --git a/benchmark/results/concurrency_sweep_latency_rand.png b/benchmark/results/concurrency_sweep_latency_rand.png new file mode 100644 index 0000000..c37a5ef --- /dev/null +++ b/benchmark/results/concurrency_sweep_latency_rand.png @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:25be813cf6daecf1e3bb4729dbb676ad6e677c4012e5f09174b406840c7f3a08 +size 462932 diff --git a/benchmark/results/concurrency_sweep_latency_seq.png b/benchmark/results/concurrency_sweep_latency_seq.png new file mode 100644 index 0000000..55107bd --- /dev/null +++ b/benchmark/results/concurrency_sweep_latency_seq.png @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:54746e1a4eccb8ba870b1e4b532bbb97e2a35885cd3540be1a16892e3b5b5288 +size 452030 diff --git a/benchmark/results/concurrency_sweep_throughput_rand.png b/benchmark/results/concurrency_sweep_throughput_rand.png new file mode 100644 index 0000000..21ccd62 --- /dev/null +++ b/benchmark/results/concurrency_sweep_throughput_rand.png @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:1d64ef620f11786da668608984885988640d56891c06e9a7f6714fd896d3c093 +size 259261 diff --git a/benchmark/results/concurrency_sweep_throughput_seq.png b/benchmark/results/concurrency_sweep_throughput_seq.png new file mode 100644 index 0000000..697fc37 --- /dev/null +++ b/benchmark/results/concurrency_sweep_throughput_seq.png @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:62236646f3bc2b31215c640f8a45d54d3a54ef47cb28ab248bc21c9be258b927 +size 260910 diff --git a/benchmark/results/latency_histograms.png b/benchmark/results/latency_histograms.png new file mode 100644 index 0000000..6155ba9 --- /dev/null +++ b/benchmark/results/latency_histograms.png @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:07fd2c31b690990a38503c72c02bba02aed6cce067f8732bed0e97225452774e +size 227561 diff --git a/benchmark/uv.lock b/benchmark/uv.lock index fc7b7ff..a8f2f20 100644 --- a/benchmark/uv.lock +++ b/benchmark/uv.lock @@ -17,6 +17,7 @@ requires-dist = [ { name = "coveralls", marker = "extra == 'develop'" }, { name = "pytest", marker = "extra == 'develop'" }, { name = "pytest-cov", marker = "extra == 'develop'" }, + { name = "pytest-rerunfailures", marker = "extra == 'develop'" }, { name = "setuptools", marker = "extra == 'develop'" }, ] provides-extras = ["develop"] @@ -49,7 +50,7 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, + { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" } }, ] sdist = { url = "https://files.pythonhosted.org/packages/66/54/eb9bfc647b19f2009dd5c7f5ec51c4e6ca831725f1aea7a993034f483147/contourpy-1.3.2.tar.gz", hash = "sha256:b6945942715a034c671b7fc54f9588126b0b8bf23db2696e3ca8328f3ff0ab54", size = 13466130, upload-time = "2025-04-15T17:47:53.79Z" } wheels = [ @@ -119,7 +120,7 @@ resolution-markers = [ "python_full_version >= '3.11'", ] dependencies = [ - { name = "numpy", version = "2.4.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, + { name = "numpy", version = "2.4.2", source = { registry = "https://pypi.org/simple" } }, ] sdist = { url = "https://files.pythonhosted.org/packages/58/01/1253e6698a07380cd31a736d248a3f2a50a7c88779a1813da27503cadc2a/contourpy-1.3.3.tar.gz", hash = "sha256:083e12155b210502d0bca491432bb04d56dc3432f95a979b429f2848c3dbe880", size = 13466174, upload-time = "2025-07-26T12:03:12.549Z" } wheels = [ From 3b92a60eb821e41fd92341765f3b6580fbdfd737 Mon Sep 17 00:00:00 2001 From: Dmitry Orlov Date: Tue, 4 Aug 2026 11:39:31 +0200 Subject: [PATCH 7/8] Address Copilot review comments on PR #72 - test_high_concurrency_stress_no_data_races: init ctx = None before the try so finally's ctx.close() can't itself raise UnboundLocalError and mask a genuine Context() construction failure. - thread_aio.c: payload getter's comment claimed a completed write's py_buffer is freed right after its callback runs - worker() actually clears it before publishing done and before the callback, matching the surrounding "publish completion only after every write completes" comment. Code was already correct, only the comment was stale. --- caio/thread_aio.c | 7 ++++--- tests/test_free_threading.py | 4 +++- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/caio/thread_aio.c b/caio/thread_aio.c index 49948ff..8cf7925 100644 --- a/caio/thread_aio.c +++ b/caio/thread_aio.c @@ -917,9 +917,10 @@ static PyObject *AIOOperation_payload_getter(AIOOperation *self, void *closure) } /* fsync/fdsync Operations never allocate a buffer, and a completed - * write's is freed right after its callback runs (see worker()) - - * matches T_OBJECT's (as opposed to T_OBJECT_EX's) own NULL-to-None - * behavior, which this getter replaces. */ + * write's is freed in worker() before done is published (i.e. before + * this getter could ever observe it) - matches T_OBJECT's (as opposed + * to T_OBJECT_EX's) own NULL-to-None behavior, which this getter + * replaces. */ if (self->py_buffer == NULL) Py_RETURN_NONE; diff --git a/tests/test_free_threading.py b/tests/test_free_threading.py index 7278747..a8c4e00 100644 --- a/tests/test_free_threading.py +++ b/tests/test_free_threading.py @@ -42,6 +42,7 @@ def test_high_concurrency_stress_no_data_races(tmp_path): path = tmp_path / "stress.bin" path.write_bytes(b"\x00" * (count * chunk)) fd = os.open(str(path), os.O_RDWR | _O_BINARY) + ctx = None try: ctx = python_aio.Context(max_requests=count, pool_size=32) expected = [bytes([i % 256]) * chunk for i in range(count)] @@ -91,7 +92,8 @@ def cb(_n): assert got == want, f"chunk {i} mismatch" finally: os.close(fd) - ctx.close() + if ctx is not None: + ctx.close() def test_same_operation_is_claimed_once_across_contexts(): From a98704503cf0b972c3efd41ed2363009a1844a5b Mon Sep 17 00:00:00 2001 From: Dmitry Orlov Date: Tue, 4 Aug 2026 11:51:03 +0200 Subject: [PATCH 8/8] Refactor free-threading test helpers into conftest.py ConcurrentThreads (worker threads that surface failures on the main thread) and a submit_and_wait fixture (submit N ops, wait for every callback, collect results) factor out the repeated boilerplate that had accumulated across tests/test_free_threading.py's stress tests. --- tests/conftest.py | 107 +++++++++ tests/test_free_threading.py | 406 +++++++++++------------------------ 2 files changed, 232 insertions(+), 281 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index e4cc32e..9dd3e52 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,4 +1,5 @@ import functools +import threading import time import types @@ -16,6 +17,112 @@ ) +class ConcurrentThreads: + """Run test workers and surface their failures in the main thread.""" + + def __init__(self, timeout=30): + self.timeout = timeout + self._barriers = [] + self._threads = [] + self._errors = [] + self._errors_lock = threading.Lock() + + def barrier(self, parties): + barrier = threading.Barrier(parties, timeout=self.timeout) + self._barriers.append(barrier) + return barrier + + def start(self, target, *args): + thread = threading.Thread( + target=self._run, + args=(target, args), + name=f"{target.__name__}-{len(self._threads)}", + ) + self._threads.append(thread) + thread.start() + return thread + + def start_many(self, target, arguments): + for args in arguments: + self.start(target, *args) + + def join(self, timeout=None): + deadline = time.monotonic() + ( + self.timeout if timeout is None else timeout + ) + for thread in self._threads: + thread.join(max(0, deadline - time.monotonic())) + + alive = [thread.name for thread in self._threads if thread.is_alive()] + assert not alive, f"worker threads did not stop: {', '.join(alive)}" + if self._errors: + raise self._errors[0] + + def close(self): + for barrier in self._barriers: + barrier.abort() + for thread in self._threads: + thread.join(timeout=5) + + def _run(self, target, args): + try: + target(*args) + except BaseException as exc: # noqa: BLE001 (surface child-thread failures) + with self._errors_lock: + self._errors.append(exc) + for barrier in self._barriers: + barrier.abort() + + +@pytest.fixture +def workers(): + threads = ConcurrentThreads() + yield threads + threads.close() + + +@pytest.fixture +def submit_and_wait(): + def submit(context, operations, result_for=None, timeout=10): + operations = tuple(operations) + count = len(operations) + results = [None] * count + remaining = [count] + errors = [] + lock = threading.Lock() + done = threading.Event() + + if result_for is None: + result_for = lambda _operation, result: result + + def make_callback(index, operation): + def callback(result): + with lock: + try: + results[index] = result_for(operation, result) + except BaseException as exc: # noqa: BLE001 (surface callback failures) + errors.append(exc) + finally: + remaining[0] -= 1 + if remaining[0] == 0: + done.set() + + return callback + + for index, operation in enumerate(operations): + operation.set_callback(make_callback(index, operation)) + assert context.submit(operation) == 1 + + assert done.wait(timeout), ( + f"only {count - remaining[0]}/{count} operations completed" + ) + if errors: + raise errors[0] + return results + + return submit + + def named_variant(name, **attrs): ns = types.SimpleNamespace(__name__=name, **attrs) return ns diff --git a/tests/test_free_threading.py b/tests/test_free_threading.py index a8c4e00..f50518f 100644 --- a/tests/test_free_threading.py +++ b/tests/test_free_threading.py @@ -27,7 +27,7 @@ def test_importing_caio_does_not_enable_gil(): ) -def test_high_concurrency_stress_no_data_races(tmp_path): +def test_high_concurrency_stress_no_data_races(tmp_path, submit_and_wait): """Hammers python_aio's own bookkeeping (the _lock-protected _in_progress counter and per-operation in_progress flag) with many concurrent writes then reads dispatched across a real multi-worker @@ -47,56 +47,30 @@ def test_high_concurrency_stress_no_data_races(tmp_path): ctx = python_aio.Context(max_requests=count, pool_size=32) expected = [bytes([i % 256]) * chunk for i in range(count)] - written = [None] * count - lock = threading.Lock() - remaining = [count] - done = threading.Event() - - def make_write_cb(i): - def cb(n): - with lock: - written[i] = n - remaining[0] -= 1 - if remaining[0] == 0: - done.set() - return cb - - for i in range(count): - op = python_aio.Operation.write(expected[i], fd, i * chunk) - op.set_callback(make_write_cb(i)) - assert ctx.submit(op) == 1 - - assert done.wait(10), f"only {count - remaining[0]}/{count} writes completed" + writes = ( + python_aio.Operation.write(payload, fd, index * chunk) + for index, payload in enumerate(expected) + ) + written = submit_and_wait(ctx, writes) assert written == [chunk] * count - read_back = [None] * count - remaining = [count] - done = threading.Event() - - def make_read_cb(i, op): - def cb(_n): - with lock: - read_back[i] = op.get_value() - remaining[0] -= 1 - if remaining[0] == 0: - done.set() - return cb - - for i in range(count): - op = python_aio.Operation.read(chunk, fd, i * chunk) - op.set_callback(make_read_cb(i, op)) - assert ctx.submit(op) == 1 - - assert done.wait(10), f"only {count - remaining[0]}/{count} reads completed" - for i, (got, want) in enumerate(zip(read_back, expected)): - assert got == want, f"chunk {i} mismatch" + reads = ( + python_aio.Operation.read(chunk, fd, index * chunk) + for index in range(count) + ) + read_back = submit_and_wait( + ctx, + reads, + result_for=lambda operation, _result: operation.get_value(), + ) + assert read_back == expected finally: os.close(fd) if ctx is not None: ctx.close() -def test_same_operation_is_claimed_once_across_contexts(): +def test_same_operation_is_claimed_once_across_contexts(workers): """An Operation's one-shot claim must be synchronized by the Operation. Context._execute() currently protects ``operation.in_progress`` with the @@ -115,31 +89,19 @@ def test_same_operation_is_claimed_once_across_contexts(): python_aio.Context(max_requests=iterations * 2), python_aio.Context(max_requests=iterations * 2), ) - start = threading.Barrier(3, timeout=30) - finished = threading.Barrier(3, timeout=30) + start = workers.barrier(3) + finished = workers.barrier(3) current = [None] submitted = [0, 0] - errors = [] def submitter(index, context): - try: - for _ in range(iterations): - start.wait() - submitted[index] += context.submit(current[0]) - finished.wait() - except Exception as exc: # noqa: BLE001 (surface child-thread failures) - errors.append(exc) - start.abort() - finished.abort() - - threads = [ - threading.Thread(target=submitter, args=(index, context)) - for index, context in enumerate(contexts) - ] + for _ in range(iterations): + start.wait() + submitted[index] += context.submit(current[0]) + finished.wait() try: - for thread in threads: - thread.start() + workers.start_many(submitter, enumerate(contexts)) for _ in range(iterations): current[0] = python_aio.Operation( @@ -148,18 +110,12 @@ def submitter(index, context): start.wait() finished.wait() - for thread in threads: - thread.join() - - assert not errors + workers.join() assert sum(submitted) == iterations, ( f"{sum(submitted) - iterations} Operations were submitted twice" ) finally: - start.abort() - finished.abort() - for thread in threads: - thread.join(timeout=5) + workers.close() for context in contexts: context.close() context.pool.join() @@ -295,6 +251,7 @@ def test_same_operation_is_claimed_once_across_contexts_all_backends( backend, ft_claimant_count, ft_claim_operation_count, + workers, ): """Every backend must synchronize a one-shot claim on the Operation. @@ -312,39 +269,25 @@ def test_same_operation_is_claimed_once_across_contexts_all_backends( backend.Context(max_requests=claim_window) for _ in range(ft_claimant_count) ) - start = threading.Barrier(ft_claimant_count + 1, timeout=30) - finished = threading.Barrier(ft_claimant_count + 1, timeout=30) + start = workers.barrier(ft_claimant_count + 1) + finished = workers.barrier(ft_claimant_count + 1) current = [None] submitted = [0] * ft_claimant_count callback_count = [0] callback_lock = threading.Lock() - errors = [] - errors_lock = threading.Lock() def on_done(_result): with callback_lock: callback_count[0] += 1 def submitter(index, context): - try: - for _ in range(iterations): - start.wait() - submitted[index] += context.submit(current[0]) - finished.wait() - except Exception as exc: # noqa: BLE001 (surface child-thread failures) - with errors_lock: - errors.append(exc) - start.abort() - finished.abort() - - threads = [ - threading.Thread(target=submitter, args=(index, context)) - for index, context in enumerate(contexts) - ] + for _ in range(iterations): + start.wait() + submitted[index] += context.submit(current[0]) + finished.wait() try: - for thread in threads: - thread.start() + workers.start_many(submitter, enumerate(contexts)) for iteration in range(iterations): operation = backend.Operation.write(b"x", fd, 0) @@ -365,10 +308,7 @@ def window_finished(expected=expected_callbacks): timeout=20, ) - for thread in threads: - thread.join() - - assert not errors + workers.join() accepted = sum(submitted) def all_callbacks_finished(): @@ -385,10 +325,7 @@ def all_callbacks_finished(): ) assert callback_count[0] == iterations finally: - start.abort() - finished.abort() - for thread in threads: - thread.join(timeout=5) + workers.close() _close_contexts(contexts) os.close(fd) @@ -399,6 +336,7 @@ def test_concurrent_submits_to_one_context_all_backends( ft_submitter_count, ft_operations_per_submit, ft_submit_rounds, + workers, ): """Distinct Operations submitted concurrently must not corrupt Context.""" _require_gil_disabled() @@ -411,13 +349,11 @@ def test_concurrent_submits_to_one_context_all_backends( path = tmp_path / "one-context-submit.bin" fd = os.open(path, os.O_RDWR | os.O_CREAT | _O_BINARY, 0o600) context = backend.Context(max_requests=total) - submit_start = threading.Barrier(worker_count, timeout=30) + submit_start = workers.barrier(worker_count) operations = [] callback_results = [None] * total callback_lock = threading.Lock() submitted = [0] * worker_count - errors = [] - errors_lock = threading.Lock() def make_callback(index, operation): def callback(result): @@ -433,37 +369,26 @@ def callback(result): def submitter(worker_index): first = worker_index * operations_per_worker_total - try: - accepted = 0 - for round_index in range(submit_rounds): - batch_first = first + round_index * operations_per_worker - batch_last = batch_first + operations_per_worker - # Every round stages a batch after the same barrier. Multiple - # rounds vary scheduling and repeatedly collide inside the - # Context's SQ-tail read/write window. - submit_start.wait() - accepted += context.submit( - *operations[batch_first:batch_last], - ) - submitted[worker_index] = accepted - except Exception as exc: # noqa: BLE001 (surface child-thread failures) - with errors_lock: - errors.append(exc) - submit_start.abort() - - threads = [ - threading.Thread(target=submitter, args=(index,)) - for index in range(worker_count) - ] + accepted = 0 + for round_index in range(submit_rounds): + batch_first = first + round_index * operations_per_worker + batch_last = batch_first + operations_per_worker + # Every round stages a batch after the same barrier. Multiple + # rounds vary scheduling and repeatedly collide inside the + # Context's SQ-tail read/write window. + submit_start.wait() + accepted += context.submit( + *operations[batch_first:batch_last], + ) + submitted[worker_index] = accepted try: - for thread in threads: - thread.start() - for thread in threads: - thread.join(timeout=30) + workers.start_many( + submitter, + ((index,) for index in range(worker_count)), + ) + workers.join() - assert all(not thread.is_alive() for thread in threads) - assert not errors assert sum(submitted) == total def all_callbacks_finished(): @@ -484,9 +409,7 @@ def all_callbacks_finished(): expected = b"".join(index.to_bytes(4, "little") for index in range(total)) assert os.read(fd, len(expected)) == expected finally: - submit_start.abort() - for thread in threads: - thread.join(timeout=5) + workers.close() _close_contexts((context,)) os.close(fd) @@ -497,6 +420,7 @@ def test_concurrent_process_events_delivers_each_completion_once( ft_operation_count, ft_drainer_count, ft_drain_rounds, + workers, ): """Concurrent drainers must never consume or callback one event twice.""" _require_gil_disabled() @@ -509,9 +433,7 @@ def test_concurrent_process_events_delivers_each_completion_once( context = polling_backend.Context(max_requests=operation_count) callback_counts = [0] * operation_count callback_lock = threading.Lock() - drain_round = threading.Barrier(drainer_count, timeout=30) - errors = [] - errors_lock = threading.Lock() + drain_round = workers.barrier(drainer_count) def make_callback(index): def callback(_result): @@ -542,33 +464,21 @@ def callback(_result): time.sleep(0.01) def drain(): - try: - for _ in range(drain_rounds): - # Force all drainers to enter every round together. Without - # this barrier a fast first thread can consume the whole CQ - # before the others even start, accidentally serializing the - # test and hiding a duplicate cq_head read/commit. - drain_round.wait() - context.process_events( - max_requests=operation_count, - min_requests=0, - timeout=0, - ) - except Exception as exc: # noqa: BLE001 (surface child-thread failures) - with errors_lock: - errors.append(exc) - drain_round.abort() - - threads = [threading.Thread(target=drain) for _ in range(drainer_count)] + for _ in range(drain_rounds): + # Force all drainers to enter every round together. Without + # this barrier a fast first thread can consume the whole CQ + # before the others even start, accidentally serializing the + # test and hiding a duplicate cq_head read/commit. + drain_round.wait() + context.process_events( + max_requests=operation_count, + min_requests=0, + timeout=0, + ) try: - for thread in threads: - thread.start() - for thread in threads: - thread.join(timeout=30) - - assert all(not thread.is_alive() for thread in threads) - assert not errors + workers.start_many(drain, (() for _ in range(drainer_count))) + workers.join() def all_callbacks_finished(): with callback_lock: @@ -580,9 +490,7 @@ def all_callbacks_finished(): _pump_contexts((context,), all_callbacks_finished, timeout=20) assert callback_counts == [1] * operation_count finally: - drain_round.abort() - for thread in threads: - thread.join(timeout=5) + workers.close() os.close(fd) @@ -591,6 +499,7 @@ def test_completed_operation_supports_concurrent_readers( backend, ft_reader_count, ft_reads_per_thread, + workers, ): """C backends must safely return owned result references to all threads.""" _require_gil_disabled() @@ -609,39 +518,20 @@ def test_completed_operation_supports_concurrent_readers( reader_count = ft_reader_count reads_per_thread = ft_reads_per_thread - start = threading.Barrier(reader_count + 1, timeout=30) - errors = [] - errors_lock = threading.Lock() + start = workers.barrier(reader_count + 1) def read_result(): - try: - start.wait() - for _ in range(reads_per_thread): - if operation.get_value() != payload: - raise AssertionError("concurrent get_value() returned wrong data") - except Exception as exc: # noqa: BLE001 (surface child-thread failures) - with errors_lock: - errors.append(exc) - start.abort() - - threads = [ - threading.Thread(target=read_result) - for _ in range(reader_count) - ] + start.wait() + for _ in range(reads_per_thread): + if operation.get_value() != payload: + raise AssertionError("concurrent get_value() returned wrong data") try: - for thread in threads: - thread.start() + workers.start_many(read_result, (() for _ in range(reader_count))) start.wait() - for thread in threads: - thread.join(timeout=30) - - assert all(not thread.is_alive() for thread in threads) - assert not errors + workers.join() finally: - start.abort() - for thread in threads: - thread.join(timeout=5) + workers.close() _close_contexts((context,)) os.close(fd) @@ -651,6 +541,7 @@ def test_payload_access_is_safe_while_worker_publishes_completion( pooled_backend, ft_observer_count, ft_completion_count, + workers, ): """Publishing ``done`` must not expose a buffer being cleared concurrently. @@ -705,53 +596,35 @@ def on_done(_result): assert context.submit(*operations) == operation_count - observer_start = threading.Barrier(observer_count + 1, timeout=30) - errors = [] - errors_lock = threading.Lock() + observer_start = workers.barrier(observer_count + 1) def observe(): - try: - observer_start.wait() - while not all_completed.is_set(): - for operation in operations: - try: - payload = operation.payload - if payload is not None: - bytes(payload) - except RuntimeError: - pass - - try: - operation.get_value() - except RuntimeError: - pass - except Exception as exc: # noqa: BLE001 (surface child-thread failures) - with errors_lock: - errors.append(exc) - observer_start.abort() - - observers = [ - threading.Thread(target=observe) - for _ in range(observer_count) - ] + observer_start.wait() + while not all_completed.is_set(): + for operation in operations: + try: + payload = operation.payload + if payload is not None: + bytes(payload) + except RuntimeError: + pass + + try: + operation.get_value() + except RuntimeError: + pass try: - for observer in observers: - observer.start() + workers.start_many(observe, (() for _ in range(observer_count))) observer_start.wait() blocker_release.set() assert all_completed.wait(30), "worker did not finish queued operations" - for observer in observers: - observer.join(timeout=10) + workers.join(timeout=10) - assert all(not observer.is_alive() for observer in observers) - assert not errors assert completed_count[0] == operation_count finally: blocker_release.set() - observer_start.abort() - for observer in observers: - observer.join(timeout=5) + workers.close() _close_contexts((context,)) os.close(fd) @@ -760,6 +633,7 @@ def test_concurrent_callback_replacement_releases_every_old_callback( backend, ft_callback_setter_count, ft_callback_sets_per_thread, + workers, ): """set_callback must atomically replace and release its owned reference.""" _require_gil_disabled() @@ -767,9 +641,7 @@ def test_concurrent_callback_replacement_releases_every_old_callback( setter_count = ft_callback_setter_count sets_per_thread = ft_callback_sets_per_thread operation = backend.Operation.fsync(0) - start = threading.Barrier(setter_count + 1, timeout=30) - errors = [] - errors_lock = threading.Lock() + start = workers.barrier(setter_count + 1) class Callback: def __call__(self, _result): @@ -782,30 +654,18 @@ def __call__(self, _result): callback_refs = [weakref.ref(callback) for callback in callbacks] def replace_callbacks(thread_index): - try: - start.wait() - first = thread_index * sets_per_thread - for index in range(first, first + sets_per_thread): - assert operation.set_callback(callbacks[index]) is True - except Exception as exc: # noqa: BLE001 (surface child-thread failures) - with errors_lock: - errors.append(exc) - start.abort() - - threads = [ - threading.Thread(target=replace_callbacks, args=(index,)) - for index in range(setter_count) - ] + start.wait() + first = thread_index * sets_per_thread + for index in range(first, first + sets_per_thread): + assert operation.set_callback(callbacks[index]) is True try: - for thread in threads: - thread.start() + workers.start_many( + replace_callbacks, + ((index,) for index in range(setter_count)), + ) start.wait() - for thread in threads: - thread.join(timeout=30) - - assert all(not thread.is_alive() for thread in threads) - assert not errors + workers.join() # Move the Operation off every callback in the contested set. Each # set_callback owns exactly one reference and must release the old @@ -816,14 +676,13 @@ def replace_callbacks(thread_index): leaked = sum(ref() is not None for ref in callback_refs) assert leaked == 0, f"{leaked} replaced callbacks are still referenced" finally: - start.abort() - for thread in threads: - thread.join(timeout=5) + workers.close() def test_concurrent_uring_cancel_requests_do_not_corrupt_submission_ring( ft_canceller_count, ft_cancel_rounds, + workers, ): """cancel and submit share io_uring's SQ tail and must share its lock.""" _require_gil_disabled() @@ -860,36 +719,23 @@ def callback(_result): assert context.submit(*targets) == operation_count context.flush() - start_round = threading.Barrier(canceller_count, timeout=30) - errors = [] - errors_lock = threading.Lock() + start_round = workers.barrier(canceller_count) cancelled = [0] * canceller_count def cancel(thread_index): - try: - for round_index in range(cancel_rounds): - start_round.wait() - cancelled[thread_index] += context.cancel( - targets[round_index * canceller_count + thread_index], - ) - except Exception as exc: # noqa: BLE001 (surface child-thread failures) - with errors_lock: - errors.append(exc) - start_round.abort() - - threads = [ - threading.Thread(target=cancel, args=(index,)) - for index in range(canceller_count) - ] + for round_index in range(cancel_rounds): + start_round.wait() + cancelled[thread_index] += context.cancel( + targets[round_index * canceller_count + thread_index], + ) try: - for thread in threads: - thread.start() - for thread in threads: - thread.join(timeout=30) + workers.start_many( + cancel, + ((index,) for index in range(canceller_count)), + ) + workers.join() - assert all(not thread.is_alive() for thread in threads) - assert not errors assert cancelled == [0] * canceller_count def all_targets_completed(): @@ -899,9 +745,7 @@ def all_targets_completed(): _pump_contexts((context,), all_targets_completed, timeout=10) assert callback_counts == [1] * operation_count finally: - start_round.abort() - for thread in threads: - thread.join(timeout=5) + workers.close() # Release any read whose cancel SQE was lost so Context teardown # cannot leave the kernel holding pointers to live Operations. try: