fix(py): restore push send throughput by avoiding per-send GIL release

- remove py.allow_threads from single-message PyRuntime send path
- add regression test for push burst GIL interleaving behavior
- add push regression benchmark script for cross-commit ARM/x86 comparison

Co-authored-by: Iris Seravelle <iris.seravelle@gmail.com>

Author: SSL-ACTX

src/py/pool.rs

@@ -212,7 +212,7 @@ mod tests {
                 Arc::new(AtomicU64::new(0)),
                 Arc::new(rt.clone()),
             )
-                .expect("max_threads=0 should force shared pool mode without strict error");
+            .expect("max_threads=0 should force shared pool mode without strict error");
             assert!(ch.is_none());
         });
     }

src/py/runtime.rs

@@ -1080,9 +1080,11 @@ impl PyRuntime {
     }
 
     fn send(&self, py: Python, pid: u64, data: &PyAny) -> PyResult<bool> {
+        let _ = py;
         let msg = py_any_to_bytes(data)?;
-        let rt = self.inner.clone();
-        Ok(py.allow_threads(move || rt.send_user(pid, msg).is_ok()))
+        // Keep GIL for single-message sends so hot push-actor loops do not
+        // interleave callback execution on every call.
+        Ok(self.inner.send_user(pid, msg).is_ok())
     }
 
     /// Batch-send bytes-like payloads to a PID.

tests/benchmarks/benchmark_push_regression.py

@@ -0,0 +1,135 @@
+import argparse
+import time
+from statistics import mean
+
+import iris
+
+
+def spawn_push_actors(rt, count: int, budget: int):
+    pids = []
+    t0 = time.perf_counter()
+
+    def no_op_handler(_msg):
+        return None
+
+    for _ in range(count):
+        pids.append(rt.spawn(no_op_handler, budget=budget, release_gil=False))
+
+    elapsed = time.perf_counter() - t0
+    return pids, elapsed
+
+
+def send_one_per_actor(rt, pids, payload: bytes):
+    t0 = time.perf_counter()
+    for pid in pids:
+        ok = rt.send(pid, payload)
+        if not ok:
+            raise RuntimeError(f"send failed for pid={pid}")
+    elapsed = time.perf_counter() - t0
+    return elapsed
+
+
+def spawn_pull_actors(rt, count: int, budget: int):
+    pids = []
+    t0 = time.perf_counter()
+
+    def mailbox_handler(mailbox):
+        while True:
+            msg = mailbox.recv()
+            if msg is None:
+                break
+
+    for _ in range(count):
+        pids.append(rt.spawn_with_mailbox(mailbox_handler, budget=budget))
+
+    elapsed = time.perf_counter() - t0
+    return pids, elapsed
+
+
+def cleanup(rt, pids):
+    for pid in pids:
+        rt.stop(pid)
+
+
+def fmt_rate(total: int, secs: float) -> str:
+    if secs <= 0:
+        return "inf"
+    return f"{total / secs:,.0f}"
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="One-shot benchmark for push actor send regression checks (ARM-friendly)."
+    )
+    parser.add_argument("--count", type=int, default=100_000, help="Number of actors/messages")
+    parser.add_argument("--budget", type=int, default=10, help="Actor budget")
+    parser.add_argument(
+        "--payload-size",
+        type=int,
+        default=4,
+        help="Payload size in bytes",
+    )
+    parser.add_argument(
+        "--rounds",
+        type=int,
+        default=3,
+        help="How many rounds to run for the push path",
+    )
+    parser.add_argument(
+        "--include-pull",
+        action="store_true",
+        help="Also benchmark pull/mailbox actors in same run",
+    )
+    args = parser.parse_args()
+
+    payload = b"p" * args.payload_size
+
+    print("--- Iris Push Regression Benchmark ---")
+    print(f"version={iris.version()}")
+    print(
+        f"count={args.count:,}, budget={args.budget}, payload_size={args.payload_size}, rounds={args.rounds}"
+    )
+    print("Tip: run this script on two commits and compare send msg/s.")
+
+    push_spawn_times = []
+    push_send_times = []
+
+    for i in range(args.rounds):
+        rt = iris.Runtime()
+        pids = []
+        try:
+            pids, spawn_t = spawn_push_actors(rt, args.count, args.budget)
+            send_t = send_one_per_actor(rt, pids, payload)
+            push_spawn_times.append(spawn_t)
+            push_send_times.append(send_t)
+
+            print(
+                f"round {i + 1} push: spawn={spawn_t:.3f}s ({fmt_rate(args.count, spawn_t)} actors/s) | "
+                f"send={send_t:.3f}s ({fmt_rate(args.count, send_t)} msg/s)"
+            )
+        finally:
+            cleanup(rt, pids)
+
+    print("\n--- Push Summary (mean) ---")
+    mean_spawn = mean(push_spawn_times)
+    mean_send = mean(push_send_times)
+    print(f"spawn: {mean_spawn:.3f}s ({fmt_rate(args.count, mean_spawn)} actors/s)")
+    print(f"send : {mean_send:.3f}s ({fmt_rate(args.count, mean_send)} msg/s)")
+
+    if args.include_pull:
+        rt = iris.Runtime()
+        pids = []
+        try:
+            pids, pull_spawn_t = spawn_pull_actors(rt, args.count, args.budget)
+            pull_send_t = send_one_per_actor(rt, pids, payload)
+            print("\n--- Pull (mailbox) One Shot ---")
+            print(
+                f"spawn={pull_spawn_t:.3f}s ({fmt_rate(args.count, pull_spawn_t)} actors/s) | "
+                f"send={pull_send_t:.3f}s ({fmt_rate(args.count, pull_send_t)} msg/s)"
+            )
+        finally:
+            cleanup(rt, pids)
+
+
+if __name__ == "__main__":
+    main()

