refactor: overhaul transport controller to eliminate deadlock bugs

Signed-off-by: Christian Stewart <[email protected]>

Author: paralin

go.mod

@@ -3,12 +3,12 @@ module github.com/aperturerobotics/bifrost
 go 1.22
 
 require (
-	github.com/aperturerobotics/common v0.18.0 // latest
-	github.com/aperturerobotics/controllerbus v0.47.0 // latest
+	github.com/aperturerobotics/common v0.18.1 // latest
+	github.com/aperturerobotics/controllerbus v0.47.2 // latest
 	github.com/aperturerobotics/entitygraph v0.9.1 // latest
 	github.com/aperturerobotics/protobuf-go-lite v0.6.5 // latest
 	github.com/aperturerobotics/starpc v0.33.6 // latest
-	github.com/aperturerobotics/util v1.25.3 // latest
+	github.com/aperturerobotics/util v1.25.7-0.20240802075949-372f05d3a13f // latest
 )
 
 // aperture: use compatibility forks
@@ -22,7 +22,7 @@ replace (
 	github.com/nats-io/nats-server/v2 => github.com/aperturerobotics/bifrost-nats-server/v2 v2.1.8-0.20221228081037-b7c2df0c151f // aperture-2.0
 	github.com/nats-io/nats.go => github.com/aperturerobotics/bifrost-nats-client v1.10.1-0.20200831103200-24c3d0464e58 // aperture-2.0
 	github.com/nats-io/nkeys => github.com/nats-io/nkeys v0.3.0 // indirect: used by bifrost-nats-server
-	github.com/quic-go/quic-go => github.com/aperturerobotics/quic-go v0.45.1-0.20240701204210-82dc570e7aa0 // aperture
+	github.com/quic-go/quic-go => github.com/aperturerobotics/quic-go v0.45.1-0.20240802054753-f83427ffc2c6 // aperture
 	github.com/sirupsen/logrus => github.com/aperturerobotics/logrus v1.9.4-0.20240119050608-13332fb58195 // aperture
 )
 

go.sum

@@ -4,10 +4,10 @@ github.com/aperturerobotics/bifrost-nats-client v1.10.1-0.20200831103200-24c3d04
 github.com/aperturerobotics/bifrost-nats-client v1.10.1-0.20200831103200-24c3d0464e58/go.mod h1:ougcjYEZDYV8pVtaNbA5sgYDukkYHyKtSsW/T3B13j0=
 github.com/aperturerobotics/bifrost-nats-server/v2 v2.1.8-0.20221228081037-b7c2df0c151f h1:bmScByQNGDPPy9T+zdwu816XaCbFtD5UDyqZMRiHJ80=
 github.com/aperturerobotics/bifrost-nats-server/v2 v2.1.8-0.20221228081037-b7c2df0c151f/go.mod h1:kIcZtLpq4UIZzOqduYLm1mYU1nuMBtN6XuDCtQ21QT8=
-github.com/aperturerobotics/common v0.18.0 h1:AoeDkZcpg83+OjwUXGQD2VlUEDUl03IZBJiHW7pXtHE=
-github.com/aperturerobotics/common v0.18.0/go.mod h1:qLWqm2T9MQmxDtE1ddNdWCHnUj1Z66GzNxllkZ3Fj7Y=
-github.com/aperturerobotics/controllerbus v0.47.0 h1:rQ9f0nd53GEYoZYqQ+lN9CaKcFot4RBATY/TosT8+j8=
-github.com/aperturerobotics/controllerbus v0.47.0/go.mod h1:pT96tRLE9bCk73Osc/DAhYr7rK4wDcypdy8FqAUunk8=
+github.com/aperturerobotics/common v0.18.1 h1:zXFzVFOPe/11YZv1EUevvTmoWRVhX+B+S/5oUk9emGQ=
+github.com/aperturerobotics/common v0.18.1/go.mod h1:qLWqm2T9MQmxDtE1ddNdWCHnUj1Z66GzNxllkZ3Fj7Y=
+github.com/aperturerobotics/controllerbus v0.47.2 h1:BKPoRY5FRhS0AWTe0DgpsyvKy+invnFgC8ZHSuz8W9M=
+github.com/aperturerobotics/controllerbus v0.47.2/go.mod h1:bbLYOlppStRVjtRbbXSB72W4aFrPMKx7oUb814A8/dI=
 github.com/aperturerobotics/entitygraph v0.9.1 h1:bCl0Q9N1hkQnOtdCYz7pNSYqH+5CY3EUCuD19dQmJeI=
 github.com/aperturerobotics/entitygraph v0.9.1/go.mod h1:40VoNXSslWQieSS4V3HyzqOOZnRiGxItj/qZYSHg8Rw=
 github.com/aperturerobotics/go-libp2p v0.33.1-0.20240511223728-e0b67c111765 h1:KTSK6vPrh6gaT6CH0G2yAnmlcdjsc9Bk5JCqSeePrUw=
