diff --git a/pkg/tcpip/link/channel/channel.go b/pkg/tcpip/link/channel/channel.go index 1bfb8537d..0f3337ab7 100644 --- a/pkg/tcpip/link/channel/channel.go +++ b/pkg/tcpip/link/channel/channel.go @@ -138,13 +138,14 @@ var _ stack.GSOEndpoint = (*Endpoint)(nil) // +stateify savable type Endpoint struct { mtu uint32 - linkAddr tcpip.LinkAddress LinkEPCapabilities stack.LinkEndpointCapabilities SupportedGSOKind stack.SupportedGSO mu sync.RWMutex `state:"nosave"` // +checklocks:mu dispatcher stack.NetworkDispatcher + // +checklocks:mu + linkAddr tcpip.LinkAddress // Outbound packet queue. q *queue @@ -249,6 +250,8 @@ func (*Endpoint) MaxHeaderLength() uint16 { // LinkAddress returns the link address of this endpoint. func (e *Endpoint) LinkAddress() tcpip.LinkAddress { + e.mu.RLock() + defer e.mu.RUnlock() return e.linkAddr } diff --git a/pkg/tcpip/link/fdbased/endpoint.go b/pkg/tcpip/link/fdbased/endpoint.go index e284a931f..237fc0716 100644 --- a/pkg/tcpip/link/fdbased/endpoint.go +++ b/pkg/tcpip/link/fdbased/endpoint.go @@ -127,9 +127,6 @@ type endpoint struct { // is added/removed; otherwise an ethernet header is used. hdrSize int - // addr is the address of the endpoint. - addr tcpip.LinkAddress - // caps holds the endpoint capabilities. caps stack.LinkEndpointCapabilities @@ -170,6 +167,11 @@ type endpoint struct { // maxSyscallHeaderBytes, it falls back to writing the packet using writev // via WritePacket.) writevMaxIovs int + + // addr is the address of the endpoint. + // + // +checklocks:mu + addr tcpip.LinkAddress } // Options specify the details about the fd-based endpoint to be created. @@ -470,6 +472,8 @@ func (e *endpoint) MaxHeaderLength() uint16 { // LinkAddress returns the link address of this endpoint. func (e *endpoint) LinkAddress() tcpip.LinkAddress { + e.mu.RLock() + defer e.mu.RUnlock() return e.addr } diff --git a/pkg/tcpip/link/pipe/pipe.go b/pkg/tcpip/link/pipe/pipe.go index 3e4522077..7cdf422b3 100644 --- a/pkg/tcpip/link/pipe/pipe.go +++ b/pkg/tcpip/link/pipe/pipe.go @@ -45,13 +45,14 @@ func New(linkAddr1, linkAddr2 tcpip.LinkAddress, mtu uint32) (*Endpoint, *Endpoi // // +stateify savable type Endpoint struct { - linked *Endpoint - linkAddr tcpip.LinkAddress - mtu uint32 + linked *Endpoint + mtu uint32 mu sync.RWMutex `state:"nosave"` // +checklocks:mu dispatcher stack.NetworkDispatcher + // +checklocks:mu + linkAddr tcpip.LinkAddress } func (e *Endpoint) deliverPackets(pkts stack.PacketBufferList) { @@ -115,6 +116,8 @@ func (*Endpoint) MaxHeaderLength() uint16 { // LinkAddress implements stack.LinkEndpoint. func (e *Endpoint) LinkAddress() tcpip.LinkAddress { + e.mu.RLock() + defer e.mu.RUnlock() return e.linkAddr } diff --git a/pkg/tcpip/link/sharedmem/sharedmem.go b/pkg/tcpip/link/sharedmem/sharedmem.go index 9e8a8552a..7a619d4dd 100644 --- a/pkg/tcpip/link/sharedmem/sharedmem.go +++ b/pkg/tcpip/link/sharedmem/sharedmem.go @@ -156,10 +156,6 @@ type endpoint struct { // bufferSize is immutable. bufferSize uint32 - // addr is the local address of this endpoint. - // addr is immutable. - addr tcpip.LinkAddress - // peerFD is an fd to the peer that can be used to detect when the // peer is gone. // peerFD is immutable. @@ -196,7 +192,7 @@ type endpoint struct { onClosed func(tcpip.Error) `state:"nosave"` // mu protects the following fields. - mu sync.Mutex `state:"nosave"` + mu sync.RWMutex `state:"nosave"` // tx is the transmit queue. // +checklocks:mu @@ -205,6 +201,11 @@ type endpoint struct { // workerStarted specifies whether the worker goroutine was started. // +checklocks:mu workerStarted bool + + // addr is the local address of this endpoint. + // + // +checklocks:mu + addr tcpip.LinkAddress } // New creates a new shared-memory-based endpoint. Buffers will be broken up @@ -342,6 +343,8 @@ func (e *endpoint) MaxHeaderLength() uint16 { // LinkAddress implements stack.LinkEndpoint.LinkAddress. It returns the local // link address. func (e *endpoint) LinkAddress() tcpip.LinkAddress { + e.mu.RLock() + defer e.mu.RUnlock() return e.addr } @@ -354,6 +357,8 @@ func (e *endpoint) SetLinkAddress(addr tcpip.LinkAddress) { // AddHeader implements stack.LinkEndpoint.AddHeader. func (e *endpoint) AddHeader(pkt *stack.PacketBuffer) { + e.mu.RLock() + defer e.mu.RUnlock() // Add ethernet header if needed. if len(e.addr) == 0 { return @@ -374,6 +379,8 @@ func (e *endpoint) parseHeader(pkt *stack.PacketBuffer) bool { // ParseHeader implements stack.LinkEndpoint.ParseHeader. func (e *endpoint) ParseHeader(pkt *stack.PacketBuffer) bool { + e.mu.RLock() + defer e.mu.RUnlock() // Add ethernet header if needed. if len(e.addr) == 0 { return true @@ -475,7 +482,10 @@ func (e *endpoint) dispatchLoop(d stack.NetworkDispatcher) { } var proto tcpip.NetworkProtocolNumber - if len(e.addr) != 0 { + e.mu.RLock() + addrLen := len(e.addr) + e.mu.RUnlock() + if addrLen != 0 { if !e.parseHeader(pkt) { pkt.DecRef() continue @@ -500,7 +510,6 @@ func (e *endpoint) dispatchLoop(d stack.NetworkDispatcher) { continue } } - // Send packet up the stack. d.DeliverNetworkPacket(proto, pkt) pkt.DecRef() diff --git a/pkg/tcpip/link/sharedmem/sharedmem_server.go b/pkg/tcpip/link/sharedmem/sharedmem_server.go index 08900cfb6..9a72f6ad9 100644 --- a/pkg/tcpip/link/sharedmem/sharedmem_server.go +++ b/pkg/tcpip/link/sharedmem/sharedmem_server.go @@ -36,10 +36,6 @@ type serverEndpoint struct { // bufferSize is immutable. bufferSize uint32 - // addr is the local address of this endpoint. - // addr is immutable - addr tcpip.LinkAddress - // rx is the receive queue. rx serverRx @@ -70,7 +66,7 @@ type serverEndpoint struct { onClosed func(tcpip.Error) // mu protects the following fields. - mu sync.Mutex + mu sync.RWMutex // tx is the transmit queue. // +checklocks:mu @@ -79,6 +75,11 @@ type serverEndpoint struct { // workerStarted specifies whether the worker goroutine was started. // +checklocks:mu workerStarted bool + + // addr is the local address of this endpoint. + // + // +checklocks:mu + addr tcpip.LinkAddress } // NewServerEndpoint creates a new shared-memory-based endpoint. Buffers will be @@ -199,6 +200,8 @@ func (e *serverEndpoint) MaxHeaderLength() uint16 { // LinkAddress implements stack.LinkEndpoint.LinkAddress. It returns the local // link address. func (e *serverEndpoint) LinkAddress() tcpip.LinkAddress { + e.mu.RLock() + defer e.mu.RUnlock() return e.addr } @@ -211,6 +214,8 @@ func (e *serverEndpoint) SetLinkAddress(addr tcpip.LinkAddress) { // AddHeader implements stack.LinkEndpoint.AddHeader. func (e *serverEndpoint) AddHeader(pkt *stack.PacketBuffer) { + e.mu.RLock() + defer e.mu.RUnlock() // Add ethernet header if needed. if len(e.addr) == 0 { return @@ -231,6 +236,8 @@ func (e *serverEndpoint) parseHeader(pkt *stack.PacketBuffer) bool { // ParseHeader implements stack.LinkEndpoint.ParseHeader. func (e *serverEndpoint) ParseHeader(pkt *stack.PacketBuffer) bool { + e.mu.RLock() + defer e.mu.RUnlock() // Add ethernet header if needed. if len(e.addr) == 0 { return true @@ -326,7 +333,10 @@ func (e *serverEndpoint) dispatchLoop(d stack.NetworkDispatcher) { } } var proto tcpip.NetworkProtocolNumber - if len(e.addr) != 0 { + e.mu.RLock() + addrLen := len(e.addr) + e.mu.RUnlock() + if addrLen != 0 { if !e.parseHeader(pkt) { pkt.DecRef() continue diff --git a/pkg/tcpip/link/veth/veth.go b/pkg/tcpip/link/veth/veth.go index 3a3eb19f1..3c9226229 100644 --- a/pkg/tcpip/link/veth/veth.go +++ b/pkg/tcpip/link/veth/veth.go @@ -43,10 +43,6 @@ type Endpoint struct { backlogQueue *chan vethPacket - // linkAddr is the local address of this endpoint. - // linkaddr is immutable. - linkAddr tcpip.LinkAddress - mu sync.RWMutex `state:"nosave"` // +checklocks:mu dispatcher stack.NetworkDispatcher @@ -55,6 +51,10 @@ type Endpoint struct { stack *stack.Stack // +checklocks:mu idx tcpip.NICID + // linkAddr is the local address of this endpoint. + // + // +checklocks:mu + linkAddr tcpip.LinkAddress } // NewPair creates a new veth pair. @@ -167,6 +167,8 @@ func (*Endpoint) MaxHeaderLength() uint16 { // LinkAddress returns the link address of this endpoint. func (e *Endpoint) LinkAddress() tcpip.LinkAddress { + e.mu.RLock() + defer e.mu.RUnlock() return e.linkAddr } diff --git a/pkg/tcpip/link/xdp/endpoint.go b/pkg/tcpip/link/xdp/endpoint.go index c47ba81b6..a053fbcf7 100644 --- a/pkg/tcpip/link/xdp/endpoint.go +++ b/pkg/tcpip/link/xdp/endpoint.go @@ -45,9 +45,6 @@ type endpoint struct { // fd is the underlying AF_XDP socket. fd int - // addr is the address of the endpoint. - addr tcpip.LinkAddress - // caps holds the endpoint capabilities. caps stack.LinkEndpointCapabilities @@ -68,6 +65,11 @@ type endpoint struct { // stopFD is used to stop the dispatch loop. stopFD stopfd.StopFD + + // addr is the address of the endpoint. + // + // +checklocks:mu + addr tcpip.LinkAddress } // Options specify the details about the fd-based endpoint to be created. @@ -236,6 +238,8 @@ func (ep *endpoint) MaxHeaderLength() uint16 { // LinkAddress returns the link address of this endpoint. func (ep *endpoint) LinkAddress() tcpip.LinkAddress { + ep.mu.RLock() + defer ep.mu.RUnlock() return ep.addr }