Use a single shared linkStateWatcher

dataplane/forwarding/fwdport/ports/common.go

@@ -18,12 +18,14 @@ package ports
 
 import (
 	"net"
+	"sync"
 
 	log "github.com/golang/glog"
 	"github.com/vishvananda/netlink"
 
 	"github.com/openconfig/lemming/dataplane/forwarding/fwdport"
 	"github.com/openconfig/lemming/dataplane/forwarding/infra/fwdcontext"
+	"github.com/openconfig/lemming/dataplane/kernel"
 	fwdpb "github.com/openconfig/lemming/proto/forwarding"
 )
 
@@ -86,45 +88,100 @@ func getAndSetState(name string, ifMgr stateIfaceMgr, pi *fwdpb.PortInfo) (*fwdp
 	default:
 		return reply, nil
 	}
-	if err == nil {
-		reply.Status.AdminStatus = pi.AdminStatus
+	if err != nil {
+		return nil, err
 	}
-	return reply, err
+	reply.Status.AdminStatus = pi.AdminStatus
+	return reply, nil
+}
+
+type linkSubscriber interface {
+	LinkSubscribe(ch chan<- netlink.LinkUpdate, done <-chan struct{}) error
+}
+
+type portTarget struct {
+	port fwdport.Port
+	ctx  *fwdcontext.Context
 }
 
