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/.github/workflows/ci.yml b/.github/workflows/ci.yml index f74d6fd..2f1d0fb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -44,6 +44,17 @@ jobs: os: ubuntu-latest - python: "3.14" os: ubuntu-latest + # 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" @@ -75,12 +86,41 @@ 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 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/.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/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 = [ 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 de42f00..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: @@ -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: - self._in_progress -= 1 + with operation._lock, self._lock: operation.in_progress = False + self._in_progress -= 1 def _execute(self, operation: "Operation") -> bool: """ @@ -133,12 +134,11 @@ 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: + # 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 @@ -150,8 +150,8 @@ def on_success(result): "Maximum simultaneous requests have been reached", ) - self._in_progress += 1 operation.in_progress = True + self._in_progress += 1 try: self.pool.apply_async( @@ -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) @@ -305,6 +319,7 @@ def __init__( self.callback: Callable[[int], Any] | None = None self.in_progress = False + self._lock = Lock() self.buffer: bytes = buffer self.opcode = opcode @@ -391,5 +406,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..8cf7925 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" @@ -878,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; @@ -1009,6 +1049,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/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/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 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 new file mode 100644 index 0000000..f50518f --- /dev/null +++ b/tests/test_free_threading.py @@ -0,0 +1,761 @@ +import gc +import os +import select +import sys +import sysconfig +import threading +import time +import weakref + +import pytest + +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.""" + if not sysconfig.get_config_var("Py_GIL_DISABLED"): + pytest.skip("not a free-threaded build") + 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, 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 + 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 | _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)] + + 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 + + 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(workers): + """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 = workers.barrier(3) + finished = workers.barrier(3) + current = [None] + submitted = [0, 0] + + def submitter(index, context): + for _ in range(iterations): + start.wait() + submitted[index] += context.submit(current[0]) + finished.wait() + + try: + workers.start_many(submitter, enumerate(contexts)) + + for _ in range(iterations): + current[0] = python_aio.Operation( + 0, None, None, python_aio.OpCode.NOOP, + ) + start.wait() + finished.wait() + + workers.join() + assert sum(submitted) == iterations, ( + f"{sum(submitted) - iterations} Operations were submitted twice" + ) + finally: + workers.close() + 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, + workers, +): + """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 | _O_BINARY, 0o600) + contexts = tuple( + backend.Context(max_requests=claim_window) + for _ in range(ft_claimant_count) + ) + 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() + + def on_done(_result): + with callback_lock: + callback_count[0] += 1 + + def submitter(index, context): + for _ in range(iterations): + start.wait() + submitted[index] += context.submit(current[0]) + finished.wait() + + try: + workers.start_many(submitter, enumerate(contexts)) + + 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, + ) + + workers.join() + 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: + workers.close() + _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, + workers, +): + """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 | _O_BINARY, 0o600) + context = backend.Context(max_requests=total) + submit_start = workers.barrier(worker_count) + operations = [] + callback_results = [None] * total + callback_lock = threading.Lock() + submitted = [0] * worker_count + + 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 + 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: + workers.start_many( + submitter, + ((index,) for index in range(worker_count)), + ) + workers.join() + + 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: + workers.close() + _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, + workers, +): + """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 | _O_BINARY, 0o600) + context = polling_backend.Context(max_requests=operation_count) + callback_counts = [0] * operation_count + callback_lock = threading.Lock() + drain_round = workers.barrier(drainer_count) + + 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(): + 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: + workers.start_many(drain, (() for _ in range(drainer_count))) + workers.join() + + 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: + workers.close() + os.close(fd) + + +def test_completed_operation_supports_concurrent_readers( + tmp_path, + backend, + ft_reader_count, + ft_reads_per_thread, + workers, +): + """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 | _O_BINARY) + 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 = workers.barrier(reader_count + 1) + + def read_result(): + start.wait() + for _ in range(reads_per_thread): + if operation.get_value() != payload: + raise AssertionError("concurrent get_value() returned wrong data") + + try: + workers.start_many(read_result, (() for _ in range(reader_count))) + start.wait() + workers.join() + finally: + workers.close() + _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, + workers, +): + """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 | _O_BINARY, 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 = workers.barrier(observer_count + 1) + + def observe(): + 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: + 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" + workers.join(timeout=10) + + assert completed_count[0] == operation_count + finally: + blocker_release.set() + workers.close() + _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, + workers, +): + """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 = workers.barrier(setter_count + 1) + + 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): + 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: + workers.start_many( + replace_callbacks, + ((index,) for index in range(setter_count)), + ) + start.wait() + workers.join() + + # 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: + 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() + 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 = workers.barrier(canceller_count) + cancelled = [0] * canceller_count + + def cancel(thread_index): + for round_index in range(cancel_rounds): + start_round.wait() + cancelled[thread_index] += context.cancel( + targets[round_index * canceller_count + thread_index], + ) + + try: + workers.start_many( + cancel, + ((index,) for index in range(canceller_count)), + ) + workers.join() + + 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: + workers.close() + # 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)