@@ -24,12 +24,12 @@ github.com/aperturerobotics/logrus v1.9.4-0.20240119050608-13332fb58195 h1:uyeD1
 github.com/aperturerobotics/logrus v1.9.4-0.20240119050608-13332fb58195/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
 github.com/aperturerobotics/protobuf-go-lite v0.6.5 h1:AuPPcZ7ZaJe9ZYYC4gF7/5/Xbn9Mt9uXyV3+ADWy+Ys=
 github.com/aperturerobotics/protobuf-go-lite v0.6.5/go.mod h1:YTbfnUj3feSULhs8VgepAHFnF3wUc0CPj4jd2axy21I=
-github.com/aperturerobotics/quic-go v0.45.1-0.20240701204210-82dc570e7aa0 h1:KH1Zk+zgI+jKmc/oRUEXEdlW00sMpy/WZbnOChW+nPk=
-github.com/aperturerobotics/quic-go v0.45.1-0.20240701204210-82dc570e7aa0/go.mod h1:X095EBMI8M7riYQRvUgegHFkEkgM2QKLvyGHyAcOw/Q=
+github.com/aperturerobotics/quic-go v0.45.1-0.20240802054753-f83427ffc2c6 h1:MZNodbX53cJQ6UuSIut4FvJWstxn+8/E1TpF1ceYjy0=
+github.com/aperturerobotics/quic-go v0.45.1-0.20240802054753-f83427ffc2c6/go.mod h1:X095EBMI8M7riYQRvUgegHFkEkgM2QKLvyGHyAcOw/Q=
 github.com/aperturerobotics/starpc v0.33.6 h1:noc/MnmIMTek9bdEvd88QiD1p9KzEV8CUOBIoKmGgm0=
 github.com/aperturerobotics/starpc v0.33.6/go.mod h1:4IYcbulEzqhPT5jKaDeL1BJPFd8WVWZ7Ugu0/348/Is=
-github.com/aperturerobotics/util v1.25.3 h1:8YwVRPnRYs/s8dRRhCPWCH0zF19fk8P4zVQtaIec73o=
-github.com/aperturerobotics/util v1.25.3/go.mod h1:m/paprtgaTiGfc4X3LkXpeseK9hfQA7QBI3cKsE/h3Y=
+github.com/aperturerobotics/util v1.25.7-0.20240802075949-372f05d3a13f h1:YIqFbDLHyly/Nmwo0+EYA/W8YEkopcMHDWhih0BNajw=
+github.com/aperturerobotics/util v1.25.7-0.20240802075949-372f05d3a13f/go.mod h1:DdflF5HvWF01S4MNnSV9O5LR0SGajkx0wDIiXTsNtFc=
 github.com/blang/semver v3.5.1+incompatible h1:cQNTCjp13qL8KC3Nbxr/y2Bqb63oX6wdnnjpJbkM4JQ=
 github.com/blang/semver v3.5.1+incompatible/go.mod h1:kRBLl5iJ+tD4TcOOxsy/0fnwebNt5EWlYSAyrTnjyyk=
 github.com/cenkalti/backoff v2.2.1+incompatible h1:tNowT99t7UNflLxfYYSlKYsBpXdEet03Pg2g16Swow4=

peer/api/api_srpc.pb.go

