PyAny and "future cannot be sent between threads safely"

[vitalik]

Hi

I'm pretty new to Rust, and I'm trying to implement some simple http server that will do some numbercunching using rust and then when done - call some python callback function

so that from python side my module will look like this:

import asyncio
import myrustmodule


def callback(path: str):
    print("callback called: ", path)


async def main():
    await myrustmodule.start_server(3000, callback) # start http server on port 3000 and call callback 

asyncio.run(main())

so I took example from hyper and pyo3 :

#![deny(warnings)]

use std::convert::Infallible;
use std::net::SocketAddr;

use bytes::Bytes;
use http_body_util::Full;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Request, Response};
use tokio::net::TcpListener;

use pyo3::{prelude::*, wrap_pyfunction};
use pyo3::exceptions::PyRuntimeError;



async fn some_work(request: Request<hyper::body::Incoming>, callback: &PyAny) -> Result<Response<Full<Bytes>>, Infallible> {
    callback.call0();
    Ok(Response::new(Full::new(Bytes::from("OK"))))
}

async fn start_hyper_server(port: u16, callback: &PyAny) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    pretty_env_logger::init();

    // This address is localhost
    let addr: SocketAddr = ([127, 0, 0, 1], port).into();

    // Bind to the port and listen for incoming TCP connections
    let listener = TcpListener::bind(addr).await?;
    println!("Listening on http://{}", addr);
    loop {
        let (stream, _) = listener.accept().await?;

        tokio::task::spawn(async move {
            // Handle the connection from the client using HTTP1 and pass any
            // HTTP requests received on that connection to the `hello` function
            if let Err(err) = http1::Builder::new()
                .serve_connection(stream, service_fn(|request| {
                    some_work(request, callback)
                }))
                .await
            {
                println!("Error serving connection: {:?}", err);
            }
        });
    }
}

#[pyfunction]
fn start_server<'a>(py: Python<'a>, port: u16, callback: &'a PyAny) -> PyResult<&'a PyAny> {
    pyo3_asyncio::tokio::future_into_py(py, async move {
        match start_hyper_server(port, callback).await {
            Ok(_) => Ok(Python::with_gil(|py| py.None())),
            Err(err) => {
                let msg = err.to_string();
                let pyerr: PyErr = PyRuntimeError::new_err(msg).into();
                Err(pyerr)
            }
        }
    })
}



/// A Python module implemented in Rust.
#[pymodule]
fn myrustmodule(_py: Python, m: &PyModule) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(start_server, m)?)?;
    Ok(())
}

but that gives me compilation errors:

error: future cannot be sent between threads safelyyperpy                                                                                              
   --> src/lib.rs:36:28
    |
36  |           tokio::task::spawn(async move {
    |  ____________________________^
37  | |             // Handle the connection from the client using HTTP1 and pass any
38  | |             // HTTP requests received on that connection to the `hello` function
39  | |             if let Err(err) = http1::Builder::new()
...   |
46  | |             }
47  | |         });
    | |_________^ future created by async block is not `Send`

can some one help me fix this code

and is it actually safe calling python functions from this or it should be wrapped somehow ?

Thank you

[adamreichold]

&PyAny is a GIL-bound reference, you need a GIL-independent Py<PyAny> a.k.a. PyObject to send to another thread. On that thread, you need to use Python::with_gil to acquire the GIL and then Py::as_ref to to access &PyAny given Py<PyAny>. https://pyo3.rs/v0.18.3/memory has more details on this.

[vitalik]

Hi @adamreichold

thank you for reply, but not sure I understand how the gil should be used

let's take simple example:

use pyo3::prelude::*;

/// Formats the sum of two numbers as string.
#[pyfunction]
fn function_in_rust(py: Python, a: usize, b: usize, callback_obj: PyObject) -> PyResult<String> {
    callback_obj.as_ref(py).call0()?;
    Ok((a + b).to_string())
}

/// A Python module implemented in Rust.
#[pymodule]
fn callbacktest(_py: Python, m: &PyModule) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(function_in_rust, m)?)?;
    Ok(())
}

Am I required to wrap into some gil my callback_obj.call0 ? and how do I do that ?

[adamreichold]

Am I required to wrap into some gil my callback_obj.call0 ? and how do I do that ?

You already have: py: Python is a GIL token, a zero-sized type acting as a proof that you have acquired the GIL. If you wanted to do this on another thread (where you cannot send that py token), you would need to acquire the GIL explicitly, e.g.

#[pyfunction]
fn function_in_rust(py: Python, a: usize, b: usize, callback_obj: PyObject) -> PyResult<String> {
    std::thread::spawn(move || {
        Python::with_gil(|py| {
            callback_obj.as_ref(py).call0().unwrap();
        });
    });
    Ok((a + b).to_string())
}