fix(connection): don't block in set_query_timeout() while a query runs

The engine applies the query timeout and the per-query thread limit under the
same per-connection lock that a running query holds for its whole duration.
Connection.set_query_timeout() and set_max_threads_for_exec() called the
engine right away, so with a query running on the connection they blocked
until it finished. On the pybind backend they also held the GIL while waiting,
which froze every Python thread, including an asyncio event loop that called
AsyncConnection.set_query_timeout().

Both setters now record the value and return immediately. The value is
applied before the next execute(), query_as_arrow() or Arrow-table call on
that connection, by one thread at a time so that the latest value wins. The
pybind bindings of the two native setters release the GIL, so a thread that
does wait for the lock does not stall other threads.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: 2 people

src_cpp/py_connection.cpp

@@ -47,10 +47,11 @@ void PyConnection::initialize(py::handle& m) {
         .def("query_as_arrow", &PyConnection::queryAsArrow, py::arg("statement"),
             py::arg("chunk_size"))
         .def("set_max_threads_for_exec", &PyConnection::setMaxNumThreadForExec,
-            py::arg("num_threads"))
+            py::arg("num_threads"), py::call_guard<py::gil_scoped_release>())
         .def("prepare", &PyConnection::prepare, py::arg("query"),
             py::arg("parameters") = py::dict())
-        .def("set_query_timeout", &PyConnection::setQueryTimeout, py::arg("timeout_in_ms"))
+        .def("set_query_timeout", &PyConnection::setQueryTimeout, py::arg("timeout_in_ms"),
+            py::call_guard<py::gil_scoped_release>())
         .def("interrupt", &PyConnection::interrupt)
         .def("get_num_nodes", &PyConnection::getNumNodes, py::arg("node_name"))
         .def("get_num_rels", &PyConnection::getNumRels, py::arg("rel_name"))

src_py/connection.py

@@ -411,6 +411,10 @@ def __init__(self, database: Database, num_threads: int = 0):
         self._calls_in_flight = 0
         self._prefer_pybind = False
         self._query_timeout_ms = 0
+        # Native setter name -> value, applied before the next query.
+        self._pending_settings: dict[str, int] = {}
+        self._pending_settings_lock = threading.Lock()
+        self._apply_settings_lock = threading.Lock()
         self._query_results: WeakSet[QueryResult] = WeakSet()
         self._capi_scan_tables: set[str] = set()
         # Implicit prepared-statement cache, shared by the pybind and
@@ -464,14 +468,27 @@ def set_max_threads_for_exec(self, num_threads: int) -> None:
         """
         Set the maximum number of threads for executing queries.
 
+        The setting applies from the next query on this connection; this call
+        does not wait for a running query.
+
         Parameters
         ----------
         num_threads : int
             Maximum number of threads to use for executing queries.
 
         """
         self.init_connection()
-        self._connection.set_max_threads_for_exec(num_threads)
+        with self._pending_settings_lock:
+            self._pending_settings["set_max_threads_for_exec"] = int(num_threads)
+
+    def _apply_pending_settings(self) -> None:
+        # The engine applies settings under the lock a running query holds, so
+        # they are applied here, before a query, rather than in the setters.
+        with self._apply_settings_lock:
+            with self._pending_settings_lock:
+                pending, self._pending_settings = self._pending_settings, {}
+            for setter, value in pending.items():
+                getattr(self._connection, setter)(value)
 
     def _register_query_result(self, query_result: QueryResult) -> None:
         self._query_results.add(query_result)
