Aren't the asynchronous runtime in pyo3 and the runtime in create webframe actually identical?

[AriBermeki]

Aren't the asynchronous runtime in pyo3 and the runtime in create webframe actually identical? If not, how can I fix this? Python sends the data and executes the function correctly, but the data doesn't arrive at the other runtime.

use std::sync::{Arc, Mutex};
use std::sync::mpsc::{self, Receiver as StdReceiver, Sender as StdSender};

use pyo3::prelude::*;
use tokio::sync::mpsc::{self as tokio_mpsc, Receiver as TokioReceiver, Sender as TokioSender};
use tokio::sync::Mutex as TokioMutex;
use tao::{
    event_loop::{ControlFlow, EventLoopBuilder},
    window::WindowBuilder,
};

#[derive(Debug, Clone)]
pub enum Event {
    FromSync(String),
    FromAsync(String),
}

#[derive(Debug, Clone)]
pub struct EventBridge {
    pub to_sync: StdSender<Event>,
    pub from_sync: TokioSender<Event>,
}

impl EventBridge {
    pub fn new() -> (Self, StdReceiver<Event>, TokioReceiver<Event>) {
        let (tx_sync, rx_sync) = mpsc::channel();
        let (tx_tokio, rx_tokio) = tokio_mpsc::channel(32);
        (
            Self {
                to_sync: tx_sync,
                from_sync: tx_tokio,
            },
            rx_sync,
            rx_tokio,
        )
    }
}

#[pyclass]
#[derive(Clone)]
pub struct EventBus {
    bridge: EventBridge,
    #[allow(dead_code)]
    sync_receiver: Arc<Mutex<StdReceiver<Event>>>,
    async_receiver: Arc<TokioMutex<TokioReceiver<Event>>>,
}

#[pymethods]
impl EventBus {
    #[new]
    pub fn new() -> PyResult<Self> {
        let (bridge, sync_receiver, async_receiver) = EventBridge::new();
        Ok(Self {
            bridge,
            sync_receiver: Arc::new(Mutex::new(sync_receiver)),
            async_receiver: Arc::new(TokioMutex::new(async_receiver)),
        })
    }

    pub fn emit_to_rust<'a>(&self, py: Python<'a>, msg: &'a str) ->  PyResult<Bound<'a,PyAny>> {
        let sender = self.bridge.from_sync.clone();
        let msg_str = msg.to_string();
        let fut = async move {
            println!("[msg_str to tao eventloop] Received: {msg_str:?}");
            if let Err(e) = sender.send(Event::FromAsync(msg_str)).await {
                eprintln!("Tokio send error: {:?}", e);
            }
            Ok(())
        };
        Ok(pyo3_async_runtimes::tokio::future_into_py(py, fut)?)
    }
}

#[derive(Debug, Clone)]
pub enum UserEventFromPy {
    SetTitle(String),
    CloseWindow,
}

#[pyfunction]
fn create_webframe(bus: &EventBus) {
    let event_loop = EventLoopBuilder::<UserEventFromPy>::with_user_event().build();
    let proxy = event_loop.create_proxy();
    let window = WindowBuilder::new()
        .with_title("Default Title")
        .build(&event_loop)
        .unwrap();

    let async_rx = bus.async_receiver.clone();
    let proxy_clone = proxy.clone();

    std::thread::spawn(move || {
        let rt = tokio::runtime::Runtime::new().unwrap();
        rt.block_on(async move {
            let mut rx = async_rx.lock().await;
            while let Some(event) = rx.recv().await {
                println!("[event] Received: {event:?}");
                match event {
                    Event::FromSync(msg) | Event::FromAsync(msg) => {
                        println!("[FromSync] Received: {msg}");
                        println!("[FromAsync] Received: {msg}");
                        match msg.as_str() {
                            "title" => {
                                proxy_clone
                                    .send_event(UserEventFromPy::SetTitle("Titel from Python".to_string()))
                                    .ok();
                            }
                            "close" => {
                                proxy_clone
                                    .send_event(UserEventFromPy::CloseWindow)
                                    .ok();
                            }
                            other => {
                                println!("Unknown command: {other}");
                            }
                        }
                    }
                }
            }
        });
    });

    event_loop.run(move |event, _, control_flow| {
        *control_flow = ControlFlow::Wait;

        match event {
            tao::event::Event::UserEvent(user_event) => {
                match user_event {
                    UserEventFromPy::SetTitle(title) => {
                        println!("Set window title: {title}");
                        window.set_title(&title);
                    }
                    UserEventFromPy::CloseWindow => {
                        println!("close window");
                        *control_flow = ControlFlow::Exit;
                    }
                }
            }
            tao::event::Event::WindowEvent {
                event: tao::event::WindowEvent::CloseRequested,
                ..
            } => {
                println!("Window was closed");
                *control_flow = ControlFlow::Exit;
            }
            _ => {}
        }
    });
}

