Tokio timeout doesn't wake up when streaming while another tokio thread is running

Open
  • Is something wrong?
#3,489 2 comments 0 reactions 0 assignees View on GitHub

Maintainers usually reply within 1 day

Nobody has claimed this yet.

Assessment

Difficulty
4/5
Estimated time
3-5 days
Newbie friendliness
35/100
Issue type
Bug
Clarity
Mostly clear
Activity status
Stale
Tech stack
rust
Domain
api, backend, web-dev

Research direction

Start by running the two supplied examples and comparing the Tokio entry point with the Actix Web serve handler and its HttpResponse::streaming call. Trace the do_work task, Command::output, and the timeout around rx.recv; done means the streaming response emits heartbeat fragments on schedule while the long-running command is still active.

Written by the indexing model from the issue text.

Description

Expected Behavior

The tokio timeout and sleep should work and trigger after the supplied time.

Current Behavior

They are "blocked" by another tokio thread executing and do not trigger before that thread is done.

Steps to Reproduce (for bugs)

I created two examples; one where the problem occurs when using actix_web, and one where it doesn't occur when using tokio on very similar code. See Context for explanation.

Tokio example: working code

use futures_util::{Stream, StreamExt};
use std::convert::Infallible;
use std::{process::Command, time::Duration};
use tokio::pin;
use tokio::sync::mpsc::Receiver;

use futures_util::stream::unfold;
use tokio::sync::mpsc::Sender;
use tokio::time::timeout;

#[tokio::main]
async fn main() {
    let stream = tokio::spawn(serve()).await;
    let stream = stream.unwrap();
    pin!(stream);
    loop {
        let item = stream.next().await;
        dbg!(item);
    }
}
// Whether this is async or not doesn't make a difference; just to make them more similar
async fn serve() -> impl Stream<Item = Result<String, Infallible>> {
    let c = |state: Option<Receiver<String>>| async move {
        let mut rx = if state.is_none() {
            let (tx, rx) = tokio::sync::mpsc::channel(100); // Construct a new reciever if it's not
                                                            // initialized yet
            println!("created channel");
            tokio::spawn(do_work(tx));
            println!("spawned work");
            rx
        } else {
            state.unwrap()
        };
        // vvv Problem line: timeout doesn't happen; the command is instead being waited on
        let t = timeout(Duration::from_secs(1), rx.recv()).await;
        // Tokio select! and Tokio sleep also don't work here. 
        dbg!(&t);
        if t == Ok(None) {
            // Stop the stream if the command ended
            return None;
        }
        Some((
            // Send a message on timeout
            Ok::<String, Infallible>("Stream fragment\n".to_owned()),
            Some(rx),
        ))
    };
    let stream = unfold(None, c); // Construct stream from closure
    stream
}

async fn do_work(tx: Sender<String>) {
    println!("Running Command");
    let _ = Command::new("sh").arg("-c").arg("sleep 10").output();
    println!("Command finished!");
    if let Err(e) = tx.send("".to_owned()).await {
        eprintln!("Sending errored! {:?}", e);
    }
}
// Expected Output: ten or nine lines of "Stream Fragment", and one is sent every second; followed by a panic! because the stream was polled after it ended
// Actual output: as expected

Actix_web example: Has the problem

use actix_web::body::MessageBody;
use actix_web::web::Bytes;
use actix_web::{services, web, HttpResponse, Responder};
use actix_web::{App, HttpServer};
use std::convert::Infallible;
use std::{process::Command, time::Duration};
use tokio::sync::mpsc::Receiver;

use futures_util::stream::unfold;
use tokio::sync::mpsc::Sender;
use tokio::time::timeout;

#[tokio::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(|| {
        let services = services![web::scope("/a").route("/b", web::get().to(serve))]; // Available
                                                                                      // at localhost:8080/a/b
        App::new().service(services)
    })
    .bind(("localhost", 8080))
    .unwrap()
    .run()
    .await
}
async fn serve() -> impl Responder {
    let c = |state: Option<Receiver<String>>| async move {
        let mut rx = if state.is_none() {
            let (tx, rx) = tokio::sync::mpsc::channel(100); // Construct a new reciever if it's not
                                                            // initialized yet
            println!("created channel");
            //tokio::spawn(do_work(tx));
            actix_web::rt::spawn(do_work(tx)); // Both have the same problem
            println!("spawned work");
            rx
        } else {
            state.unwrap()
        };
        // vvv Problem line: timeout doesn't happen; the command is instead being waited on
        let t = timeout(Duration::from_secs(1), rx.recv()).await;
        // Tokio select! and Tokio sleep also don't work here. 
        dbg!(&t);
        if t == Ok(None) {
            // Stop the stream if the command ended
            return None;
        }
        Some((
            // Send a message on timeout
            Ok::<Bytes, Infallible>("Stream fragment\n".to_owned().try_into_bytes().unwrap()),
            Some(rx),
        ))
    };
    let stream = unfold(None, c); // Construct stream from closure
    HttpResponse::Ok().streaming(stream) // Stream out the answer
}
async fn do_work(tx: Sender<String>) {
    println!("Running Command");
    let _ = Command::new("sh").arg("-c").arg("sleep 10").output();
    println!("Command finished!");
    if let Err(e) = tx.send("".to_owned()).await {
        eprintln!("Sending errored! {:?}", e);
    }
}
// Expected Output: ten or nine lines of "Stream Fragment", and one is sent every second
// Actual output: a single line of "Stream Fragment" is output at the very end.

Edit: Using async_process::Command and then awaiting instead of using std::process::Command does solve this particular case.

Context

I am writing a REST API in actix_web that is heavily relying on streams to the client.

While I was adding a heartbeat so that the connection to the client wouldn't time out and the client could be sure that the server was still alive, I came across the issue that some tokio functionality like sleep, select! (with sleep) and timeout don't properly work.

I managed to reduce it to the two examples above.

In both cases, a channel is spawned inside an async closure, which sender is given to a new tokio task that takes long to finish.
If the Receiver is then awaited using a tokio timeout (or a tokio sleep is called), the timeout (or sleep) doesn't trigger after the given time, but only after the Receiver is available.

In the context I used it, the async closure was then unfolded into a stream, which is then served to a client.
This causes the stream to become "stuck" and not send a heartbeat, until the Receiver can recieve.

In the tokio example, the stream is instead given to the main function, pinned, and iterated through until completion. Here, the stream does not become "stuck" and instead sends a heartbeat once every second, as expected.

(I put this issue on actix-web because it seems to happen because of the way streaming is handeled and because the tokio example works)

Your Environment

The given examples work on fresh install.

  • Rust Version (I.e, output of rustc -V): rustc 1.81.0 (eeb90cda1 2024-09-04)
  • Actix Web Version: 4.9.0
Dominant language
Rust
Stars
24.9k
Forks
2k
Avg merge
19h 9m
Merged PRs (30d)
61

Getting set up

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

More from actix/actix-web

All issues in actix/actix-web

Similar issues

More Rust issues

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.