Repository: actix/actix (language: rust)

Issue:
`ActorStream` blocks other tasks in the same thread until `ctx.waiting()` returns `true` or the `Stream` is pending.
## Version

* actix v0.9.0

## Platform

* any

## Description

`ActorStream` blocks other tasks in the same thread until `ctx.waiting()` returns `true` or the `Stream` is pending.
https://github.com/actix/actix/blob/master/src/stream.rs#L137

If the `Stream` can keep providing items for a long time, other tasks have no chance to run unless `ctx.wait()` is called in `StreamHandler::handle()` .
https://github.com/masnagam/rust-case-studies/blob/master/actix-actor-stream-hang/src/main.rs#L13

The asynchronous task system used in Rust is some kind of [cooperative multitasking system](https://en.wikipedia.org/wiki/Cooperative_multitasking).  So, I think that it might be better to provide a method to voluntarily yield control.

At this point, `ctx.wait(actix::fut::wrap_future(actix::clock::delay_for(Duration::from_secs(0))))` is one of ways to pause the context and yield control.

A similar situation occurs in other types. For example:
https://github.com/actix/actix/blob/master/src/mailbox.rs#L89


Discussion / hints:
The way we `poll` the `Stream` seams right to me, as we will ensure we consume all currently available data when polling the stream. Same goes for the `Mailbox` as we can ensure all messages get spawned.

If I get it right, your point is that we should offer a better way to yield control when handling a infinite stream? I see how this can be a valid use case and we should offer better support for it. Using `ctx.wait(tokio::task::yield_now().into_actor(self));` also does the trick, having a `yeld_now` method for `AsyncContext` semas good to me, I'll look into it, but I wanna be sure I got right the problem you're reporting.
Thank you for your quick reply.

> we should offer a better way to yield control when handling a infinite stream?

No.  The current implementation is acceptable to me when handling an infinite stream.  For simplicity, I used `futures::stream::repeat()` in the example.

I'm thinking of a case where there are many items in the buffer of the stream object.  In this case, the `ActorStream::poll()` blocks for a long time.  This might cause an issue if there are other higher priority async tasks than it in the same thread.

Libraries (actix, tokio) have no way to detect priority of each async tasks at this point (and this may be same in the future).  Only the application knows priority of each tasks created by the application.  So, the application should yield control if needed.

> Using ctx.wait(tokio::task::yield_now().into_actor(self)); also does the trick

I confirmed that it works fine, too.  I updated the example in github.

Fix patch:
diff --git a/src/stream.rs b/src/stream.rs
index c4a23de6a..786dcb157 100644
--- a/src/stream.rs
+++ b/src/stream.rs
@@ -134,20 +134,20 @@ where
             <A as StreamHandler<M>>::started(act, ctx);
         }
 
-        loop {
-            match this.stream.as_mut().poll_next(task) {
-                Poll::Ready(Some(msg)) => {
-                    A::handle(act, msg, ctx);
-                    if ctx.waiting() {
-                        return Poll::Pending;
-                    }
+        match this.stream.as_mut().poll_next(task) {
+            Poll::Ready(Some(msg)) => {
+                A::handle(act, msg, ctx);
+                if !ctx.waiting() {
+                    // Let the future's context know that this future might be polled right the way
+                    task.waker().wake_by_ref();
                 }
-                Poll::Ready(None) => {
-                    A::finished(act, ctx);
-                    return Poll::Ready(());
-                }
-                Poll::Pending => return Poll::Pending,
+                return Poll::Pending;
+            }
+            Poll::Ready(None) => {
+                A::finished(act, ctx);
+                return Poll::Ready(());
             }
+            Poll::Pending => return Poll::Pending,
         }
     }
 }


Test patch:
diff --git a/tests/test_actor.rs b/tests/test_actor.rs
index 2695da64c..4d403b145 100644
--- a/tests/test_actor.rs
+++ b/tests/test_actor.rs
@@ -5,7 +5,7 @@ use std::sync::Arc;
 use actix::prelude::*;
 use tokio::time::{delay_for, Duration, Instant};
 
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 struct Num(usize);
 
 impl Message for Num {
@@ -53,6 +53,68 @@ async fn test_stream() {
     assert!(err.load(Ordering::Relaxed));
 }
 
+#[derive(Message)]
+#[rtype(result = "()")]
+struct Stop;
+
+struct StopOnRequest(Arc<AtomicUsize>, Arc<AtomicBool>, Arc<AtomicBool>);
+
+impl Actor for StopOnRequest {
+    type Context = actix::Context<Self>;
+}
+
+impl StreamHandler<Num> for StopOnRequest {
+    fn handle(&mut self, msg: Num, _: &mut Self::Context) {
+        self.0.fetch_add(msg.0, Ordering::Relaxed);
+    }
+
+    fn finished(&mut self, _: &mut Self::Context) {
+        self.2.store(true, Ordering::Relaxed);
+    }
+}
+
+impl actix::Handler<Stop> for StopOnRequest {
+    type Result = ();
+
+    fn handle(&mut self, _: Stop, ctx: &mut Self::Context) {
+        self.1.store(true, Ordering::Relaxed);
+        ctx.stop();
+    }
+}
+
+#[actix_rt::test]
+async fn test_infinite_stream() {
+    let count = Arc::new(AtomicUsize::new(0));
+    let stopped = Arc::new(AtomicBool::new(false));
+    let finished = Arc::new(AtomicBool::new(false));
+
+    let act_count = Arc::clone(&count);
+    let act_stopped = Arc::clone(&stopped);
+    let act_finished = Arc::clone(&finished);
+
+    let addr = StopOnRequest::create(move |ctx| {
+        StopOnRequest::add_stream(futures_util::stream::repeat(Num(1)), ctx);
+        StopOnRequest(act_count, act_stopped, act_finished)
+    });
+
+    delay_for(Duration::new(0, 1_000_000)).await;
+
+    addr.send(Stop).await.unwrap();
+
+    assert!(
+        count.load(Ordering::Relaxed) > 0,
+        "Some items should be processed as actor has ran for some time"
+    );
+    assert!(
+        stopped.load(Ordering::Relaxed),
+        "Actor should have processed the Stop message"
+    );
+    assert!(
+        !finished.load(Ordering::Relaxed),
+        "As the stream is infinite, finished should never be called"
+    );
+}
+
 struct MySyncActor {
     started: Arc<AtomicUsize>,
     stopping: Arc<AtomicUsize>,