POLLRDHUP delivery fix

Fixes https://github.com/google/gvisor/issues/6015 | Added EventRdHUp to waiter events | Added POLLRDHUP delivery on half-closed connection detection | Added POLLRDHUP delivery on local read end shutdown

pkg/sentry/syscalls/linux/sys_read_write.go

@@ -29,7 +29,7 @@ import (
 )
 
 const (
-	eventMaskRead  = waiter.EventRdNorm | waiter.EventIn | waiter.EventHUp | waiter.EventErr
+	eventMaskRead  = waiter.EventRdNorm | waiter.EventIn | waiter.EventHUp | waiter.EventErr | waiter.EventRdHUp
 	eventMaskWrite = waiter.EventWrNorm | waiter.EventOut | waiter.EventHUp | waiter.EventErr
 )
 

pkg/tcpip/transport/tcp/endpoint.go

@@ -77,6 +77,17 @@ const (
 	SegOverheadFactor = 2
 )
 
+type connDirectionState uint32
+
+// Connection direction states used for directionState checks in endpoint struct
+// to detect half-closed connection and deliver POLLRDHUP
+const (
+	connDirectionStateOpen      connDirectionState = 0
+	connDirectionStateRcvClosed connDirectionState = 1
+	connDirectionStateSndClosed connDirectionState = 2
+	connDirectionStateAll       connDirectionState = connDirectionStateOpen | connDirectionStateRcvClosed | connDirectionStateSndClosed
+)
+
 // connected returns true when s is one of the states representing an
 // endpoint connected to a peer.
 func (s EndpointState) connected() bool {
@@ -399,6 +410,10 @@ type endpoint struct {
 	// methods.
 	state atomicbitops.Uint32 `state:".(EndpointState)"`
 
+	// connectionDirectionState holds current state of send and receive,
+	// accessed atomically
+	connectionDirectionState atomicbitops.Uint32
+
 	// origEndpointState is only used during a restore phase to save the
 	// endpoint state at restore time as the socket is moved to it's correct
 	// state.
@@ -940,6 +955,9 @@ func (e *endpoint) Readiness(mask waiter.EventMask) waiter.EventMask {
 			if e.sndQueueInfo.SndClosed || e.sndQueueInfo.SndBufUsed < sndBufSize {
 				result |= waiter.WritableEvents
 			}
+			if e.sndQueueInfo.SndClosed {
+				e.updateConnDirectionState(connDirectionStateSndClosed)
+			}
 			e.sndQueueInfo.sndQueueMu.Unlock()
 		}
 
@@ -949,10 +967,18 @@ func (e *endpoint) Readiness(mask waiter.EventMask) waiter.EventMask {
 			if e.RcvBufUsed > 0 || e.RcvClosed {
 				result |= waiter.ReadableEvents
 			}
+			if e.RcvClosed {
+				e.updateConnDirectionState(connDirectionStateRcvClosed)
+			}
 			e.rcvQueueMu.Unlock()
 		}
 	}
 
+	// Determine whether endpoint is half-closed with rcv shutdown
+	if e.connDirectionState() == connDirectionStateRcvClosed {
+		result |= waiter.EventRdHUp
+	}
+
 	return result
 }
 
@@ -2509,7 +2535,13 @@ func (e *endpoint) shutdownLocked(flags tcpip.ShutdownFlags) tcpip.Error {
 			}
 			// Wake up any readers that maybe waiting for the stream to become
 			// readable.
-			e.waiterQueue.Notify(waiter.ReadableEvents)
+			events := waiter.ReadableEvents
+			if e.shutdownFlags&tcpip.ShutdownWrite == 0 {
+				// If ShutdownWrite is not set, write end won't close and
+				// we end up with a half-closed connection
+				events |= waiter.EventRdHUp
+			}
+			e.waiterQueue.Notify(events)
 		}
 
 		// Close for write.
@@ -2597,6 +2629,7 @@ func (e *endpoint) listen(backlog int) tcpip.Error {
 		}
 
 		e.shutdownFlags = 0
+		e.updateConnDirectionState(connDirectionStateOpen)
 		e.rcvQueueMu.Lock()
 		e.RcvClosed = false
 		e.rcvQueueMu.Unlock()
@@ -3019,6 +3052,16 @@ func (e *endpoint) maxReceiveBufferSize() int {
 	return rs.Max
 }
 
+// directionState returns the close state of send and receive part of the endpoint
+func (e *endpoint) connDirectionState() connDirectionState {
+	return connDirectionState(e.connectionDirectionState.Load())
+}
+
+// updateDirectionState updates the close state of send and receive part of the endpoint
+func (e *endpoint) updateConnDirectionState(state connDirectionState) connDirectionState {
+	return connDirectionState(e.connectionDirectionState.Swap(uint32(e.connDirectionState() | state)))
+}
+
 // rcvWndScaleForHandshake computes the receive window scale to offer to the
 // peer when window scaling is enabled (true by default). If auto-tuning is
 // disabled then the window scaling factor is based on the size of the

pkg/tcpip/transport/tcp/rcv.go

@@ -293,6 +293,7 @@ func (r *receiver) consumeSegment(s *segment, segSeq seqnum.Value, segLen seqnum
 			r.pendingRcvdSegments[i] = nil
 		}
 		r.pendingRcvdSegments = r.pendingRcvdSegments[:first]
+		r.ep.updateConnDirectionState(connDirectionStateRcvClosed)
 
 		return true
 	}

pkg/tcpip/transport/tcp/snd.go

@@ -799,6 +799,7 @@ func (s *sender) maybeSendSegment(seg *segment, limit int, end seqnum.Value) (se
 		segEnd = seg.sequenceNumber.Add(1)
 		// Update the state to reflect that we have now
 		// queued a FIN.
+		s.ep.updateConnDirectionState(connDirectionStateSndClosed)
 		switch s.ep.EndpointState() {
 		case StateCloseWait:
 			s.ep.setEndpointState(StateLastAck)

test/syscalls/linux/socket_inet_loopback.cc

@@ -438,6 +438,149 @@ TEST_P(SocketInetLoopbackTest, TCPListenClose) {
   }
 }
 
+TEST_P(SocketInetLoopbackTest, TCPUnblockWaitOnLocalRdHUp) {
+  SocketInetTestParam const& param = GetParam();
+  TestAddress const& listener = param.listener;
+  TestAddress const& connector = param.connector;
+  constexpr int kTimeout = 100000;
+
+  // Setup listening socket
+  FileDescriptor const listen_fd = ASSERT_NO_ERRNO_AND_VALUE(
+      Socket(listener.family(), SOCK_STREAM, IPPROTO_TCP));
+  sockaddr_storage listen_addr = listener.addr;
+  FileDescriptor accepted;
+
+    // Bind and listen on socket
+  ASSERT_THAT(
+      bind(listen_fd.get(), AsSockAddr(&listen_addr), listener.addr_len),
+      SyscallSucceeds());
+  ASSERT_THAT(listen(listen_fd.get(), SOMAXCONN), SyscallSucceeds());
+
+  ScopedThread t1([&] {
+    // Accept connections
+    accepted = ASSERT_NO_ERRNO_AND_VALUE(Accept(listen_fd.get(), nullptr, nullptr));
+    int data = 1234;
+    ASSERT_THAT(RetryEINTR(recv)(accepted.get(), &data, sizeof(data), 0),
+              SyscallSucceedsWithValue(0));
+  });
+
+  ScopedThread t2([&] {
+    // Get the port bound by the listening socket.
+    socklen_t addrlen = listener.addr_len;
+    ASSERT_THAT(getsockname(listen_fd.get(), AsSockAddr(&listen_addr), &addrlen),
+                SyscallSucceeds());
+    uint16_t const port =
+          ASSERT_NO_ERRNO_AND_VALUE(AddrPort(listener.family(), listen_addr));
+    FileDescriptor conn_fd = ASSERT_NO_ERRNO_AND_VALUE(
+        Socket(connector.family(), SOCK_STREAM, IPPROTO_TCP));
+    sockaddr_storage conn_addr = connector.addr;
+    ASSERT_NO_ERRNO(SetAddrPort(connector.family(), &conn_addr, port));
+
+    for (int i = 0; i < 10; i++) {
+      // Connect to listening socket
+      int ret;
+      ASSERT_THAT(ret = RetryEINTR(connect)(conn_fd.get(), AsSockAddr(&conn_addr),
+                          connector.addr_len), SyscallSucceeds());
+      if (ret == 0) {
+        // Connect succeeded
+        break;
+      }
+
+      // Connect failed
+      EXPECT_THAT(ret, SyscallFailsWithErrno(EINPROGRESS));
+      // Sleep to wait for Accept on another thread
+      // since we got errno=Connection refused
+      absl::SleepFor(absl::Milliseconds(50));
+    }
+
+    // Shutdown read
+    shutdown(accepted.get(), SHUT_RD);
+  });
+  t1.Join();
+  t2.Join();
+  // Poll accepted fd for POLLRDHUP
+  struct pollfd pfd = {
+      .fd = accepted.get(),
+      .events = POLLIN | POLLRDHUP,
+  };
+  ASSERT_THAT(RetryEINTR(poll)(&pfd, 1, kTimeout), SyscallSucceedsWithValue(1));
+  ASSERT_EQ(pfd.revents, POLLIN | POLLRDHUP);
+}
+
+TEST_P(SocketInetLoopbackTest, TCPUnblockWaitOnRemoteRdHUp) {
+  SocketInetTestParam const& param = GetParam();
+  TestAddress const& listener = param.listener;
+  TestAddress const& connector = param.connector;
+  constexpr int kTimeout = 10000;
+
+  // Setup listening socket
+  FileDescriptor const listen_fd = ASSERT_NO_ERRNO_AND_VALUE(
+      Socket(listener.family(), SOCK_STREAM, IPPROTO_TCP));
+  sockaddr_storage listen_addr = listener.addr;
+  FileDescriptor accepted;
+
+  // Bind and listen on socket
+  ASSERT_THAT(
+      bind(listen_fd.get(), AsSockAddr(&listen_addr), listener.addr_len),
+      SyscallSucceeds());
+  ASSERT_THAT(listen(listen_fd.get(), SOMAXCONN), SyscallSucceeds());
+
+  ScopedThread t1([&] {
+    // Accept connections
+    auto accepted =
+        ASSERT_NO_ERRNO_AND_VALUE(Accept(listen_fd.get(), nullptr, nullptr));
+    int data = 1234;
+    ASSERT_THAT(RetryEINTR(recv)(accepted.get(), &data, sizeof(data), 0),
+              SyscallSucceedsWithValue(0));
+    // Poll accepted fd for POLLRDHUP
+    // Thread is unblocked at this point
+    struct pollfd pfd = {
+        .fd = accepted.get(),
+        .events = POLLIN | POLLRDHUP,
+    };
+    ASSERT_THAT(RetryEINTR(poll)(&pfd, 1, kTimeout), SyscallSucceedsWithValue(1));
+    ASSERT_EQ(pfd.revents, POLLIN | POLLRDHUP);
+  });
+
+  ScopedThread t2([&] {
+    // Sleep to wait for Accept on another thread
+    // otherwise the test may fail on connect with errno=Connection refused
+    absl::SleepFor(absl::Milliseconds(500));
+    // Get the port bound by the listening socket.
+    socklen_t addrlen = listener.addr_len;
+    ASSERT_THAT(getsockname(listen_fd.get(), AsSockAddr(&listen_addr), &addrlen),
+                SyscallSucceeds());
+    uint16_t const port =
+          ASSERT_NO_ERRNO_AND_VALUE(AddrPort(listener.family(), listen_addr));
+    FileDescriptor conn_fd = ASSERT_NO_ERRNO_AND_VALUE(
+        Socket(connector.family(), SOCK_STREAM, IPPROTO_TCP));
+    sockaddr_storage conn_addr = connector.addr;
+    ASSERT_NO_ERRNO(SetAddrPort(connector.family(), &conn_addr, port));
+
+    for (int i = 0; i < 10; i++) {
+      // Connect to listening socket
+      int ret;
+      ASSERT_THAT(ret = RetryEINTR(connect)(conn_fd.get(), AsSockAddr(&conn_addr),
+                          connector.addr_len), SyscallSucceeds());
+      if (ret == 0) {
+        // Connect succeeded
+        break;
+      }
+
+      // Connect failed
+      EXPECT_THAT(ret, SyscallFailsWithErrno(EINPROGRESS));
+      // Sleep to wait for Accept on another thread
+      // since we got errno=Connection refused
+      absl::SleepFor(absl::Milliseconds(50));
+    }
+
+    // Shutdown write
+    shutdown(conn_fd.get(), SHUT_WR);
+  });
+  t1.Join();
+  t2.Join();
+}
+
 // Test the protocol state information returned by TCPINFO.
 TEST_P(SocketInetLoopbackTest, TCPInfoState) {
   SocketInetTestParam const& param = GetParam();
@@ -498,13 +641,7 @@ TEST_P(SocketInetLoopbackTest, TCPInfoState) {
   int n = poll(&pfd, 1, kTimeout);
   ASSERT_GE(n, 0) << strerror(errno);
   ASSERT_EQ(n, 1);
-  if (IsRunningOnGvisor() && !IsRunningWithHostinet() &&
-      GvisorPlatform() != Platform::kFuchsia) {
-    // TODO(gvisor.dev/issue/6015): Notify POLLRDHUP on incoming FIN.
-    ASSERT_EQ(pfd.revents, POLLIN);
-  } else {
-    ASSERT_EQ(pfd.revents, POLLIN | POLLRDHUP);
-  }
+  ASSERT_EQ(pfd.revents, POLLIN | POLLRDHUP);
 
   ASSERT_THAT(state(conn_fd.get()), TCP_CLOSE_WAIT);
   ASSERT_THAT(close(conn_fd.release()), SyscallSucceeds());
@@ -752,14 +889,7 @@ TEST_P(SocketInetLoopbackTest, TCPNonBlockingConnectClose) {
     int n = poll(&pfd, 1, kTimeout);
     ASSERT_GE(n, 0) << strerror(errno);
     ASSERT_EQ(n, 1);
-
-    if (IsRunningOnGvisor() && !IsRunningWithHostinet() &&
-        GvisorPlatform() != Platform::kFuchsia) {
-      // TODO(gvisor.dev/issue/6015): Notify POLLRDHUP on incoming FIN.
-      ASSERT_EQ(pfd.revents, POLLIN);
-    } else {
-      ASSERT_EQ(pfd.revents, POLLIN | POLLRDHUP);
-    }
+    ASSERT_EQ(pfd.revents, POLLIN | POLLRDHUP);
     ASSERT_THAT(close(accepted.release()), SyscallSucceeds());
   }
 }