Fix racy behavior in tcp Read().

The startRead and commitRead with RcvReadMu is not safe anymore with the
change to Close() that purges the read queue.

This may reduce performance of highly concurrent reads.

PiperOrigin-RevId: 452175522

pkg/tcpip/transport/tcp/accept.go

@@ -214,7 +214,10 @@ func (l *listenContext) createConnectingEndpoint(s *segment, rcvdSynOpts header.
 	// Bootstrap the auto tuning algorithm. Starting at zero will result in
 	// a large step function on the first window adjustment causing the
 	// window to grow to a really large value.
-	n.rcvQueueInfo.RcvAutoParams.PrevCopiedBytes = n.initialReceiveWindow()
+	initWnd := n.initialReceiveWindow()
+	n.rcvQueueMu.Lock()
+	n.RcvAutoParams.PrevCopiedBytes = initWnd
+	n.rcvQueueMu.Unlock()
 
 	return n, nil
 }
@@ -432,9 +435,9 @@ func (a *acceptQueue) isFull() bool {
 //
 // +checklocks:e.mu
 func (e *endpoint) handleListenSegment(ctx *listenContext, s *segment) tcpip.Error {
-	e.rcvQueueInfo.rcvQueueMu.Lock()
-	rcvClosed := e.rcvQueueInfo.RcvClosed
-	e.rcvQueueInfo.rcvQueueMu.Unlock()
+	e.rcvQueueMu.Lock()
+	rcvClosed := e.RcvClosed
+	e.rcvQueueMu.Unlock()
 	if rcvClosed || s.flags.Contains(header.TCPFlagSyn|header.TCPFlagAck) {
 		// If the endpoint is shutdown, reply with reset.
 		//

pkg/tcpip/transport/tcp/connect.go

@@ -654,13 +654,13 @@ func (h *handshake) transitionToStateEstablishedLocked(s *segment) {
 		h.ep.snd.updateRTO(rtt)
 	}
 
-	h.ep.rcvQueueInfo.rcvQueueMu.Lock()
+	h.ep.rcvQueueMu.Lock()
 	h.ep.rcv = newReceiver(h.ep, h.ackNum-1, h.rcvWnd, h.effectiveRcvWndScale())
 	// Bootstrap the auto tuning algorithm. Starting at zero will
 	// result in a really large receive window after the first auto
 	// tuning adjustment.
-	h.ep.rcvQueueInfo.RcvAutoParams.PrevCopiedBytes = int(h.rcvWnd)
-	h.ep.rcvQueueInfo.rcvQueueMu.Unlock()
+	h.ep.RcvAutoParams.PrevCopiedBytes = int(h.rcvWnd)
+	h.ep.rcvQueueMu.Unlock()
 
 	h.ep.setEndpointState(StateEstablished)
 

pkg/tcpip/transport/tcp/rcv.go

@@ -153,11 +153,11 @@ func (r *receiver) getSendParams() (RcvNxt seqnum.Value, rcvWnd seqnum.Size) {
 	// Keep advertising zero receive window up until the new window reaches a
 	// threshold.
 	if r.rcvWnd == 0 && newWnd != 0 {
-		r.ep.rcvQueueInfo.rcvQueueMu.Lock()
+		r.ep.rcvQueueMu.Lock()
 		if crossed, above := r.ep.windowCrossedACKThresholdLocked(int(newWnd), int(r.ep.ops.GetReceiveBufferSize())); !crossed && !above {
 			newWnd = 0
 		}
-		r.ep.rcvQueueInfo.rcvQueueMu.Unlock()
+		r.ep.rcvQueueMu.Unlock()
 	}
 
 	// Stash away the non-scaled receive window as we use it for measuring
@@ -328,36 +328,36 @@ func (r *receiver) updateRTT() {
 	// estimate the round-trip time by observing the time between when a byte
 	// is first acknowledged and the receipt of data that is at least one
 	// window beyond the sequence number that was acknowledged.
-	r.ep.rcvQueueInfo.rcvQueueMu.Lock()
-	if r.ep.rcvQueueInfo.RcvAutoParams.RTTMeasureTime == (tcpip.MonotonicTime{}) {
+	r.ep.rcvQueueMu.Lock()
+	if r.ep.RcvAutoParams.RTTMeasureTime == (tcpip.MonotonicTime{}) {
 		// New measurement.
-		r.ep.rcvQueueInfo.RcvAutoParams.RTTMeasureTime = r.ep.stack.Clock().NowMonotonic()
-		r.ep.rcvQueueInfo.RcvAutoParams.RTTMeasureSeqNumber = r.RcvNxt.Add(r.rcvWnd)
-		r.ep.rcvQueueInfo.rcvQueueMu.Unlock()
+		r.ep.RcvAutoParams.RTTMeasureTime = r.ep.stack.Clock().NowMonotonic()
+		r.ep.RcvAutoParams.RTTMeasureSeqNumber = r.RcvNxt.Add(r.rcvWnd)
+		r.ep.rcvQueueMu.Unlock()
 		return
 	}
-	if r.RcvNxt.LessThan(r.ep.rcvQueueInfo.RcvAutoParams.RTTMeasureSeqNumber) {
-		r.ep.rcvQueueInfo.rcvQueueMu.Unlock()
+	if r.RcvNxt.LessThan(r.ep.RcvAutoParams.RTTMeasureSeqNumber) {
+		r.ep.rcvQueueMu.Unlock()
 		return
 	}
-	rtt := r.ep.stack.Clock().NowMonotonic().Sub(r.ep.rcvQueueInfo.RcvAutoParams.RTTMeasureTime)
+	rtt := r.ep.stack.Clock().NowMonotonic().Sub(r.ep.RcvAutoParams.RTTMeasureTime)
 	// We only store the minimum observed RTT here as this is only used in
 	// absence of a SRTT available from either timestamps or a sender
 	// measurement of RTT.
-	if r.ep.rcvQueueInfo.RcvAutoParams.RTT == 0 || rtt < r.ep.rcvQueueInfo.RcvAutoParams.RTT {
-		r.ep.rcvQueueInfo.RcvAutoParams.RTT = rtt
+	if r.ep.RcvAutoParams.RTT == 0 || rtt < r.ep.RcvAutoParams.RTT {
+		r.ep.RcvAutoParams.RTT = rtt
 	}
-	r.ep.rcvQueueInfo.RcvAutoParams.RTTMeasureTime = r.ep.stack.Clock().NowMonotonic()
-	r.ep.rcvQueueInfo.RcvAutoParams.RTTMeasureSeqNumber = r.RcvNxt.Add(r.rcvWnd)
-	r.ep.rcvQueueInfo.rcvQueueMu.Unlock()
+	r.ep.RcvAutoParams.RTTMeasureTime = r.ep.stack.Clock().NowMonotonic()
+	r.ep.RcvAutoParams.RTTMeasureSeqNumber = r.RcvNxt.Add(r.rcvWnd)
+	r.ep.rcvQueueMu.Unlock()
 }
 
 // +checklocks:r.ep.mu
 // +checklocksalias:r.ep.snd.ep.mu=r.ep.mu
 func (r *receiver) handleRcvdSegmentClosing(s *segment, state EndpointState, closed bool) (drop bool, err tcpip.Error) {
-	r.ep.rcvQueueInfo.rcvQueueMu.Lock()
-	rcvClosed := r.ep.rcvQueueInfo.RcvClosed || r.closed
-	r.ep.rcvQueueInfo.rcvQueueMu.Unlock()
+	r.ep.rcvQueueMu.Lock()
+	rcvClosed := r.ep.RcvClosed || r.closed
+	r.ep.rcvQueueMu.Unlock()
 
 	// If we are in one of the shutdown states then we need to do
 	// additional checks before we try and process the segment.
@@ -487,9 +487,9 @@ func (r *receiver) handleRcvdSegment(s *segment) (drop bool, err tcpip.Error) {
 			// segments to arrive allowing pending segments to be processed and
 			// delivered to the user.
 			if rcvBufSize := r.ep.ops.GetReceiveBufferSize(); rcvBufSize > 0 && (r.PendingBufUsed+int(segLen)) < int(rcvBufSize)>>2 {
-				r.ep.rcvQueueInfo.rcvQueueMu.Lock()
+				r.ep.rcvQueueMu.Lock()
 				r.PendingBufUsed += s.segMemSize()
-				r.ep.rcvQueueInfo.rcvQueueMu.Unlock()
+				r.ep.rcvQueueMu.Unlock()
 				s.IncRef()
 				heap.Push(&r.pendingRcvdSegments, s)
 				UpdateSACKBlocks(&r.ep.sack, segSeq, segSeq.Add(segLen), r.RcvNxt)
@@ -523,9 +523,9 @@ func (r *receiver) handleRcvdSegment(s *segment) (drop bool, err tcpip.Error) {
 		}
 
 		heap.Pop(&r.pendingRcvdSegments)
-		r.ep.rcvQueueInfo.rcvQueueMu.Lock()
+		r.ep.rcvQueueMu.Lock()
 		r.PendingBufUsed -= s.segMemSize()
-		r.ep.rcvQueueInfo.rcvQueueMu.Unlock()
+		r.ep.rcvQueueMu.Unlock()
 		s.DecRef()
 	}
 	return false, nil

pkg/tcpip/transport/tcp/segment.go

@@ -185,6 +185,7 @@ func (s *segment) setOwner(ep *endpoint, qFlags queueFlags) {
 func (s *segment) DecRef() {
 	s.segmentRefs.DecRef(func() {
 		defer s.pkt.DecRef()
+		s.pkt = nil
 		if s.ep != nil {
 			switch s.qFlags {
 			case recvQ: