Repository: actix/actix (language: rust)

Issue:
Future cancellation doesn't work with attached UDP stream
Using actix (d28d286) is see some behaviour that I cannot explain. My code opens a udp socket, attaches a Tokio codec and add that stream to a Actor.  Whenever a packet is received (regardless of the payload for this test) a timeout is scheduled and the corresponding `SpawnHandle` stored in a `Option`. If the `Option` is `Some` the previous timeout is canceled via `Context::cancel_future`.
I would expect that once the last packet of a sequence is received the timeout happens once. Turns out that in this scenario the timeouts aren't cancelled as expected. I couldn't reproduce the behaviour with a plain `futures::Stream::iter` - only with the UDP stuff. 

I boiled the stuff down to this test case:

```rust
extern crate actix;
extern crate futures;
extern crate tokio;

use actix::prelude::*;
use futures::prelude::*;
use std::net::SocketAddr;
use tokio::codec::BytesCodec;
use tokio::net::{UdpFramed, UdpSocket};

const ADDRESS: &str = "[::1]:42556";
const DESTINATION: &str = "[::1]:42557";

#[derive(actix::Message)]
struct Message;

#[derive(actix::Message)]
struct Timeout;

#[derive(Default)]
struct SomeActor {
    timeout: Option<SpawnHandle>,
    timeout_happened: bool,
}

impl Actor for SomeActor {
    type Context = Context<Self>;

    fn started(&mut self, ctx: &mut Self::Context) {
        let addr: SocketAddr = DESTINATION.parse().unwrap();
        let sock = UdpSocket::bind(&addr).unwrap();
        let (_sink, stream) = UdpFramed::new(sock, BytesCodec::new()).split();
        ctx.add_stream(stream.map(|(_, _)| Message));

        ctx.run_later(std::time::Duration::from_secs(1), |_, _| {
            println!("Sending");
            let socket = std::net::UdpSocket::bind(ADDRESS).expect("Failed to bind address");
            for _ in 0..10 {
                socket.send_to(&[b'0'], &DESTINATION).expect("Failed to send");
            }
        });

        ctx.run_later(std::time::Duration::from_secs(3), |_, ctx| {
            ctx.stop();
        });
    }
}

impl StreamHandler<Message, std::io::Error> for SomeActor {
    fn handle(&mut self, _: Message, ctx: &mut Context<Self>) {
        if let Some(handle) = self.timeout {
            println!("Cancelling timeout");
            ctx.cancel_future(handle);
        }
        println!("Starting timeout");
        let handle = ctx.notify_later(Timeout, std::time::Duration::from_secs(1));
        self.timeout = Some(handle);
    }
}

impl Handler<Timeout> for SomeActor {
    type Result = ();

    fn handle(&mut self, _: Timeout, _: &mut Context<Self>) -> Self::Result {
        println!("Timeout! Already there? {}", self.timeout_happened);
        assert!(!self.timeout_happened);
        self.timeout_happened = true;
    }
}

#[test]
fn test_udp_stream_cancel() {
    actix::System::run(|| {
        let _ = SomeActor::default().start();
    });
}
```  

The output looks like this:

```
running 1 test
Sending
Starting timeout
Cancelling timeout
Starting timeout
Cancelling timeout
Starting timeout
Cancelling timeout
Starting timeout
Cancelling timeout
Starting timeout
Cancelling timeout
Starting timeout
Cancelling timeout
Starting timeout
Cancelling timeout
Starting timeout
Cancelling timeout
Starting timeout
Cancelling timeout
Starting timeout
Timeout! Already there? false
Timeout! Already there? true
thread 'test_udp_stream_cancel' panicked at 'assertion failed: !self.timeout_happened', tests/test_stream.rs:66:9
note: Run with `RUST_BACKTRACE=1` for a backtrace.
Panic in Arbiter thread, shutting down system.
test test_udp_stream_cancel ... FAILED
```

What am I doing wrong? For me it looks like a bug.
Future cancellation doesn't work with attached UDP stream
Using actix (d28d286) is see some behaviour that I cannot explain. My code opens a udp socket, attaches a Tokio codec and add that stream to a Actor.  Whenever a packet is received (regardless of the payload for this test) a timeout is scheduled and the corresponding `SpawnHandle` stored in a `Option`. If the `Option` is `Some` the previous timeout is canceled via `Context::cancel_future`.
I would expect that once the last packet of a sequence is received the timeout happens once. Turns out that in this scenario the timeouts aren't cancelled as expected. I couldn't reproduce the behaviour with a plain `futures::Stream::iter` - only with the UDP stuff. 

I boiled the stuff down to this test case:

```rust
extern crate actix;
extern crate futures;
extern crate tokio;

use actix::prelude::*;
use futures::prelude::*;
use std::net::SocketAddr;
use tokio::codec::BytesCodec;
use tokio::net::{UdpFramed, UdpSocket};

const ADDRESS: &str = "[::1]:42556";
const DESTINATION: &str = "[::1]:42557";

#[derive(actix::Message)]
struct Message;

#[derive(actix::Message)]
struct Timeout;

#[derive(Default)]
struct SomeActor {
    timeout: Option<SpawnHandle>,
    timeout_happened: bool,
}

impl Actor for SomeActor {
    type Context = Context<Self>;

    fn started(&mut self, ctx: &mut Self::Context) {
        let addr: SocketAddr = DESTINATION.parse().unwrap();
        let sock = UdpSocket::bind(&addr).unwrap();
        let (_sink, stream) = UdpFramed::new(sock, BytesCodec::new()).split();
        ctx.add_stream(stream.map(|(_, _)| Message));

        ctx.run_later(std::time::Duration::from_secs(1), |_, _| {
            println!("Sending");
            let socket = std::net::UdpSocket::bind(ADDRESS).expect("Failed to bind address");
            for _ in 0..10 {
                socket.send_to(&[b'0'], &DESTINATION).expect("Failed to send");
            }
        });

        ctx.run_later(std::time::Duration::from_secs(3), |_, ctx| {
            ctx.stop();
        });
    }
}

impl StreamHandler<Message, std::io::Error> for SomeActor {
    fn handle(&mut self, _: Message, ctx: &mut Context<Self>) {
        if let Some(handle) = self.timeout {
            println!("Cancelling timeout");
            ctx.cancel_future(handle);
        }
        println!("Starting timeout");
        let handle = ctx.notify_later(Timeout, std::time::Duration::from_secs(1));
        self.timeout = Some(handle);
    }
}

impl Handler<Timeout> for SomeActor {
    type Result = ();

    fn handle(&mut self, _: Timeout, _: &mut Context<Self>) -> Self::Result {
        println!("Timeout! Already there? {}", self.timeout_happened);
        assert!(!self.timeout_happened);
        self.timeout_happened = true;
    }
}

#[test]
fn test_udp_stream_cancel() {
    actix::System::run(|| {
        let _ = SomeActor::default().start();
    });
}
```  

The output looks like this:

```
running 1 test
Sending
Starting timeout
Cancelling timeout
Starting timeout
Cancelling timeout
Starting timeout
Cancelling timeout
Starting timeout
Cancelling timeout
Starting timeout
Cancelling timeout
Starting timeout
Cancelling timeout
Starting timeout
Cancelling timeout
Starting timeout
Cancelling timeout
Starting timeout
Cancelling timeout
Starting timeout
Timeout! Already there? false
Timeout! Already there? true
thread 'test_udp_stream_cancel' panicked at 'assertion failed: !self.timeout_happened', tests/test_stream.rs:66:9
note: Run with `RUST_BACKTRACE=1` for a backtrace.
Panic in Arbiter thread, shutting down system.
test test_udp_stream_cancel ... FAILED
```

What am I doing wrong? For me it looks like a bug.


Discussion / hints:
Not sure what's going on there(need to check code) but you could try to check return value of `cancel_future` out of curiosity.


Hi, yes. Checked that already - it's `true`. No surprise because of the implementation of `ContextParts`:

```rust
    #[inline]
    /// Cancel previously scheduled future.
    pub fn cancel_future(&mut self, handle: SpawnHandle) -> bool {
        self.handles.push(handle);
        true
    }
```
I have also encountered this issue. Here is my minimal reproducible example:

