Ctrl-C stops a statement, and so does another thread

README.md

@@ -85,6 +85,20 @@ result.record_batches()  # a reader, for a result larger than memory
 
 `Result` implements `__arrow_c_stream__`, so anything that reads the protocol reads a result directly and none of the four methods above is needed: `pyarrow.table(result)` and `polars.DataFrame(result)` both work. Batches are 65,536 rows. A column holds one type, which the values decide, and integers beside floats are the one mixture that widens rather than being refused. Nodes, rels and paths go across as structs. The copy runs with the GIL released, and on this machine 300,000 rows across three columns take 44 ms as Arrow against 67 ms as Python objects, and a single integer column takes 13.8 ms against 44.5 ms.
 
+## Stopping a statement
+
+A statement that is running can be stopped two ways, and neither of them closes the connection: the session, its plans and its warm readers are all there afterwards, which is the whole difference between stopping a statement and starting again.
+
+```python
+conn.execute(long_one)  # Ctrl-C raises KeyboardInterrupt here
+conn.interrupt()  # from another thread, raises zudb.Interrupted there
+conn.rows_read  # how far the statement running now has got
+```
+
+`Ctrl-C` is the one a person presses, and it raises `KeyboardInterrupt` on the thread that called `execute`, measured at 5 ms from the press on this machine against a budget of 50. Python only delivers a signal to the main thread between two bytecodes, so a statement called from the main thread runs on a thread this client keeps for it and the main thread waits and asks for signals while it does. That thread is kept rather than made per statement, because making one costs 30 microseconds against a small statement that costs 10, and a statement called from any other thread runs inline where a signal was never going to arrive anyway.
+
+`interrupt()` is the one a program calls, from a thread that is not the one inside `execute`, and it raises `zudb.Interrupted` there. It is one of the three calls that may be made on a connection while a statement is running, with `rows_read` and `closed`, and none of the three waits for it: a progress bar drawn from `rows_read` is a poll of an atomic, not a queue behind the executor.
+
 ## Types
 
 The wheel carries `py.typed` and a stub for the compiled module, so mypy, pyright and an editor's completion all work with nothing else installed. `zudb.Value` is the union a row holds and a parameter takes, for code that passes rows around and wants to say so.
@@ -93,7 +107,7 @@ The stub is checked against the module it describes in CI: griffe reads the stub
 
 ## What works today
 
-The list above is what this client is for. What it does so far is the core of it: `connect`, `execute` and `sql` with named parameters, results that iterate and fetch, values as Python objects both ways including dates, times, datetimes and durations, `Node`, `Rel` and `Path` as classes, `load` for building a graph with edges in it, an appender for growing one, every condition as an exception class carrying its code, its position and its documentation link, results as Arrow columns and as pandas and polars frames, stubs inside the wheel with a gate that keeps them true, and the GIL released around every statement, every load and every copy out. `register` and the interrupt are next, and each one lands with the tests that say it works.
+The list above is what this client is for. What it does so far is the core of it: `connect`, `execute` and `sql` with named parameters, results that iterate and fetch, values as Python objects both ways including dates, times, datetimes and durations, `Node`, `Rel` and `Path` as classes, `load` for building a graph with edges in it, an appender for growing one, every condition as an exception class carrying its code, its position and its documentation link, results as Arrow columns and as pandas and polars frames, stubs inside the wheel with a gate that keeps them true, the GIL released around every statement, every load and every copy out, and `Ctrl-C` and `interrupt()` stopping a statement without touching the connection under it. `register` is next, and each one lands with the tests that say it works.
 
 ## Wheels
 

python/zudb/_zudb.pyi

@@ -63,6 +63,13 @@ class Connection:
     def closed(self) -> bool:
         """Whether this connection is still open."""
 
+    @property
+    def rows_read(self) -> int:
+        """How many rows the statement running on this connection has read out of storage."""
+
+    def interrupt(self) -> None:
+        """Asks the statement running on this connection to stop."""
+
     def execute(self, statement: str, params: Mapping[str, Value] | None = None) -> Result:
         """Runs one statement and gives back its rows."""
 

src/conn.rs

@@ -9,16 +9,18 @@
 
 use std::ffi::CStr;
 use std::path::PathBuf;
