diff --git a/pkg/tcpip/link/channel/channel.go b/pkg/tcpip/link/channel/channel.go index c1e48db20..e8c9d4374 100644 --- a/pkg/tcpip/link/channel/channel.go +++ b/pkg/tcpip/link/channel/channel.go @@ -135,12 +135,15 @@ var _ stack.GSOEndpoint = (*Endpoint)(nil) // Endpoint is link layer endpoint that stores outbound packets in a channel // and allows injection of inbound packets. type Endpoint struct { - dispatcher stack.NetworkDispatcher mtu uint32 linkAddr tcpip.LinkAddress LinkEPCapabilities stack.LinkEndpointCapabilities SupportedGSOKind stack.SupportedGSO + mu sync.RWMutex + // +checklocks:mu + dispatcher stack.NetworkDispatcher + // Outbound packet queue. q *queue } @@ -191,17 +194,24 @@ func (e *Endpoint) NumQueued() int { // InjectInbound injects an inbound packet. func (e *Endpoint) InjectInbound(protocol tcpip.NetworkProtocolNumber, pkt stack.PacketBufferPtr) { - e.dispatcher.DeliverNetworkPacket(protocol, pkt) + e.mu.RLock() + d := e.dispatcher + e.mu.RUnlock() + d.DeliverNetworkPacket(protocol, pkt) } // Attach saves the stack network-layer dispatcher for use later when packets // are injected. func (e *Endpoint) Attach(dispatcher stack.NetworkDispatcher) { + e.mu.Lock() + defer e.mu.Unlock() e.dispatcher = dispatcher } // IsAttached implements stack.LinkEndpoint.IsAttached. func (e *Endpoint) IsAttached() bool { + e.mu.RLock() + defer e.mu.RUnlock() return e.dispatcher != nil } diff --git a/pkg/tcpip/link/fdbased/endpoint.go b/pkg/tcpip/link/fdbased/endpoint.go index c8c162efb..adc97926c 100644 --- a/pkg/tcpip/link/fdbased/endpoint.go +++ b/pkg/tcpip/link/fdbased/endpoint.go @@ -135,7 +135,10 @@ type endpoint struct { closed func(tcpip.Error) inboundDispatchers []linkDispatcher - dispatcher stack.NetworkDispatcher + + mu sync.RWMutex + // +checklocks:mu + dispatcher stack.NetworkDispatcher // packetDispatchMode controls the packet dispatcher used by this // endpoint. @@ -405,6 +408,8 @@ func isSocketFD(fd int) (bool, error) { // // Attach implements stack.LinkEndpoint.Attach. func (e *endpoint) Attach(dispatcher stack.NetworkDispatcher) { + e.mu.Lock() + defer e.mu.Unlock() // nil means the NIC is being removed. if dispatcher == nil && e.dispatcher != nil { for _, dispatcher := range e.inboundDispatchers { @@ -431,6 +436,8 @@ func (e *endpoint) Attach(dispatcher stack.NetworkDispatcher) { // IsAttached implements stack.LinkEndpoint.IsAttached. func (e *endpoint) IsAttached() bool { + e.mu.RLock() + defer e.mu.RUnlock() return e.dispatcher != nil } @@ -765,18 +772,25 @@ func (e *endpoint) ARPHardwareType() header.ARPHardwareType { type InjectableEndpoint struct { endpoint + mu sync.RWMutex + // +checklocks:mu dispatcher stack.NetworkDispatcher } // Attach saves the stack network-layer dispatcher for use later when packets // are injected. func (e *InjectableEndpoint) Attach(dispatcher stack.NetworkDispatcher) { + e.mu.Lock() + defer e.mu.Unlock() e.dispatcher = dispatcher } // InjectInbound injects an inbound packet. func (e *InjectableEndpoint) InjectInbound(protocol tcpip.NetworkProtocolNumber, pkt stack.PacketBufferPtr) { - e.dispatcher.DeliverNetworkPacket(protocol, pkt) + e.mu.RLock() + d := e.dispatcher + e.mu.RUnlock() + d.DeliverNetworkPacket(protocol, pkt) } // NewInjectable creates a new fd-based InjectableEndpoint. diff --git a/pkg/tcpip/link/fdbased/mmap.go b/pkg/tcpip/link/fdbased/mmap.go index 743bb912c..b7c756730 100644 --- a/pkg/tcpip/link/fdbased/mmap.go +++ b/pkg/tcpip/link/fdbased/mmap.go @@ -198,6 +198,9 @@ func (d *packetMMapDispatcher) dispatch() (bool, tcpip.Error) { panic(fmt.Sprintf("LinkHeader().Consume(%d) must succeed", d.e.hdrSize)) } } - d.e.dispatcher.DeliverNetworkPacket(p, pbuf) + d.e.mu.RLock() + dsp := d.e.dispatcher + d.e.mu.RUnlock() + dsp.DeliverNetworkPacket(p, pbuf) return true, nil } diff --git a/pkg/tcpip/link/fdbased/packet_dispatchers.go b/pkg/tcpip/link/fdbased/packet_dispatchers.go index 13bbf494e..f5d3eb809 100644 --- a/pkg/tcpip/link/fdbased/packet_dispatchers.go +++ b/pkg/tcpip/link/fdbased/packet_dispatchers.go @@ -212,7 +212,10 @@ func (d *readVDispatcher) dispatch() (bool, tcpip.Error) { } } - d.e.dispatcher.DeliverNetworkPacket(p, pkt) + d.e.mu.RLock() + dsp := d.e.dispatcher + d.e.mu.RUnlock() + dsp.DeliverNetworkPacket(p, pkt) return true, nil } @@ -291,6 +294,10 @@ func (d *recvMMsgDispatcher) dispatch() (bool, tcpip.Error) { // Keep a list of packets so we can DecRef outside of the loop. var pkts stack.PacketBufferList + d.e.mu.RLock() + dsp := d.e.dispatcher + d.e.mu.RUnlock() + defer func() { pkts.DecRef() }() for k := 0; k < nMsgs; k++ { n := int(d.msgHdrs[k].Len) @@ -329,7 +336,7 @@ func (d *recvMMsgDispatcher) dispatch() (bool, tcpip.Error) { } } - d.e.dispatcher.DeliverNetworkPacket(p, pkt) + dsp.DeliverNetworkPacket(p, pkt) } return true, nil diff --git a/pkg/tcpip/link/loopback/loopback.go b/pkg/tcpip/link/loopback/loopback.go index 35cdbdff4..d3fb48dfc 100644 --- a/pkg/tcpip/link/loopback/loopback.go +++ b/pkg/tcpip/link/loopback/loopback.go @@ -21,12 +21,16 @@ package loopback import ( + "sync" + "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/header" "gvisor.dev/gvisor/pkg/tcpip/stack" ) type endpoint struct { + mu sync.RWMutex + // +checklocks:mu dispatcher stack.NetworkDispatcher } @@ -39,11 +43,15 @@ func New() stack.LinkEndpoint { // Attach implements stack.LinkEndpoint.Attach. It just saves the stack network- // layer dispatcher for later use when packets need to be dispatched. func (e *endpoint) Attach(dispatcher stack.NetworkDispatcher) { + e.mu.Lock() + defer e.mu.Unlock() e.dispatcher = dispatcher } // IsAttached implements stack.LinkEndpoint.IsAttached. func (e *endpoint) IsAttached() bool { + e.mu.RLock() + defer e.mu.RUnlock() return e.dispatcher != nil } @@ -75,6 +83,9 @@ func (*endpoint) Wait() {} // WritePackets implements stack.LinkEndpoint.WritePackets. func (e *endpoint) WritePackets(pkts stack.PacketBufferList) (int, tcpip.Error) { + e.mu.RLock() + d := e.dispatcher + e.mu.RUnlock() for _, pkt := range pkts.AsSlice() { // In order to properly loop back to the inbound side we must create a // fresh packet that only contains the underlying payload with no headers @@ -82,7 +93,7 @@ func (e *endpoint) WritePackets(pkts stack.PacketBufferList) (int, tcpip.Error) newPkt := stack.NewPacketBuffer(stack.PacketBufferOptions{ Payload: pkt.ToBuffer(), }) - e.dispatcher.DeliverNetworkPacket(pkt.NetworkProtocolNumber, newPkt) + d.DeliverNetworkPacket(pkt.NetworkProtocolNumber, newPkt) newPkt.DecRef() } return pkts.Len(), nil diff --git a/pkg/tcpip/link/muxed/injectable.go b/pkg/tcpip/link/muxed/injectable.go index 5c421d0b5..014057892 100644 --- a/pkg/tcpip/link/muxed/injectable.go +++ b/pkg/tcpip/link/muxed/injectable.go @@ -16,6 +16,8 @@ package muxed import ( + "sync" + "gvisor.dev/gvisor/pkg/bufferv2" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/header" @@ -27,7 +29,10 @@ import ( // will be written to. Note that HandleLocal works differently for this // endpoint (see WritePacket). type InjectableEndpoint struct { - routes map[tcpip.Address]stack.InjectableLinkEndpoint + routes map[tcpip.Address]stack.InjectableLinkEndpoint + + mu sync.RWMutex + // +checklocks:mu dispatcher stack.NetworkDispatcher } @@ -72,17 +77,24 @@ func (m *InjectableEndpoint) Attach(dispatcher stack.NetworkDispatcher) { for _, endpoint := range m.routes { endpoint.Attach(dispatcher) } + m.mu.Lock() m.dispatcher = dispatcher + m.mu.Unlock() } // IsAttached implements stack.LinkEndpoint. func (m *InjectableEndpoint) IsAttached() bool { + m.mu.RLock() + defer m.mu.RUnlock() return m.dispatcher != nil } // InjectInbound implements stack.InjectableLinkEndpoint. func (m *InjectableEndpoint) InjectInbound(protocol tcpip.NetworkProtocolNumber, pkt stack.PacketBufferPtr) { - m.dispatcher.DeliverNetworkPacket(protocol, pkt) + m.mu.RLock() + d := m.dispatcher + m.mu.RUnlock() + d.DeliverNetworkPacket(protocol, pkt) } // WritePackets writes outbound packets to the appropriate diff --git a/pkg/tcpip/link/pipe/pipe.go b/pkg/tcpip/link/pipe/pipe.go index aad1e2d5d..789c62a0a 100644 --- a/pkg/tcpip/link/pipe/pipe.go +++ b/pkg/tcpip/link/pipe/pipe.go @@ -17,6 +17,8 @@ package pipe import ( + "sync" + "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/header" "gvisor.dev/gvisor/pkg/tcpip/stack" @@ -41,10 +43,13 @@ func New(linkAddr1, linkAddr2 tcpip.LinkAddress, mtu uint32) (*Endpoint, *Endpoi // Endpoint is one end of a pipe. type Endpoint struct { + linked *Endpoint + linkAddr tcpip.LinkAddress + mtu uint32 + + mu sync.RWMutex + // +checklocks:mu dispatcher stack.NetworkDispatcher - linked *Endpoint - linkAddr tcpip.LinkAddress - mtu uint32 } func (e *Endpoint) deliverPackets(pkts stack.PacketBufferList) { @@ -59,7 +64,10 @@ func (e *Endpoint) deliverPackets(pkts stack.PacketBufferList) { newPkt := stack.NewPacketBuffer(stack.PacketBufferOptions{ Payload: pkt.ToBuffer(), }) - e.linked.dispatcher.DeliverNetworkPacket(pkt.NetworkProtocolNumber, newPkt) + e.linked.mu.RLock() + d := e.linked.dispatcher + e.linked.mu.RUnlock() + d.DeliverNetworkPacket(pkt.NetworkProtocolNumber, newPkt) newPkt.DecRef() } } @@ -73,11 +81,15 @@ func (e *Endpoint) WritePackets(pkts stack.PacketBufferList) (int, tcpip.Error) // Attach implements stack.LinkEndpoint. func (e *Endpoint) Attach(dispatcher stack.NetworkDispatcher) { + e.mu.Lock() + defer e.mu.Unlock() e.dispatcher = dispatcher } // IsAttached implements stack.LinkEndpoint. func (e *Endpoint) IsAttached() bool { + e.mu.RLock() + defer e.mu.RUnlock() return e.dispatcher != nil } diff --git a/pkg/tcpip/link/waitable/waitable.go b/pkg/tcpip/link/waitable/waitable.go index 25d9a7fc3..ec08bf805 100644 --- a/pkg/tcpip/link/waitable/waitable.go +++ b/pkg/tcpip/link/waitable/waitable.go @@ -34,7 +34,10 @@ var _ stack.LinkEndpoint = (*Endpoint)(nil) // Endpoint is a waitable link-layer endpoint. type Endpoint struct { dispatchGate sync.Gate - dispatcher stack.NetworkDispatcher + + mu sync.RWMutex + // +checklocks:mu + dispatcher stack.NetworkDispatcher writeGate sync.Gate lower stack.LinkEndpoint @@ -57,8 +60,10 @@ func (e *Endpoint) DeliverNetworkPacket(protocol tcpip.NetworkProtocolNumber, pk if !e.dispatchGate.Enter() { return } - - e.dispatcher.DeliverNetworkPacket(protocol, pkt) + e.mu.RLock() + d := e.dispatcher + e.mu.RUnlock() + d.DeliverNetworkPacket(protocol, pkt) e.dispatchGate.Leave() } @@ -67,8 +72,10 @@ func (e *Endpoint) DeliverLinkPacket(protocol tcpip.NetworkProtocolNumber, pkt s if !e.dispatchGate.Enter() { return } - - e.dispatcher.DeliverLinkPacket(protocol, pkt) + e.mu.RLock() + d := e.dispatcher + e.mu.RUnlock() + d.DeliverLinkPacket(protocol, pkt) e.dispatchGate.Leave() } @@ -76,12 +83,16 @@ func (e *Endpoint) DeliverLinkPacket(protocol tcpip.NetworkProtocolNumber, pkt s // registers with the lower endpoint as its dispatcher so that "e" is called // for inbound packets. func (e *Endpoint) Attach(dispatcher stack.NetworkDispatcher) { + e.mu.Lock() e.dispatcher = dispatcher + e.mu.Unlock() e.lower.Attach(e) } // IsAttached implements stack.LinkEndpoint.IsAttached. func (e *Endpoint) IsAttached() bool { + e.mu.RLock() + defer e.mu.RUnlock() return e.dispatcher != nil } diff --git a/pkg/tcpip/link/xdp/endpoint.go b/pkg/tcpip/link/xdp/endpoint.go index e53bbbdab..515296feb 100644 --- a/pkg/tcpip/link/xdp/endpoint.go +++ b/pkg/tcpip/link/xdp/endpoint.go @@ -54,6 +54,8 @@ type endpoint struct { // its end of the communication pipe. closed func(tcpip.Error) + mu sync.RWMutex + // +checkloks:mu networkDispatcher stack.NetworkDispatcher // wg keeps track of running goroutines. @@ -169,6 +171,8 @@ func New(opts *Options) (stack.LinkEndpoint, error) { // // Attach implements stack.LinkEndpoint.Attach. func (ep *endpoint) Attach(networkDispatcher stack.NetworkDispatcher) { + ep.mu.Lock() + defer ep.mu.Unlock() // nil means the NIC is being removed. if networkDispatcher == nil && ep.IsAttached() { ep.stopFD.Stop() @@ -199,6 +203,8 @@ func (ep *endpoint) Attach(networkDispatcher stack.NetworkDispatcher) { // IsAttached implements stack.LinkEndpoint.IsAttached. func (ep *endpoint) IsAttached() bool { + ep.mu.RLock() + defer ep.mu.RUnlock() return ep.networkDispatcher != nil } @@ -341,6 +347,9 @@ func (ep *endpoint) dispatch() (bool, tcpip.Error) { ep.control.UMEM.Unlock() // Process each packet. + ep.mu.RLock() + d := ep.networkDispatcher + ep.mu.RUnlock() for i := uint32(0); i < nReceived; i++ { view := views[i] data := view.AsSlice() @@ -355,7 +364,7 @@ func (ep *endpoint) dispatch() (bool, tcpip.Error) { if _, ok := pkt.LinkHeader().Consume(header.EthernetMinimumSize); !ok { panic(fmt.Sprintf("LinkHeader().Consume(%d) must succeed", header.EthernetMinimumSize)) } - ep.networkDispatcher.DeliverNetworkPacket(netProto, pkt) + d.DeliverNetworkPacket(netProto, pkt) pkt.DecRef() } // Tell the kernel that we're done with these