diff --git a/pkg/tcpip/transport/tcp/endpoint.go b/pkg/tcpip/transport/tcp/endpoint.go index faab4d70d..f43b2cdf5 100644 --- a/pkg/tcpip/transport/tcp/endpoint.go +++ b/pkg/tcpip/transport/tcp/endpoint.go @@ -1184,6 +1184,7 @@ func (e *endpoint) cleanupLocked() { e.snd.resendTimer.cleanup() e.snd.probeTimer.cleanup() e.snd.reorderTimer.cleanup() + e.snd.corkTimer.cleanup() } if e.finWait2Timer != nil { @@ -1762,6 +1763,7 @@ func (e *endpoint) OnCorkOptionSet(v bool) { if !v { e.LockUser() defer e.UnlockUser() + e.snd.corkTimer.disable() // Handle the corked data. if e.EndpointState().connected() { e.sendData(nil /* next */) diff --git a/pkg/tcpip/transport/tcp/endpoint_state.go b/pkg/tcpip/transport/tcp/endpoint_state.go index f23a22b03..bb0bc8c3e 100644 --- a/pkg/tcpip/transport/tcp/endpoint_state.go +++ b/pkg/tcpip/transport/tcp/endpoint_state.go @@ -130,6 +130,7 @@ func (e *endpoint) Restore(s *stack.Stack) { snd.resendTimer.init(s.Clock(), timerHandler(e, e.snd.retransmitTimerExpired)) snd.reorderTimer.init(s.Clock(), timerHandler(e, e.snd.rc.reorderTimerExpired)) snd.probeTimer.init(s.Clock(), timerHandler(e, e.snd.probeTimerExpired)) + snd.corkTimer.init(s.Clock(), timerHandler(e, e.snd.corkTimerExpired)) } e.stack = s e.protocol = protocolFromStack(s) @@ -196,6 +197,11 @@ func (e *endpoint) Restore(s *stack.Stack) { e.timeWaitTimer = e.stack.Clock().AfterFunc(e.getTimeWaitDuration(), e.timeWaitTimerExpired) } + if e.ops.GetCorkOption() { + // Rearm the timer if TCP_CORK is enabled which will + // drain all the segments in the queue after restore. + e.snd.corkTimer.enable(MinRTO) + } e.mu.Unlock() connectedLoading.Done() case epState == StateListen: diff --git a/pkg/tcpip/transport/tcp/snd.go b/pkg/tcpip/transport/tcp/snd.go index e3315a66d..f9f61a309 100644 --- a/pkg/tcpip/transport/tcp/snd.go +++ b/pkg/tcpip/transport/tcp/snd.go @@ -151,6 +151,13 @@ type sender struct { // segment after entering an RTO for the first time as described in // RFC3522 Section 3.2. retransmitTS uint32 + + // startCork start corking the segments. + startCork bool + + // corkTimer is used to drain the segments which are held when TCP_CORK + // option is enabled. + corkTimer timer `state:"nosave"` } // rtt is a synchronization wrapper used to appease stateify. See the comment @@ -208,6 +215,7 @@ func newSender(ep *endpoint, iss, irs seqnum.Value, sndWnd seqnum.Size, mss uint s.resendTimer.init(s.ep.stack.Clock(), timerHandler(s.ep, s.retransmitTimerExpired)) s.reorderTimer.init(s.ep.stack.Clock(), timerHandler(s.ep, s.rc.reorderTimerExpired)) s.probeTimer.init(s.ep.stack.Clock(), timerHandler(s.ep, s.probeTimerExpired)) + s.corkTimer.init(s.ep.stack.Clock(), timerHandler(s.ep, s.corkTimerExpired)) s.ep.AssertLockHeld(ep) s.updateMaxPayloadSize(int(ep.route.MTU()), 0) @@ -776,10 +784,20 @@ func (s *sender) maybeSendSegment(seg *segment, limit int, end seqnum.Value) (se } // With TCP_CORK, hold back until minimum of the available // send space and MSS. - // TODO(gvisor.dev/issue/2833): Drain the held segments after a - // timeout. - if seg.payloadSize() < s.MaxPayloadSize && s.ep.ops.GetCorkOption() { - return false + if s.ep.ops.GetCorkOption() { + if seg.payloadSize() < s.MaxPayloadSize { + if !s.startCork { + s.startCork = true + // Enable the timer for + // 200ms, after which + // the segments are drained. + s.corkTimer.enable(MinRTO) + } + return false + } + // Disable the TCP_CORK timer. + s.startCork = false + s.corkTimer.disable() } } } @@ -1724,3 +1742,24 @@ func (s *sender) updateWriteNext(seg *segment) { } s.writeNext = seg } + +// corkTimerExpired drains all the segments when TCP_CORK is enabled. +// +checklocks:s.ep.mu +func (s *sender) corkTimerExpired() tcpip.Error { + // Check if the timer actually expired or if it's a spurious wake due + // to a previously orphaned runtime timer. + if s.corkTimer.isUninitialized() || !s.corkTimer.checkExpiration() { + return nil + } + + // Assign sequence number and flags to the segment. + seg := s.writeNext + if seg == nil { + return nil + } + seg.sequenceNumber = s.SndNxt + seg.flags = header.TCPFlagAck | header.TCPFlagPsh + // Drain all the segments. + s.sendData() + return nil +} diff --git a/test/syscalls/linux/socket_ip_tcp_generic.cc b/test/syscalls/linux/socket_ip_tcp_generic.cc index 9dfb2a885..92e5a1501 100644 --- a/test/syscalls/linux/socket_ip_tcp_generic.cc +++ b/test/syscalls/linux/socket_ip_tcp_generic.cc @@ -1464,5 +1464,47 @@ TEST_P(TCPSocketPairTest, ResetWithSoLingerZeroTimeoutOption) { SyscallSucceedsWithValue(sizeof(buf))); } +TEST_P(TCPSocketPairTest, WaitTillMSSWithCorkOption) { + auto sockets = ASSERT_NO_ERRNO_AND_VALUE(NewSocketPair()); + ASSERT_THAT(setsockopt(sockets->first_fd(), IPPROTO_TCP, TCP_CORK, + &kSockOptOn, sizeof(kSockOptOn)), + SyscallSucceeds()); + + constexpr int kTCPMaxSeg = 1024; + EXPECT_THAT(setsockopt(sockets->first_fd(), IPPROTO_TCP, TCP_MAXSEG, + &kTCPMaxSeg, sizeof(kTCPMaxSeg)), + SyscallSucceedsWithValue(0)); + + // Use the buffer size which is guaranteed to be >= MSS. + char buffer[1024] = {}; + EXPECT_THAT(RetryEINTR(send)(sockets->first_fd(), buffer, sizeof(buffer), 0), + SyscallSucceedsWithValue(sizeof(buffer))); + + char buf[1024] = {}; + EXPECT_THAT(RetryEINTR(recv)(sockets->second_fd(), buf, sizeof(buf), 0), + SyscallSucceedsWithValue(sizeof(buf))); +} + +TEST_P(TCPSocketPairTest, WaitTillTimeoutWithCorkOption) { + auto sockets = ASSERT_NO_ERRNO_AND_VALUE(NewSocketPair()); + ASSERT_THAT(setsockopt(sockets->first_fd(), IPPROTO_TCP, TCP_CORK, + &kSockOptOn, sizeof(kSockOptOn)), + SyscallSucceeds()); + + constexpr int kTCPMaxSeg = 1024; + EXPECT_THAT(setsockopt(sockets->first_fd(), IPPROTO_TCP, TCP_MAXSEG, + &kTCPMaxSeg, sizeof(kTCPMaxSeg)), + SyscallSucceedsWithValue(0)); + + // Use buffer size less than MSS. + char buffer[512] = {}; + EXPECT_THAT(RetryEINTR(send)(sockets->first_fd(), buffer, sizeof(buffer), 0), + SyscallSucceedsWithValue(sizeof(buffer))); + + char buf[512] = {}; + EXPECT_THAT(RetryEINTR(recv)(sockets->second_fd(), buf, sizeof(buf), 0), + SyscallSucceedsWithValue(sizeof(buf))); +} + } // namespace testing } // namespace gvisor