Canceling a running future

[romansvozil]

Hi, I'm playing around with PyO3, and I'm trying to implement some scripting API for my game (let's say walking & attacking), but the issue is that sometimes I would like to cancel the already running future.

Code:

from asyncio import wait_for   
from asyncio import TimeoutError

import chrono_api

class ChronoScript: 
    async def run(self) -> None:
        try:
            await wait_for(chrono_api.walk(5), 2.)
        except TimeoutError:
            print("Walking was canceled!")

Where the walk function looks like this:

#[pyfunction]
fn walk(py: Python, duration: i32) -> PyResult<&PyAny> {
    pyo3_asyncio::tokio::future_into_py(py, async move {
        for i in 0..duration {
            println!("Walking tick: {}", i);
            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
        }
        Ok(Python::with_gil(|py| py.None()))
    })
}

Real output

Walking tick: 0
Walking tick: 1
Walking tick: 2
Walking was canceled!
Walking tick: 3
Walking tick: 4

The expected output

Walking tick: 0
Walking tick: 1
Walking tick: 2 (maybe)
Walking was canceled!
-- nothing more --

Is it possible to somehow achieve this output?
I'm sorry if I missed something 👍

(Whole repo can be found here: https://github.com/romansvozil/embedded-python/tree/master/src)

[awestlake87]

What versions of Python and pyo3-asyncio are you using? I know recently we put in a patch that should make cancellation work.

I just plugged your code into a test project and it gave the expected output for me:

# Cargo.toml

[package]
name = "pyo3-asyncio-lib"
version = "0.1.0"
authors = ["Andrew J Westlake <awestlake87@yahoo.com>"]
edition = "2018"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[lib]
name = "my_async_module"
crate-type = ["cdylib"]

[dependencies]
pyo3 = { version = "0.14", features = ["extension-module"] }
pyo3-asyncio = { version = "0.14", features = ["tokio-runtime"] }
tokio = "1.4"

//! lib.rs

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

#[pyfunction]
fn walk(py: Python, duration: i32) -> PyResult<&PyAny> {
    pyo3_asyncio::tokio::future_into_py(py, async move {
        for i in 0..duration {
            println!("Walking tick: {}", i);
            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
        }
        Ok(Python::with_gil(|py| py.None()))
    })
}
#[pymodule]
fn my_async_module(_py: Python, m: &PyModule) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(walk, m)?)?;

    Ok(())
}

# example.py
import asyncio
from asyncio import wait_for   
from asyncio import TimeoutError

import my_async_module as chrono_api

class ChronoScript: 
    async def run(self) -> None:
        try:
            await wait_for(chrono_api.walk(5), 2.)
        except TimeoutError:
            print("Walking was canceled!")

asyncio.run(ChronoScript().run())

maturin develop && python3 example.py
🔗 Found pyo3 bindings
🐍 Found CPython 3.8 at python3
    Finished dev [unoptimized + debuginfo] target(s) in 0.03s
Walking tick: 0
Walking tick: 1
Walking tick: 2
Walking was canceled!

Occasionally the order of Walking tick: 2 and Walking was canceled! are switched, but I don't see it going up to 4.

[awestlake87]

I know your project is set up a bit different. I've got some family stuff going on for most of the day, but if i find time, Ill play around with that repo and see if i can figure out the difference.

[romansvozil]

It "works" because the interpreter exits before another prints, but when you add some delay at the end, it continues printing

import asyncio
from asyncio import wait_for   
from asyncio import TimeoutError
from asyncio import sleep

import my_async_module as chrono_api

class ChronoScript: 
    async def run(self) -> None:
        try:
            await wait_for(chrono_api.walk(5), 2.)
        except TimeoutError:
            print("Walking was canceled!")
            await sleep(5)


asyncio.run(ChronoScript().run())

Output

Walking tick: 0
Walking tick: 1
Walking tick: 2
Walking was canceled!
Walking tick: 3
Walking tick: 4

[awestlake87]

Ah ok, my bad. I think I know what's wrong now. When we added support for cancellation, what we really did was fix a bug where the runtime would panic when the future was cancelled. The Rust future will continue running.

I believe what we can do is add a 'done' callback to the python future and cancel the Rust future from there. I'll be away from my computer for awhile, but i can write up a helper function for a cancellable future. It might be worth adding to the library!

[awestlake87]

Alright, I didn't get time to look at this yesterday, but this morning I added a set of conversions in #48 that should handle cancellation properly. If you replace future_into_py with cancellable_future_into_py, the Rust future should be cancelled as well when the asyncio.TimeoutError is thrown.

I've added a test for it and it seems to work on my end, but let me know if this works for you!

[romansvozil]

Works fine now 👍

[awestlake87]

Awesome! I'm gonna give the PR a bit of time in case anyone wants to give some feedback, but I'll probably release this as 0.14.1 at some point in the next week or two (unless pyo3 0.15 is planned to be released sooner rather than later).

[romansvozil]

Anyways I also noticed that the future is getting called right after creating, instead of after awaiting.

Example

class ChronoScript: 
    async def run(self) -> None:
        chrono_api.walk(5) # gets executed in the background even without an await
        await sleep(5)

asyncio.run(ChronoScript().run())

Here I would expect that there is not gonna be any output, but the chrono_api.walk(5) gets executed even without explicit await, which is probably not the right behavior (at least not expected).

But this is probably not even closely related to this issue 😄

[ShadowJonathan]

I do think that's indeed an issue (even if it's a seperate one), as in both python, and in rust, there's an implicit agreement that futures dont get run until they're awaited.

In this case, the future there gets created, but immediately "removed" when it goes out of scope, even though the garbage collector doesn't collect it until later. That can be a problem, because tasks can be "prepared", and then ran concurrently in the background from a 'correct' scope, and all sorts of other python tricks.

[awestlake87]

I believe this is something that's fundamental to the library since we have two event loops. What you're really doing during a conversion is spawning a future on another runtime and awaiting the join handle. I'm open to suggestions on fixes for this issue, but I think it's a consequence of the trade-offs we made for this library since the underlying runtime is not shared.

Something you can do to avoid starting the future until the first await is wrap it in a simple coroutine:

async def walk(duration):
     return await chrono_api.walk(duration)

This will create the coroutine when walk(5) is called, but it will only start the Rust future when the walk(5) coroutine is awaited.

This problem is related to the one documented here.

[awestlake87]

Just a heads up, I finally released this as 0.14.1. (Sorry I put it off for so long!)