tests/pyo3_runtime.rs

@@ -215,6 +215,82 @@ async fn py_send_many_accepts_bytes_like_objects() {
     assert_eq!(msgs, vec![b"one".to_vec(), b"two".to_vec(), b"three".to_vec()]);
 }
 
+#[tokio::test]
+async fn py_send_push_loop_does_not_yield_gil_mid_burst_regression() {
+    let (rt_py, pid, list_obj, old_switch): (PyObject, u64, PyObject, f64) = Python::with_gil(
+        |py| {
+            let module = iris::py::make_module(py).expect("make_module");
+            let runtime_type = module
+                .as_ref(py)
+                .getattr("PyRuntime")
+                .expect("no PyRuntime type");
+            let rt = runtime_type.call0().expect("construct PyRuntime");
+
+            let locals = pyo3::types::PyDict::new(py);
+            py.run(
+                "import sys\nold = sys.getswitchinterval()\nsys.setswitchinterval(1.0)\nitems = []\ndef cb(b):\n    items.append(b)\n",
+                Some(locals),
+                Some(locals),
+            )
+            .unwrap();
+
+            let cb = locals.get_item("cb").unwrap().unwrap();
+            let list_obj: PyObject = locals.get_item("items").unwrap().unwrap().into_py(py);
+            let old_switch: f64 = locals.get_item("old").unwrap().unwrap().extract().unwrap();
+
+            let pid: u64 = rt
+                .call_method1("spawn_py_handler", (cb, 100usize, false))
+                .unwrap()
+                .extract()
+                .unwrap();
+
+            for _ in 0..2000usize {
+                let ok: bool = rt
+                    .call_method1("send", (pid, PyBytes::new(py, b"x")))
+                    .unwrap()
+                    .extract()
+                    .unwrap();
+                assert!(ok);
+            }
+
+            let len_during_burst: usize = list_obj
+                .as_ref(py)
+                .downcast::<pyo3::types::PyList>()
+                .unwrap()
+                .len();
+            assert_eq!(
+                len_during_burst, 0,
+                "callbacks ran during send burst (send likely yielded GIL)"
+            );
+
+            (rt.into_py(py), pid, list_obj, old_switch)
+        },
+    );
+
+    tokio::time::sleep(Duration::from_millis(250)).await;
+
+    Python::with_gil(|py| {
+        let rt = rt_py.as_ref(py);
+        let _ = rt; // Keep runtime alive while assertions run.
+
+        let items = list_obj
+            .as_ref(py)
+            .downcast::<pyo3::types::PyList>()
+            .unwrap();
+        assert!(
+            items.len() > 0,
+            "expected messages to be eventually processed after burst"
+        );
+
+        let sys = py.import("sys").unwrap();
+        sys.call_method1("setswitchinterval", (old_switch,))
+            .unwrap();
+
+        // Cleanup spawned actor.
+        let _ = rt.call_method1("stop", (pid,));
+    });
+}
+
 #[tokio::test]
 async fn py_runtime_spawn_and_send() {
     // create a single PyRuntime instance and keep it alive across await points