Throttle results in 100% cpu

[tekjar]

use async_std::stream::StreamExt;
use async_std::sync::channel;
use async_std::task;

use std::thread;
use std::time::Duration;

#[async_std::main]
async fn main() {
    let (requests_tx, requests_rx) = channel::<i32>(10);
    let mut requests_rx = requests_rx.throttle(Duration::from_secs(1));

    thread::spawn(move || {
        task::block_on( async {
            for i in 0..5 {
                requests_tx.send(i).await;
            }
        });

        thread::sleep(Duration::from_secs(100));
    });

    while let Some(item) = requests_rx.next().await {
        println!("{:?}", item);
    }
}

[yoshuawuyts]

I was able to reproduce this locally in debug mode, uploaded a repro here: https://github.com/yoshuawuyts/repro-async-std-583.

edit: seems the cause is block_on + the use of thread::*: #583 (comment)

[yoshuawuyts]

Confirmed that this happens in release mode too.

I now suspect this is caused because thread::sleep is called, and our scheduler doesn't know how to handle that so it bounces the task around indefinitely. The spikes seem to be happening only once thread::sleep is hit.

[yoshuawuyts]

When applying this patch:

diff --git a/src/main.rs b/src/main.rs
index fa8dfed..b2d8e8a 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -8,7 +8,7 @@ use std::time::Duration;
 #[async_std::main]
 async fn main() {
     let (requests_tx, requests_rx) = channel::<i32>(10);
-    let mut requests_rx = requests_rx.throttle(Duration::from_secs(1));
+    let mut requests_rx = requests_rx.throttle(Duration::from_secs(10));
 
     thread::spawn(move || {
         task::block_on( async {

The output doesn't seem to spike, further corroborating the theory that this is caused by thread::sleep confusing the scheduler.

[yoshuawuyts]

Okay, no it's not the threads happening. I think you're right and it's indeed something about debounce perhaps? Will run a flame graph next.

Tested it with this patch, and got the same results:

diff --git a/src/main.rs b/src/main.rs
index fa8dfed..7c71e3a 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -2,7 +2,6 @@ use async_std::stream::StreamExt;
 use async_std::sync::channel;
 use async_std::task;
 
-use std::thread;
 use std::time::Duration;
 
 #[async_std::main]
@@ -10,14 +9,14 @@ async fn main() {
     let (requests_tx, requests_rx) = channel::<i32>(10);
     let mut requests_rx = requests_rx.throttle(Duration::from_secs(1));
 
-    thread::spawn(move || {
+    task::spawn(async move {
         task::block_on( async {
             for i in 0..5 {
                 requests_tx.send(i).await;
             }
         });
 
-        thread::sleep(Duration::from_secs(100));
+        task::sleep(Duration::from_secs(100)).await;
     });
 
     while let Some(item) = requests_rx.next().await {

[yoshuawuyts]

Yes okay, if we remove the throttle call the problem indeed goes away. So it seems to be that when throttle is in a suspended mode something is causing it to spin. Maybe we're hitting a waker loop of sorts?

diff --git a/src/main.rs b/src/main.rs
index fa8dfed..4839cc2 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -8,7 +8,7 @@ use std::time::Duration;
 #[async_std::main]
 async fn main() {
     let (requests_tx, requests_rx) = channel::<i32>(10);
-    let mut requests_rx = requests_rx.throttle(Duration::from_secs(1));
+    let mut requests_rx = requests_rx;
 
     thread::spawn(move || {
         task::block_on( async {

[yoshuawuyts]

Okay, ran the flame graph locally, but nothing stands out really. It's a flat line it seems 😅

[yoshuawuyts]

Okay, it seems that I forgot to remove the task::block_on call from task::spawn when I was testing that out. The CPU spikes go away if we remove it.

diff --git a/src/main.rs b/src/main.rs
index fa8dfed..ff7dae9 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -2,22 +2,19 @@ use async_std::stream::StreamExt;
 use async_std::sync::channel;
 use async_std::task;
 
-use std::thread;
 use std::time::Duration;
 
 #[async_std::main]
 async fn main() {
     let (requests_tx, requests_rx) = channel::<i32>(10);
-    let mut requests_rx = requests_rx.throttle(Duration::from_secs(1));
+    let mut requests_rx = requests_rx.throttle(Duration::from_secs(10));
 
-    thread::spawn(move || {
-        task::block_on( async {
-            for i in 0..5 {
-                requests_tx.send(i).await;
-            }
-        });
+    task::spawn(async move {
+        for i in 0..5 {
+            requests_tx.send(i).await;
+        }
 
-        thread::sleep(Duration::from_secs(100));
+        task::sleep(Duration::from_secs(100)).await;
     });
 
     while let Some(item) = requests_rx.next().await {

Seems like this might still be related to #358?

Either way, this doesn't seem like much of a priority to fix I think; using block_on and thread::* APIs inside of a task should be discouraged, and is definitely not using the APIs the way we intended it to. Even if it takes up more CPU than it should, at least it doesn't deadlock which seems to have been the purpose. I don't think this should be a priority to address. @stjepang what do you think?

@tekjar if you apply the patch above, the performance issues you experienced should go away (:

[tekjar]

thread::spawn(move || task::block_on(async {
        let requests_tx = requests_tx;
        for i in 0..10 {
            requests_tx.send(i).await;
        }

        task::sleep(Duration::from_secs(100)).await
    }));

Does this look like a legitimate use case? Looks so to me and cpu spikes in this case as well. task::block_on inside thread::spawn is a very similar case to using task::block_on in main.

using block_on and thread::* APIs inside of a task should be discouraged, and is definitely not using the APIs the way we intended it to

Sure. But in my head, worst case of using synchronous apis inside a cooperative eventloop should cause a block (and hence no progress), not cpu spikes.

[tekjar]

@tekjar if you apply the patch above, the performance issues you experienced should go away (:

It doesn't to me. Maybe your throttle duration is very high (10 secs) and you are not waiting enough time?

I think you should wait for 50 seconds in your case

[tekjar]

thread::spawn(move || {
    task::block_on( async {
            for i in 0..5 {
                requests_tx.send(i).await;
            }
    });

    thread::sleep(Duration::from_secs(100));
});

Is this pattern very weird in async world. To me, all the asynchronous code is encapsulated in task::block_on. I would've used task::sleep inside it. But using thread::sleep outside doesn't look like an anti pattern

[Matthias247]

Here is your busy-loop:

async-std/src/stream/stream/throttle.rs

Line 59 in 50cefce

cx.waker().wake_by_ref(); // Continue driving even though emitting Pending

Removing the line should fix things. Actually this should lead to a busy loop in every use of throttle where the polled Stream is not immediately ready. I guess @tekjar is just the first one who use it (or noticed the CPU going wild).

The usage pattern of @tekjar looks fine to me. It's doing thread blocking operations in a thread, and not in a task.