TcpListener: drain all pending accepts

This commit makes `TcpListener::accept` try to accept the next
connection without waiting for a notify, if the previous connection
couldn't be established because the peer already went away.

The change is meant to fix a bug where a pending connection sometimes
wouldn't be accepted. Consider this series of events:

 * Server runs `TcpListener::accept`. There are no connections pending,
   so it starts waiting on the `notify`.
 * Client 1 sends a SYN, which gets pushed to `ServerSocket::deque` and
   `ServerSocket::notify` is notified.
 * Client 2 sends a SYN, which gets pushed to `ServerSocket::deque` and
   `ServerSocket::notify` is notified.
 * Client 1 goes away.
 * Server wakes up because the `notify` was notified.
 * Server tries to accept the connection from client 1, which fails
   because client 1 went away.
 * Server starts waiting on the `notify` again.

We end up in a situation where the connection from client 2 is still
pending but the server doesn't accept it because it's waiting for a
notification that was already sent.

src/net/tcp/listener.rs

@@ -64,10 +64,15 @@ impl TcpListener {
     /// established, the corresponding [`TcpStream`] and the remote peer’s
     /// address will be returned.
     pub async fn accept(&self) -> Result<(TcpStream, SocketAddr)> {
+        enum Retry {
+            Now,
+            Wait,
+        }
+
         loop {
-            let maybe_accept = World::current(|world| {
+            let accept_result = World::current(|world| {
                 let host = world.current_host_mut();
-                let (syn, origin) = host.tcp.accept(self.local_addr)?;
+                let (syn, origin) = host.tcp.accept(self.local_addr).ok_or(Retry::Wait)?;
 
                 tracing::trace!(target: TRACING_TARGET, src = ?origin, dst = ?self.local_addr, protocol = %"TCP SYN", "Recv");
 
@@ -77,7 +82,7 @@ impl TcpListener {
                 tracing::trace!(target: TRACING_TARGET, src = ?self.local_addr, dst = ?origin, protocol = %"TCP SYN-ACK", "Send");
 
                 if ack.is_err() {
-                    return None;
+                    return Err(Retry::Now);
                 }
 
                 let mut my_addr = self.local_addr;
@@ -91,14 +96,19 @@ impl TcpListener {
                 let pair = SocketPair::new(my_addr, origin);
                 let rx = host.tcp.new_stream(pair);
 
-                Some((TcpStream::new(pair, rx), origin))
+                tracing::trace!(target: TRACING_TARGET, src = ?self.local_addr, dst = ?origin, "Accepted");
+                Ok((TcpStream::new(pair, rx), origin))
             });
 
-            if let Some(accepted) = maybe_accept {
-                return Ok(accepted);
+            match accept_result {
+                Ok(accepted) => return Ok(accepted),
+                Err(Retry::Now) => (),
+                Err(Retry::Wait) => {
+                    tracing::trace!(target: TRACING_TARGET, src = ?self.local_addr, "Waiting for notify");
+                    self.notify.notified().await;
+                    tracing::trace!(target: TRACING_TARGET, src = ?self.local_addr, "Received notify");
+                }
             }
-
-            self.notify.notified().await;
         }
     }