From 8696447b72f1009b97b7d1d04974beec5ad0e22a Mon Sep 17 00:00:00 2001 From: Nayana Bidari Date: Wed, 13 Mar 2024 14:10:14 -0700 Subject: [PATCH] Implement a timer when TCP_CORK is enabled. TCP_CORK option when enabled queues the segments which are less than MSS size. These segments are drained when the MSS is reached or when the 200ms timer expires. This change implements the timer which drains the segments held with TCP_CORK. Updates #2833 PiperOrigin-RevId: 615538807 --- pkg/tcpip/transport/tcp/endpoint.go | 2 + pkg/tcpip/transport/tcp/endpoint_state.go | 6 +++ pkg/tcpip/transport/tcp/snd.go | 47 ++++++++++++++++++-- test/syscalls/linux/socket_ip_tcp_generic.cc | 42 +++++++++++++++++ 4 files changed, 93 insertions(+), 4 deletions(-) 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