async/client: don't spawn per incoming frame

ClientReader::handle_msg spawns a new task per frame. For a
server-streaming RPC, the final DATA frame and the subsequent
FLAG_REMOTE_CLOSED frame then race: if the close-frame task grabs
the req_map lock first, it removes the stream from the map, and
the preceding data-frame task finds nothing and silently drops
the payload. The stream consumer sees Ok(None) (EOF) without
ever observing the payload the server sent.

The connection read loop already awaits handle_msg per frame, so
processing inline preserves per-stream wire order. It also gives
the per-stream mpsc natural back-pressure in place of the
unbounded per-frame spawning.

handle_err gets the same treatment for consistency.

Also update example/async-stream-client.rs so the existing
echo_default_value example exercises the race: wrap it in a
1000-iteration loop and flip the client runtime to multi_thread.
On a single-threaded runtime the tokio scheduler masks the race
because spawned tasks run in submission order. Without the fix,
the modified example fails on any multi-core runner; with the
fix, it passes.

Signed-off-by: Shiv Bhosale <shvbsle@amazon.com>

Author: shvbsle

src/asynchronous/client.rs

@@ -313,26 +313,27 @@ impl ReaderDelegate for ClientReader {
     async fn exit(&self) {}
 
     async fn handle_err(&self, header: MessageHeader, e: Error) {
+        // Process inline to preserve per-stream frame ordering. See `handle_msg`.
         let req_map = self.streams.clone();
-        tokio::spawn(async move {
-            if let Some(resp_tx) = get_resp_tx(req_map, &header).await {
-                resp_tx
-                    .send(Err(e))
-                    .await
-                    .unwrap_or_else(|_e| error!("The request has returned"));
-            }
-        });
+        if let Some(resp_tx) = get_resp_tx(req_map, &header).await {
+            resp_tx
+                .send(Err(e))
+                .await
+                .unwrap_or_else(|_e| error!("The request has returned"));
+        }
     }
 
     async fn handle_msg(&self, msg: GenMessage) {
+        // Do not `tokio::spawn` per frame: a `FLAG_REMOTE_CLOSED` frame could
+        // then `remove` a stream from `req_map` before the preceding DATA
+        // frame's task looked it up, silently dropping the final payload.
+        // The read loop already awaits this per frame, so inline is correct.
         let req_map = self.streams.clone();
-        tokio::spawn(async move {
-            if let Some(resp_tx) = get_resp_tx(req_map, &msg.header).await {
-                resp_tx
-                    .send(Ok(msg))
-                    .await
-                    .unwrap_or_else(|_e| error!("The request has returned"));
-            }
-        });
+        if let Some(resp_tx) = get_resp_tx(req_map, &msg.header).await {
+            resp_tx
+                .send(Ok(msg))
+                .await
+                .unwrap_or_else(|_e| error!("The request has returned"));
+        }
     }
 }