From d36d5df834e738400547eabb9c2415a269c1e8f0 Mon Sep 17 00:00:00 2001 From: gavinz-ship-it Date: Fri, 23 Oct 2026 12:55:08 -0700 Subject: [PATCH 1/3] Update common.go Use a single shared linkStateWatcher for all interfaces. --- dataplane/forwarding/fwdport/ports/common.go | 123 ++++++++++++++----- 1 file changed, 90 insertions(+), 33 deletions(-) diff --git a/dataplane/forwarding/fwdport/ports/common.go b/dataplane/forwarding/fwdport/ports/common.go index c04ceaa0..9be6d1da 100644 --- a/dataplane/forwarding/fwdport/ports/common.go +++ b/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) } } - }() + } } From 2ee8b770664c037d9c2404be179f262c829b5af9 Mon Sep 17 00:00:00 2001 From: gavinz-ship-it Date: Fri, 23 Oct 2026 12:55:53 -0700 Subject: [PATCH 2/3] Update kernel.go Update kernel.go to use a single shared linkStateWatcher. --- dataplane/forwarding/fwdport/ports/kernel.go | 32 +++++++++----------- 1 file changed, 15 insertions(+), 17 deletions(-) diff --git a/dataplane/forwarding/fwdport/ports/kernel.go b/dataplane/forwarding/fwdport/ports/kernel.go index ecc42c63..163971b6 100644 --- a/dataplane/forwarding/fwdport/ports/kernel.go +++ b/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 } From b6b63b2481f2bdb54277f758c58bc5d6196e35c5 Mon Sep 17 00:00:00 2001 From: gavinz-ship-it Date: Fri, 23 Oct 2026 12:56:19 -0700 Subject: [PATCH 3/3] Update tap.go Update tap.go to use a single shared linkStateWatcher. --- dataplane/forwarding/fwdport/ports/tap.go | 37 +++++++++++------------ 1 file changed, 17 insertions(+), 20 deletions(-) diff --git a/dataplane/forwarding/fwdport/ports/tap.go b/dataplane/forwarding/fwdport/ports/tap.go index 30e66723..9f1f60a7 100644 --- a/dataplane/forwarding/fwdport/ports/tap.go +++ b/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 }