Keep two tests stable under xdist and machine load, and judge the GIL scan tests one way

tests/_gil_stall.py

@@ -0,0 +1,77 @@
+"""Measures how much a scan stalls when a Python UDF holds the GIL at its first row.
+
+A scan that takes the GIL for every vector waits out the hold on every thread but the one that holds it; a scan that
+reads without the GIL does not. The profiler's TABLE_SCAN timing is wall time summed over the threads that ran the
+scan, so it grows by the time those threads waited.
+"""
+
+import ctypes
+import json
+
+import duckdb
+
+HOLD_SECONDS = 0.2
+ROUNDS = 5
+
+
+def scan_seconds(source, held_row, threads):
+    """Summed TABLE_SCAN seconds of one scan of the source, and how many times the UDF ran.
+
+    The UDF holds the GIL at the row whose ``i`` equals ``held_row``; a row the source does not have, such as -1,
+    gives a free run of the same query.
+    """
+    libc = ctypes.PyDLL(None)
+    libc.usleep.argtypes = [ctypes.c_uint]
+    calls = [0]
+
+    # A call through PyDLL keeps the GIL for the whole sleep.
+    def hold_gil(value):
+        calls[0] += 1
+        libc.usleep(int(HOLD_SECONDS * 1_000_000))
+        return value
+
+    with duckdb.connect(config={"threads": threads}) as con:
+        con.create_function("hold_gil", hold_gil, ["BIGINT"], "BIGINT")
+        con.register("src", source)
+        con.execute("PRAGMA enable_profiling = 'no_output'")
+        con.execute(f"SELECT count(x), sum(CASE WHEN i = {held_row} THEN hold_gil(i) ELSE i END) FROM src").fetchall()
+        operators = list(json.loads(con.get_profiling_information())["operator"])
+    seconds = 0.0
+    while operators:
+        operator = operators.pop()
+        if operator["type"] == "TABLE_SCAN":
+            seconds += operator["timing"]
+        operators.extend(operator.get("children", []))
+    return seconds, calls[0]
+
+
+class ScanTimings:
+    """Held and free scan times of one source, one of each per round, and the UDF calls each pair of runs made."""
+
+    def __init__(self, threads) -> None:
+        self.threads = threads
+        self.held = []
+        self.free = []
+        self.calls = set()
+
+    def add_round(self, source):
+        held, held_calls = scan_seconds(source, 0, self.threads)
+        free, free_calls = scan_seconds(source, -1, self.threads)
+        self.held.append(held)
+        self.free.append(free)
+        self.calls.add((held_calls, free_calls))
+
+    def growth_in_holds(self):
+        """The fastest held run minus the fastest free run, for bounding a stall from above.
+
+        Load adds time to any run, so each kind counts at its fastest.
+        """
+        return (min(self.held) - min(self.free)) / HOLD_SECONDS
+
+    def best_round_in_holds(self):
+        """The largest held minus free difference of one round, for bounding a stall from below.
+
+        One round that shows the stall is enough; load can take stall away from a held run when fewer threads wait
+        out the hold, so the fastest held run would understate it.
+        """
+        return max(held - free for held, free in zip(self.held, self.free, strict=True)) / HOLD_SECONDS

tests/fast/arrow/scan/test_arrow_scan_row_ranges.py

@@ -8,15 +8,14 @@
 
 from __future__ import annotations
 
-import ctypes
 import gc
-import json
 import re
 import sys
 import threading
 import time
 
 import pytest
+from _gil_stall import ROUNDS, ScanTimings
 from conftest import PANDAS_GE_3
 
 import duckdb
@@ -811,7 +810,6 @@ def test_verify_parallelism_keeps_the_order_of_a_one_chunk_source(self):
             assert np.array_equal(con.execute("SELECT n FROM src").fetchnumpy()["n"], arange(LARGE))
 
 
-HOLD_SECONDS = 0.2
 HOLD_ROWS = 1_000_000
 
 
@@ -829,30 +827,9 @@ def _hold_source(source):
     return table
 
 
-def _scan_seconds(obj, udf, hold):
-    query = (
-        "SELECT count(x), sum(CASE WHEN i = 0 THEN hold_gil(i) ELSE i END) FROM src"
-        if hold
-        else "SELECT count(x), sum(i) FROM src"
-    )
-    with duckdb.connect(config={"threads": THREADS}) as con:
-        con.create_function("hold_gil", udf, ["BIGINT"], "BIGINT")
-        con.register("src", obj)
-        con.execute("PRAGMA enable_profiling = 'no_output'")
-        con.execute(query).fetchall()
-        info = json.loads(con.get_profiling_information())
-    total = 0.0
-    stack = list(info["operator"])
-    while stack:
-        node = stack.pop()
-        if node.get("type") == "TABLE_SCAN":
-            total += node["timing"]
-        stack.extend(node.get("children", []))
-    return total
-
-
 class TestGil:
-    @pytest.mark.skipif(sys.platform == "win32", reason="holds the GIL through ctypes.PyDLL(None)")
+    # Linux only until a test-only counter of GIL acquisitions replaces the timing.
+    @pytest.mark.skipif(sys.platform != "linux", reason="the GIL release does not vary by platform; its timing does")
     @pytest.mark.skipif(not getattr(sys, "_is_gil_enabled", lambda: True)(), reason="needs an interpreter with a GIL")
     @pytest.mark.timeout(120)
     @pytest.mark.parametrize(
@@ -865,27 +842,15 @@ class TestGil:
         ],
     )
     def test_scanning_threads_keep_reading_while_another_holds_the_gil(self, source, takes_gil):
-        libc = ctypes.PyDLL(None)
-        libc.usleep.argtypes = [ctypes.c_uint]
-        calls = []
-
-        # A call through PyDLL keeps the GIL for the whole sleep.
-        def hold_gil(x):
-            calls.append(x)
-            libc.usleep(int(HOLD_SECONDS * 1_000_000))
-            return x
-
         obj = _hold_source(source)
-        rounds = 3 if takes_gil else 1
-        holds = max(
-            (_scan_seconds(obj, hold_gil, hold=True) - _scan_seconds(obj, hold_gil, hold=False)) / HOLD_SECONDS
-            for _ in range(rounds)
-        )
-        assert calls == [0] * rounds
+        timings = ScanTimings(THREADS)
+        for _ in range(ROUNDS):
+            timings.add_round(obj)
+        assert timings.calls == {(1, 0)}
         if takes_gil:
-            assert holds >= 1, holds
+            assert timings.best_round_in_holds() >= 1, timings.best_round_in_holds()
         else:
-            assert holds < 0.5, holds
+            assert timings.growth_in_holds() < 0.5, timings.growth_in_holds()
 
 
 STRUCT_VALUES = [{"i": 1, "s": "a"}, {"i": 2, "s": "b"}, {"i": 3, "s": "c"}]

tests/fast/arrow/scan/test_arrow_scan_streams.py

@@ -9,6 +9,7 @@
 import datetime
 import decimal
 import gc
+import itertools
 import threading
 import time
 import uuid
@@ -285,15 +286,18 @@ def batches():
         with pytest.raises(duckdb.InvalidInputException, match=r"registered as 'bad' failed: .*no more rows for you"):
             rows(con, "SELECT sum(a) FROM bad")
 
+    @pytest.mark.timeout(60)
     @pytest.mark.parametrize("pull_under_gil", [False, True])
     def test_a_python_error_in_a_pull_on_an_engine_thread_fails_the_query(self, con, pull_under_gil):
         con.execute("SET threads = 4")
         arrow_schema = pa.schema([("a", pa.int64())])
         submitting = threading.get_ident()
         raised_on = []
 
+        # The submitting thread drives the query until its first result and can pull every batch of a finite stream
+        # while the worker threads wait for a CPU, so only an engine thread's error may end this one.
         def batches():
-            for i in range(2000):
+            for i in itertools.count():
                 if threading.get_ident() != submitting:
                     raised_on.append(threading.get_ident())
                     message = "no more rows for you"
@@ -312,9 +316,10 @@ def stream(self, columns, filters) -> tuple[object, bool]:
                 return pa.RecordBatchReader.from_batches(arrow_schema, batches()).__arrow_c_stream__(), False
 
         con._register_arrow_source("t", scan_source(Pulled))
-        with pytest.raises(duckdb.InvalidInputException, match=r"registered as 't' failed: .*no more rows for you"):
+        with pytest.raises(duckdb.InvalidInputException) as error:
             rows(con, "SELECT sum(a) FROM t")
-        assert raised_on, "every pull ran on the submitting thread, so no engine thread raised"
+        assert raised_on, "no engine thread pulled, so the error was raised on the submitting thread"
+        assert error.match(r"registered as 't' failed: .*no more rows for you")
         assert rows(con, "SELECT 42") == [(42,)]
         con.register("good", pa.table({"a": [1, 2, 3]}))
         assert rows(con, "SELECT sum(a) FROM good") == [(6,)]

tests/fast/pandas/test_pandas_string.py

@@ -1,19 +1,18 @@
-import ctypes
 import datetime
 import gc
 import importlib.metadata
 import importlib.util
 import itertools
 import json
 import re
-import statistics
 import subprocess
 import sys
 import textwrap
 import tracemalloc
 import warnings
 
 import pytest
+from _gil_stall import ROUNDS, ScanTimings
 from conftest import PANDAS_GE_3
 from packaging.version import Version
 
@@ -881,42 +880,46 @@ def test_str_invalid_utf8_below_the_threshold_fails_at_bind(pa, pieces, setup, u
     assert re.search(r"Wrapping .* failed", message)
 
 
-GIL_HOLD_SECONDS = 0.2
 GIL_ROWS = 1_000_000
-# A single measurement can miss the scan, so the object and str columns are measured back to back in each round and
-# the medians are compared.
-GIL_ROUNDS = 3
 
 
-def _scan_seconds(frame, held_row):
-    libc = ctypes.PyDLL(None)
-    libc.usleep.argtypes = [ctypes.c_uint]
-    holds = []
-
-    def hold_gil(value):
-        holds.append(value)
-        libc.usleep(int(GIL_HOLD_SECONDS * 1_000_000))
-        return value
-
-    with duckdb.connect(config={"threads": 4}) as con:
-        con.create_function("hold_gil", hold_gil, ["BIGINT"], "BIGINT")
-        con.register("frame", frame)
-        con.execute("PRAGMA enable_profiling = 'no_output'")
-        con.execute(f"SELECT count(x), sum(CASE WHEN i = {held_row} THEN hold_gil(i) ELSE i END) FROM frame").fetchall()
-        operators = list(json.loads(con.get_profiling_information())["operator"])
-    seconds = 0.0
-    while operators:
-        operator = operators.pop()
-        if operator["type"] == "TABLE_SCAN":
-            seconds += operator["timing"]
-        operators.extend(operator.get("children", []))
-    return seconds, len(holds)
+def _judge_str_scan(calibration_growth, growth):
+    """Fails a str scan that stalled behind the hold; skips when the measurement could not have shown a stall."""
+    # A scan that takes the GIL reads about three holds, as three of its four threads wait out the hold; load alone
+    # has read up to one hold on a scan without the GIL.
+    assert growth < 2, f"the hold stalled the str column scan by {growth:.2f} holds"
+    # An object column takes the GIL for every vector, so its scan must stall behind the hold in every held run;
+    # otherwise a low str reading says nothing.
+    if calibration_growth < 1:
+        pytest.skip(
+            f"inconclusive: the hold stalled the object column scan by only {calibration_growth:.2f} holds "
+            f"and the str column scan by {growth:.2f}"
+        )
+    assert growth < 0.5, growth
 
 
-def _scan_growth_in_holds(frame):
-    held, held_calls = _scan_seconds(frame, 0)
-    free, free_calls = _scan_seconds(frame, -1)
-    return (held - free) / GIL_HOLD_SECONDS, (held_calls, free_calls)
+@pytest.mark.parametrize(
+    ("calibration_growth", "growth", "outcome"),
+    [
+        pytest.param(2.96, 0.00, None, id="calibration_met_str_free"),
+        pytest.param(1.00, 0.49, None, id="both_at_their_bounds"),
+        pytest.param(2.96, 0.52, AssertionError, id="calibration_met_str_stalled"),
+        pytest.param(0.61, 3.15, AssertionError, id="calibration_missed_str_stalled"),
+        pytest.param(2.96, 0.50, AssertionError, id="str_at_the_bound"),
+        pytest.param(0.61, 2.00, AssertionError, id="calibration_missed_str_at_two_holds"),
+        pytest.param(2.96, 2.00, AssertionError, id="calibration_met_str_at_two_holds"),
+        pytest.param(0.61, 0.45, pytest.skip.Exception, id="calibration_missed_str_free"),
+        pytest.param(0.99, 0.00, pytest.skip.Exception, id="calibration_just_missed"),
+        pytest.param(-0.02, 0.96, pytest.skip.Exception, id="calibration_missed_str_slowed_by_load"),
+        pytest.param(0.61, 1.99, pytest.skip.Exception, id="calibration_missed_str_under_two_holds"),
+    ],
+)
+def test_str_scan_judgement(calibration_growth, growth, outcome):
+    if outcome is None:
+        _judge_str_scan(calibration_growth, growth)
+        return
+    with pytest.raises(outcome):
+        _judge_str_scan(calibration_growth, growth)
 
 
 GIL_COLUMNS = [
@@ -938,7 +941,8 @@ def _scan_growth_in_holds(frame):
 ]
 
 
-@pytest.mark.skipif(sys.platform == "win32", reason="ctypes cannot load the C library through a null name on Windows")
+# Linux only until a test-only counter of GIL acquisitions replaces the timing.
+@pytest.mark.skipif(sys.platform != "linux", reason="the GIL release does not vary by platform; its timing does")
 @pytest.mark.skipif(not getattr(sys, "_is_gil_enabled", lambda: True)(), reason="the interpreter has no GIL")
 @pytest.mark.timeout(120)
 @pytest.mark.parametrize("make_column", GIL_COLUMNS)
@@ -947,15 +951,9 @@ def test_arrow_str_scan_runs_without_gil(pa, make_column):
     index = numpy.arange(GIL_ROWS, dtype=numpy.int64)
     calibration = pd.DataFrame({"x": pd.Series(values, dtype=object), "i": index})
     frame = pd.DataFrame({"x": make_column(pa, values), "i": index})
-    calibration_growths, growths, calls = [], [], set()
-    for _ in range(GIL_ROUNDS):
-        calibration_growth, _ = _scan_growth_in_holds(calibration)
-        growth, round_calls = _scan_growth_in_holds(frame)
-        calibration_growths.append(calibration_growth)
-        growths.append(growth)
-        calls.add(round_calls)
-    # An object column takes the GIL for every vector, so its scan must stall behind the hold; otherwise the hold
-    # did not overlap the scan and the measurement says nothing.
-    assert statistics.median(calibration_growths) >= 1
-    assert calls == {(1, 0)}
-    assert statistics.median(growths) < 0.5
+    calibration_timings, timings = ScanTimings(4), ScanTimings(4)
+    for _ in range(ROUNDS):
+        calibration_timings.add_round(calibration)
+        timings.add_round(frame)
+    assert timings.calls == {(1, 0)}
+    _judge_str_scan(calibration_timings.growth_in_holds(), timings.growth_in_holds())