netstack/icmp: mv icmp responder to processors

Author: ignoramous

intra/netstack/dispatchers.go

@@ -178,7 +178,6 @@ type readVDispatcher struct {
 	closed atomic.Bool  // closed is set to true when fd is closed.
 	once   sync.Once    // Ensures stop() is called only once.
 	mgr    *supervisor
-	icmp   icmpResponder
 }
 
 var _ linkDispatcher = (*readVDispatcher)(nil)
@@ -187,10 +186,9 @@ var _ linkDispatcher = (*readVDispatcher)(nil)
 // fd and dispatches them to endpoint e. It assumes ownership of fd but not of e.
 func newReadVDispatcher(f *fds, e *endpoint) (linkDispatcher, error) {
 	d := &readVDispatcher{
-		e:    e,
-		buf:  newIovecBuffer(bufcfg),
-		mgr:  newSupervisor(e, f.tun()),
-		icmp: newICMPResponder(e),
+		e:   e,
+		buf: newIovecBuffer(bufcfg),
+		mgr: newSupervisor(e, icmp, f.tun()),
 	}
 	d.mgr.start()
 
@@ -319,12 +317,6 @@ func (d *readVDispatcher) io(fds *fds) (bool, tcpip.Error) {
 	})
 	defer pkt.DecRef()
 
-	if d.icmp.ok() {
-		if d.icmp.respond(pkt) {
-			return cont, nil
-		}
-	}
-
 	var iseth = d.e.hdrSize > 0 // hdrSize always zero; unused
 	if iseth {
 		if !d.e.parseHeader(pkt) {

intra/netstack/forwarders.go

@@ -134,6 +134,7 @@ type processor struct {
 	pkts stack.PacketBufferList
 
 	e           stack.InjectableLinkEndpoint
+	icmp        *icmpResponder
 	sleeper     sleep.Sleeper
 	packetWaker sleep.Waker
 	closeWaker  sleep.Waker
@@ -176,6 +177,9 @@ func (p *processor) deliverPackets() {
 		pkt := p.pkts.PopFront()
 		p.mu.Unlock()
 		if pkt != nil {
+			if p.icmp.respond(pkt) {
+				return
+			}
 			p.e.InjectInbound(pkt.NetworkProtocolNumber, pkt)
 			pkt.DecRef()
 		}
@@ -191,6 +195,7 @@ func (p *processor) deliverPackets() {
 // goroutines.
 type supervisor struct {
 	processors []processor
+	icmp       *icmpResponder
 	seed       uint32
 	wg         sync.WaitGroup
 	sid        *core.Volatile[int] // tun fd for diagnostics
@@ -199,21 +204,24 @@ type supervisor struct {
 
 // newSupervisor creates a new supervisor for the processors of endpoint e.
 func newSupervisor(e stack.InjectableLinkEndpoint, sid int) *supervisor {
+	icmp := newICMPResponder(e)
+
 	m := &supervisor{
 		seed:       rand.Uint32(),
 		sid:        core.NewVolatile(sid),
 		ready:      make([]bool, maxForwarders),
 		processors: make([]processor, maxForwarders),
+		icmp:       &icmp,
 		wg:         sync.WaitGroup{},
 	}
 
 	m.wg.Add(maxForwarders)
-
 	for i := range m.processors {
 		p := &m.processors[i]
 		p.sleeper.AddWaker(&p.packetWaker)
 		p.sleeper.AddWaker(&p.closeWaker)
 		p.e = e
+		p.icmp = &icmp
 	}
 
 	return m
@@ -314,6 +322,7 @@ func (m *supervisor) queuePacket(pkt *stack.PacketBuffer, hasEthHeader bool) {
 
 // stop stops all processor goroutines.
 func (m *supervisor) stop() {
+	m.icmp.stop()
 	sid := m.tunid()
 	start := time.Now()
 	log.D("ns: tun(%d): forwarder: stopping %d procs", sid, len(m.processors))

intra/netstack/icmp.go

@@ -44,6 +44,7 @@ func OutboundICMP(id string, s *stack.Stack, hdl GICMPHandler) {
 	forwarder := newIcmpForwarder(id, s, hdl)
 	s.SetTransportProtocolHandler(icmp.ProtocolNumber4, forwarder.reply4)
 	s.SetTransportProtocolHandler(icmp.ProtocolNumber6, forwarder.reply6)
+	// TODO: the handler must only be set for the "main" netstack
 	setICMPEchoHandler(forwarder)
 }
 

intra/netstack/icmpecho.go

@@ -36,17 +36,16 @@ func setICMPEchoHandler(h *icmpForwarder) {
 // ICMP implementation) to forward the ping and then injects the response back
 // into the TUN device.
 type icmpResponder struct {
-	ep   *endpoint
+	ep   stack.LinkEndpoint
 	open atomic.Bool
 }
 
 func (r *icmpResponder) stop() {
-	r.ep = nil
 	r.open.Store(false)
 }
 
-func newICMPResponder(ep *endpoint) (r icmpResponder) {
-	if ep == nil {
+func newICMPResponder(ep stack.LinkEndpoint) (r icmpResponder) {
+	if ep == nil || core.IsNil(ep) {
 		return
 	}
 	r.ep = ep
@@ -55,7 +54,7 @@ func newICMPResponder(ep *endpoint) (r icmpResponder) {
 }
 
 func (r *icmpResponder) ok() bool {
-	return r != nil && r.open.Load() && r.ep != nil
+	return r != nil && r.open.Load()
 }
 
 func (r *icmpResponder) respond(pkt *stack.PacketBuffer) (handled bool) {