feat(pyo3): migrate to pyo3 0.28.3 and pyo3-async-runtimes 0.28.0 (p1)

- Upgrade pyo3 to 0.28.3 and replace pyo3-asyncio with pyo3-async-runtimes 0.28.0.
- Migrate codebase to use the modern Bound API (Python::attach, py.detach, etc.).
- Modernize #[pyclass] with (from_py_object) and replace deprecated downcast with cast.
- Update tests to align with new PyO3 APIs and remove outdated initialization calls.
- Fix clippy warnings and remove unnecessary unsafe blocks in Python bindings.
- Note: JIT and Vortex features are pending full migration in p2.

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

Author: SSL-ACTX

Cargo.toml

@@ -24,8 +24,8 @@ crossbeam-channel = "0.5"
 crossbeam-queue = "0.3.11"
 
 # Python (Default)
-pyo3 = { version = "0.20", features = ["auto-initialize"], optional = true}
-pyo3-asyncio = { version = "0.20", features = ["tokio-runtime"], optional = true }
+pyo3 = { version = "0.28.3", features = ["auto-initialize"], optional = true}
+pyo3-async-runtimes = { version = "0.28.0", features = ["tokio-runtime"], optional = true }
 
 # JIT (Optional and experimental)
 cranelift = { version = "0.87", optional = true }
@@ -43,7 +43,7 @@ napi-build = { version = "2.0", optional = true }
 
 [features]
 default = ["pyo3"]
-pyo3 = ["dep:pyo3", "dep:pyo3-asyncio"]
+pyo3 = ["dep:pyo3", "dep:pyo3-async-runtimes"]
 extension-module = ["pyo3/extension-module"]
 jit = ["dep:cranelift", "dep:cranelift-module", "dep:cranelift-jit", "dep:cranelift-native", "dep:wide"]
 # node = ["dep:napi", "dep:napi-derive", "dep:napi-build"]
@@ -52,7 +52,7 @@ vortex = ["pyo3"]
 [dev-dependencies]
 futures = "0.3"
 tracing-subscriber = "0.3"
-pyo3 = { version = "0.20", features = ["auto-initialize"] }
+pyo3 = { version = "0.28.3", features = ["auto-initialize"] }
 
 [profile.release]
 opt-level = 3

src/core/mod.rs

