Communication between Python and Rust threads/asyncloops

[roma-glushko]

Hey folks 👋

I'm a new guy to the PyO3 and its community, but I have gotten an interesting case to implement and wanted to consult with you on where to start.

I want to implement a class in Rust that would either run in a separate thread or thread + event loop. When started, the class is going to do some work (until explicitly stopped) and will put the results somewhere (normally, it would be some sort of queue).

At the same time, Python will be running in its own thread and will be pulling those results and processing them further.

So both parties (Python & Rust) will be running at their own paces and would be sharing just some sort of queue.

What is the best way to archive this sort of evilibrium? I would appreciate any information, examples that would seed my further investigation.

Thank you in advance 🙏

[davidhewitt]

Sorry for the delayed reply. Are you using async? If so, pyo3-asyncio may be able to help solve your problems and provide examples.

If you want to send Python and Rust jobs between threads, maybe #576 (comment) will be interesting to you.

[roma-glushko]

@davidhewitt I feel like I came really close to it but I have one issue that perplexes me.

I ended up using the crossbeam-channels lib to communicate between threads. They took the idea of Golang channels (that's another way to explain what I wanted to attain) and ported to Rust. Then I created a new thread scope and add the message generation logic there:

https://github.com/roma-glushko/inotify-toolkit/blob/c6316829098ca47fcfcf5ba6f1b1174901cbb010/src/watcher.rs#L406-L415

Although, the issue is that the whole thing gets hung up when the new thread is up. Seems like it does not really give any control back to the Python's main thread for some reason. Like it's trying to join the new thread immediately.

Somewhat equivalent Rust program works well though:

use notify::{Config, RecommendedWatcher, RecursiveMode, Watcher};
use std::path::Path;
use std::thread;

use crossbeam_channel::{unbounded, Sender, Receiver};

fn main() {
    if let Err(e) = watch("./wd") {
        println!("{:?}", e)
    }
}

fn watch<P: AsRef<Path>>(path: P) -> notify::Result<()> {
    let (tx, rx) = unbounded();

    let mut watcher = RecommendedWatcher::new(tx, Config::default())?;
    watcher.watch(path.as_ref(), RecursiveMode::Recursive)?;

    thread::scope(|scope| {
        scope.spawn(move || {
            for res in rx {
                match res {
                    Ok(event) => println!("Change: {event:?}"),
                    Err(error) => println!("Error: {error:?}"),
                }
            }
        });
    });

    Ok(())
}

Have you seen this problem before, David?

[davidhewitt]

Are you potentially missing a call to Python::allow_threads around the point where the thread blocks to wait on the channel?

[roma-glushko]

@davidhewitt this one is not that easy to hop over. Have tried to all allow_threads() all around the places, but still could not fix it 🤔

roma-glushko/notifykit@228025c

[davidhewitt]

Your choice of crossbeam_thread::scope is probably the cause. That requires the watcher thread to finish before it will ever return. Probably you just need to use std::thread::spawn (and save a reference to that thread if you need to kill it later).

You probably need to change your __enter__ method to the signature:

pub fn __enter__(slf: Py<self>, py: Python<'_>) -> PyResult<()> {

so that you're able to move the slf smart pointer inside the thread.

[roma-glushko]

@davidhewitt thank you for the help David!

What I ended up doing was separating the main logic out of PyO3 wrapper: https://github.com/roma-glushko/notifykit/blob/main/src/watcher.rs#L53

That really helped me to get reference passing right and now the main logic works 🙌