@@ -975,6 +992,7 @@ def execute(
             parameters = {}
 
         self.init_connection()
+        self._apply_pending_settings()
         if not isinstance(parameters, dict):
             msg = f"Parameters must be a dict; found {type(parameters)}."
             raise RuntimeError(msg)  # noqa: TRY004
@@ -1068,6 +1086,7 @@ def query_as_arrow(self, query: str, chunk_size: int) -> ArrowQueryResult:
         This is the efficient path for CSR-aware Arrow export.
         """
         self.init_connection()
+        self._apply_pending_settings()
         if not self._using_pybind_backend():
             query_result_internal = self._connection.query(query)
         else:
@@ -1186,6 +1205,9 @@ def set_query_timeout(self, timeout_in_ms: int) -> None:
         """
         Set the query timeout value in ms for executing queries.
 
+        The timeout applies from the next query on this connection; this call
+        does not wait for a running query.
+
         Parameters
         ----------
         timeout_in_ms : int
@@ -1194,7 +1216,8 @@ def set_query_timeout(self, timeout_in_ms: int) -> None:
         """
         self.init_connection()
         self._query_timeout_ms = int(timeout_in_ms)
-        self._connection.set_query_timeout(timeout_in_ms)
+        with self._pending_settings_lock:
+            self._pending_settings["set_query_timeout"] = self._query_timeout_ms
 
     def interrupt(self) -> None:
         """
@@ -1317,6 +1340,7 @@ def create_arrow_table(
 
         """
         self.init_connection()
+        self._apply_pending_settings()
         try:
             query_result_internal = self._connection.create_arrow_table(
                 table_name, dataframe
@@ -1350,6 +1374,7 @@ def drop_arrow_table(self, table_name: str) -> QueryResult:
 
         """
         self.init_connection()
+        self._apply_pending_settings()
         try:
             query_result_internal = self._connection.drop_arrow_table(table_name)
         except NotImplementedError:
@@ -1411,6 +1436,7 @@ def create_arrow_rel_table(
 
         """
         self.init_connection()
+        self._apply_pending_settings()
         layout_value = (
             layout.value if isinstance(layout, ArrowRelTableLayout) else str(layout)
         ).upper()

test/test_timeout.py

@@ -1,8 +1,17 @@
 from __future__ import annotations
 
+import contextlib
+import threading
+import time
+from typing import TYPE_CHECKING
+
+import ladybug as lb
 import pytest
 from type_aliases import ConnDB
 
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
 
 def test_timeout(conn_db_readonly: ConnDB) -> None:
     conn, _ = conn_db_readonly
@@ -11,3 +20,63 @@ def test_timeout(conn_db_readonly: ConnDB) -> None:
         conn.execute(
             "UNWIND RANGE(1,100000) AS x UNWIND RANGE(1, 100000) AS y RETURN COUNT(x + y);"
         )
+
+
+# Runs for seconds without materializing large lists.
+LONG_QUERY = "WITH range(1, 30000) AS r UNWIND r AS x UNWIND r AS y RETURN sum(x * y)"
+
+
+def _execute_until_interrupted(conn: lb.Connection, query: str) -> None:
+    with contextlib.suppress(RuntimeError):
+        conn.execute(query)
+
+
+def _interrupt_until_done(conn: lb.Connection, *threads: threading.Thread) -> None:
+    # An interrupt sent before execution starts is cleared, so repeat it.
+    for thread in threads:
+        while thread.is_alive():
+            conn.interrupt()
+            thread.join(0.05)
+
+
+@pytest.fixture
+def conn() -> Iterator[lb.Connection]:
+    # A fresh connection: the shared read-only one carries other tests' settings.
+    db = lb.Database(":memory:")
+    conn = lb.Connection(db)
+    yield conn
+    conn.close()
+    db.close()
+
+
+def test_settings_do_not_wait_for_running_query(conn: lb.Connection) -> None:
+    worker = threading.Thread(
+        target=_execute_until_interrupted, args=(conn, LONG_QUERY)
+    )
+    worker.start()
+    time.sleep(0.3)
+    started = time.monotonic()
+    conn.set_query_timeout(60_000)
+    conn.set_max_threads_for_exec(2)
+    elapsed = time.monotonic() - started
+    _interrupt_until_done(conn, worker)
+    assert elapsed < 1, elapsed
+
+
+def test_applying_settings_lets_other_threads_run(conn: lb.Connection) -> None:
+    running = threading.Thread(
+        target=_execute_until_interrupted, args=(conn, LONG_QUERY)
+    )
+    running.start()
+    time.sleep(0.3)
+    conn.set_query_timeout(60_000)
+    # This call applies the new timeout, which waits for the running query.
+    waiting = threading.Thread(
+        target=_execute_until_interrupted, args=(conn, "RETURN 1")
+    )
+    started = time.monotonic()
+    waiting.start()
+    time.sleep(0.4)
+    elapsed = time.monotonic() - started
+    _interrupt_until_done(conn, running, waiting)
+    assert elapsed < 1, elapsed