@@ -126,7 +126,8 @@ impl Runtime {
 
         #[cfg(feature = "pyo3")]
         {
-            pyo3::prepare_freethreaded_python();
+            // In modern PyO3, this is often handled by auto-initialize or not needed for library usage.
+            // pyo3::prepare_freethreaded_python();
         }
 
         let (telemetry_mgr, _event_rx) = telemetry::TelemetryManager::new();

src/py/jit_stub.rs

@@ -13,42 +13,42 @@ use pyo3::wrap_pyfunction;
 #[cfg(feature = "pyo3")]
 #[pyfunction]
 fn register_offload(
-    func: PyObject,
+    func: Py<PyAny>,
     _strategy: Option<String>,
     _return_type: Option<String>,
     _source_expr: Option<String>,
     _arg_names: Option<Vec<String>>,
-) -> PyResult<PyObject> {
+) -> PyResult<Py<PyAny>> {
     Ok(func)
 }
 
 #[cfg(feature = "pyo3")]
 #[pyfunction]
 fn offload_call(
-    py: Python,
-    func: PyObject,
-    args: &PyTuple,
-    kwargs: Option<&PyDict>,
-) -> PyResult<PyObject> {
-    func.as_ref(py).call(args, kwargs).map(|v| v.into_py(py))
+    py: Python<'_>,
+    func: Py<PyAny>,
+    args: &Bound<'_, PyTuple>,
+    kwargs: Option<&Bound<'_, PyDict>>,
+) -> PyResult<Py<PyAny>> {
+    func.bind(py).call(args, kwargs).map(|v| v.unbind())
 }
 
 #[cfg(feature = "pyo3")]
 #[pyfunction]
 fn call_jit(
-    _py: Python,
-    _func: PyObject,
-    _args: &PyTuple,
-    _kwargs: Option<&PyDict>,
-) -> PyResult<PyObject> {
+    _py: Python<'_>,
+    _func: Py<PyAny>,
+    _args: &Bound<'_, PyTuple>,
+    _kwargs: Option<&Bound<'_, PyDict>>,
+) -> PyResult<Py<PyAny>> {
     Err(pyo3::exceptions::PyRuntimeError::new_err(
         "JIT feature is disabled in this build",
     ))
 }
 
 #[cfg(feature = "pyo3")]
 #[pyfunction]
-fn call_jit_step_loop_f64(_func: PyObject, _seed: f64, _count: usize) -> PyResult<f64> {
+fn call_jit_step_loop_f64(_func: Py<PyAny>, _seed: f64, _count: usize) -> PyResult<f64> {
     Err(pyo3::exceptions::PyRuntimeError::new_err(
         "JIT feature is disabled in this build",
     ))
@@ -83,13 +83,13 @@ fn is_quantum_speculation_enabled() -> PyResult<bool> {
 
 #[cfg(feature = "pyo3")]
 #[pyfunction]
-fn get_quantum_profile(_func: PyObject) -> PyResult<Vec<(usize, f64, u64, u64)>> {
+fn get_quantum_profile(_func: Py<PyAny>) -> PyResult<Vec<(usize, f64, u64, u64)>> {
     Ok(Vec::new())
 }
 
 #[cfg(feature = "pyo3")]
 #[pyfunction]
-fn seed_quantum_profile(_func: PyObject, _rows: Vec<(usize, f64, u64, u64)>) -> PyResult<bool> {
+fn seed_quantum_profile(_func: Py<PyAny>, _rows: Vec<(usize, f64, u64, u64)>) -> PyResult<bool> {
     Ok(false)
 }
 
@@ -158,7 +158,7 @@ fn get_quantum_cooldown() -> PyResult<(u64, u64)> {
 }
 
 #[cfg(feature = "pyo3")]
-pub(crate) fn init_py(m: &PyModule) -> PyResult<()> {
+pub(crate) fn init_py(m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_function(wrap_pyfunction!(register_offload, m)?)?;
     m.add_function(wrap_pyfunction!(offload_call, m)?)?;
     m.add_function(wrap_pyfunction!(call_jit, m)?)?;

src/py/mailbox.rs

@@ -9,7 +9,7 @@ use tokio::sync::Mutex as TokioMutex;
 
 /// A wrapper around a live MailboxReceiver for Python actors.
 /// Revamped: Now purely blocking/synchronous to Python, running in dedicated threads.
-#[pyclass]
+#[pyclass(from_py_object)]
 #[derive(Clone)]
 pub struct PyMailbox {
     pub(crate) inner: Arc<TokioMutex<crate::mailbox::MailboxReceiver>>,
@@ -19,7 +19,8 @@ pub struct PyMailbox {
 impl PyMailbox {
     /// Receive the next message (Blocking).
     /// Releases the GIL while waiting. Checks for Python signals cleanly to allow Ctrl+C escapes.
-    fn recv(&self, py: Python, timeout: Option<f64>) -> PyResult<PyObject> {
+    #[pyo3(signature = (timeout=None))]
+    fn recv(&self, py: Python<'_>, timeout: Option<f64>) -> PyResult<Py<PyAny>> {
         let rx = self.inner.clone();
         let start = std::time::Instant::now();
         let timeout_dur = timeout.map(std::time::Duration::from_secs_f64);
@@ -40,7 +41,7 @@ impl PyMailbox {
             };
 
             // Release GIL to allow other threads to run while we block on the channel
-            let res = py.allow_threads(|| {
+            let res = py.detach(|| {
                 crate::RUNTIME.block_on(async {
                     let fut = async {
                         let mut guard = rx.lock().await;
@@ -68,12 +69,13 @@ impl PyMailbox {
 
     /// Selectively receive a message matching a Python predicate (Blocking).
     /// Releases the GIL while waiting. Checks for Python signals cleanly.
+    #[pyo3(signature = (matcher, timeout=None))]
     fn selective_recv(
         &self,
-        py: Python,
-        matcher: PyObject,
+        py: Python<'_>,
+        matcher: Py<PyAny>,
         timeout: Option<f64>,
-    ) -> PyResult<PyObject> {
+    ) -> PyResult<Py<PyAny>> {
         let rx = self.inner.clone();
         let start = std::time::Instant::now();
         let timeout_dur = timeout.map(std::time::Duration::from_secs_f64);
@@ -92,13 +94,16 @@ impl PyMailbox {
                 wait_time
             };
 
-            let res = py.allow_threads(|| {
+            let res = py.detach(|| {
                 crate::RUNTIME.block_on(async {
                     let fut = async {
                         let mut guard = rx.lock().await;
                         guard
                             .selective_recv(|msg| {
-                                Python::with_gil(|py| run_python_matcher(py, &matcher, msg))
+                                Python::attach(|py| {
+                                    Ok::<bool, PyErr>(run_python_matcher(py, matcher.bind(py), msg))
+                                })
+                                .unwrap_or(false)
                             })
                             .await
                     };

src/py/pool.rs

@@ -13,19 +13,18 @@ use crate::Runtime;
 #[cfg(feature = "pyo3")]
 use pyo3::prelude::*;
 use pyo3::types::PyBytes;
-use pyo3::PyObject;
 
 /// Task variants sent to dedicated or pooled GIL workers.
 #[cfg(feature = "pyo3")]
 pub(crate) enum PoolTask {
     Execute {
-        behavior: Arc<parking_lot::RwLock<PyObject>>,
+        behavior: Arc<parking_lot::RwLock<Py<PyAny>>>,
         bytes: bytes::Bytes,
         pid_holder: Arc<AtomicU64>,
         rt: Arc<Runtime>,
     },
     HotSwap {
-        behavior: Arc<parking_lot::RwLock<PyObject>>,
+        behavior: Arc<parking_lot::RwLock<Py<PyAny>>>,
         ptr: usize,
     },
 }
@@ -68,7 +67,7 @@ impl GilPool {
                             }
                             let success = crate::py::utils::run_python_callback(|py| {
                                 let guard = behavior.read();
-                                let cb = guard.as_ref(py);
+                                let cb = guard.bind(py);
                                 let pybytes = PyBytes::new(py, &bytes);
                                 cb.call1((pybytes,)).map(|_| ())
                             });
@@ -83,11 +82,14 @@ impl GilPool {
                             if unsafe { pyo3::ffi::Py_IsInitialized() } == 0 {
                                 break;
                             }
-                            Python::with_gil(|py| unsafe {
+                            Python::attach(|py| unsafe {
                                 let new_obj =
-                                    PyObject::from_owned_ptr(py, ptr as *mut pyo3::ffi::PyObject);
+                                    Bound::from_owned_ptr(py, ptr as *mut pyo3::ffi::PyObject)
+                                        .unbind();
                                 *behavior.write() = new_obj;
-                            });
+                                Ok::<(), PyErr>(())
+                            })
+                            .unwrap();
                         }
                     },
                     Err(cb_channel::RecvTimeoutError::Timeout) => continue,
@@ -107,7 +109,7 @@ impl GilPool {
 pub(crate) fn make_release_gil_channel(
     rt: &Runtime,
     release: bool,
-    behavior: Arc<parking_lot::RwLock<PyObject>>,
+    behavior: Arc<parking_lot::RwLock<Py<PyAny>>>,
     pid_holder: Arc<AtomicU64>,
     rt_arc: Arc<Runtime>,
 ) -> PyResult<Option<cb_channel::Sender<DedicatedPoolTask>>> {
@@ -160,7 +162,7 @@ pub(crate) fn make_release_gil_channel(
                     }
                     let success = crate::py::utils::run_python_callback(|py| {
                         let guard = _b_thread.read();
-                        let cb = guard.as_ref(py);
+                        let cb = guard.bind(py);
                         let pybytes = PyBytes::new(py, &bytes);
                         cb.call1((pybytes,)).map(|_| ())
                     });
@@ -175,10 +177,13 @@ pub(crate) fn make_release_gil_channel(
                     if unsafe { pyo3::ffi::Py_IsInitialized() } == 0 {
                         continue;
                     }
-                    Python::with_gil(|py| unsafe {
-                        let new_obj = PyObject::from_owned_ptr(py, ptr as *mut pyo3::ffi::PyObject);
+                    Python::attach(|py| unsafe {
+                        let new_obj =
+                            Bound::from_owned_ptr(py, ptr as *mut pyo3::ffi::PyObject).unbind();
                         *_b_thread.write() = new_obj;
-                    });
+                        Ok::<(), PyErr>(())
+                    })
+                    .unwrap();
                 }
             },
             Err(cb_channel::RecvTimeoutError::Timeout) => continue,
@@ -197,14 +202,12 @@ mod tests {
 
     #[test]
     fn max_threads_zero_forces_shared_pool_even_in_strict_mode() {
-        pyo3::prepare_freethreaded_python();
-
         let rt = Runtime::new();
         rt.set_release_gil_limits(0, 1);
         rt.set_release_gil_strict(true);
 
-        pyo3::Python::with_gil(|py| {
-            let behavior = Arc::new(parking_lot::RwLock::new(py.None().into_py(py)));
+        Python::attach(|py| {
+            let behavior = Arc::new(parking_lot::RwLock::new(py.None()));
             let ch = make_release_gil_channel(
                 &rt,
                 true,
@@ -214,6 +217,8 @@ mod tests {
             )
             .expect("max_threads=0 should force shared pool mode without strict error");
             assert!(ch.is_none());
-        });
+            Ok::<(), PyErr>(())
+        })
+        .unwrap();
     }
 }