Confused by tokio::time::timeout (not timeout-ing)

[hidden ravine]

I wrote the following test

    #[tokio::test(flavor = "multi_thread")]
    async fn test_processor_stops() {
        let running = Arc::new(AtomicBool::new(true));
        let item = serde_json::json!("{}");
        let stream: MessageStream<serde_json::Value> = stream::repeat(item).boxed();

        // let r = running.clone();
        // tokio::spawn(async move {
        //     tokio::time::sleep(Duration::from_secs(5)).await;
        //     r.store(false, Ordering::Relaxed);
        // });
        let result = tokio::time::timeout(
            Duration::from_secs(1),
            ProcessingPipeline::run(&running, stream, IdentityProcessor {}, DevNullSink {}),
        )
        .await;

        assert!(result.is_ok());
    }

With the commented code present the run method terminates gracefully after five seconds/after I set the AtomicBool and the test succeeds (it was initially confusing to me that this didn't work, but since I learned that I need the flavor in the test attribute for this).

In the state as posted above, I would expect result to be an Err variant after a timeout, failing the test. Instead, this test will never terminate.

The run method is basically

        let mut stream = source.get_stream().await.unwrap();
        while running.load(Ordering::Relaxed) {
            let msg = match tokio::time::timeout(Duration::from_secs(2), stream.next()).await {
                Ok(Some(msg)) => msg,
                Ok(None) => break,
                Err(_) => continue, // Timeout
            };

            // two more await points here 
        }

Is the behavior I'm seeing expected? I thought that any of the three await points (listed above, the timeout for stream.next. Omitted two following ones in the loop) would allow for this task to be dropped/cancelled with the timeout in the first snippet?

[white hollow]

Tokio runtime doesn't poll the expired timers on each .await point.

It only polls the timer under specific conditions.

• No pending tasks in the queue
• Every time event_interval tasks are polled
• more ...

        let msg = match tokio::time::timeout(Duration::from_secs(2), stream.next()).await {

The stream.next() looks problematic.

The stream.next() might always return Poll::Ready. If so, the async runtime cannot get the control back, and also no chance to poll the timers.

        let mut stream = source.get_stream().await.unwrap();
        while running.load(Ordering::Relaxed) {
            tokio::task::coop::consume_budget().await; // <--- 
            let msg = match tokio::time::timeout(Duration::from_secs(2), stream.next()).await {
                Ok(Some(msg)) => msg,
                Ok(None) => break,
                Err(_) => continue, // Timeout
            };

            // two more await points here 
        }

You could try the cooperative scheduling to see if this still happens.