-func startStateWatch(updCh chan netlink.LinkUpdate, doneCh chan struct{}, devName string, port fwdport.Port, ctx *fwdcontext.Context) {
-	go func() {
-		for {
-			select {
-			case <-doneCh:
-				log.Warningf("state watch stopping for dev: %v", devName)
+type linkStateWatcher struct {
+	mu       sync.Mutex
+	ifaceMgr linkSubscriber
+	ports    map[string]portTarget
+	doneCh   chan struct{}
+}
+
+var defaultLinkWatcher = &linkStateWatcher{
+	ifaceMgr: &kernel.Interfaces{},
+	ports:    make(map[string]portTarget),
+}
+
+func (w *linkStateWatcher) watch(devName string, port fwdport.Port, ctx *fwdcontext.Context) error {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+
+	if len(w.ports) == 0 {
+		updCh := make(chan netlink.LinkUpdate)
+		doneCh := make(chan struct{})
+		if err := w.ifaceMgr.LinkSubscribe(updCh, doneCh); err != nil {
+			return err
+		}
+		w.doneCh = doneCh
+		go w.run(updCh, doneCh)
+	}
+	w.ports[devName] = portTarget{port: port, ctx: ctx}
+	return nil
+}
+
+func (w *linkStateWatcher) unwatch(devName string) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+
+	delete(w.ports, devName)
+	if len(w.ports) == 0 && w.doneCh != nil {
+		close(w.doneCh)
+		w.doneCh = nil
+	}
+}
+
+func (w *linkStateWatcher) run(updCh <-chan netlink.LinkUpdate, doneCh <-chan struct{}) {
+	for {
+		select {
+		case <-doneCh:
+			log.Warningf("shared link state watch stopping")
+			return
+		case upd, ok := <-updCh:
+			if !ok {
+				log.Infof("shared link update chan closed")
 				return
-			case upd, ok := <-updCh:
-				if !ok {
-					log.Infof("update chan close for dev: %v", devName)
-					return
-				}
-				if upd.Attrs() == nil || upd.Attrs().Name != devName {
-					continue
-				}
-				admin, oper := stateFromAttrs(upd.Attrs())
-				log.V(1).Infof("dataplane receive link update: port %q, admin %v, oper %v", devName, admin.String(), oper.String())
-				if err := ctx.Notify(&fwdpb.EventDesc{
-					Event: fwdpb.Event_EVENT_PORT,
-					Desc: &fwdpb.EventDesc_Port{
-						Port: &fwdpb.PortEventDesc{
-							Context: &fwdpb.ContextId{Id: ctx.ID},
-							PortId:  &fwdpb.PortId{ObjectId: &fwdpb.ObjectId{Id: string(port.ID())}},
-							PortInfo: &fwdpb.PortInfo{
-								AdminStatus: admin,
-								OperStatus:  oper,
-							},
+			}
+			if upd.Attrs() == nil {
+				continue
+			}
+			devName := upd.Attrs().Name
+			w.mu.Lock()
+			target, ok := w.ports[devName]
+			w.mu.Unlock()
+			if !ok {
+				continue
+			}
+			admin, oper := stateFromAttrs(upd.Attrs())
+			log.V(1).Infof("dataplane receive link update: port %q, admin %v, oper %v", devName, admin.String(), oper.String())
+			if err := target.ctx.Notify(&fwdpb.EventDesc{
+				Event: fwdpb.Event_EVENT_PORT,
+				Desc: &fwdpb.EventDesc_Port{
+					Port: &fwdpb.PortEventDesc{
+						Context: &fwdpb.ContextId{Id: target.ctx.ID},
+						PortId:  &fwdpb.PortId{ObjectId: &fwdpb.ObjectId{Id: string(target.port.ID())}},
+						PortInfo: &fwdpb.PortInfo{
+							AdminStatus: admin,
+							OperStatus:  oper,
 						},
 					},
-				}); err != nil {
-					log.Warningf("failed to notify port state update: %v", err)
-				}
+				},
+			}); err != nil {
+				log.Errorf("failed to notify port state update: %v", err)
 			}
 		}
-	}()
+	}
 }

dataplane/forwarding/fwdport/ports/kernel.go

@@ -45,15 +45,14 @@ func init() {
 // kernelPort is a ports that receives from and writes a linux network device.
 type kernelPort struct {
 	fwdobject.Base
-	devName      string
-	input        fwdaction.Actions
-	output       fwdaction.Actions
-	desc         *fwdpb.PortDesc
-	ctx          *fwdcontext.Context // Forwarding context containing the port
-	handle       packetHandle
-	doneCh       chan struct{}
-	linkUpdateCh chan netlink.LinkUpdate
-	ifaceMgr     kernel.Interfaces
+	devName  string
+	input    fwdaction.Actions
+	output   fwdaction.Actions
+	desc     *fwdpb.PortDesc
+	ctx      *fwdcontext.Context // Forwarding context containing the port
+	handle   packetHandle
+	doneCh   chan struct{}
+	ifaceMgr kernel.Interfaces
 }
 
 type packetHandle interface {
@@ -82,6 +81,7 @@ func (p *kernelPort) Type() fwdpb.PortType {
 func (p *kernelPort) Cleanup() {
 	p.input.Cleanup()
 	p.output.Cleanup()
+	defaultLinkWatcher.unwatch(p.devName)
 	close(p.doneCh)
 	p.input = nil
 	p.output = nil
@@ -111,7 +111,6 @@ func (p *kernelPort) Update(upd *fwdpb.PortUpdateDesc) error {
 }
 
 func (p *kernelPort) process() {
-	startStateWatch(p.linkUpdateCh, p.doneCh, p.devName, p, p.ctx)
 	go func() {
 		for {
 			select {
@@ -205,18 +204,17 @@ func (kernelBuilder) Build(portDesc *fwdpb.PortDesc, ctx *fwdcontext.Context) (f
 		}
 	}
 	p := &kernelPort{
-		ctx:          ctx,
-		handle:       handle,
-		devName:      kp.Kernel.DeviceName,
-		desc:         portDesc,
-		doneCh:       make(chan struct{}),
-		linkUpdateCh: make(chan netlink.LinkUpdate),
+		ctx:     ctx,
+		handle:  handle,
+		devName: kp.Kernel.DeviceName,
+		desc:    portDesc,
+		doneCh:  make(chan struct{}),
 	}
 	list := append(fwdport.CounterList, fwdaction.CounterList...)
 	if err := p.InitCounters("", list...); err != nil {
 		return nil, err
 	}
-	if err := p.ifaceMgr.LinkSubscribe(p.linkUpdateCh, p.doneCh); err != nil {
+	if err := defaultLinkWatcher.watch(p.devName, p, p.ctx); err != nil {
 		return nil, err
 	}
 

dataplane/forwarding/fwdport/ports/tap.go

@@ -23,7 +23,6 @@ import (
 	"os"
 	"time"
 
-	"github.com/vishvananda/netlink"
 	"golang.org/x/sys/unix"
 
 	"github.com/openconfig/lemming/dataplane/forwarding/fwdaction"
@@ -50,16 +49,15 @@ func init() {
 // this ensures kernel correctly responds to lower level protocols such as ARP, ICMP, etc.
 type tapPort struct {
 	fwdobject.Base
-	input        fwdaction.Actions
-	output       fwdaction.Actions
-	desc         *fwdpb.PortDesc
-	ctx          *fwdcontext.Context // Forwarding context containing the port
-	fd           int
-	devName      string
-	doneCh       chan struct{}
-	linkUpdateCh chan netlink.LinkUpdate
-	file         *os.File
-	ifaceMgr     kernel.Interfaces
+	input    fwdaction.Actions
+	output   fwdaction.Actions
+	desc     *fwdpb.PortDesc
+	ctx      *fwdcontext.Context // Forwarding context containing the port
+	fd       int
+	devName  string
+	doneCh   chan struct{}
+	file     *os.File
+	ifaceMgr kernel.Interfaces
 }
 
 // Desc returns the port description proto.
@@ -82,6 +80,7 @@ func (p *tapPort) Type() fwdpb.PortType {
 func (p *tapPort) Cleanup() {
 	p.input.Cleanup()
 	p.output.Cleanup()
+	defaultLinkWatcher.unwatch(p.devName)
 	close(p.doneCh)
 	p.input = nil
 	p.output = nil
@@ -112,7 +111,6 @@ func (p *tapPort) Update(upd *fwdpb.PortUpdateDesc) error {
 }
 
 func (p *tapPort) process() {
-	startStateWatch(p.linkUpdateCh, p.doneCh, p.devName, p, p.ctx)
 	go func() {
 		buf := make([]byte, packetBufferSize)
 		for {
@@ -204,19 +202,18 @@ func (tp tapBuilder) Build(portDesc *fwdpb.PortDesc, ctx *fwdcontext.Context) (f
 	}
 	file := os.NewFile(uintptr(fd), "/dev/tun")
 	p := &tapPort{
-		ctx:          ctx,
-		file:         file,
-		fd:           fd,
-		devName:      kp.Tap.GetDeviceName(),
-		desc:         portDesc,
-		doneCh:       make(chan struct{}),
-		linkUpdateCh: make(chan netlink.LinkUpdate),
+		ctx:     ctx,
+		file:    file,
+		fd:      fd,
+		devName: kp.Tap.GetDeviceName(),
+		desc:    portDesc,
+		doneCh:  make(chan struct{}),
 	}
 	list := append(fwdport.CounterList, fwdaction.CounterList...)
 	if err := p.InitCounters("", list...); err != nil {
 		return nil, err
 	}
-	if err := tp.ifaceMgr.LinkSubscribe(p.linkUpdateCh, p.doneCh); err != nil {
+	if err := defaultLinkWatcher.watch(p.devName, p, p.ctx); err != nil {
 		return nil, err
 	}