-use std::sync::Mutex;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex, OnceLock};
 
 use pyo3::prelude::*;
 use pyo3::types::{PyCapsule, PyDict, PyList, PyTuple};
 use zudb::query::{QueryResult, Value};
-use zudb::{Config, Database};
+use zudb::{Config, Database, Interrupt};
 
 use crate::appender::Appender;
 use crate::columns;
 use crate::error::{closed, to_py_err};
+use crate::interrupt;
 use crate::value::{Names, from_py, to_py};
 
 /// What a capsule holding an Arrow stream is called. The name is part
@@ -44,7 +46,25 @@ pub struct Connection {
     /// thread in the process for the length of somebody else's
     /// statement, and would deadlock against the thread inside that
     /// statement, which needs the GIL back to return.
-    pub(crate) inner: Mutex<Option<zudb::Connection>>,
+    ///
+    /// Counted rather than plain, because the thread a statement runs
+    /// on takes a share of it: a job handed to that thread outlives the
+    /// call that made it in the type system even though it never does
+    /// in fact.
+    pub(crate) inner: Arc<Mutex<Option<zudb::Connection>>>,
+    /// The thread statements run on where a `Ctrl-C` can arrive,
+    /// started at the first one that needs it and kept for the rest.
+    runner: OnceLock<interrupt::Runner>,
+    /// The word the running statement reads, held out here rather than
+    /// reached through the lock. A stop that had to wait for the
+    /// connection to be free could only ever arrive after the statement
+    /// it was meant to stop.
+    stop: Interrupt,
+    /// Whether the connection is still open, kept beside the lock for
+    /// the same reason: asking a connection whether it is closed, or
+    /// asking it to stop, should not queue behind a ten second
+    /// statement.
+    alive: AtomicBool,
     #[pyo3(get)]
     path: PathBuf,
     #[pyo3(get)]
@@ -67,27 +87,30 @@ impl Connection {
         params: Option<&Bound<'_, PyDict>>,
     ) -> PyResult<Result> {
         let params = bind(params)?;
-        let borrowed: Vec<(&str, Value)> = params
-            .iter()
-            .map(|(name, value)| (name.as_str(), value.clone()))
-            .collect();
+        // Owned rather than borrowed, both of them, because the thread
+        // that runs the statement is not this one and the borrow would
+        // have to say so. It is two allocations against a statement,
+        // which is nothing beside the parse it is about to have.
+        let statement = statement.to_string();
         // The GIL goes down for the whole statement, waiting for the
         // connection's own lock included. That is the point of a
         // compiled engine in a Python process: another thread runs
-        // while this one is inside the executor, and the signal
-        // handler gets to run too, which is what lets a `Ctrl-C`
-        // arrive at all. Waiting for the lock with the GIL held would
-        // be worse than slow: the thread inside the statement has to
-        // take the GIL back to return, and it could not.
-        let (result, names) = py
-            .detach(|| -> std::result::Result<_, Trouble> {
-                let mut held = self.inner.lock().map_err(|_| Trouble::Closed)?;
-                let conn = held.as_mut().ok_or(Trouble::Closed)?;
+        // while this one is inside the executor. Where a `Ctrl-C` can
+        // arrive the statement goes on the connection's own thread and
+        // this one waits for it, which is the only way a press is felt
+        // before the statement ends.
+        let (result, names) =
+            interrupt::watched(py, &self.runner, &self.inner, &self.stop, move |conn| {
+                let borrowed: Vec<(&str, Value)> = params
+                    .iter()
+                    .map(|(name, value)| (name.as_str(), value.clone()))
+                    .collect();
                 let names = Names::of(conn.session_mut().catalog());
-                let result = conn.query_with(statement, &borrowed)?;
-                Ok((result, names))
+                conn.query_with(&statement, &borrowed)
+                    .map(|result| (result, names))
             })
-            .map_err(|trouble| trouble.raise(py))?;
+            .map_err(|stopped| stopped.raise(py))?
+            .map_err(|err| to_py_err(py, err))?;
         Ok(Result {
             result,
             names,
@@ -119,6 +142,40 @@ impl Connection {
         Appender::open(py, slf, table)
     }
 
+    /// Asks the statement running on this connection to stop.
+    ///
+    /// The one call meant to be made from another thread while the
+    /// connection is busy, which is why it waits for nothing: the
+    /// statement it stops is the one holding everything a call that
+    /// waited would be waiting for. The statement raises
+    /// `zudb.Interrupted` and the connection is exactly as it was, so
+    /// the next statement on it starts warm. That is the difference
+    /// between stopping a statement and closing a connection.
+    ///
+    /// With nothing running this does nothing. It does not arm a stop
+    /// for the next statement, because a statement nobody has run yet
+    /// is not one anybody has waited too long for.
+    fn interrupt(&self, py: Python<'_>) -> PyResult<()> {
+        if !self.alive.load(Ordering::Acquire) {
+            return Err(closed(py, "this connection"));
+        }
+        self.stop.stop();
+        Ok(())
+    }
+
+    /// How many rows the statement running on this connection has read
+    /// out of storage, for showing a person that something is
+    /// happening.
+    ///
+    /// Rows read rather than rows answered, because the statement
+    /// somebody is waiting on is exactly the one that reads a hundred
+    /// million rows to answer one. It starts at zero at each statement
+    /// and holds its last value once one ends.
+    #[getter]
+    fn rows_read(&self) -> u64 {
+        self.stop.rows()
+    }
+
     /// Closes the connection and frees what it held.
     ///
     /// Doing it twice is not an error, because a `with` block that
@@ -131,13 +188,21 @@ impl Connection {
             if let Ok(mut held) = self.inner.lock() {
                 drop(held.take());
             }
+            // Written after the drop rather than before it, so that a
+            // connection reports itself open until it really is not,
+            // and a poisoned lock reports itself closed because
+            // nothing can be run on one.
+            self.alive.store(false, Ordering::Release);
         });
     }
 
     /// Whether this connection is still open.
+    ///
+    /// Answered from a word beside the lock rather than through it, so
+    /// that asking does not queue behind whatever is running.
     #[getter]
-    fn closed(&self, py: Python<'_>) -> bool {
-        py.detach(|| self.inner.lock().map(|held| held.is_none()).unwrap_or(true))
+    fn closed(&self) -> bool {
+        !self.alive.load(Ordering::Acquire)
     }
 
     fn __enter__(slf: PyRef<'_, Self>) -> PyRef<'_, Self> {
@@ -153,8 +218,8 @@ impl Connection {
         false
     }
 
-    fn __repr__(&self, py: Python<'_>) -> String {
-        let state = if self.closed(py) { ", closed" } else { "" };
+    fn __repr__(&self) -> String {
+        let state = if self.closed() { ", closed" } else { "" };
         format!("<zudb.Connection {}{state}>", self.path.display())
     }
 }
@@ -189,39 +254,21 @@ impl Connection {
             }
             .and_then(|db| db.connect())
         });
+        let opened = opened.map_err(|err| to_py_err(py, err))?;
         Ok(Connection {
-            inner: Mutex::new(Some(opened.map_err(|err| to_py_err(py, err))?)),
+            // Taken here, once, because every later reader of it wants
+            // it while the connection is busy and taking it then would
+            // mean waiting for the statement it is there to stop.
+            stop: opened.interrupt(),
+            inner: Arc::new(Mutex::new(Some(opened))),
+            runner: OnceLock::new(),
+            alive: AtomicBool::new(true),
             path,
             read_only,
         })
     }
 }
 
-/// What can go wrong inside a statement, with the GIL down and no way
-/// to build a Python exception yet.
-enum Trouble {
-    Closed,
-    Engine(zudb::ZuError),
-}
-
-impl From<zudb::ZuError> for Trouble {
-    fn from(err: zudb::ZuError) -> Trouble {
-        Trouble::Engine(err)
-    }
-}
-
-impl Trouble {
-    fn raise(self, py: Python<'_>) -> PyErr {
-        match self {
-            // A connection whose lock a panic left poisoned is a
-            // connection nothing can be run on again, which is the
-            // same fact as a closed one and reads better as one.
-            Trouble::Closed => closed(py, "this connection"),
-            Trouble::Engine(err) => to_py_err(py, err),
-        }
-    }
-}
-
 /// The rows a statement gave back.
 ///
 /// Held as the engine produced them and turned into Python objects