```rust
struct MyActor;

impl Actor for MyActor {
    type Context = Context<Self>;

    fn started(&mut self, ctx: &mut Self::Context) {
        ctx.add_stream(Box::pin(futures::stream::repeat(())).then(|_| async {
            tokio::time::sleep(std::time::Duration::from_millis(200)).await;
        }));
    }
}

impl StreamHandler<()> for MyActor {
    fn handle(&mut self, _: (), ctx: &mut Self::Context) {
        let handle = ctx.run_later(std::time::Duration::from_millis(10), |_, _| {
            panic!("this should not happen");
        });
        ctx.cancel_future(handle);
    }
}

#[actix_rt::main]
async fn main() {
    MyActor.start();
    sleep(std::time::Duration::from_secs(3)).await;
}
```

Ive also tried to debug this issue a little bit, and im pretty sure that it comes down to [impl Future for ContextFut](https://github.com/actix/actix/blob/d40b6103602acecb3930f3192d17d3da3e532781/src/contextimpl.rs#L345).

It seems that `poll` does not handles situation when item creates new item and cancels it (by pushing asociated `SpawnHandle` to handles). So when poll enters `clean_canceled_handle`, it [pops](https://github.com/actix/actix/blob/d40b6103602acecb3930f3192d17d3da3e532781/src/contextimpl.rs#L324) handle from `ContextParts`, but associated item is not yet merged into `ContextFut.items`, therefor handle is dropped but item is not. 

I can think of few ways to fix this:
1. Adding additional `merge()` call somewhere.
2. Return handle back to handles list if associated item is not found in `ContextFut.items` (item not merged yet).
3. Remove item associated with popped `Spawnhandle` from `ContextParts.items` if it is not found in `ContextFut.items`. 

I believe that 3rd option is the way to go and i can prepare PR, but it would be nice to have some of the maintainers take a look at this.
Not sure what's going on there(need to check code) but you could try to check return value of `cancel_future` out of curiosity.


Hi, yes. Checked that already - it's `true`. No surprise because of the implementation of `ContextParts`:

```rust
    #[inline]
    /// Cancel previously scheduled future.
    pub fn cancel_future(&mut self, handle: SpawnHandle) -> bool {
        self.handles.push(handle);
        true
    }
```
I have also encountered this issue. Here is my minimal reproducible example:

```rust
struct MyActor;

impl Actor for MyActor {
    type Context = Context<Self>;

    fn started(&mut self, ctx: &mut Self::Context) {
        ctx.add_stream(Box::pin(futures::stream::repeat(())).then(|_| async {
            tokio::time::sleep(std::time::Duration::from_millis(200)).await;
        }));
    }
}

impl StreamHandler<()> for MyActor {
    fn handle(&mut self, _: (), ctx: &mut Self::Context) {
        let handle = ctx.run_later(std::time::Duration::from_millis(10), |_, _| {
            panic!("this should not happen");
        });
        ctx.cancel_future(handle);
    }
}

#[actix_rt::main]
async fn main() {
    MyActor.start();
    sleep(std::time::Duration::from_secs(3)).await;
}
```

Ive also tried to debug this issue a little bit, and im pretty sure that it comes down to [impl Future for ContextFut](https://github.com/actix/actix/blob/d40b6103602acecb3930f3192d17d3da3e532781/src/contextimpl.rs#L345).

It seems that `poll` does not handles situation when item creates new item and cancels it (by pushing asociated `SpawnHandle` to handles). So when poll enters `clean_canceled_handle`, it [pops](https://github.com/actix/actix/blob/d40b6103602acecb3930f3192d17d3da3e532781/src/contextimpl.rs#L324) handle from `ContextParts`, but associated item is not yet merged into `ContextFut.items`, therefor handle is dropped but item is not. 

I can think of few ways to fix this:
1. Adding additional `merge()` call somewhere.
2. Return handle back to handles list if associated item is not found in `ContextFut.items` (item not merged yet).
3. Remove item associated with popped `Spawnhandle` from `ContextParts.items` if it is not found in `ContextFut.items`. 

I believe that 3rd option is the way to go and i can prepare PR, but it would be nice to have some of the maintainers take a look at this.

Fix patch:
diff --git a/actix/CHANGES.md b/actix/CHANGES.md
index d5aa3207c..cedfc4284 100644
--- a/actix/CHANGES.md
+++ b/actix/CHANGES.md
@@ -1,6 +1,10 @@
 # CHANGES
 
 ## Unreleased - 2021-xx-xx
+### Fixed
+* Instant cancellation of a spawned `Future`. [#484]
+
+[#484]: https://github.com/actix/actix/pull/484
 
 
 ## 0.11.0 - 2021-03-21
diff --git a/actix/src/contextimpl.rs b/actix/src/contextimpl.rs
index 4b8e8cebb..a755268b3 100644
--- a/actix/src/contextimpl.rs
+++ b/actix/src/contextimpl.rs
@@ -315,16 +315,31 @@ where
     }
 
     fn clean_canceled_handle(&mut self) {
-        while self.ctx.parts().handles.len() > 2 {
-            let handle = self.ctx.parts().handles.pop().unwrap();
+        fn remove_item_by_handle<C>(
+            items: &mut SmallVec<[Item<C>; 3]>,
+            handle: &SpawnHandle,
+        ) -> bool {
             let mut idx = 0;
-            while idx < self.items.len() {
-                if self.items[idx].0 == handle {
-                    self.items.swap_remove(idx);
+            let mut removed = false;
+            while idx < items.len() {
+                if &items[idx].0 == handle {
+                    items.swap_remove(idx);
+                    removed = true;
                 } else {
                     idx += 1;
                 }
             }
+            removed
+        }
+
+        while self.ctx.parts().handles.len() > 2 {
+            let handle = self.ctx.parts().handles.pop().unwrap();
+            // remove item from ContextFut.items in case associated item is already merged
+            if !remove_item_by_handle(&mut self.items, &handle) {
+                // item is not merged into ContextFut.items yet,
+                // so it should be in ContextParts.items
+                remove_item_by_handle(&mut self.ctx.parts().items, &handle);
+            }
         }
     }
 }


Test patch:
diff --git a/actix/tests/test_context.rs b/actix/tests/test_context.rs
index 2b0060761..fc21e9eab 100644
--- a/actix/tests/test_context.rs
+++ b/actix/tests/test_context.rs
@@ -500,3 +500,60 @@ fn test_cancel_completed_with_no_context_item() {
         });
     });
 }
+
+mod scheduled_task_is_cancelled_properly {
+    use tokio::sync::oneshot;
+    use tokio::task::yield_now;
+    use tokio::time::timeout;
+
+    use super::*;
+
+    struct TestActor;
+
+    impl Actor for TestActor {
+        type Context = Context<Self>;
+
+        fn started(&mut self, ctx: &mut Self::Context) {
+            ctx.spawn(yield_now().into_actor(self));
+        }
+    }
+
+    #[derive(Message)]
+    #[rtype(result = "()")]
+    struct Msg(Option<oneshot::Sender<()>>);
+
+    impl Drop for Msg {
+        fn drop(&mut self) {
+            self.0.take().unwrap().send(()).unwrap();
+        }
+    }
+
+    impl Handler<Msg> for TestActor {
+        type Result = ();
+
+        fn handle(&mut self, msg: Msg, ctx: &mut Self::Context) -> Self::Result {
+            let handle = ctx.run_later(Duration::from_millis(0), |_, _| {
+                let _msg = msg;
+                System::current().stop_with_code(1);
+                panic!("This closure must not be executed");
+            });
+            ctx.cancel_future(handle);
+        }
+    }
+
+    #[test]
+    fn cancels() {
+        let sys = System::new();
+        sys.block_on(async {
+            let addr = TestActor.start();
+            let (tx, rx) = oneshot::channel();
+            addr.do_send(Msg(Some(tx)));
+            timeout(Duration::from_millis(500), rx)
+                .await
+                .unwrap()
+                .unwrap();
+            System::current().stop();
+        });
+        sys.run().unwrap();
+    }
+}