Allow specifying a minimum number of underlays for a channel, regardless of underlay type. Fixes #226

Author: plorenz

multi.go

@@ -139,9 +139,8 @@ func NewMultiChannel(config *MultiChannelConfig) (MultiChannel, error) {
 	}
 
 	impl.startMultiplex(config.Underlay)
-	go impl.underlayHandler.Start(impl)
-
 	impl.underlayHandler.HandleUnderlayAccepted(impl, config.Underlay)
+	go impl.underlayHandler.Start(impl)
 
 	return impl, nil
 }
@@ -563,6 +562,7 @@ type underlayConstraint struct {
 
 type UnderlayConstraints struct {
 	types           map[string]underlayConstraint
+	minTotal        uint32
 	applyInProgress atomic.Bool
 	lastDial        concurrenz.AtomicValue[time.Time]
 }
@@ -571,6 +571,10 @@ func (self *UnderlayConstraints) LastDialTime() time.Time {
 	return self.lastDial.Load()
 }
 
+func (self *UnderlayConstraints) SetMinTotal(minTotal uint32) {
+	self.minTotal = minTotal
+}
+
 func (self *UnderlayConstraints) AddConstraint(underlayType string, numDesired int, minAllowed int) {
 	if self.types == nil {
 		self.types = make(map[string]underlayConstraint)
@@ -599,6 +603,24 @@ func (self *UnderlayConstraints) countsShowValidState(ch MultiChannel, counts ma
 			return false
 		}
 	}
+
+	totalCount := 0
+	for _, count := range counts {
+		totalCount += count
+	}
+
+	if uint32(totalCount) < self.minTotal {
+		pfxlog.Logger().WithField("conn", ch.LogicalName()).
+			WithField("channelId", ch.ConnectionId()).
+			WithField("label", ch.Label()).
+			WithField("underlays", counts).
+			Info("not enough total open underlays, closing multi-underlay channel")
+		if err := ch.Close(); err != nil {
+			pfxlog.Logger().WithError(err).Error("error closing underlay")
+		}
+		return false
+	}
+
 	return true
 }
 

multi_listener.go

@@ -17,8 +17,9 @@
 package channel
 
 import (
-	"github.com/michaelquigley/pfxlog"
 	"sync"
+
+	"github.com/michaelquigley/pfxlog"
 )
 
 type MultiChannelFactory func(underlay Underlay, closeCallback func()) (MultiChannel, error)
@@ -47,6 +48,7 @@ func (self *MultiListener) AcceptUnderlay(underlay Underlay) {
 	}
 
 	chId := underlay.ConnectionId()
+	isFirst, _ := Headers(underlay.Headers()).GetBoolHeader(IsFirstGroupConnection)
 
 	self.lock.Lock()
 	mc, ok := self.channels[chId]
@@ -58,7 +60,7 @@ func (self *MultiListener) AcceptUnderlay(underlay Underlay) {
 			log.WithError(err).Error("error accepting underlay")
 		}
 	} else {
-		if isFirst, _ := Headers(underlay.Headers()).GetBoolHeader(IsFirstGroupConnection); !isFirst {
+		if !isFirst {
 			log.Info("no existing channel found for underlay, but isFirstGroupConnection not set, closing connection")
 			if err := underlay.Close(); err != nil {
 				log.Info("error closing underlay")