How does asyncio interact/work?

[leeian1011]

Disclaimer: This question maybe unrelated to PyO3 in general as it is questions regarding how async works between Python and Rust

So I have a sample code here:

use std::{future::Future, pin::Pin, sync::Arc};

use pyo3::prelude::*;

#[derive(Debug)]
struct MyError;

impl std::fmt::Display for MyError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "MyError")
    }
}

impl std::error::Error for MyError {
    fn cause(&self) -> Option<&dyn std::error::Error> {
        None
    }

    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        None
    }

    fn description(&self) -> &str {
        "something happened"
    }
}

struct Handler {
    pub handle_something: Arc<
        dyn Fn(
            PyObject,
        ) -> Pin<
            Box<dyn Future<Output = Result<Py<PyAny>, Box<dyn std::error::Error>>> + Send + Sync>,
        > + Send + Sync,
    >,
}

impl Handler {
    fn new() -> Self {
        Self {
            handle_something: Arc::new(move |obj| {
                Box::pin(async move {
                    Python::with_gil(|py| {
                        let asloop = py.import_bound("asyncio")?;
                        Ok(asloop.call_method1("run", (obj.call_method1(py, "async_work", (100,)).unwrap(),)).unwrap().into_py(py))
                    })
                })
            }),
        }
    }
}


#[pyfunction]
fn start(py_obj_with_method: PyObject, py: Python<'_>) -> PyResult<Bound<PyAny>> {
    pyo3_asyncio::tokio::future_into_py(
        py,
        async move { Ok(do_something(py_obj_with_method).await) },
    )
}

async fn do_something(obj: PyObject) {
    let handler = Handler::new();
    match (handler.handle_something)(obj).await {
        Ok(_) => {},
        Err(e) => println!("error:{:?}", e),
    };
}

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

What I am trying to achieve here is to essentially have a python function that starts a rust 'runtime' of sorts that will invoke async python functions. The flow would be something similar to this:

When I try to run this code block in the handle_something definition such as:

impl Handler {
    fn new() -> Self {
        Self {
            handle_something: Arc::new(move |obj| {
                Box::pin(async move {
                    Python::with_gil(|py| {
                        pyo3_asyncio::tokio::into_future(
                            obj.call_method1(py, "async_work", (100,))?.into_bound(py),
                        )
                    })?
                    .await
                    .map(|_| ())
                    .map_err(|e| {
                        println!("error: {:?}", e);
                        Box::new(MyError) as Box<dyn std::error::Error>
                    })
                })
            }),
        }
    }
}

as well as changing the type for handle_something to

struct Handler {
    pub handle_something: Arc<
        dyn Fn(
            PyObject,
        ) -> Pin<
            Box<dyn Future<Output = Result<(), Box<dyn std::error::Error>>> + Send + Sync>,
        > + Send + Sync,
    >,
}

I will hit a no running event loop error.

When I face the no running event loop error, isn't there an event loop already running when I call asyncio.run(main())?
Why do I have to RPC asyncio.run() through rust in order to run the the python async function?
Why can't I run pyo3_asyncio::into_future?

Perhaps my understanding of how asyncio works is lacking.

I understand that this might be a post more related for pyo3_asyncio, just thought I'd ask here as it seems more likely I'll get help here!
Thank you for reading and for any help!!

[davidhewitt]

So at the moment in PyO3 we don't try to manage any event loop for either language in our experimental-async feature. I wonder, is https://pyo3.rs/v0.21.2/ecosystem/async-await#managing-event-loops perhaps a useful starting point for you to read from?

[leeian1011]

I've read through the pyo3 documentation that you linked as well as some other asyncio documentations and it's answered my initial questions by explaining how asyncio is rather sensitive to which thread is invoking it, thanks a lot for the reply I appreciate it!

[davidhewitt]

Great, glad to have helped!