std: Relax an assertion in oneshot selection

The assertion was erroneously ensuring that there was no data on the port when
the port had selection aborted on it. This assertion was written in error
because it's possible for data to be waiting on a port, even after it was
disconnected. When aborting selection, if we see that there's data on the port,
then we return true that data is available on the port.

Closes #12802

Author: alexcrichton

src/libstd/comm/oneshot.rs

@@ -339,14 +339,19 @@ impl<T: Send> Packet<T> {
             DATA => Ok(true),
 
             // If the other end has hung up, then we have complete ownership
-            // of the port. We need to check to see if there was an upgrade
-            // requested, and if so, the other end needs to have its selection
-            // aborted.
+            // of the port. First, check if there was data waiting for us. This
+            // is possible if the other end sent something and then hung up.
+            //
+            // We then need to check to see if there was an upgrade requested,
+            // and if so, the upgraded port needs to have its selection aborted.
             DISCONNECTED => {
-                assert!(self.data.is_none());
-                match mem::replace(&mut self.upgrade, SendUsed) {
-                    GoUp(port) => Err(port),
-                    _ => Ok(true),
+                if self.data.is_some() {
+                    Ok(true)
+                } else {
+                    match mem::replace(&mut self.upgrade, SendUsed) {
+                        GoUp(port) => Err(port),
+                        _ => Ok(true),
+                    }
                 }
             }
 

src/libstd/comm/select.rs

@@ -597,4 +597,56 @@ mod test {
         unsafe { h.add(); }
         assert_eq!(s.wait2(false), h.id);
     })
+
+    test!(fn oneshot_data_waiting() {
+        let (p, c) = Chan::new();
+        let (p2, c2) = Chan::new();
+        spawn(proc() {
+            select! {
+                () = p.recv() => {}
+            }
+            c2.send(());
+        });
+
+        for _ in range(0, 100) { task::deschedule() }
+        c.send(());
+        p2.recv();
+    })
+
+    test!(fn stream_data_waiting() {
+        let (p, c) = Chan::new();
+        let (p2, c2) = Chan::new();
+        c.send(());
+        c.send(());
+        p.recv();
+        p.recv();
+        spawn(proc() {
+            select! {
+                () = p.recv() => {}
+            }
+            c2.send(());
+        });
+
+        for _ in range(0, 100) { task::deschedule() }
+        c.send(());
+        p2.recv();
+    })
+
+    test!(fn shared_data_waiting() {
+        let (p, c) = Chan::new();
+        let (p2, c2) = Chan::new();
+        drop(c.clone());
+        c.send(());
+        p.recv();
+        spawn(proc() {
+            select! {
+                () = p.recv() => {}
+            }
+            c2.send(());
+        });
+
+        for _ in range(0, 100) { task::deschedule() }
+        c.send(());
+        p2.recv();
+    })
 }