rework error handling

Author: matklad

src/main.rs

@@ -26,16 +26,11 @@ async fn server(addr: impl ToSocketAddrs) -> Result<()> {
 
     let (broker_sender, broker_receiver) = mpsc::unbounded();
     let broker = task::spawn(broker(broker_receiver));
-    let mut readers = Vec::new();
     let mut incoming = listener.incoming();
     while let Some(stream) = incoming.next().await {
         let stream = stream?;
         println!("Accepting from: {}", stream.peer_addr()?);
-        let handle = task::spawn(client(broker_sender.clone(), stream));
-        readers.push(handle);
-    }
-    for reader in readers {
-        reader.await?;
+        spawn_and_log_error(client(broker_sender.clone(), stream));
     }
     broker.await?;
     Ok(())
@@ -96,7 +91,6 @@ enum Event {
 
 async fn broker(mut events: Receiver<Event>) -> Result<()> {
     let mut peers: HashMap<String, Sender<String>> = HashMap::new();
-    let mut writers = Vec::new();
 
     while let Some(event) = events.next().await {
         match event {
@@ -110,14 +104,20 @@ async fn broker(mut events: Receiver<Event>) -> Result<()> {
             Event::NewPeer { name, stream} => {
                 let (client_sender, client_receiver) = mpsc::unbounded();
                 peers.insert(name.clone(), client_sender);
-                let handle = task::spawn(client_writer(client_receiver, stream));
-                writers.push(handle);
+                spawn_and_log_error(client_writer(client_receiver, stream));
             }
         }
     }
-    drop(peers);
-    for writer in writers {
-        writer.await?;
-    }
     Ok(())
 }
+
+fn spawn_and_log_error<F>(fut: F) -> task::JoinHandle<()>
+where
+    F: Future<Output = Result<()>> + Send + 'static,
+{
+    task::spawn(async move {
+        if let Err(e) = fut.await {
+            eprintln!("{}", e)
+        }
+    })
+}

tutorial.adoc

@@ -215,6 +215,42 @@ async fn client(stream: TcpStream) -> Result<()> {
 
 <5> Finally, we parse each line into a list of destination logins and the message itself.
 
+== Managing Errors
+
+One serious problem in the above solution is that, while we correctly propagate errors in the `client`, we just drop the error on the floor afterwards!
+That is, `task::spawn` does not return error immediately (it can't, it needs to run the future to completion first), only after it is joined.
+We can "fix" it by waiting for the task to be joined, like this:
+
+[source,rust]
+----
+let handle = task::spawn(client(stream)); <1>
+handle.await?
+----
+
+The `.await` waits until the client finishes, and `?` propagates the result.
+
+There are two problems with this solution however!
+_First_, because we immediately await the client, we can only handle one client at time, and that completely defeats the purpose of async!
+_Second_, if a client encounters an IO error, the whole server immediately exits.
+That is, a flaky internet connection of one peer brings down the whole chat room!
+
+A correct way to handle client errors in this case is log them, and continue serving other clients.
+So let's use a helper function for this:
+
+[source,rust]
+----
+fn spawn_and_log_error<F>(fut: F) -> task::JoinHandle<()>
+where
+    F: Future<Output = Result<()>> + Send + 'static,
+{
+    task::spawn(async move {
+        if let Err(e) = fut.await {
+            eprintln!("{}", e)
+        }
+    })
+}
+----
+
 == Sending Messages
 
 Now it's time to implement the other half -- sending messages.
@@ -295,7 +331,7 @@ async fn broker(mut events: Receiver<Event>) -> Result<()> {
             Event::NewPeer { name, stream} => {
                 let (client_sender, client_receiver) = mpsc::unbounded();
                 peers.insert(name.clone(), client_sender); <4>
-                let _handle = task::spawn(client_writer(client_receiver, stream)); <5>
+                spawn_and_log_error(client_writer(client_receiver, stream)); <5>
             }
         }
     }
@@ -348,7 +384,7 @@ async fn server(addr: impl ToSocketAddrs) -> Result<()> {
     while let Some(stream) = incoming.next().await {
         let stream = stream?;
         println!("Accepting from: {}", stream.peer_addr()?);
-        let _handle = task::spawn(client(broker_sender.clone(), stream));
+        spawn_and_log_error(client(broker_sender.clone(), stream));
     }
     Ok(())
 }
@@ -422,7 +458,7 @@ async fn broker(mut events: Receiver<Event>) -> Result<()> {
             Event::NewPeer { name, stream} => {
                 let (client_sender, client_receiver) = mpsc::unbounded();
                 peers.insert(name.clone(), client_sender);
-                let _handle = task::spawn(client_writer(client_receiver, stream));
+                spawn_and_log_error(client_writer(client_receiver, stream));
             }
         }
     }
@@ -466,18 +502,13 @@ async fn server(addr: impl ToSocketAddrs) -> Result<()> {
 
     let (broker_sender, broker_receiver) = mpsc::unbounded();
     let broker = task::spawn(broker(broker_receiver));
-    let mut readers = Vec::new();
     let mut incoming = listener.incoming();
     while let Some(stream) = incoming.next().await {
         let stream = stream?;
         println!("Accepting from: {}", stream.peer_addr()?);
-        let handle = task::spawn(client(broker_sender.clone(), stream));
-        readers.push(handle);
-    }
-    drop(broker_sender);
-    for reader in readers {
-        reader.await?; <1>
+        spawn_and_log_error(client(broker_sender.clone(), stream));
     }
+    drop(broker_sender); <1>
     broker.await?; <5>
     Ok(())
 }
@@ -503,13 +534,13 @@ async fn broker(mut events: Receiver<Event>) -> Result<()> {
             Event::NewPeer { name, stream} => {
                 let (client_sender, client_receiver) = mpsc::unbounded();
                 peers.insert(name.clone(), client_sender);
-                let handle = task::spawn(client_writer(client_receiver, stream));
+                let handle = spawn_and_log_error(client_writer(client_receiver, stream));
                 writers.push(handle);
             }
         }
     }
     drop(peers); <3>
-    for writer in writers {
+    for writer in writers { <4>
         writer.await?;
     }
     Ok(())
@@ -518,10 +549,10 @@ async fn broker(mut events: Receiver<Event>) -> Result<()> {
 
 Notice what happens with all of the channels once we exit the accept loop:
 
-<1> First, we drop the main broker's sender and join all of the readers.
-    When the readers are done, there's no sender for the broker's channel.
+<1> First, we drop the main broker's sender.
+    That way when the readers are done, there's no sender for the broker's channel, and the chanel closes.
 <2> Next, the broker exits `while let Some(event) = events.next().await` loop.
 <3> It's crucial that, at this stage, we drop the `peers` map.
     This drops writer's senders.
 <4> Now we can join all of the writers.
-<5> Finally, we join the broker, which also gurantess that all the writes have terminated.
+<5> Finally, we join the broker, which also guarantees that all the writes have terminated.