#[pymodule]
fn webruntime(_m: &Bound<'_, PyModule>) -> PyResult<()> {
    _m.add_function(wrap_pyfunction!(create_webframe, _m)?)?;
    _m.add_class::<EventBus>()?;
    Ok(())
}

PS C:\Users\MalekAli\Desktop\test_gg> & C:/Users/MalekAli/Desktop/test_gg/.venv/Scripts/python.exe c:/Users/MalekAli/Desktop/test_gg/main.py
NiceGUI ready to go on http://127.0.0.1:8000
[msg_str to tao eventloop] Received: "title"
[msg_str to tao eventloop] Received: "title"
[msg_str to tao eventloop] Received: "title"
[msg_str to tao eventloop] Received: "title"
[msg_str to tao eventloop] Received: "title"
[msg_str to tao eventloop] Received: "title"
[msg_str to tao eventloop] Received: "title"
[msg_str to tao eventloop] Received: "title"
[msg_str to tao eventloop] Received: "title"
[msg_str to tao eventloop] Received: "title"

[davidhewitt]

There's a lot going on here tbh that isn't specific to PyO3, e.g. the tao crate I have no idea about.

I can't see where you feed the sync_receiver into the Tao event loop, and you've even marked that with #[allow(dead_code)], so I have to just assume the implementation is incomplete and those messages are going nowhere.

[AriBermeki]

Imagine the code as if you start a Tk GUI, and Tk GUI has an EventloopProxy which is implemented with crossbeam-channel, and you want to send asynchronous events to eventloop via some way, so I researched whether this is possible with crossbeam-channel, then I came across this comment which answered it

How to use CrossBeam-Channel in Tokio?

So I implemented this EventBus, as shown above, and it didn't work. Then I looked at the Tauri code; they used CrossBeam channels. Then I tried it myself, as shown below, and it worked.

#[derive(Debug)]
pub enum FrameWindowEvent {
    GetTitle(String),
    SetTitle(String, String),
}

#[derive(Debug)]
pub enum AppEvent {
    WindowEvent(FrameWindowEvent),
}
#[pyclass]
#[derive(Debug, Clone)]
pub struct WindowHandle {
    proxy: std::sync::Arc<std::sync::Mutex<Option<tao::event_loop::EventLoopProxy<AppEvent>>>>,
}
#[pymethods]
impl WindowHandle {
    #[new]
    pub fn new() -> Self {
        Self {
            proxy: std::sync::Arc::new(std::sync::Mutex::new(None)),
        }
    }

    pub fn set_title<'py>(
        &self,
        py: Python<'py>,
        window_id: String,
        new_title: String,
    ) -> PyResult<Bound<'py, PyAny>> {
        let proxy = self.proxy.clone();

        pyo3_async_runtimes::tokio::future_into_py(py, async move {
            let proxy = proxy.lock().unwrap();
            if let Some(proxy) = proxy.as_ref() {
                let event = AppEvent::WindowEvent(FrameWindowEvent::SetTitle(window_id, new_title));
                proxy.send_event(event).map_err(|err| {
                    pyo3::exceptions::PyRuntimeError::new_err(format!("Send failed: {err:?}"))
                })?;
                Ok(())
            } else {
                Err(pyo3::exceptions::PyRuntimeError::new_err("No proxy bound"))
            }
        })
    }
}
impl WindowHandle {
    pub fn bind_proxy(&mut self, proxy: tao::event_loop::EventLoopProxy<AppEvent>) -> Result<()> {
        let mut guard = self.proxy.lock().unwrap();
        *guard = Some(proxy);
        Ok(())
    }
}

@davidhewitt
you are definitely better than me in CPython
Now I have one more question, Dawid. Every GUI has an OS event loop created by the respective operating system. This means that the event loop is triggered as soon as a window is opened. The question is, how can I intentionally abort the asynchronous event loop when the last window is closed or when the user closes the Python script with Ctrl-C to stop the program, which also stops _run_tk_gui.

fn _run_tk_gui(_py: Python) {
    use tao::{
        event::{Event, WindowEvent},
        event_loop::{ControlFlow, EventLoop},
        window::WindowBuilder,
    };
    let event_loop = EventLoop::new();
    let _window = WindowBuilder::new()
        .with_title("A fantastic window!")
        .with_inner_size(tao::dpi::LogicalSize::new(300.0, 300.0))
        .with_min_inner_size(tao::dpi::LogicalSize::new(200.0, 200.0))
        .build(&event_loop)
        .unwrap();

    event_loop.run(move |event, _, control_flow| {
        *control_flow = ControlFlow::Wait;

        match event {
            Event::WindowEvent {
                event: WindowEvent::CloseRequested,
                window_id: _,
                ..
            } => {
                // when the user closes the Python script with Ctrl-C to stop the program, which also stops _run_tk_gui.
                // if asyncio_event_loop.is_err() {
                //     *control_flow = ControlFlow::Exit;
                // } else { // when the user has closed the last window
                //     send Signale to Asyncio to stop the loop
                //     *control_flow = ControlFlow::Exit;
                // }

                
                *control_flow = ControlFlow::Exit;
            }


            _ => (),
        }
    });
}
#[pyfunction]
fn async_webframe_runtime(py: Python) -> PyResult<Bound<'_, PyAny>> {
    let start_gui = Python::with_gil(|_py| _run_tk_gui(_py));

    let fut = pyo3_async_runtimes::tokio::future_into_py(py, async move {
        Python::with_gil(|py| -> PyResult<PyObject> {
            let asyncio = py.import("asyncio")?;
            let concurrent_futures = py.import("concurrent.futures")?;
            let loop_obj = asyncio.call_method0("new_event_loop")?;
            asyncio.call_method1("set_event_loop", (loop_obj.clone(),))?;
            let executor = concurrent_futures
                .getattr("ProcessPoolExecutor")?
                .call1((1,))?;

            let blocked_future = loop_obj.call_method1("run_in_executor", (executor, start_gui))?;
            let fut_list = pyo3::types::PyList::new(py, &[blocked_future])?;
            let kwargs = pyo3::types::PyDict::new(py);
            kwargs.set_item("return_when", asyncio.getattr("FIRST_COMPLETED")?)?;

            let result = asyncio
                .call_method("wait", (fut_list,), Some(&kwargs))?
                .into_bound();
            Ok(result.into())
        })
    })?;
    Ok(fut)
}

Note

The reason for using ProcessPoolExecutor at program startup is that the GUI frameworks on Linux (X11/Wayland/GTK), Windows (Win32/WinRT), and macOS (AppKit/Cocoa) all require a synchronous event loop, which is necessarily executed in the main thread and therefore blocks it. This means that starting the event loop in a separate thread is not possible..

# main.py

async def main():
    await async_webframe_runtime()






if __name__ == "__main__":
    asyncio.run(main())

[davidhewitt]

Respectfully, you've just dropped a ton of code here which isn't tied to a specific PyO3 behaviour or feature but rather a downstream use case wiring up tao with Python, which I have no experience of.

I'll give a shot in the dark - it seems like what you're doing makes sense, you could try moving the ProcessPoolExecutor bit into Python and make run_tk_gui the bit the #[pyfunction], would be easier to debug.

I'll repeat again as I said in PyO3/pyo3-async-runtimes#49 (comment) - please stop dumping such large code blobs and asking for others to understand and debug these highly specific problems in your app. If you spent time minimising the examples you'd probably figure out the answer while also being respectful of other's time.

I will stop helping on these issues if this pattern continues.