Edge case: Only provide epoll notification for read if it emptied the buffer

Author: tiif

src/shims/unix/socket.rs

@@ -111,16 +111,6 @@ impl FileDescription for SocketPair {
             return Ok(Ok(0));
         }
 
-        // In libc socketpair, reading from a previously full socketpair buffer won't trigger
-        // epoll notification. We want to inform the user about this inconsistency in libc and
-        // hope to receive a bug report if someone has a use case for this.
-        if readbuf.buf.len() == MAX_SOCKETPAIR_BUFFER_CAPACITY {
-            throw_unsup_format!(
-                "A previously full socketpair has been read, please submit a \
-            bug report in https://github.com/rust-lang/miri/issues/new"
-            );
-        }
-
         if readbuf.buf.is_empty() {
             if !readbuf.buf_has_writer {
                 // Socketpair with no writer and empty buffer.
@@ -146,9 +136,21 @@ impl FileDescription for SocketPair {
         // FIXME: this over-synchronizes; a more precise approach would be to
         // only sync with the writes whose data we will read.
         ecx.acquire_clock(&readbuf.clock);
+
         // Do full read / partial read based on the space available.
         // Conveniently, `read` exists on `VecDeque` and has exactly the desired behavior.
         let actual_read_size = readbuf.buf.read(bytes).unwrap();
+
+        // Notification should be provided for peer fd when read emptied the buffer.
+        if let Some(peer_fd) = self.peer_fd.upgrade() {
+            if readbuf.buf.is_empty() {
+                // The readbuf needs to be explicitly dropped because it will cause panic when
+                // check_and_update_readiness borrows it again.
+                drop(readbuf);
+                peer_fd.check_and_update_readiness(ecx)?;
+            }
+        }
+
         return Ok(Ok(actual_read_size));
     }
 

tests/fail-dep/tokio/read_from_full_socketpair.rs

@@ -16,6 +16,7 @@ fn main() {
     test_epoll_ctl_del();
     test_pointer();
     test_two_same_fd_in_same_epoll_instance();
+    test_socketpair_read();
 }
 
 fn check_epoll_wait<const N: usize>(
@@ -339,16 +340,18 @@ fn test_epoll_socketpair_special_case() {
         vec![(expected_event1, expected_value1), (expected_event0, expected_value0)]
     ));
 
-    // Read from fds[0]
+    // Read from fds[0].
     let mut buf: [u8; 5] = [0; 5];
     res = unsafe {
         libc::read(fds[0], buf.as_mut_ptr().cast(), buf.len() as libc::size_t).try_into().unwrap()
     };
     assert_eq!(res, 5);
     assert_eq!(buf, "abcde".as_bytes());
 
-    // No notification should be returned.
-    assert!(check_epoll_wait::<8>(epfd, vec![]));
+    // Notification should be provided for fds[1].
+    let expected_event = u32::try_from(libc::EPOLLOUT).unwrap();
+    let expected_value = fds[1] as u64;
+    assert!(check_epoll_wait::<8>(epfd, vec![(expected_event, expected_value)]));
 }
 
 // When file description is fully closed, epoll_wait should not provide any notification for
@@ -481,3 +484,64 @@ fn test_event_overwrite() {
     let expected_value = u64::try_from(fd).unwrap();
     assert!(check_epoll_wait::<8>(epfd, vec![(expected_event, expected_value)]));
 }
+
+// read in socketpair will only provide epoll notification if it emptied the buffer.
+fn test_socketpair_read() {
+    // Create an epoll instance.
+    let epfd = unsafe { libc::epoll_create1(0) };
+    assert_ne!(epfd, -1);
+
+    // Create a socketpair instance.
+    let mut fds = [-1, -1];
+    let res = unsafe { libc::socketpair(libc::AF_UNIX, libc::SOCK_STREAM, 0, fds.as_mut_ptr()) };
+    assert_eq!(res, 0);
+
+    // Register both fd to the same epoll instance.
+    let epollet = libc::EPOLLET as u32;
+    let flags = u32::try_from(libc::EPOLLIN | libc::EPOLLOUT).unwrap() | epollet;
+    let mut ev = libc::epoll_event { events: u32::try_from(flags).unwrap(), u64: fds[0] as u64 };
+    let mut res = unsafe { libc::epoll_ctl(epfd, libc::EPOLL_CTL_ADD, fds[0], &mut ev) };
+    assert_ne!(res, -1);
+    let mut ev = libc::epoll_event { events: u32::try_from(flags).unwrap(), u64: fds[1] as u64 };
+    res = unsafe { libc::epoll_ctl(epfd, libc::EPOLL_CTL_ADD, fds[1], &mut ev) };
+    assert_ne!(res, -1);
+
+    // Write 5 bytes to fds[1].
+    let data = "abcde".as_bytes().as_ptr();
+    res = unsafe { libc::write(fds[1], data as *const libc::c_void, 5).try_into().unwrap() };
+    assert_eq!(res, 5);
+
+    //Two notification should be received.
+    let expected_event0 = u32::try_from(libc::EPOLLIN | libc::EPOLLOUT).unwrap();
+    let expected_value0 = fds[0] as u64;
+    let expected_event1 = u32::try_from(libc::EPOLLOUT).unwrap();
+    let expected_value1 = fds[1] as u64;
+    assert!(check_epoll_wait::<8>(
+        epfd,
+        vec![(expected_event1, expected_value1), (expected_event0, expected_value0)]
+    ));
+
+    // Read 3 bytes from fds[0].
+    let mut buf: [u8; 3] = [0; 3];
+    res = unsafe {
+        libc::read(fds[0], buf.as_mut_ptr().cast(), buf.len() as libc::size_t).try_into().unwrap()
+    };
+    assert_eq!(res, 3);
+    assert_eq!(buf, "abc".as_bytes());
+
+    // No notification should be provided.
+    assert!(check_epoll_wait::<8>(epfd, vec![]));
+
+    // Read until the buffer is empty.
+    let mut buf: [u8; 2] = [0; 2];
+    res = unsafe {
+        libc::read(fds[0], buf.as_mut_ptr().cast(), buf.len() as libc::size_t).try_into().unwrap()
+    };
+    assert_eq!(res, 2);
+    assert_eq!(buf, "de".as_bytes());
+
+    // Notification should be provided.
+    let expected_event = u32::try_from(libc::EPOLLOUT).unwrap();
+    let expected_value = fds[1] as u64;
+    assert!(check_epoll_wait::<8>(epfd, vec![(expected_event, expected_value)]));
+}