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

@@ -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 */)

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:

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
+}

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