@@ -2,7 +2,6 @@ package pubsub_controller
 
 import (
 	"context"
-	"sync"
 
 	"github.com/aperturerobotics/bifrost/link"
 	"github.com/aperturerobotics/bifrost/peer"
@@ -47,9 +46,7 @@ type Controller struct {
 	// peerCtr holds the peer
 	peerCtr *ccontainer.CContainer[*peer.Peer]
 
-	// mtx guards below fields
-	mtx sync.Mutex
-	// bcast wakes the controller
+	// bcast guards below fields
 	bcast broadcast.Broadcast
 	// cleanupRefs are the refs to cleanup
 	cleanupRefs []directive.Reference
@@ -141,46 +138,47 @@ func (c *Controller) Execute(ctx context.Context) error {
 	}()
 
 	for {
-		c.mtx.Lock()
-		for _, vl := range c.incLinks {
-			tpl := pubsub.NewPeerLinkTuple(vl)
-			if e, ok := c.links[tpl]; ok {
-				e.ctxCancel()
-			}
-			tlCtx, tlCtxCancel := context.WithCancel(ctx)
-			tl := &trackedLink{
-				c:         c,
-				ctxCancel: tlCtxCancel,
-				tpl:       tpl,
-				lnk:       vl,
-				le: c.le.
-					WithField("link-uuid", vl.GetUUID()).
-					WithField("link-remote-peer", vl.GetRemotePeer().String()),
-			}
-			c.links[tpl] = tl
-			go func() {
-				err := tl.trackLink(tlCtx)
-				tlCtxCancel()
-				if err != context.Canceled && err != nil {
-					tl.le.WithError(err).Warn("link tracker returned fatal error")
+		var waitCh <-chan struct{}
+		c.bcast.HoldLock(func(broadcast func(), getWaitCh func() <-chan struct{}) {
+			for _, vl := range c.incLinks {
+				tpl := pubsub.NewPeerLinkTuple(vl)
+				if e, ok := c.links[tpl]; ok {
+					e.ctxCancel()
 				}
-				c.mtx.Lock()
-				if ol, ok := c.links[tpl]; ok && ol == tl {
-					delete(c.links, tpl)
+				tlCtx, tlCtxCancel := context.WithCancel(ctx)
+				tl := &trackedLink{
+					c:         c,
+					ctxCancel: tlCtxCancel,
+					tpl:       tpl,
+					lnk:       vl,
+					le: c.le.
+						WithField("link-uuid", vl.GetUUID()).
+						WithField("link-remote-peer", vl.GetRemotePeer().String()),
 				}
-				c.mtx.Unlock()
-			}()
-		}
-		c.incLinks = nil
-		wake := c.bcast.GetWaitCh()
-		c.mtx.Unlock()
+				c.links[tpl] = tl
+				go func() {
+					err := tl.trackLink(tlCtx)
+					tlCtxCancel()
+					if err != context.Canceled && err != nil {
+						tl.le.WithError(err).Warn("link tracker returned fatal error")
+					}
+					c.bcast.HoldLock(func(broadcast func(), getWaitCh func() <-chan struct{}) {
+						if ol, ok := c.links[tpl]; ok && ol == tl {
+							delete(c.links, tpl)
+						}
+					})
+				}()
+			}
+			c.incLinks = nil
+			waitCh = getWaitCh()
+		})
 
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
 		case err := <-psErr:
 			return err
-		case <-wake:
+		case <-waitCh:
 		}
 	}
 }
@@ -211,7 +209,7 @@ func (c *Controller) HandleDirective(ctx context.Context, di directive.Instance)
 	dir := di.GetDirective()
 	switch d := dir.(type) {
 	case link.EstablishLinkWithPeer:
-		c.handleEstablishLink(ctx, di, d)
+		c.handleEstablishLink(di)
 	case link.HandleMountedStream:
 		return c.handleMountedStream(ctx, di, d)
 	case pubsub.BuildChannelSubscription:
@@ -224,16 +222,16 @@ func (c *Controller) HandleDirective(ctx context.Context, di directive.Instance)
 // Close releases any resources used by the controller.
 // Error indicates any issue encountered releasing.
 func (c *Controller) Close() error {
-	c.mtx.Lock()
-	for _, ref := range c.cleanupRefs {
-		ref.Release()
-	}
-	c.cleanupRefs = nil
-	for k, l := range c.links {
-		l.ctxCancel()
-		delete(c.links, k)
-	}
-	c.mtx.Unlock()
+	c.bcast.HoldLock(func(broadcast func(), getWaitCh func() <-chan struct{}) {
+		for _, ref := range c.cleanupRefs {
+			ref.Release()
+		}
+		c.cleanupRefs = nil
+		for k, l := range c.links {
+			l.ctxCancel()
+			delete(c.links, k)
+		}
+	})
 
 	_ = c.pubSubCtr.SwapValue(func(val *pubsub.PubSub) *pubsub.PubSub {
 		if val != nil {
@@ -245,10 +243,5 @@ func (c *Controller) Close() error {
 	return nil
 }
 
-// wake wakes the controller
-func (c *Controller) wake() {
-	c.bcast.Broadcast()
-}
-
 // _ is a type assertion
 var _ pubsub.Controller = ((*Controller)(nil))

pubsub/api/api_srpc.pb.go

@@ -1,8 +1,6 @@
 package pubsub_controller
 
 import (
-	"context"
-
 	"github.com/aperturerobotics/bifrost/link"
 	"github.com/aperturerobotics/bifrost/pubsub"
 	"github.com/aperturerobotics/controllerbus/directive"
@@ -24,11 +22,7 @@ func newEstablishLinkHandler(c *Controller) *establishLinkHandler {
 }
 
 // handleEstablishLink handles an EstablishLink directive.
-func (c *Controller) handleEstablishLink(
-	ctx context.Context,
-	di directive.Instance,
-	d link.EstablishLinkWithPeer,
-) {
+func (c *Controller) handleEstablishLink(di directive.Instance) {
 	handler := newEstablishLinkHandler(c)
 	ref := di.AddReference(handler, true)
 	if ref == nil {
@@ -48,10 +42,10 @@ func (e *establishLinkHandler) HandleValueAdded(inst directive.Instance, val dir
 	e.c.le.Debugf("got link with uuid %v", vl.GetUUID())
 
 	// Attempt to open the stream.
-	e.c.mtx.Lock()
-	e.c.incLinks = append(e.c.incLinks, vl)
-	e.c.mtx.Unlock()
-	e.c.wake()
+	e.c.bcast.HoldLock(func(broadcast func(), getWaitCh func() <-chan struct{}) {
+		e.c.incLinks = append(e.c.incLinks, vl)
+		broadcast()
+	})
 }
 
 // HandleValueRemoved is called when a value is removed from the directive.
@@ -62,19 +56,19 @@ func (e *establishLinkHandler) HandleValueRemoved(inst directive.Instance, val d
 	}
 	e.c.le.Debugf("lost link with uuid %v", vl.GetUUID())
 	tpl := pubsub.NewPeerLinkTuple(vl)
-	e.c.mtx.Lock()
-	for i, l := range e.c.incLinks {
-		if l == vl {
-			e.c.incLinks[i] = e.c.incLinks[len(e.c.incLinks)-1]
-			e.c.incLinks[len(e.c.incLinks)-1] = nil
-			e.c.incLinks = e.c.incLinks[:len(e.c.incLinks)-1]
-			break
+	e.c.bcast.HoldLock(func(broadcast func(), getWaitCh func() <-chan struct{}) {
+		for i, l := range e.c.incLinks {
+			if l == vl {
+				e.c.incLinks[i] = e.c.incLinks[len(e.c.incLinks)-1]
+				e.c.incLinks[len(e.c.incLinks)-1] = nil
+				e.c.incLinks = e.c.incLinks[:len(e.c.incLinks)-1]
+				break
+			}
 		}
-	}
-	if v, ok := e.c.links[tpl]; ok {
-		v.ctxCancel()
-	}
-	e.c.mtx.Unlock()
+		if v, ok := e.c.links[tpl]; ok {
+			v.ctxCancel()
+		}
+	})
 }
 
 // HandleInstanceDisposed is called when a directive instance is disposed.
@@ -86,18 +80,18 @@ func (e *establishLinkHandler) HandleInstanceDisposed(inst directive.Instance) {
 	}
 	e.ref = nil
 
-	e.c.mtx.Lock()
-	for i, ref := range e.c.cleanupRefs {
-		if ref == eref {
-			a := e.c.cleanupRefs
-			a[i] = a[len(a)-1]
-			a[len(a)-1] = nil
-			a = a[:len(a)-1]
-			e.c.cleanupRefs = a
-			break
+	e.c.bcast.HoldLock(func(broadcast func(), getWaitCh func() <-chan struct{}) {
+		for i, ref := range e.c.cleanupRefs {
+			if ref == eref {
+				a := e.c.cleanupRefs
+				a[i] = a[len(a)-1]
+				a[len(a)-1] = nil
+				a = a[:len(a)-1]
+				e.c.cleanupRefs = a
+				break
+			}
 		}
-	}
-	e.c.mtx.Unlock()
+	})
 }
 
 var _ directive.ReferenceHandler = ((*establishLinkHandler)(nil))

pubsub/controller/controller.go

@@ -53,8 +53,12 @@ func (s *AccessRpcServiceServer) LookupRpcService(
 		}
 	}
 
+	var waitCh <-chan struct{}
+	bcast.HoldLock(func(broadcast func(), getWaitCh func() <-chan struct{}) {
+		waitCh = getWaitCh()
+	})
+
 	dir := bifrost_rpc.NewLookupRpcService(req.GetServiceId(), serverID)
-	waitCh := bcast.GetWaitCh()
 	vals := make(map[uint32]struct{})
 	di, ref, err := s.b.AddDirective(dir, bus.NewCallbackHandler(
 		func(av directive.AttachedValue) {