From 38cd5e5b1bc78941dff78cad529c65686457903d Mon Sep 17 00:00:00 2001 From: Nico Ritschel Date: Tue, 20 Oct 2026 09:50:28 -0700 Subject: [PATCH 1/2] Prove concurrent API reads with a barrier instead of timings --- tests/server/test_concurrency.py | 66 +++++++++++++++++++------------- 1 file changed, 40 insertions(+), 26 deletions(-) diff --git a/tests/server/test_concurrency.py b/tests/server/test_concurrency.py index 968da920b..3599bc671 100644 --- a/tests/server/test_concurrency.py +++ b/tests/server/test_concurrency.py @@ -1,16 +1,15 @@ """Concurrency proof for the HTTP API server. Verifies that read-only query handlers no longer serialize on a single lock / -connection: multiple slow queries fired concurrently complete in far less than -the sum of their individual runtimes because each request executes on its own -``adapter.cursor()`` (an independent DuckDB handle over the same database). +connection: multiple queries fired concurrently are inside execute at the same +time because each request executes on its own ``adapter.cursor()`` (an +independent DuckDB handle over the same database). """ # ruff: noqa: E402 from __future__ import annotations -import statistics import threading import time from concurrent.futures import ThreadPoolExecutor @@ -88,34 +87,49 @@ def _post_slow_sql(client: TestClient) -> float: return elapsed -def test_concurrent_reads_do_not_serialize(tmp_path): - client = _build_client(tmp_path) +class _RendezvousCursor: + """Cursor proxy whose ``execute`` waits until every concurrent read is executing.""" + + def __init__(self, cursor, barrier: threading.Barrier): + self._cursor = cursor + self._barrier = barrier + + def execute(self, *args, **kwargs): + self._barrier.wait() + return self._cursor.execute(*args, **kwargs) + + def __getattr__(self, name): + return getattr(self._cursor, name) - # Warm up (first query pays DuckDB init cost) and measure single-request time. - _post_slow_sql(client) - single_times = [_post_slow_sql(client) for _ in range(2)] - single = statistics.median(single_times) +def test_concurrent_reads_do_not_serialize(tmp_path, monkeypatch): + client = _build_client(tmp_path) + adapter = client.app.state.layer.adapter + _post_slow_sql(client) # warm up before patching + + # Every read blocks in execute until all of them are executing at once. Reads + # serialized on a lock or shared connection can never all arrive, so the barrier + # times out; unlike a wall-clock ratio, this does not depend on runner load. concurrency = 4 - start = time.perf_counter() + barrier = threading.Barrier(concurrency, timeout=30) + opened = [] + open_cursor = adapter.cursor + + def rendezvous_cursor(): + cursor = open_cursor() + opened.append(cursor) + return _RendezvousCursor(cursor, barrier) + + monkeypatch.setattr(adapter, "cursor", rendezvous_cursor) + with ThreadPoolExecutor(max_workers=concurrency) as pool: futures = [pool.submit(_post_slow_sql, client) for _ in range(concurrency)] - per_request = [f.result() for f in futures] - concurrent_total = time.perf_counter() - start + for future in futures: + future.result() - print( - f"\nconcurrency proof: single={single * 1000:.0f}ms " - f"concurrent_total({concurrency})={concurrent_total * 1000:.0f}ms " - f"per_request={[f'{t * 1000:.0f}ms' for t in per_request]}" - ) - - # If reads serialized, concurrent_total would approach concurrency * single. - # With per-request cursors it should stay well under 2.5x single. The bound - # is generous to absorb CI runners with fewer cores than requests. - assert concurrent_total < 2.5 * single, ( - f"concurrent total {concurrent_total * 1000:.0f}ms not < 2.5x single " - f"{single * 1000:.0f}ms -- reads appear serialized" - ) + assert not barrier.broken, "reads did not execute concurrently" + # Each request executes on its own independent handle. + assert len({id(cursor) for cursor in opened}) == concurrency def test_mutation_style_endpoint_works_while_queries_run(tmp_path): From 3fe0d932636d230a33069a2a3369be1b653de259 Mon Sep 17 00:00:00 2001 From: Nico Ritschel Date: Tue, 20 Oct 2026 09:59:14 -0700 Subject: [PATCH 2/2] Rendezvous inside DuckDB execution so lock-wrapped cursors fail --- tests/server/test_concurrency.py | 46 +++++++++++--------------------- 1 file changed, 16 insertions(+), 30 deletions(-) diff --git a/tests/server/test_concurrency.py b/tests/server/test_concurrency.py index 3599bc671..0a7d408df 100644 --- a/tests/server/test_concurrency.py +++ b/tests/server/test_concurrency.py @@ -87,49 +87,35 @@ def _post_slow_sql(client: TestClient) -> float: return elapsed -class _RendezvousCursor: - """Cursor proxy whose ``execute`` waits until every concurrent read is executing.""" - - def __init__(self, cursor, barrier: threading.Barrier): - self._cursor = cursor - self._barrier = barrier - - def execute(self, *args, **kwargs): - self._barrier.wait() - return self._cursor.execute(*args, **kwargs) - - def __getattr__(self, name): - return getattr(self._cursor, name) - - -def test_concurrent_reads_do_not_serialize(tmp_path, monkeypatch): +def test_concurrent_reads_do_not_serialize(tmp_path): client = _build_client(tmp_path) adapter = client.app.state.layer.adapter - _post_slow_sql(client) # warm up before patching - # Every read blocks in execute until all of them are executing at once. Reads - # serialized on a lock or shared connection can never all arrive, so the barrier - # times out; unlike a wall-clock ratio, this does not depend on runner load. + # Each query calls a UDF that blocks inside DuckDB execution until every + # concurrent query is executing too. Reads serialized anywhere (an app lock, + # a lock-wrapped cursor, or one shared connection) can never all arrive, so + # the barrier times out; unlike a wall-clock ratio, this ignores runner load. concurrency = 4 barrier = threading.Barrier(concurrency, timeout=30) - opened = [] - open_cursor = adapter.cursor - def rendezvous_cursor(): - cursor = open_cursor() - opened.append(cursor) - return _RendezvousCursor(cursor, barrier) + def rendezvous(value: int) -> int: + barrier.wait() + return value + + # Python UDFs registered on the connection are visible to its cursors. + adapter.conn.create_function("rendezvous", rendezvous) - monkeypatch.setattr(adapter, "cursor", rendezvous_cursor) + def post_rendezvous() -> None: + response = client.post("/raw", json={"query": "select rendezvous(1) as c"}, headers=_AUTH) + assert response.status_code == 200, response.text + assert response.json()["rows"][0]["c"] == 1 with ThreadPoolExecutor(max_workers=concurrency) as pool: - futures = [pool.submit(_post_slow_sql, client) for _ in range(concurrency)] + futures = [pool.submit(post_rendezvous) for _ in range(concurrency)] for future in futures: future.result() assert not barrier.broken, "reads did not execute concurrently" - # Each request executes on its own independent handle. - assert len({id(cursor) for cursor in opened}) == concurrency def test_mutation_style_endpoint_works_while_queries_run(tmp_path):