From d02faa3034bbb2b2c83a089cc51209e78a457c82 Mon Sep 17 00:00:00 2001 From: Darek Date: Tue, 6 Oct 2026 03:57:34 +0200 Subject: [PATCH 1/2] fix(pybind): let Ctrl-C interrupt a running query Python runs its SIGINT handler only between bytecodes. While the main thread waited in the engine for query(), execute() or query_as_arrow(), Ctrl-C therefore took effect only after the query had finished: a 15 s query ran to the end before KeyboardInterrupt appeared. During such a wait on the main thread, a scoped SIGINT handler now interrupts the connection's query and then calls Python's own handler. Once the call returns, the pending signal is raised as KeyboardInterrupt instead of the engine's "Interrupted." error, and the connection stays usable. The handler is installed only while Python's default handler (KeyboardInterrupt) is active, so a custom SIGINT handler keeps its behaviour, and only on POSIX systems. That check goes through the driver's import cache and never fails a query, e.g. one run during interpreter shutdown. A Ctrl-C that lands while the query is still compiling is dropped by the engine, as any interrupt is; a second Ctrl-C stops the query. The engine interrupts a connection, not a query: if the main thread's call is waiting behind another thread's query on the same connection, Ctrl-C stops that query instead, and the main thread's query runs before KeyboardInterrupt is raised. The new tests send SIGINT one second into a long query through query(), execute() with parameters and query_as_arrow(), and expect KeyboardInterrupt within seconds; they fail on main. A custom handler test checks that the query then still finishes and the handler runs. The C-API backend calls the engine through ctypes and is not covered, so the interrupt tests are listed in capi_xfails. Co-Authored-By: Claude Opus 5.5 --- .../include/cached_import/py_cached_import.h | 2 + .../include/cached_import/py_cached_modules.h | 18 ++++ src_cpp/py_connection.cpp | 96 +++++++++++++++++++ test/capi_xfails.py | 4 + test/test_keyboard_interrupt.py | 79 +++++++++++++++ 5 files changed, 199 insertions(+) create mode 100644 test/test_keyboard_interrupt.py diff --git a/src_cpp/include/cached_import/py_cached_import.h b/src_cpp/include/cached_import/py_cached_import.h index 734550b..b85c74a 100644 --- a/src_cpp/include/cached_import/py_cached_import.h +++ b/src_cpp/include/cached_import/py_cached_import.h @@ -25,6 +25,8 @@ class PythonCachedImport { PandasCachedItem pandas; PolarsCachedItem polars; PyarrowCachedItem pyarrow; + SignalCachedItem signal; + ThreadingCachedItem threading; UUIDCachedItem uuid; private: diff --git a/src_cpp/include/cached_import/py_cached_modules.h b/src_cpp/include/cached_import/py_cached_modules.h index 0d18bde..dbf840e 100644 --- a/src_cpp/include/cached_import/py_cached_modules.h +++ b/src_cpp/include/cached_import/py_cached_modules.h @@ -154,6 +154,24 @@ class PyarrowCachedItem : public PythonCachedItem { LibCachedItem lib; }; +class SignalCachedItem : public PythonCachedItem { +public: + SignalCachedItem() + : PythonCachedItem("signal"), getsignal("getsignal", this), sigint("SIGINT", this), + default_int_handler("default_int_handler", this) {} + + PythonCachedItem getsignal; + PythonCachedItem sigint; + PythonCachedItem default_int_handler; +}; + +class ThreadingCachedItem : public PythonCachedItem { +public: + ThreadingCachedItem() : PythonCachedItem("threading"), main_thread("main_thread", this) {} + + PythonCachedItem main_thread; +}; + class UUIDCachedItem : public PythonCachedItem { public: UUIDCachedItem() : PythonCachedItem("uuid"), UUID("UUID", this) {} diff --git a/src_cpp/py_connection.cpp b/src_cpp/py_connection.cpp index de7c09d..477151b 100644 --- a/src_cpp/py_connection.cpp +++ b/src_cpp/py_connection.cpp @@ -1,9 +1,14 @@ #include "include/py_connection.h" #include +#include #include #include +#ifndef _WIN32 +#include +#endif + #include "cached_import/py_cached_import.h" #include "common/constants.h" #include "common/exception/not_implemented.h" @@ -183,6 +188,91 @@ void PyConnection::interrupt() { static std::unordered_map> transformPythonParameters( const py::dict& params, Connection* conn); +namespace { + +// Python runs its SIGINT handler only between bytecodes, so not while the main thread waits in +// the engine. During such a wait, SIGINT interrupts the connection's query and then calls +// Python's handler. This is only done on the main thread while Python's default handler +// (KeyboardInterrupt) is active, so custom handlers keep their behaviour. +class SigintInterruptsQuery { +public: + explicit SigintInterruptsQuery([[maybe_unused]] main::ClientContext* context) { +#ifndef _WIN32 + if (target.load() != nullptr || !onMainThreadWithDefaultHandler()) { + return; + } + if (sigaction(SIGINT, nullptr, &previous) != 0) { + return; + } + struct sigaction action {}; + action.sa_sigaction = handle; + action.sa_flags = SA_SIGINFO | (previous.sa_flags & SA_ONSTACK); + sigemptyset(&action.sa_mask); + received = false; + target = context; + installed = sigaction(SIGINT, &action, nullptr) == 0; + if (!installed) { + target = nullptr; + } +#endif + } + + ~SigintInterruptsQuery() { +#ifndef _WIN32 + if (installed) { + sigaction(SIGINT, &previous, nullptr); + target = nullptr; + } +#endif + } + + // Call with the GIL held once the query has returned. + void raiseIfInterrupted() const { +#ifndef _WIN32 + if (installed && received && PyErr_CheckSignals() != 0) { + throw py::error_already_set(); + } +#endif + } + +private: +#ifndef _WIN32 + static bool onMainThreadWithDefaultHandler() { + // The check must never fail a query, e.g. once imports stop working at shutdown. + try { + auto mainThread = importCache->threading.main_thread()(); + if (mainThread.attr("ident").cast() != PyThread_get_thread_ident()) { + return false; + } + auto& signal = importCache->signal; + return signal.getsignal()(signal.sigint()).is(signal.default_int_handler()); + } catch (py::error_already_set&) { + return false; + } + } + + static void handle(int signum, siginfo_t* info, void* ucontext) { + if (auto* context = target.load()) { + context->interrupt(); + } + received = true; + // Python's handler only records the signal; KeyboardInterrupt is raised later. + if (previous.sa_flags & SA_SIGINFO) { + previous.sa_sigaction(signum, info, ucontext); + } else if (previous.sa_handler != SIG_DFL && previous.sa_handler != SIG_IGN) { + previous.sa_handler(signum); + } + } + + static inline std::atomic target{nullptr}; + static inline std::atomic received{false}; + static inline struct sigaction previous {}; + bool installed = false; +#endif +}; + +} // namespace + std::unique_ptr PyConnection::execute(PyPreparedStatement* preparedStatement, const py::dict& params) { auto& stateRef = refState(); @@ -190,27 +280,33 @@ std::unique_ptr PyConnection::execute(PyPreparedStatement* prepar throw RuntimeException("Prepared statement is closed."); } auto parameters = transformPythonParameters(params, &stateRef.ref()); + SigintInterruptsQuery sigint(stateRef.ref().getClientContext()); py::gil_scoped_release release; auto queryResult = stateRef.ref().executeWithParams(&preparedStatement->state->ref(), std::move(parameters)); py::gil_scoped_acquire acquire; + sigint.raiseIfInterrupted(); return checkAndWrapQueryResult(queryResult, state); } std::unique_ptr PyConnection::query(const std::string& statement) { auto& stateRef = refState(); + SigintInterruptsQuery sigint(stateRef.ref().getClientContext()); py::gil_scoped_release release; auto queryResult = stateRef.ref().query(statement); py::gil_scoped_acquire acquire; + sigint.raiseIfInterrupted(); return checkAndWrapQueryResult(queryResult, state); } std::unique_ptr PyConnection::queryAsArrow(const std::string& statement, int64_t chunkSize) { auto& stateRef = refState(); + SigintInterruptsQuery sigint(stateRef.ref().getClientContext()); py::gil_scoped_release release; auto queryResult = stateRef.ref().queryAsArrow(statement, chunkSize); py::gil_scoped_acquire acquire; + sigint.raiseIfInterrupted(); return checkAndWrapQueryResult(queryResult, state); } diff --git a/test/capi_xfails.py b/test/capi_xfails.py index c928e6f..c47026a 100644 --- a/test/capi_xfails.py +++ b/test/capi_xfails.py @@ -69,6 +69,10 @@ # C API scan rewriting uses temporary Arrow-backed tables, which cannot be # created through a read-only connection. "test/test_scan_pyarrow.py::test_pyarrow_basic", + # Ctrl-C interrupts a running query only on the pybind backend. + "test/test_keyboard_interrupt.py::test_keyboard_interrupt_stops_the_query[execute_with_parameters]", + "test/test_keyboard_interrupt.py::test_keyboard_interrupt_stops_the_query[query]", + "test/test_keyboard_interrupt.py::test_keyboard_interrupt_stops_the_query[query_as_arrow]", # UDF registration is still routed through pybind. "test/test_blob_parameter.py::test_bytes_param_udf", "test/test_udf.py::test_udf", diff --git a/test/test_keyboard_interrupt.py b/test/test_keyboard_interrupt.py new file mode 100644 index 0000000..3062a06 --- /dev/null +++ b/test/test_keyboard_interrupt.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +import subprocess +import sys +import textwrap +from typing import TYPE_CHECKING + +import pytest + +if TYPE_CHECKING: + from pathlib import Path + +# Each scenario sends SIGINT to itself one second into a query that runs for many +# seconds, and reports how the query ended. +PRELUDE = """ +import os, signal, sys, threading, time +import ladybug as lb + +db = lb.Database(sys.argv[1]) +conn = lb.Connection(db) +LONG_QUERY = "WITH range(1, $n) AS r UNWIND r AS x UNWIND r AS y RETURN sum(x * y)" +threading.Timer(1.0, os.kill, (os.getpid(), signal.SIGINT)).start() +""" + +SCENARIOS = { + "query": "conn.execute(LONG_QUERY.replace('$n', '40000'))", + "execute_with_parameters": "conn.execute(LONG_QUERY, {'n': 40000})", + "query_as_arrow": "conn.query_as_arrow(LONG_QUERY.replace('$n', '40000'), 1000)", +} + +RUN = """ +started = time.monotonic() +try: + {call} + print("OUTCOME finished") +except KeyboardInterrupt: + print("OUTCOME KeyboardInterrupt") +print("ELAPSED", time.monotonic() - started) +print("AFTER", conn.execute("RETURN 1").get_next()[0]) +""" + +CUSTOM_HANDLER = """ +received = [] +signal.signal(signal.SIGINT, lambda signum, frame: received.append(signum)) +result = conn.execute(LONG_QUERY.replace("$n", "20000")).get_next()[0] +time.sleep(0.2) +print("OUTCOME", "finished" if result else "empty", "handled" if received else "missed") +""" + + +def run_script(tmp_path: Path, body: str) -> str: + script = textwrap.dedent(PRELUDE) + textwrap.dedent(body) + try: + proc = subprocess.run( + [sys.executable, "-c", script, str(tmp_path / "db")], + capture_output=True, + text=True, + timeout=120, + ) + except subprocess.TimeoutExpired: + pytest.fail("the query was not interrupted") + assert proc.returncode == 0, proc.stderr[-2000:] + return proc.stdout + + +@pytest.mark.skipif(sys.platform == "win32", reason="SIGINT is POSIX only") +@pytest.mark.parametrize("scenario", sorted(SCENARIOS)) +def test_keyboard_interrupt_stops_the_query(tmp_path: Path, scenario: str) -> None: + stdout = run_script(tmp_path, RUN.format(call=SCENARIOS[scenario])) + assert "OUTCOME KeyboardInterrupt" in stdout, stdout + assert float(stdout.split("ELAPSED")[1].split()[0]) < 10, stdout + assert "AFTER 1" in stdout, stdout + + +@pytest.mark.skipif(sys.platform == "win32", reason="SIGINT is POSIX only") +def test_custom_sigint_handler_still_runs(tmp_path: Path) -> None: + stdout = run_script(tmp_path, CUSTOM_HANDLER) + # A custom handler is left alone: the query finishes, then the handler runs. + assert "OUTCOME finished handled" in stdout, stdout From 5319fde1d918a5cb3a72ea7ec4c68b2c6917d8ee Mon Sep 17 00:00:00 2001 From: Darek Date: Tue, 13 Oct 2026 00:54:41 +0200 Subject: [PATCH 2/2] fix(pybind): keep the connection alive across a Ctrl-C guarded call query(), execute() and query_as_arrow() now hold the connection state for the whole call, so the ClientContext the SIGINT handler interrupts outlives the guard even if another thread closes the connection. Before, a close() from another thread destroyed the Connection while the main thread was still inside the query, after waiting for it with the GIL held. The handler's signal safety is pinned: static_asserts that both atomics it touches are lock-free, and a comment that ClientContext::interrupt() must stay a store to an atomic flag. Comments also state that prepare() is not guarded and why the handler swap is not wrapped in a signal mask. A failure of the main-thread/handler check is reported as unraisable, except during interpreter shutdown, instead of being swallowed; the query still runs. Co-Authored-By: Claude Opus 5.5 --- src_cpp/py_connection.cpp | 36 ++++++++++++++++++++++---- test/capi_xfails.py | 2 ++ test/test_keyboard_interrupt.py | 46 +++++++++++++++++++++++++++++++++ 3 files changed, 79 insertions(+), 5 deletions(-) diff --git a/src_cpp/py_connection.cpp b/src_cpp/py_connection.cpp index 477151b..d330063 100644 --- a/src_cpp/py_connection.cpp +++ b/src_cpp/py_connection.cpp @@ -9,6 +9,10 @@ #include #endif +#if PY_VERSION_HEX < 0x030D0000 +#define Py_IsFinalizing _Py_IsFinalizing +#endif + #include "cached_import/py_cached_import.h" #include "common/constants.h" #include "common/exception/not_implemented.h" @@ -193,7 +197,14 @@ namespace { // Python runs its SIGINT handler only between bytecodes, so not while the main thread waits in // the engine. During such a wait, SIGINT interrupts the connection's query and then calls // Python's handler. This is only done on the main thread while Python's default handler -// (KeyboardInterrupt) is active, so custom handlers keep their behaviour. +// (KeyboardInterrupt) is active, so custom handlers keep their behaviour. Only query(), +// execute() and query_as_arrow() install it. +// +// target is set before the handler is installed and cleared after the previous one is +// restored, so a SIGINT in between reaches Python's handler, as without this guard. A signal +// mask would not help: masks are per thread, and a process-directed SIGINT runs the handler on +// whichever thread does not block it. The statics are per process; this single-phase-init +// module cannot be imported into subinterpreters. class SigintInterruptsQuery { public: explicit SigintInterruptsQuery([[maybe_unused]] main::ClientContext* context) { @@ -246,11 +257,17 @@ class SigintInterruptsQuery { } auto& signal = importCache->signal; return signal.getsignal()(signal.sigint()).is(signal.default_int_handler()); - } catch (py::error_already_set&) { + } catch (py::error_already_set& e) { + if (!Py_IsFinalizing()) { + e.discard_as_unraisable("checking the SIGINT handler before a query"); + } return false; } } + // Runs in signal context, so it may only use async-signal-safe operations: lock-free atomics + // and Python's own handler, which is written for signal context. ClientContext::interrupt() + // is a store to an std::atomic in the engine and must stay one. static void handle(int signum, siginfo_t* info, void* ucontext) { if (auto* context = target.load()) { context->interrupt(); @@ -266,6 +283,8 @@ class SigintInterruptsQuery { static inline std::atomic target{nullptr}; static inline std::atomic received{false}; + static_assert(std::atomic::is_always_lock_free); + static_assert(std::atomic::is_always_lock_free); static inline struct sigaction previous {}; bool installed = false; #endif @@ -280,40 +299,47 @@ std::unique_ptr PyConnection::execute(PyPreparedStatement* prepar throw RuntimeException("Prepared statement is closed."); } auto parameters = transformPythonParameters(params, &stateRef.ref()); + // Keeps the ClientContext alive past the guard, even if another thread closes this + // connection while the GIL is released. + auto keepAlive = state; SigintInterruptsQuery sigint(stateRef.ref().getClientContext()); py::gil_scoped_release release; auto queryResult = stateRef.ref().executeWithParams(&preparedStatement->state->ref(), std::move(parameters)); py::gil_scoped_acquire acquire; sigint.raiseIfInterrupted(); - return checkAndWrapQueryResult(queryResult, state); + return checkAndWrapQueryResult(queryResult, keepAlive); } std::unique_ptr PyConnection::query(const std::string& statement) { auto& stateRef = refState(); + auto keepAlive = state; SigintInterruptsQuery sigint(stateRef.ref().getClientContext()); py::gil_scoped_release release; auto queryResult = stateRef.ref().query(statement); py::gil_scoped_acquire acquire; sigint.raiseIfInterrupted(); - return checkAndWrapQueryResult(queryResult, state); + return checkAndWrapQueryResult(queryResult, keepAlive); } std::unique_ptr PyConnection::queryAsArrow(const std::string& statement, int64_t chunkSize) { auto& stateRef = refState(); + auto keepAlive = state; SigintInterruptsQuery sigint(stateRef.ref().getClientContext()); py::gil_scoped_release release; auto queryResult = stateRef.ref().queryAsArrow(statement, chunkSize); py::gil_scoped_acquire acquire; sigint.raiseIfInterrupted(); - return checkAndWrapQueryResult(queryResult, state); + return checkAndWrapQueryResult(queryResult, keepAlive); } void PyConnection::setMaxNumThreadForExec(uint64_t numThreads) { refState().ref().setMaxNumThreadForExec(numThreads); } +// Not guarded by SigintInterruptsQuery: a Ctrl-C while the statement compiles takes effect when +// prepare() returns. PyPreparedStatement PyConnection::prepare(const std::string& query, const py::dict& parameters) { auto& stateRef = refState(); auto params = transformPythonParameters(parameters, &stateRef.ref()); diff --git a/test/capi_xfails.py b/test/capi_xfails.py index c47026a..fe82ec7 100644 --- a/test/capi_xfails.py +++ b/test/capi_xfails.py @@ -73,6 +73,8 @@ "test/test_keyboard_interrupt.py::test_keyboard_interrupt_stops_the_query[execute_with_parameters]", "test/test_keyboard_interrupt.py::test_keyboard_interrupt_stops_the_query[query]", "test/test_keyboard_interrupt.py::test_keyboard_interrupt_stops_the_query[query_as_arrow]", + "test/test_keyboard_interrupt.py::test_keyboard_interrupt_after_close_from_another_thread", + "test/test_keyboard_interrupt.py::test_failed_handler_check_is_reported_not_raised", # UDF registration is still routed through pybind. "test/test_blob_parameter.py::test_bytes_param_udf", "test/test_udf.py::test_udf", diff --git a/test/test_keyboard_interrupt.py b/test/test_keyboard_interrupt.py index 3062a06..a43a5f0 100644 --- a/test/test_keyboard_interrupt.py +++ b/test/test_keyboard_interrupt.py @@ -77,3 +77,49 @@ def test_custom_sigint_handler_still_runs(tmp_path: Path) -> None: stdout = run_script(tmp_path, CUSTOM_HANDLER) # A custom handler is left alone: the query finishes, then the handler runs. assert "OUTCOME finished handled" in stdout, stdout + + +CLOSE_DURING_QUERY = """ +threading.Timer(0.5, conn._connection.close).start() +started = time.monotonic() +try: + conn.execute(LONG_QUERY.replace('$n', '40000')) + print("OUTCOME finished") +except KeyboardInterrupt: + print("OUTCOME KeyboardInterrupt") +print("ELAPSED", time.monotonic() - started) +""" + + +@pytest.mark.skipif(sys.platform == "win32", reason="SIGINT is POSIX only") +def test_keyboard_interrupt_after_close_from_another_thread(tmp_path: Path) -> None: + # The running call keeps the connection alive, so closing it from another thread + # returns at once instead of waiting for the query with the GIL held. (Whether the + # context was freed under the running query would need ASan to observe.) + stdout = run_script(tmp_path, CLOSE_DURING_QUERY) + assert "OUTCOME KeyboardInterrupt" in stdout, stdout + assert float(stdout.split("ELAPSED")[1].split()[0]) < 10, stdout + + +@pytest.mark.skipif(sys.platform == "win32", reason="SIGINT is POSIX only") +def test_failed_handler_check_is_reported_not_raised(tmp_path: Path) -> None: + script = """ +import signal, sys +import ladybug as lb + +def broken(signum): + raise RuntimeError("getsignal is broken") + +signal.getsignal = broken +conn = lb.Connection(lb.Database(sys.argv[1])) +print("RESULT", conn.execute("RETURN 1").get_next()[0]) +""" + proc = subprocess.run( + [sys.executable, "-c", script, str(tmp_path / "db")], + capture_output=True, + text=True, + timeout=120, + ) + assert proc.returncode == 0, proc.stderr[-2000:] + assert "RESULT 1" in proc.stdout, proc.stdout + assert "getsignal is broken" in proc.stderr, proc.stderr