tcp: use lockdep mutexes

PiperOrigin-RevId: 761300509

Author: avagin

pkg/tcpip/transport/tcp/BUILD

@@ -1,3 +1,4 @@
+load("//pkg/sync/locking:locking.bzl", "declare_mutex", "declare_rwmutex")
 load("//tools:defs.bzl", "go_library", "go_test")
 load("//tools/go_generics:defs.bzl", "go_template_instance")
 
@@ -6,6 +7,104 @@ package(
     licenses = ["notice"],
 )
 
+declare_mutex(
+    name = "snd_queue_mutex",
+    out = "snd_queue_mutex.go",
+    package = "tcp",
+    prefix = "sndQueue",
+)
+
+declare_mutex(
+    name = "pending_processing_mutex",
+    out = "pending_processing_mutex.go",
+    package = "tcp",
+    prefix = "pendingProcessing",
+)
+
+declare_mutex(
+    name = "last_error_mutex",
+    out = "last_error_mutex.go",
+    package = "tcp",
+    prefix = "lastError",
+)
+
+declare_mutex(
+    name = "rcv_queue_mutex",
+    out = "rcv_queue_mutex.go",
+    package = "tcp",
+    prefix = "rcvQueue",
+)
+
+declare_mutex(
+    name = "accept_mutex",
+    out = "accept_mutex.go",
+    package = "tcp",
+    prefix = "accept",
+)
+
+declare_mutex(
+    name = "keepalive_mutex",
+    out = "keepalive_mutex.go",
+    package = "tcp",
+    prefix = "keepalive",
+)
+
+declare_mutex(
+    name = "hasher_mutex",
+    out = "hasher_mutex.go",
+    package = "tcp",
+    prefix = "hasher",
+)
+
+declare_mutex(
+    name = "forwarder_mutex",
+    out = "forwarder_mutex.go",
+    package = "tcp",
+    prefix = "forwarder",
+)
+
+declare_mutex(
+    name = "forwarder_request_mutex",
+    out = "forwarder_request_mutex.go",
+    package = "tcp",
+    prefix = "forwarderRequest",
+)
+
+declare_mutex(
+    name = "segment_queue_mutex",
+    out = "segment_queue_mutex.go",
+    package = "tcp",
+    prefix = "segmentQueue",
+)
+
+declare_mutex(
+    name = "ep_queue_mutex",
+    out = "ep_queue_mutex.go",
+    package = "tcp",
+    prefix = "epQueue",
+)
+
+declare_mutex(
+    name = "dispatcher_mutex",
+    out = "dispatcher_mutex.go",
+    package = "tcp",
+    prefix = "dispatcher",
+)
+
+declare_mutex(
+    name = "rtt_mutex",
+    out = "rtt_mutex.go",
+    package = "tcp",
+    prefix = "rtt",
+)
+
+declare_rwmutex(
+    name = "protocol_mutex",
+    out = "protocol_mutex.go",
+    package = "tcp",
+    prefix = "protocol",
+)
+
 go_template_instance(
     name = "tcp_segment_list",
     out = "tcp_segment_list.go",
@@ -45,27 +144,41 @@ go_library(
     name = "tcp",
     srcs = [
         "accept.go",
+        "accept_mutex.go",
         "connect.go",
         "connect_unsafe.go",
         "cubic.go",
         "dispatcher.go",
+        "dispatcher_mutex.go",
         "endpoint.go",
         "endpoint_state.go",
+        "ep_queue_mutex.go",
         "forwarder.go",
+        "forwarder_mutex.go",
+        "forwarder_request_mutex.go",
+        "hasher_mutex.go",
+        "keepalive_mutex.go",
+        "last_error_mutex.go",
+        "pending_processing_mutex.go",
         "protocol.go",
+        "protocol_mutex.go",
         "rack.go",
         "rcv.go",
+        "rcv_queue_mutex.go",
         "reno.go",
         "reno_recovery.go",
+        "rtt_mutex.go",
         "sack.go",
         "sack_recovery.go",
         "sack_scoreboard.go",
         "segment.go",
         "segment_heap.go",
         "segment_queue.go",
+        "segment_queue_mutex.go",
         "segment_state.go",
         "segment_unsafe.go",
         "snd.go",
+        "snd_queue_mutex.go",
         "state.go",
         "tcp_endpoint_list.go",
         "tcp_segment_list.go",
@@ -81,6 +194,7 @@ go_library(
         "//pkg/refs",
         "//pkg/sleep",
         "//pkg/sync",
+        "//pkg/sync/locking",
         "//pkg/tcpip",
         "//pkg/tcpip/checksum",
         "//pkg/tcpip/hash/jenkins",

pkg/tcpip/transport/tcp/accept.go

@@ -23,7 +23,6 @@ import (
 	"io"
 	"time"
 
-	"gvisor.dev/gvisor/pkg/sync"
 	"gvisor.dev/gvisor/pkg/tcpip"
 	"gvisor.dev/gvisor/pkg/tcpip/header"
 	"gvisor.dev/gvisor/pkg/tcpip/ports"
@@ -88,7 +87,7 @@ type listenContext struct {
 	listenEP *Endpoint
 
 	// hasherMu protects hasher.
-	hasherMu sync.Mutex
+	hasherMu hasherMutex
 	// hasher is the hash function used to generate a SYN cookie.
 	hasher hash.Hash
 

pkg/tcpip/transport/tcp/dispatcher.go

@@ -32,7 +32,7 @@ import (
 //
 // +stateify savable
 type epQueue struct {
-	mu   sync.Mutex `state:"nosave"`
+	mu   epQueueMutex `state:"nosave"`
 	list endpointList
 }
 
@@ -365,7 +365,7 @@ type dispatcher struct {
 	processors []processor
 	wg         sync.WaitGroup `state:"nosave"`
 	hasher     jenkinsHasher
-	mu         sync.Mutex `state:"nosave"`
+	mu         dispatcherMutex `state:"nosave"`
 	// +checklocks:mu
 	paused bool
 	// +checklocks:mu

pkg/tcpip/transport/tcp/endpoint.go

@@ -285,7 +285,7 @@ func (*Stats) IsEndpointStats() {}
 //
 // +stateify savable
 type sndQueueInfo struct {
-	sndQueueMu sync.Mutex `state:"nosave"`
+	sndQueueMu sndQueueMutex `state:"nosave"`
 	TCPSndBufState
 }
 
@@ -354,7 +354,7 @@ type Endpoint struct {
 	endpointEntry `state:"nosave"`
 
 	// pendingProcessingMu protects pendingProcessing.
-	pendingProcessingMu sync.Mutex `state:"nosave"`
+	pendingProcessingMu pendingProcessingMutex `state:"nosave"`
 
 	// pendingProcessing is true if this endpoint is queued for processing
 	// to a TCP processor.
@@ -374,10 +374,10 @@ type Endpoint struct {
 
 	// lastError represents the last error that the endpoint reported;
 	// access to it is protected by the following mutex.
-	lastErrorMu sync.Mutex `state:"nosave"`
+	lastErrorMu lastErrorMutex `state:"nosave"`
 	lastError   tcpip.Error
 
-	rcvQueueMu sync.Mutex `state:"nosave"`
+	rcvQueueMu rcvQueueMutex `state:"nosave"`
 
 	// +checklocks:rcvQueueMu
 	TCPRcvBufState
@@ -834,11 +834,11 @@ func calculateTTL(route *stack.Route, ipv4TTL uint8, ipv6HopLimit int16) uint8 {
 //
 // +stateify savable
 type keepalive struct {
-	sync.Mutex `state:"nosave"`
-	idle       time.Duration
-	interval   time.Duration
-	count      int
-	unacked    int
+	keepaliveMutex `state:"nosave"`
+	idle           time.Duration
+	interval       time.Duration
+	count          int
+	unacked        int
 	// should never be a zero timer if the endpoint is not closed.
 	timer timer       `state:"nosave"`
 	waker sleep.Waker `state:"nosave"`

pkg/tcpip/transport/tcp/forwarder.go

@@ -18,7 +18,6 @@ import (
 	"fmt"
 
 	"gvisor.dev/gvisor/pkg/buffer"
-	"gvisor.dev/gvisor/pkg/sync"
 	"gvisor.dev/gvisor/pkg/tcpip"
 	"gvisor.dev/gvisor/pkg/tcpip/header"
 	"gvisor.dev/gvisor/pkg/tcpip/seqnum"
@@ -38,7 +37,7 @@ type Forwarder struct {
 	maxInFlight int
 	handler     func(*ForwarderRequest)
 
-	mu       sync.Mutex
+	mu       forwarderMutex
 	inFlight map[stack.TransportEndpointID]struct{}
 	listen   *listenContext
 }
@@ -111,7 +110,7 @@ func (f *Forwarder) HandlePacket(id stack.TransportEndpointID, pkt *stack.Packet
 // and passed to the client. Clients must eventually call Complete() on it, and
 // may optionally create an endpoint to represent it via CreateEndpoint.
 type ForwarderRequest struct {
-	mu         sync.Mutex
+	mu         forwarderRequestMutex
 	forwarder  *Forwarder
 	segment    *segment
 	synOptions header.TCPSynOptions

pkg/tcpip/transport/tcp/protocol.go

@@ -23,7 +23,6 @@ import (
 	"strings"
 	"time"
 
-	"gvisor.dev/gvisor/pkg/sync"
 	"gvisor.dev/gvisor/pkg/tcpip"
 	"gvisor.dev/gvisor/pkg/tcpip/header"
 	"gvisor.dev/gvisor/pkg/tcpip/header/parse"
@@ -90,7 +89,7 @@ const (
 type protocol struct {
 	stack *stack.Stack
 
-	mu                         sync.RWMutex `state:"nosave"`
+	mu                         protocolRWMutex `state:"nosave"`
 	sackEnabled                bool
 	recovery                   tcpip.TCPRecovery
 	delayEnabled               bool

pkg/tcpip/transport/tcp/segment_queue.go

@@ -14,16 +14,12 @@
 
 package tcp
 
-import (
-	"gvisor.dev/gvisor/pkg/sync"
-)
-
 // segmentQueue is a bounded, thread-safe queue of TCP segments.
 //
 // +stateify savable
 type segmentQueue struct {
-	mu     sync.Mutex  `state:"nosave"`
-	list   segmentList `state:"wait"`
+	mu     segmentQueueMutex `state:"nosave"`
+	list   segmentList       `state:"wait"`
 	ep     *Endpoint
 	frozen bool
 }

pkg/tcpip/transport/tcp/snd.go

@@ -21,7 +21,6 @@ import (
 	"time"
 
 	"gvisor.dev/gvisor/pkg/buffer"
-	"gvisor.dev/gvisor/pkg/sync"
 	"gvisor.dev/gvisor/pkg/tcpip"
 	"gvisor.dev/gvisor/pkg/tcpip/header"
 	"gvisor.dev/gvisor/pkg/tcpip/seqnum"
@@ -237,7 +236,7 @@ func (wl *protectedWriteList) InsertAfter(before, seg *segment) {
 //
 // +stateify savable
 type rtt struct {
-	sync.Mutex `state:"nosave"`
+	rttMutex `state:"nosave"`
 
 	TCPRTTState
 }