diff --git a/pkg/sentry/syscalls/linux/sys_read_write.go b/pkg/sentry/syscalls/linux/sys_read_write.go index d014ae87b..fd0f25ad9 100644 --- a/pkg/sentry/syscalls/linux/sys_read_write.go +++ b/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 ) diff --git a/pkg/tcpip/transport/tcp/endpoint.go b/pkg/tcpip/transport/tcp/endpoint.go index 6dcf557e2..7ae7043ec 100644 --- a/pkg/tcpip/transport/tcp/endpoint.go +++ b/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 diff --git a/pkg/tcpip/transport/tcp/rcv.go b/pkg/tcpip/transport/tcp/rcv.go index 98e4d12fb..9981e4d88 100644 --- a/pkg/tcpip/transport/tcp/rcv.go +++ b/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 } diff --git a/pkg/tcpip/transport/tcp/snd.go b/pkg/tcpip/transport/tcp/snd.go index 9c7306911..a90f46e3f 100644 --- a/pkg/tcpip/transport/tcp/snd.go +++ b/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) diff --git a/test/syscalls/linux/socket_inet_loopback.cc b/test/syscalls/linux/socket_inet_loopback.cc index 10ef1e702..14373a0b8 100644 --- a/test/syscalls/linux/socket_inet_loopback.cc +++ b/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()); } }