Script 'mail_helper' called by obssrc Hello community, here is the log from the commit of package python-caio for openSUSE:Factory checked in at 2026-09-28 10:46:00 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ Comparing /work/SRC/openSUSE:Factory/python-caio (Old) and /work/SRC/openSUSE:Factory/.python-caio.new.383539 (New) ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Package is "python-caio" Mon Sep 28 10:46:00 2026 rev:8 rq:1380831 version:0.12.9 Changes: -------- --- /work/SRC/openSUSE:Factory/python-caio/python-caio.changes 2026-09-08 17:01:28.948408970 +0200 +++ /work/SRC/openSUSE:Factory/.python-caio.new.383539/python-caio.changes 2026-09-28 10:46:37.846400359 +0200 @@ -1,0 +2,20 @@ +Sat Sep 26 16:16:05 UTC 2026 - Martin Pluskal <[email protected]> + +- Update to 0.12.9: + * Fix a race in the io_uring backend's blocking wait that made it + return 0 or block forever on a single-CPU system + * thread_aio: fix a pool leak of about 9 MB per Context when the last + reference is dropped from a worker thread, and stop destroying a + locked mutex on close (undefined behaviour, reported by + ThreadSanitizer on every close) + * Fix C type handling on big-endian platforms, buffer size and + timeout handling on 32-bit, and the SQPOLL fallback on permission + errors + * Declare support for Python 3.15 + * Skip backend-specific tests when the kernel lacks the backend + instead of reporting them as errors +- Replace the tracked source tarball caio-0.12.4.tar.gz with + caio-0.12.9.tar.gz; the file named by Source0 has to be tracked or + source_validator aborts the build + +------------------------------------------------------------------- Old: ---- caio-0.12.4.tar.gz New: ---- caio-0.12.9.tar.gz ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ Other differences: ------------------ ++++++ python-caio.spec ++++++ --- /var/tmp/diff_new_pack.ZXit2A/_old 2026-09-28 10:46:38.439425204 +0200 +++ /var/tmp/diff_new_pack.ZXit2A/_new 2026-09-28 10:46:38.441425288 +0200 @@ -17,7 +17,7 @@ Name: python-caio -Version: 0.12.4 +Version: 0.12.9 Release: 0 Summary: Asynchronous file IO for Linux MacOS or Windows License: Apache-2.0 @@ -25,6 +25,8 @@ # PyPI sdist (graft tests in MANIFEST.in). 0.12.3 had none and used the # GitHub tag archive; 0.12.4 publishes a real sdist with the correct # [project] version, so no %%prep rewrite is needed. +# The tarball is also committed to the package: source_validator aborts the +# build on a Source0 whose file is not tracked. Source: https://files.pythonhosted.org/packages/source/c/caio/caio-%{version}.tar.gz BuildRequires: %{python_module devel} BuildRequires: %{python_module pip} ++++++ caio-0.12.4.tar.gz -> caio-0.12.9.tar.gz ++++++ diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/caio-0.12.4/PKG-INFO new/caio-0.12.9/PKG-INFO --- old/caio-0.12.4/PKG-INFO 2026-09-07 08:50:42.403425500 +0200 +++ new/caio-0.12.9/PKG-INFO 2026-09-26 11:48:40.259000800 +0200 @@ -1,6 +1,6 @@ Metadata-Version: 2.4 Name: caio -Version: 0.12.4 +Version: 0.12.9 Summary: Asynchronous file IO for Linux MacOS or Windows. Author-email: Dmitry Orlov <[email protected]> License-Expression: Apache-2.0 @@ -19,6 +19,7 @@ Classifier: Programming Language :: Python :: 3.12 Classifier: Programming Language :: Python :: 3.13 Classifier: Programming Language :: Python :: 3.14 +Classifier: Programming Language :: Python :: 3.15 Classifier: Programming Language :: Python :: Implementation :: CPython Classifier: Programming Language :: Python :: Free Threading :: 2 - Beta Requires-Python: >=3.10 diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/caio-0.12.4/caio/linux_aio.c new/caio-0.12.9/caio/linux_aio.c --- old/caio-0.12.4/caio/linux_aio.c 2026-09-07 08:50:37.244893800 +0200 +++ new/caio-0.12.9/caio/linux_aio.c 2026-09-26 11:48:32.227241300 +0200 @@ -45,6 +45,12 @@ static const unsigned EV_MAX_REQUESTS_DEFAULT = 512; static int kernel_support = -1; +/* Raw io_getevents uses the kernel's two-long timeout layout. */ +struct caio_timespec { + long tv_sec; + long tv_nsec; +}; + inline static int io_setup(unsigned nr, aio_context_t *ctxp) { return syscall(__NR_io_setup, nr, ctxp); } @@ -57,7 +63,7 @@ inline static int io_getevents( aio_context_t ctx, long min_nr, long max_nr, - struct io_event *events, struct timespec *timeout + struct io_event *events, struct caio_timespec *timeout ) { return syscall(__NR_io_getevents, ctx, min_nr, max_nr, events, timeout); } @@ -257,8 +263,9 @@ return NULL; } return PyUnicode_FromFormat( - "<%s as %p: max_requests=%i, ctx=%lli>", - Py_TYPE(self)->tp_name, self, self->max_requests, self->ctx + "<%s as %p: max_requests=%u, ctx=%llu>", + Py_TYPE(self)->tp_name, self, self->max_requests, + (unsigned long long) self->ctx ); } @@ -462,7 +469,7 @@ uint32_t min_requests = 0; uint32_t max_requests = 0; int32_t tv_sec = 0; - struct timespec timeout = {0, 0}; + struct caio_timespec timeout = {0, 0}; static char *kwlist[] = {"max_requests", "min_requests", "timeout", NULL}; @@ -476,7 +483,7 @@ * (unlike linux_uring, whose io_uring_enter() has no native timeout * parameter at all). timeout=0 is a non-blocking check; timeout>0 * bounds the wait. */ - struct timespec *timeout_arg = NULL; + struct caio_timespec *timeout_arg = NULL; if (tv_sec >= 0) { timeout.tv_sec = tv_sec; timeout_arg = &timeout; @@ -628,7 +635,7 @@ }, { "max_requests", - T_USHORT, + T_UINT, offsetof(AIOContext, max_requests), READONLY, "max requests" @@ -744,9 +751,11 @@ } return PyUnicode_FromFormat( - "<%s at %p: mode=\"%s\", fd=%i, offset=%i, buffer=%p>", + "<%s at %p: mode=\"%s\", fd=%u, offset=%lld, buffer=%p>", Py_TYPE(self)->tp_name, self, mode, - self->iocb.aio_fildes, self->iocb.aio_offset, self->iocb.aio_buf + self->iocb.aio_fildes, + (long long) self->iocb.aio_offset, + (void *)(uintptr_t) self->iocb.aio_buf ); } @@ -801,12 +810,21 @@ return NULL; } + if (nbytes > (uint64_t) PY_SSIZE_T_MAX) { + Py_DECREF(self); + PyErr_SetString( + PyExc_OverflowError, + "nbytes does not fit in Py_ssize_t" + ); + return NULL; + } + /* PyMem_Calloc can return NULL for a large enough (or just OOM-at-the- * time) nbytes - proceeding with a NULL buf would hand the kernel (via * aio_buf) and PyMemoryView_FromMemory a NULL pointer with a nonzero * declared size, corrupting memory instead of raising a catchable * error. */ - self->buffer = PyMem_Calloc(nbytes, sizeof(char)); + self->buffer = PyMem_Calloc((size_t) nbytes, sizeof(char)); if (self->buffer == NULL && nbytes > 0) { Py_DECREF(self); PyErr_NoMemory(); @@ -814,7 +832,9 @@ } self->iocb.aio_buf = (uint64_t)(uintptr_t) self->buffer; self->iocb.aio_nbytes = nbytes; - self->py_buffer = PyMemoryView_FromMemory(self->buffer, nbytes, PyBUF_READ); + self->py_buffer = PyMemoryView_FromMemory( + self->buffer, (Py_ssize_t) nbytes, PyBUF_READ + ); if (self->py_buffer == NULL) { Py_DECREF(self); return NULL; diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/caio-0.12.4/caio/linux_uring.c new/caio-0.12.9/caio/linux_uring.c --- old/caio-0.12.4/caio/linux_uring.c 2026-09-07 08:50:37.244893800 +0200 +++ new/caio-0.12.9/caio/linux_uring.c 2026-09-26 11:48:32.227241300 +0200 @@ -219,6 +219,15 @@ return NULL; } + if (nbytes > UINT32_MAX || nbytes > (uint64_t) PY_SSIZE_T_MAX) { + Py_DECREF(self); + PyErr_SetString( + PyExc_OverflowError, + "nbytes exceeds the io_uring read limit" + ); + return NULL; + } + /* Allocate the result bytes object directly — the kernel writes into * its internal buffer, so get_value() can return it with no copy. * PyBytes_FromStringAndSize(NULL, n) leaves the memory uninitialized @@ -600,6 +609,12 @@ uint8_t no_sqarray; /* IORING_SETUP_NO_SQARRAY was used */ uint8_t sqpoll; /* IORING_SETUP_SQPOLL was used */ + /* Total CQEs retired by uring_drain_cq() over the Context lifetime. + * A blocking process_events() compares it with the value at entry, so + * a completion that flush() or another thread drained first still + * ends the wait instead of leaving it blocked on the eventfd. */ + uint64_t cq_consumed; + PyObject *weakreflist; } AIOContext; @@ -636,6 +651,7 @@ self->sqes = MAP_FAILED; self->no_sqarray = 0; self->sqpoll = 0; + self->cq_consumed = 0; } return (PyObject *) self; } @@ -673,8 +689,8 @@ * * Opt-in (sqpoll=True): SQPOLL kernel thread polls SQ ring; io_uring_enter * only needed to wake a sleeping thread. Eliminates per-op syscall - * overhead at sustained high QD. EPERM on pre-5.11 kernels without - * CAP_SYS_NICE is treated like EINVAL (try next entry). + * overhead at sustained high QD. Permission errors are treated like + * EINVAL (try next entry), allowing fallback to a plain ring. * * IORING_SETUP_SINGLE_ISSUER deliberately never tried: it pins the ring * to whichever thread's io_uring_setup()/io_uring_enter() call created @@ -710,7 +726,7 @@ flags_used = params.flags; break; } - if (errno != EINVAL && errno != EPERM) { + if (errno != EINVAL && errno != EPERM && errno != EACCES) { PyErr_SetFromErrno(PyExc_SystemError); return -1; } @@ -861,6 +877,7 @@ 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; + uint32_t start_head = head; uint32_t avail = tail - head; uint32_t cap = avail < max ? avail : max; @@ -911,6 +928,7 @@ /* Ring state fully committed - reentrant callers now see this whole * batch as already consumed, before a single callback has run. */ + __atomic_add_fetch(&self->cq_consumed, head - start_head, __ATOMIC_RELEASE); __atomic_store_n(self->cq_head, head, __ATOMIC_RELEASE); CAIO_END_CRITICAL_SECTION(); @@ -1184,8 +1202,33 @@ " indefinitely. Both are for manual synchronous polling from a plain\n" " thread - do not also select()/poll() on `.fileno` from elsewhere\n" " while relying on this to wait, the two waiting mechanisms are\n" - " alternatives, not meant to be combined on the same Context." + " alternatives, not meant to be combined on the same Context.\n" + " The wait ends when min_requests completions occurred, even if a\n" + " concurrent flush() or process_events() already collected them -\n" + " the return value then counts only what this call collected, so\n" + " it can be lower than min_requests, including 0." ); +/* + * Return nonzero when at least min_requests CQEs became available since a + * blocking process_events() started to wait. Count the CQEs still in the + * ring plus the CQEs that some drain (flush() in another thread, a + * reentrant process_events()) retired after `consumed_at_entry`. A wait + * that only watched the ring could miss a completion twice: it wakes, + * finds the ring already empty, and blocks again on an eventfd counter + * that it drained itself - or it sees the CQE, loses the GIL race to + * flush()'s own drain, and returns 0 without any signal that the + * completion happened. Both showed up deterministically on a single CPU. + */ +static inline int uring_wait_satisfied( + AIOContext *self, uint32_t min_requests, uint64_t consumed_at_entry +) { + uint32_t head = __atomic_load_n(self->cq_head, __ATOMIC_RELAXED); + uint32_t tail = __atomic_load_n(self->cq_tail, __ATOMIC_ACQUIRE); + uint64_t consumed = __atomic_load_n(&self->cq_consumed, __ATOMIC_ACQUIRE); + uint64_t seen = (uint64_t) (tail - head) + (consumed - consumed_at_entry); + return seen >= min_requests; +} + static PyObject *AIOContext_process_events( AIOContext *self, PyObject *args, PyObject *kwds ) { @@ -1241,11 +1284,11 @@ } int saved_errno = 0; + uint64_t consumed_at_entry = + __atomic_load_n(&self->cq_consumed, __ATOMIC_ACQUIRE); Py_BEGIN_ALLOW_THREADS for (;;) { - uint32_t head = __atomic_load_n(self->cq_head, __ATOMIC_RELAXED); - uint32_t tail = __atomic_load_n(self->cq_tail, __ATOMIC_ACQUIRE); - if (tail - head >= min_requests) + if (uring_wait_satisfied(self, min_requests, consumed_at_entry)) break; int ret = io_uring_enter( @@ -1256,9 +1299,7 @@ break; } - head = __atomic_load_n(self->cq_head, __ATOMIC_RELAXED); - tail = __atomic_load_n(self->cq_tail, __ATOMIC_ACQUIRE); - if (tail - head >= min_requests) + if (uring_wait_satisfied(self, min_requests, consumed_at_entry)) break; int poll_timeout_ms = -1; /* poll()'s own "block indefinitely" */ diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/caio-0.12.4/caio/src/threadpool/threadpool.c new/caio-0.12.9/caio/src/threadpool/threadpool.c --- old/caio-0.12.4/caio/src/threadpool/threadpool.c 2026-09-07 08:50:37.246019800 +0200 +++ new/caio-0.12.9/caio/src/threadpool/threadpool.c 2026-09-26 11:48:32.228455500 +0200 @@ -82,6 +82,9 @@ int count; int shutdown; int started; + /* Set by threadpool_destroy() when a worker thread destroys its own + pool. The last worker to exit then frees the pool. */ + int free_on_exit; }; /** @@ -112,6 +115,7 @@ pool->queue_size = queue_size; pool->head = pool->tail = pool->count = 0; pool->shutdown = pool->started = 0; + pool->free_on_exit = 0; /* Allocate thread and task queue */ pool->threads = (pthread_t *)malloc(sizeof(pthread_t) * thread_count); @@ -199,6 +203,8 @@ int threadpool_destroy(threadpool_t *pool, int flags) { int i, err = 0; + int self_in_pool = 0; + pthread_t self = pthread_self(); if(pool == NULL) { return threadpool_invalid; @@ -225,19 +231,40 @@ break; } - /* Join all worker thread */ + /* Join all worker threads. The caller can be one of them: a task + can release the last reference to the object that owns this + pool, so the owner's destructor runs on that worker. A thread + cannot join itself (pthread_join returns EDEADLK), which used + to skip threadpool_free() and leak the pool and the thread. + Detach that thread instead; it frees the pool when it exits + (see threadpool_thread). */ for(i = 0; i < pool->thread_count; i++) { + if(pthread_equal(pool->threads[i], self)) { + pthread_detach(pool->threads[i]); + self_in_pool = 1; + continue; + } if(pthread_join(pool->threads[i], NULL) != 0) { err = threadpool_thread_failure; } } } while(0); - /* Only if everything went well do we deallocate the pool */ - if(!err) { - threadpool_free(pool); + if(err) { + return err; } - return err; + + if(self_in_pool) { + /* Every other worker is joined; only the calling worker still + runs. It reads free_on_exit under the lock when it exits. */ + pthread_mutex_lock(&(pool->lock)); + pool->free_on_exit = 1; + pthread_mutex_unlock(&(pool->lock)); + return 0; + } + + threadpool_free(pool); + return 0; } int threadpool_free(threadpool_t *pool) @@ -251,10 +278,11 @@ free(pool->threads); free(pool->queue); - /* Because we allocate pool->threads after initializing the - mutex and condition variable, we're sure they're - initialized. Let's lock the mutex just in case. */ - pthread_mutex_lock(&(pool->lock)); + /* pool->threads is allocated after the mutex and the condition + variable are initialized, so both exist here. Every worker has + been joined (started == 0), so no thread holds the lock. Do not + lock it: pthread_mutex_destroy() on a locked mutex is undefined + behavior, and ThreadSanitizer reports it on every close(). */ pthread_mutex_destroy(&(pool->lock)); pthread_cond_destroy(&(pool->notify)); } @@ -298,8 +326,14 @@ } pool->started--; + /* The worker that destroyed its own pool frees it once it is the last + one out. No other thread touches the pool after this point. */ + int free_pool = pool->free_on_exit && pool->started == 0; pthread_mutex_unlock(&(pool->lock)); + if(free_pool) { + threadpool_free(pool); + } pthread_exit(NULL); return(NULL); } diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/caio-0.12.4/caio/thread_aio.c new/caio-0.12.9/caio/thread_aio.c --- old/caio-0.12.4/caio/thread_aio.c 2026-09-07 08:50:37.246019800 +0200 +++ new/caio-0.12.9/caio/thread_aio.c 2026-09-26 11:48:32.228455500 +0200 @@ -47,7 +47,7 @@ PyObject_HEAD threadpool_t* pool; uint16_t max_requests; - uint8_t pool_size; + uint16_t pool_size; PyObject* weakreflist; } AIOContext; @@ -153,6 +153,7 @@ self->pool = NULL; self->max_requests = 0; + self->pool_size = 0; if (!PyArg_ParseTupleAndKeywords( args, kwds, "|HH", kwlist, @@ -205,9 +206,9 @@ return NULL; } return PyUnicode_FromFormat( - "<%s as %p: max_requests=%i, pool_size=%i, ctx=%lli>", + "<%s as %p: max_requests=%u, pool_size=%u, ctx=%p>", Py_TYPE(self)->tp_name, self, self->max_requests, - self->pool_size, self->pool + self->pool_size, (void *) self->pool ); } @@ -456,7 +457,7 @@ static PyMemberDef AIOContext_members[] = { { "pool_size", - T_INT, + T_USHORT, offsetof(AIOContext, pool_size), READONLY, "pool_size" @@ -581,9 +582,9 @@ } return PyUnicode_FromFormat( - "<%s at %p: mode=\"%s\", fd=%i, offset=%i, result=%i, buffer=%p>", + "<%s at %p: mode=\"%s\", fd=%u, offset=%lld, result=%d, buffer=%p>", Py_TYPE(self)->tp_name, self, mode, - self->fileno, self->offset, self->result, self->buf + self->fileno, (long long) self->offset, self->result, self->buf ); } @@ -636,18 +637,27 @@ return NULL; } + if (nbytes > (uint64_t) PY_SSIZE_T_MAX) { + Py_DECREF(self); + PyErr_SetString( + PyExc_OverflowError, + "nbytes does not fit in Py_ssize_t" + ); + return NULL; + } + // PyMem_Calloc can return NULL for a large enough (or just // OOM-at-the-time) nbytes - proceeding with a NULL buf would hand the // kernel (via pread() in worker()) and PyMemoryView_FromMemory a NULL // pointer with a nonzero declared size, corrupting memory instead of // raising a catchable error. - self->buf = PyMem_Calloc(nbytes, sizeof(char)); + self->buf = PyMem_Calloc((size_t) nbytes, sizeof(char)); if (self->buf == NULL && nbytes > 0) { Py_DECREF(self); PyErr_NoMemory(); return NULL; } - self->buf_size = nbytes; + self->buf_size = (Py_ssize_t) nbytes; self->py_buffer = PyMemoryView_FromMemory( self->buf, @@ -992,7 +1002,7 @@ READONLY, "offset" }, { - "nbytes", T_ULONGLONG, + "nbytes", T_PYSSIZET, offsetof(AIOOperation, buf_size), READONLY, "nbytes" }, diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/caio-0.12.4/caio.egg-info/PKG-INFO new/caio-0.12.9/caio.egg-info/PKG-INFO --- old/caio-0.12.4/caio.egg-info/PKG-INFO 2026-09-07 08:50:42.379699200 +0200 +++ new/caio-0.12.9/caio.egg-info/PKG-INFO 2026-09-26 11:48:40.237560700 +0200 @@ -1,6 +1,6 @@ Metadata-Version: 2.4 Name: caio -Version: 0.12.4 +Version: 0.12.9 Summary: Asynchronous file IO for Linux MacOS or Windows. Author-email: Dmitry Orlov <[email protected]> License-Expression: Apache-2.0 @@ -19,6 +19,7 @@ Classifier: Programming Language :: Python :: 3.12 Classifier: Programming Language :: Python :: 3.13 Classifier: Programming Language :: Python :: 3.14 +Classifier: Programming Language :: Python :: 3.15 Classifier: Programming Language :: Python :: Implementation :: CPython Classifier: Programming Language :: Python :: Free Threading :: 2 - Beta Requires-Python: >=3.10 diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/caio-0.12.4/pyproject.toml new/caio-0.12.9/pyproject.toml --- old/caio-0.12.4/pyproject.toml 2026-09-07 08:50:38.943673400 +0200 +++ new/caio-0.12.9/pyproject.toml 2026-09-26 11:48:34.141161000 +0200 @@ -4,7 +4,7 @@ [project] name = "caio" -version = "0.12.4" +version = "0.12.9" license = "Apache-2.0" license-files = ["COPYING"] description = "Asynchronous file IO for Linux MacOS or Windows." @@ -26,6 +26,7 @@ "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", + "Programming Language :: Python :: 3.15", "Programming Language :: Python :: Implementation :: CPython", "Programming Language :: Python :: Free Threading :: 2 - Beta", ] diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/caio-0.12.4/tests/conftest.py new/caio-0.12.9/tests/conftest.py --- old/caio-0.12.4/tests/conftest.py 2026-09-07 08:50:37.246019800 +0200 +++ new/caio-0.12.9/tests/conftest.py 2026-09-26 11:48:32.228455500 +0200 @@ -1,4 +1,5 @@ import functools +import importlib import threading import time import types @@ -17,6 +18,20 @@ ) +def import_backend_or_skip(name): + """Import a caio backend module, or skip the test when the host lacks it. + + pytest.importorskip() skips only on ModuleNotFoundError since pytest + 8.2. A backend whose kernel probe fails (io_uring blocked by seccomp, + kernel AIO missing under qemu-user) raises a plain ImportError with an + explanation, so importorskip() reports an error instead of a skip. + """ + try: + return importlib.import_module(name) + except ImportError as exc: + pytest.skip(f"{name} is unavailable here: {exc}") + + class ConcurrentThreads: """Run test workers and surface their failures in the main thread.""" diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/caio-0.12.4/tests/test_aio_context.py new/caio-0.12.9/tests/test_aio_context.py --- old/caio-0.12.4/tests/test_aio_context.py 2026-09-07 08:50:37.246884000 +0200 +++ new/caio-0.12.9/tests/test_aio_context.py 2026-09-26 11:48:32.229390000 +0200 @@ -10,7 +10,7 @@ ctx = context_maker(1) assert ctx is not None - ctx = context_maker(32218) + ctx = context_maker(128) assert ctx is not None diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/caio-0.12.4/tests/test_asyncio_adapter.py new/caio-0.12.9/tests/test_asyncio_adapter.py --- old/caio-0.12.4/tests/test_asyncio_adapter.py 2026-09-07 08:50:37.246884000 +0200 +++ new/caio-0.12.9/tests/test_asyncio_adapter.py 2026-09-26 11:48:32.229390000 +0200 @@ -5,11 +5,12 @@ import aiomisc import pytest +from conftest import import_backend_or_skip @aiomisc.timeout(5) async def test_linux_uring_asyncio_forwards_context_kwargs(): - uring_asyncio = pytest.importorskip("caio.linux_uring_asyncio") + uring_asyncio = import_backend_or_skip("caio.linux_uring_asyncio") async with uring_asyncio.AsyncioContext( max_requests=8, @@ -137,15 +138,26 @@ async def test_cancel_before_first_step_runs(tmp_path, async_context, asyncio_exception_handler): """Cancelling right after the op's own first step (submit queued, still suspended at `await future`) - covers context.cancel() raising ValueError - for an op the backend never actually got to submit to the kernel yet.""" + for an op the backend never actually got to submit to the kernel yet. + + The op can also complete before cancel() runs: a non-SQPOLL flush() + finishes a page-cache write inline, and on a single CPU the SQPOLL + kernel thread usually completes it before this coroutine resumes + (GitHub #79). That outcome is valid too - then the task must return + the real write result, not swallow it. The linux_uring[deferred=True] + variant always suspends first and keeps the cancel path covered.""" context = async_context with open(str(tmp_path / "temp.bin"), "wb+") as fp: # noqa: ASYNC230 fd = fp.fileno() task = asyncio.ensure_future(context.write(b"x", fd, 0)) await asyncio.sleep(0) task.cancel() - with pytest.raises(asyncio.CancelledError): - await task + try: + written = await task + except asyncio.CancelledError: + pass + else: + assert written == 1 asyncio_exception_handler.assert_not_called() diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/caio-0.12.4/tests/test_free_threading.py new/caio-0.12.9/tests/test_free_threading.py --- old/caio-0.12.4/tests/test_free_threading.py 2026-09-07 08:50:37.246884000 +0200 +++ new/caio-0.12.9/tests/test_free_threading.py 2026-09-26 11:48:32.229390000 +0200 @@ -44,7 +44,7 @@ fd = os.open(str(path), os.O_RDWR | _O_BINARY) ctx = None try: - ctx = python_aio.Context(max_requests=count, pool_size=32) + ctx = python_aio.Context(max_requests=count, pool_size=8) expected = [bytes([i % 256]) * chunk for i in range(count)] writes = ( diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/caio-0.12.4/tests/test_raw_low_level.py new/caio-0.12.9/tests/test_raw_low_level.py --- old/caio-0.12.4/tests/test_raw_low_level.py 2026-09-07 08:50:37.246884000 +0200 +++ new/caio-0.12.9/tests/test_raw_low_level.py 2026-09-26 11:48:32.229390000 +0200 @@ -18,7 +18,7 @@ import weakref import pytest -from conftest import drain +from conftest import drain, import_backend_or_skip ABSURD_NBYTES = 2**62 @@ -379,7 +379,7 @@ objects and - until something else eventually dropped the Context - the underlying kernel AIO context/eventfd too. """ - linux_aio = pytest.importorskip("caio.linux_aio") + linux_aio = import_backend_or_skip("caio.linux_aio") with open(str(tmp_path / "temp.bin"), "wb+") as f: fd = f.fileno() @@ -416,7 +416,7 @@ mid-flight, munmapping the SQ/CQ rings and closing uring_fd while the kernel might still be touching them. """ - linux_uring = pytest.importorskip("caio.linux_uring") + linux_uring = import_backend_or_skip("caio.linux_uring") with open(str(tmp_path / "temp.bin"), "wb+") as f: fd = f.fileno() @@ -448,15 +448,15 @@ op.context clears and nothing else references the Context, so the weakref must clear too. """ - linux_uring = pytest.importorskip("caio.linux_uring") + linux_uring = import_backend_or_skip("caio.linux_uring") import fcntl # Unix-only; this whole test is uring-only (Linux-only) F_SETPIPE_SZ = 1031 r_fd, w_fd = os.pipe() try: - fcntl.fcntl(w_fd, F_SETPIPE_SZ, 4096) - os.write(w_fd, b"f" * 4096) # fill the pipe so the next write blocks until drained + pipe_size = fcntl.fcntl(w_fd, F_SETPIPE_SZ, 4096) + os.write(w_fd, b"f" * pipe_size) # fill the pipe so the next write blocks until drained ctx = linux_uring.Context(max_requests=8) ctx_ref = weakref.ref(ctx) @@ -476,7 +476,7 @@ "in flight - op.context should have kept it alive" ) - os.read(r_fd, 8192) # unblocks the pending write + os.read(r_fd, pipe_size) # unblocks the pending write drain(ctx_ref(), 1, timeout=5.0) assert op.get_value() == len(b"pending") @@ -509,7 +509,7 @@ that one, which plain refcounting alone (as exercised here, with the cyclic GC disabled) genuinely cannot free - it needs `gc.collect()`. """ - linux_aio = pytest.importorskip("caio.linux_aio") + linux_aio = import_backend_or_skip("caio.linux_aio") gc.disable() try: @@ -608,7 +608,7 @@ above: submit() must atomically claim each op, so the kernel never ends up with two concurrent iocbs pointing at the very same buffer. """ - linux_aio = pytest.importorskip("caio.linux_aio") + linux_aio = import_backend_or_skip("caio.linux_aio") with open(str(tmp_path / "temp.bin"), "wb+") as f: fd = f.fileno() @@ -719,7 +719,7 @@ ever reaching its own, symmetric completion-time error handling - so the same scenario can't be driven through linux_aio's public API. """ - linux_uring = pytest.importorskip("caio.linux_uring") + linux_uring = import_backend_or_skip("caio.linux_uring") bad_path = tmp_path / "closed.bin" good_path = tmp_path / "good.bin" @@ -773,7 +773,7 @@ isn't guaranteed), but this assertion is unconditionally the correct behavior regardless. """ - linux_uring = pytest.importorskip("caio.linux_uring") + linux_uring = import_backend_or_skip("caio.linux_uring") nbytes = 256 for _ in range(64): @@ -805,7 +805,7 @@ there is no later kernel write into it. A fresh Operation started against the same fd afterward must get its own, independent buffer. """ - linux_uring = pytest.importorskip("caio.linux_uring") + linux_uring = import_backend_or_skip("caio.linux_uring") path = tmp_path / "data.bin" path.write_bytes(b"AAAA") @@ -858,7 +858,7 @@ being asserted on - flaky, not a real bug. Capacity 1 makes "the op accepted before the overflow point" unambiguously just `ops[0]`. """ - linux_uring = pytest.importorskip("caio.linux_uring") + linux_uring = import_backend_or_skip("caio.linux_uring") with open(str(tmp_path / "temp.bin"), "wb+") as f: fd = f.fileno() @@ -891,7 +891,7 @@ reconstruct the same "kernel-owned" reference a second time, causing a double-decref once both reconstructions eventually drop. """ - linux_uring = pytest.importorskip("caio.linux_uring") + linux_uring = import_backend_or_skip("caio.linux_uring") with open(str(tmp_path / "temp.bin"), "wb+") as f: fd = f.fileno() @@ -935,7 +935,7 @@ permanently stuck in_flight - it must never have been touched at all, since the whole call raises before anything reaches io_submit(). """ - linux_aio = pytest.importorskip("caio.linux_aio") + linux_aio = import_backend_or_skip("caio.linux_aio") with open(str(tmp_path / "temp.bin"), "wb+") as f: fd = f.fileno() @@ -998,17 +998,26 @@ (not return early or misinterpret negative as "don't wait"), and it must actually wake up and return once a real completion satisfies min_requests, not stay stuck forever once given something to wait for. + + The waiter may legitimately return 0: linux_uring's flush() drains + inline completions itself, and on a single CPU that drain reliably + wins the race against the waiter (GitHub #79). The wait must still end + - the completion is verified through the operation's own callback, not + through the waiter's return value. """ with open(str(tmp_path / "temp.bin"), "wb+") as f: fd = f.fileno() ctx = polling_backend.Context(max_requests=8) result = {} + completed = threading.Event() def wait(): result["n"] = ctx.process_events(min_requests=1, timeout=-1) - t = threading.Thread(target=wait) + # daemon: a waiter that never wakes up must fail this test, not + # hang interpreter shutdown afterwards. + t = threading.Thread(target=wait, daemon=True) t.start() try: t.join(timeout=0.3) @@ -1019,6 +1028,7 @@ ) op = polling_backend.Operation.write(b"x" * 4, fd, 0) + op.set_callback(lambda _r: completed.set()) ctx.submit(op) if hasattr(ctx, "flush"): ctx.flush() @@ -1031,7 +1041,8 @@ finally: t.join(timeout=5.0) - assert result.get("n", 0) >= 1 + assert completed.wait(timeout=5.0), "the write never completed" + assert result["n"] in (0, 1), result def test_uring_process_events_max_requests_bounds_callbacks_not_just_return_value(tmp_path): @@ -1051,15 +1062,15 @@ letting max_requests bound exactly how many of them get delivered per call, not just how many get reported. """ - linux_uring = pytest.importorskip("caio.linux_uring") + linux_uring = import_backend_or_skip("caio.linux_uring") import fcntl # Unix-only; this whole test is uring-only (Linux-only) F_SETPIPE_SZ = 1031 pipes = [os.pipe() for _ in range(5)] try: for r_fd, w_fd in pipes: - fcntl.fcntl(w_fd, F_SETPIPE_SZ, 4096) - os.write(w_fd, b"f" * 4096) # fill each pipe so the next write blocks + pipe_size = fcntl.fcntl(w_fd, F_SETPIPE_SZ, 4096) + os.write(w_fd, b"f" * pipe_size) # fill each pipe so the next write blocks ctx = linux_uring.Context(max_requests=16) called = [] @@ -1078,7 +1089,7 @@ ) for r_fd, _w_fd in pipes: - os.read(r_fd, 8192) + os.read(r_fd, pipe_size) time.sleep(0.05) # let the kernel actually post the completions first = ctx.process_events(max_requests=1, min_requests=0, timeout=0) @@ -1113,7 +1124,7 @@ would satisfy min_requests=1 immediately, so the call returns 0 real completions almost instantly, well before that 0.2s delay elapses. """ - linux_uring = pytest.importorskip("caio.linux_uring") + linux_uring = import_backend_or_skip("caio.linux_uring") import fcntl # Unix-only; this whole test is uring-only (Linux-only) F_SETPIPE_SZ = 1031 @@ -1122,8 +1133,8 @@ fd = f.fileno() r_fd, w_fd = os.pipe() try: - fcntl.fcntl(w_fd, F_SETPIPE_SZ, 4096) - os.write(w_fd, b"f" * 4096) # fill the pipe so the next write blocks + pipe_size = fcntl.fcntl(w_fd, F_SETPIPE_SZ, 4096) + os.write(w_fd, b"f" * pipe_size) # fill the pipe so the next write blocks ctx = linux_uring.Context(max_requests=16) @@ -1150,7 +1161,7 @@ def drain_pipe_late(): time.sleep(0.2) drained_before_call_returned.set() - os.read(r_fd, 8192) + os.read(r_fd, pipe_size) t = threading.Thread(target=drain_pipe_late) t.start() @@ -1180,7 +1191,7 @@ whichever task's io_uring_setup()/io_uring_enter() call created it and rejects submission from any other task with -EEXIST. """ - linux_uring = pytest.importorskip("caio.linux_uring") + linux_uring = import_backend_or_skip("caio.linux_uring") with open(str(tmp_path / "temp.bin"), "wb+") as f: fd = f.fileno() @@ -1219,7 +1230,7 @@ available if some other reference to the Operation survives), just silently, with no callback call. """ - linux_uring = pytest.importorskip("caio.linux_uring") + linux_uring = import_backend_or_skip("caio.linux_uring") with open(str(tmp_path / "temp.bin"), "wb+") as f: fd = f.fileno() @@ -1307,7 +1318,7 @@ it in Engine::cancel() before ever touching that Context's own registry. """ - linux_aio = pytest.importorskip("caio.linux_aio") + linux_aio = import_backend_or_skip("caio.linux_aio") path_a = tmp_path / "a.bin" path_b = tmp_path / "b.bin" @@ -1345,7 +1356,7 @@ own, same-numbered request completes normally and is left untouched by a cancel() call made through a different Context. """ - linux_uring = pytest.importorskip("caio.linux_uring") + linux_uring = import_backend_or_skip("caio.linux_uring") path_a = tmp_path / "a.bin" path_b = tmp_path / "b.bin" diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/caio-0.12.4/tests/test_thread_aio_specific.py new/caio-0.12.9/tests/test_thread_aio_specific.py --- old/caio-0.12.4/tests/test_thread_aio_specific.py 2026-09-07 08:50:37.246884000 +0200 +++ new/caio-0.12.9/tests/test_thread_aio_specific.py 2026-09-26 11:48:32.229390000 +0200 @@ -4,7 +4,11 @@ the cross-backend parametrized suite. Skipped outright wherever thread_aio itself isn't available. """ +import gc +import sys import threading +import time +import weakref import pytest @@ -28,6 +32,12 @@ most one op can ever be raced away like that, making the LAST op in a 5-op batch guaranteed to overflow regardless of exactly how that race resolves for the first one. + + The raced-away first op means a second op can sit in the one-slot + queue when the first op's callback fires, so the resubmit below can + still hit a full queue for a moment - reliably so on a single CPU + (GitHub #79). Retry until it is accepted: the contract under test is + "not permanently stuck", not "accepted on the first try". """ with open(str(tmp_path / "temp.bin"), "wb+") as f: fd = f.fileno() @@ -52,7 +62,15 @@ done = threading.Event() rejected.set_callback(lambda _r: done.set()) - resubmitted = ctx.submit(rejected) + deadline = time.monotonic() + 30.0 + while True: + try: + resubmitted = ctx.submit(rejected) + except RuntimeError: + assert time.monotonic() < deadline, "queue never drained for the retry" + time.sleep(0.001) + else: + break assert resubmitted == 1, "a queue-rejected operation must not be permanently stuck" assert done.wait(timeout=30.0), "retried operation must actually run" assert rejected.result == len(payload), f"expected a full write, got result={rejected.result}" @@ -62,3 +80,73 @@ assert rf.read(len(payload)) == payload, ( "resubmitted operation must write its ORIGINAL payload, not lost/empty data" ) + + +def _vm_data_mb(): + """Virtual data size of this process in MiB, from /proc (Linux only).""" + with open("/proc/self/status") as status: + for line in status: + if line.startswith("VmData:"): + return int(line.split()[1]) // 1024 + raise RuntimeError("VmData not found") + + [email protected](sys.platform != "linux", reason="uses /proc/self/status") +def test_context_released_by_its_own_worker_frees_the_pool(tmp_path): + """A completion callback can drop the last reference to the Context. + The worker that runs the callback then runs the Context destructor + and tears the pool down from inside one of its own threads. + + pthread_join() of the calling thread fails with EDEADLK. That used to + skip freeing the pool, so every such Context leaked its queue and the + unjoined worker's 8 MiB stack. VmData counts that memory even though + it is never touched, so the growth over 40 iterations is ~360 MiB + with the leak and a few MiB without it. + + The callback blocks until the main thread has dropped its reference, + so the worker's own DECREF is the last one. On a free-threaded build + biased reference counting can still move the dealloc to the owning + thread, so this test proves the leak only on GIL builds; it still + exercises the teardown path on both. + """ + iterations = 40 + with open(str(tmp_path / "temp.bin"), "wb+") as f: + fd = f.fileno() + keep = [] + dealloc_threads = set() + + def one(): + done = threading.Event() + main_dropped = threading.Event() + ctx = thread_aio.Context(max_requests=60000, pool_size=2) + weakref.finalize( + ctx, + lambda: dealloc_threads.add(threading.current_thread().name), + ) + op = thread_aio.Operation.write(b"x", fd, 0) + + def callback(_result): + main_dropped.wait(5) + done.set() + + op.set_callback(callback) + assert ctx.submit(op) == 1 + del ctx + main_dropped.set() + assert done.wait(5), "write never completed" + keep.append(op) + + one() # warm up allocator arenas before measuring + gc.collect() + before = _vm_data_mb() + for _ in range(iterations): + one() + gc.collect() + time.sleep(0.2) + growth = _vm_data_mb() - before + + assert dealloc_threads, "no Context was deallocated" + assert growth < 100, ( + f"VmData grew by {growth} MiB over {iterations} Contexts released " + f"on {sorted(dealloc_threads)} - the pool or a worker stack leaks" + )
