Repository: tokio-rs/tokio (language: rust)

Issue:
Using timeout hangs a program on wasm32-wasip1
**Version**
```
tokio-hang v0.1.0 (/data/surban/dev/rust-wasi-web/tokio-hang)
└── tokio v1.42.0 (/data/surban/dev/tokio/tokio)
    └── tokio-macros v2.4.0 (proc-macro) (/data/surban/dev/tokio/tokio-macros)
```

**Platform**
`wasm32-wasip1` and `wasm32-wasip1-threads`
using wasmtime executor or wasmer

**Description**
Using the `timeout` function makes the following program hang on `wasm32-wasip1`.
The program works fine either on
   - target `x86_64-unknown-linux-gnu`
   - also on `wasm32-wasip1` without the `timeout` wrapper

I tried this code:

Available at https://github.com/surban/tokio-hang

```rust
use std::time::Duration;

use futures::{SinkExt, StreamExt};
use tokio::time::timeout;

async fn send_task(mut tx: futures::channel::mpsc::Sender<u8>) {
    println!("feeding 0");
    tx.feed(0).await.unwrap();
    println!("flushing 0");
    tx.flush().await.unwrap();

    println!("feeding 1");
    tx.feed(1).await.unwrap();
    println!("flushing 1");
    tx.flush().await.unwrap();
}

async fn recv_task(mut rx: futures::channel::mpsc::Receiver<u8>) {
    loop {
        println!("waiting to receive");
        match rx.next().await {
            Some(msg) => println!("received: {msg}"),
            None => break,
        }
    }

    println!("end of receive");
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let qlen = 0;
    let (a_tx, b_rx) = futures::channel::mpsc::channel::<u8>(qlen);

    let task = async move { tokio::join!(send_task(a_tx), recv_task(b_rx)) };
    timeout(Duration::from_secs(60), task).await.unwrap();

    // works without timeout:
    // task.await;
}
```

I expected to see the following output (it works on `x86_64-unknown-linux-gnu`):
```
feeding 0
flushing 0
waiting to receive
received: 0
waiting to receive
feeding 1
flushing 1
received: 1
waiting to receive
end of receive
```

Instead, this happened on `wasm32-wasip1`: 
```
feeding 0
flushing 0
waiting to receive
received: 0
waiting to receive
```
Then it hangs.



Discussion / hints:

Fix patch:
diff --git a/tokio/src/runtime/park.rs b/tokio/src/runtime/park.rs
index c260b7512be..c5c8b1307d0 100644
--- a/tokio/src/runtime/park.rs
+++ b/tokio/src/runtime/park.rs
@@ -65,12 +65,7 @@ impl ParkThread {
     pub(crate) fn park_timeout(&mut self, duration: Duration) {
         #[cfg(loom)]
         CURRENT_THREAD_PARK_COUNT.with(|count| count.fetch_add(1, SeqCst));
-
-        // Wasm doesn't have threads, so just sleep.
-        #[cfg(not(target_family = "wasm"))]
         self.inner.park_timeout(duration);
-        #[cfg(target_family = "wasm")]
-        std::thread::sleep(duration);
     }
 
     pub(crate) fn shutdown(&mut self) {
@@ -158,12 +153,20 @@ impl Inner {
             Err(actual) => panic!("inconsistent park_timeout state; actual = {actual}"),
         }
 
+        #[cfg(not(all(target_family = "wasm", not(target_feature = "atomics"))))]
         // Wait with a timeout, and if we spuriously wake up or otherwise wake up
         // from a notification, we just want to unconditionally set the state back to
         // empty, either consuming a notification or un-flagging ourselves as
         // parked.
         let (_m, _result) = self.condvar.wait_timeout(m, dur).unwrap();
 
+        #[cfg(all(target_family = "wasm", not(target_feature = "atomics")))]
+        // Wasm without atomics doesn't have threads, so just sleep.
+        {
+            let _m = m;
+            std::thread::sleep(dur);
+        }
+
         match self.state.swap(EMPTY, SeqCst) {
             NOTIFIED => {} // got a notification, hurray!
             PARKED => {}   // no notification, alas


Test patch: