Auto merge of #197 - antrik:fix-notification-tests, r=jdm

Fix notification handling tests and `unix` implementation

The first commit attempts to make notification tests more robust. (Hopefully fixing intermittent failures on Linux.) The other commits add some additional related tests, and fix a (minor) issue uncovered by them.

Author: bors-servo

src/platform/test.rs

@@ -706,11 +706,64 @@ fn no_receiver_notification() {
     let (sender, receiver) = platform::channel().unwrap();
     drop(receiver);
     let data: &[u8] = b"1234567";
+    loop {
+        if let Err(err) = sender.send(data, vec![], vec![]) {
+            // We don't have an actual method for distinguishing a "broken pipe" error --
+            // but at least it's not supposed to signal the same condition as closing the sender.
+            assert!(!err.channel_is_closed());
+            break;
+        }
+    }
+}
+
+/// Checks for broken pipe notification when receiver is closed
+/// while there are pending unreceived messages.
+///
+/// This can result in a different error condition
+/// than dropping the receiver before a send is attempted.
+/// (Linux reports `ECONNRESET` instead of `EPIPE` in this case.)
+#[test]
+fn no_receiver_notification_pending() {
+    let (sender, receiver) = platform::channel().unwrap();
+    let data: &[u8] = b"1234567";
+
     let result = sender.send(data, vec![], vec![]);
-    assert!(result.is_err());
-    // We don't have an actual method for distinguishing a "broken pipe" error --
-    // but at least it's not supposed to signal the same condition as closing the sender.
-    assert!(!result.unwrap_err().channel_is_closed());
+    assert!(result.is_ok());
+
+    drop(receiver);
+    loop {
+        if let Err(err) = sender.send(data, vec![], vec![]) {
+            // We don't have an actual method for distinguishing a "broken pipe" error --
+            // but at least it's not supposed to signal the same condition as closing the sender.
+            assert!(!err.channel_is_closed());
+            break;
+        }
+    }
+}
+
+/// Checks for broken pipe notification when receiver is closed after a delay.
+///
+/// This might uncover some timing-related issues.
+#[test]
+fn no_receiver_notification_delayed() {
+    let (sender, receiver) = platform::channel().unwrap();
+
+    let thread = thread::spawn(move || {
+        thread::sleep(Duration::from_millis(42));
+        drop(receiver);
+    });
+
+    let data: &[u8] = b"1234567";
+    loop {
+        if let Err(err) = sender.send(data, vec![], vec![]) {
+            // We don't have an actual method for distinguishing a "broken pipe" error --
+            // but at least it's not supposed to signal the same condition as closing the sender.
+            assert!(!err.channel_is_closed());
+            break;
+        }
+    }
+
+    thread.join().unwrap();
 }
 
 #[test]
@@ -756,9 +809,39 @@ fn no_senders_notification_try_recv() {
     assert!(result.is_err());
     assert!(!result.unwrap_err().channel_is_closed());
     drop(sender);
+    loop {
+        let result = receiver.try_recv();
+        assert!(result.is_err());
+        if result.unwrap_err().channel_is_closed() {
+            break;
+        }
+    }
+}
+
+/// Checks for channel closed notification when receiver is closed after a delay.
+///
+/// This might uncover some timing-related issues.
+#[test]
+fn no_senders_notification_try_recv_delayed() {
+    let (sender, receiver) = platform::channel().unwrap();
     let result = receiver.try_recv();
     assert!(result.is_err());
-    assert!(result.unwrap_err().channel_is_closed());
+    assert!(!result.unwrap_err().channel_is_closed());
+
+    let thread = thread::spawn(move || {
+        thread::sleep(Duration::from_millis(42));
+        drop(sender);
+    });
+
+    loop {
+        let result = receiver.try_recv();
+        assert!(result.is_err());
+        if result.unwrap_err().channel_is_closed() {
+            break;
+        }
+    }
+
+    thread.join().unwrap();
 }
 
 #[test]

src/platform/unix/mod.rs

@@ -343,7 +343,7 @@ impl OsIpcSender {
                     // using a reduced send buffer size.
                     //
                     // Any other errors we might get here are non-recoverable.
-                    if !(error.0 == libc::ENOBUFS
+                    if !(error == UnixError::Errno(libc::ENOBUFS)
                          && downsize(&mut sendbuf_size, data.len()).is_ok()) {
                         return Err(error)
                     }
@@ -382,7 +382,7 @@ impl OsIpcSender {
             };
 
             if let Err(error) = result {
-                if error.0 == libc::ENOBUFS
+                if error == UnixError::Errno(libc::ENOBUFS)
                    && downsize(&mut sendbuf_size, end_byte_position - byte_position).is_ok() {
                     // If the kernel failed to allocate a buffer large enough for the packet,
                     // retry with a smaller size (if possible).
@@ -807,17 +807,20 @@ impl OsIpcSharedMemory {
     }
 }
 
-#[derive(Copy, Clone, Debug)]
-pub struct UnixError(c_int);
+#[derive(Copy, Clone, Debug, PartialEq)]
+pub enum UnixError {
+    Errno(c_int),
+    ChannelClosed,
+}
 
 impl UnixError {
     fn last() -> UnixError {
-        UnixError(Error::last_os_error().raw_os_error().unwrap())
+        UnixError::Errno(Error::last_os_error().raw_os_error().unwrap())
     }
 
     #[allow(dead_code)]
     pub fn channel_is_closed(&self) -> bool {
-        self.0 == libc::ECONNRESET
+        *self == UnixError::ChannelClosed
     }
 }
 
@@ -829,13 +832,22 @@ impl From<UnixError> for bincode::Error {
 
 impl From<UnixError> for Error {
     fn from(unix_error: UnixError) -> Error {
-        Error::from_raw_os_error(unix_error.0)
+        match unix_error {
+            UnixError::Errno(errno) => Error::from_raw_os_error(errno),
+            UnixError::ChannelClosed => Error::new(ErrorKind::ConnectionReset,
+                                                   "All senders for this socket closed"),
+        }
     }
 }
 
 impl From<Error> for UnixError {
     fn from(e: Error) -> UnixError {
-        UnixError(e.raw_os_error().unwrap())
+        if let Some(errno) = e.raw_os_error() {
+            UnixError::Errno(errno)
+        } else {
+            assert!(e.kind() == ErrorKind::ConnectionReset);
+            UnixError::ChannelClosed
+        }
     }
 }
 
@@ -932,7 +944,7 @@ fn recv(fd: c_int, blocking_mode: BlockingMode)
         };
 
         if result == 0 {
-            return Err(UnixError(libc::ECONNRESET))
+            return Err(UnixError::ChannelClosed)
         } else if result < 0 {
             return Err(UnixError::last())
         };
@@ -1011,7 +1023,7 @@ impl UnixCmsg {
         let result = if result > 0 {
             Ok(result as usize)
         } else if result == 0 {
-            Err(UnixError(libc::ECONNRESET))
+            Err(UnixError::ChannelClosed)
         } else {
             Err(UnixError::last())
         };