Use a go-routine pool for payload ingest. Fixes #763

Author: plorenz

edge-apis/pool.go

@@ -24,6 +24,7 @@ import (
 	"math/rand/v2"
 	"net"
 	"net/url"
+	"slices"
 	"sync/atomic"
 	"time"
 )
@@ -35,7 +36,7 @@ type ApiClientTransport struct {
 
 // ClientTransportPool abstracts the concept of multiple `runtime.ClientTransport` (openapi interface) representing one
 // target OpenZiti network. In situations where controllers are running in HA mode (multiple controllers) this
-// interface can attempt to try different controller during outages or partitioning.
+// interface can attempt to try a different controller during outages or partitioning.
 type ClientTransportPool interface {
 	runtime.ClientTransport
 
@@ -265,6 +266,5 @@ func selectAndRemoveRandom[T any](slice []T, zero T) (selected T, modifiedSlice
 	rng := rand.New(rand.NewPCG(seed, seed))
 	index := rng.IntN(len(slice))
 	selected = slice[index]
-	modifiedSlice = append(slice[:index], slice[index+1:]...)
-	return selected, modifiedSlice
+	return selected, slices.Delete(slice, index, index+1)
 }

xgress/circuit_inspections.go

@@ -62,10 +62,12 @@ type InspectDetail struct {
 	Goroutines            []string          `json:"goroutines"`
 	Sequence              uint64            `json:"sequence"`
 	Flags                 string            `json:"flags"`
+	LastSizeSent          uint32            `json:"lastSizeSent"`
 }
 
 type SendBufferDetail struct {
 	WindowSize            uint32  `json:"windowSize"`
+	QueuedPayloadCount    int     `json:"queuedPayloadCount"`
 	LinkSendBufferSize    uint32  `json:"linkSendBufferSize"`
 	LinkRecvBufferSize    uint32  `json:"linkRecvBufferSize"`
 	Accumulator           uint32  `json:"accumulator"`
@@ -85,7 +87,6 @@ type SendBufferDetail struct {
 type RecvBufferDetail struct {
 	Size           uint32 `json:"size"`
 	PayloadCount   uint32 `json:"payloadCount"`
-	LastSizeSent   uint32 `json:"lastSizeSent"`
 	Sequence       int32  `json:"sequence"`
 	MaxSequence    int32  `json:"maxSequence"`
 	NextPayload    string `json:"nextPayload"`

xgress/link_receive_buffer.go

@@ -21,22 +21,24 @@ import (
 	"github.com/emirpasic/gods/trees/btree"
 	"github.com/emirpasic/gods/utils"
 	"github.com/michaelquigley/pfxlog"
+	"sync"
 	"sync/atomic"
-	"time"
 )
 
 type LinkReceiveBuffer struct {
-	tree               *btree.Tree
-	sequence           int32
-	maxSequence        int32
-	size               uint32
-	lastBufferSizeSent uint32
+	sync.Mutex
+	tree        *btree.Tree
+	sequence    int32
+	maxSequence int32
+	size        uint32
+	txQueue     chan *Payload
 }
 
-func NewLinkReceiveBuffer() *LinkReceiveBuffer {
+func NewLinkReceiveBuffer(txQueueSize int32) *LinkReceiveBuffer {
 	return &LinkReceiveBuffer{
 		tree:     btree.NewWith(10240, utils.Int32Comparator),
 		sequence: -1,
+		txQueue:  make(chan *Payload, txQueueSize),
 	}
 }
 
@@ -45,6 +47,9 @@ func (buffer *LinkReceiveBuffer) Size() uint32 {
 }
 
 func (buffer *LinkReceiveBuffer) ReceiveUnordered(x *Xgress, payload *Payload, maxSize uint32) bool {
+	buffer.Lock()
+	defer buffer.Unlock()
+
 	if payload.GetSequence() <= buffer.sequence {
 		x.dataPlane.GetMetrics().MarkDuplicatePayload()
 		return true
@@ -67,47 +72,56 @@ func (buffer *LinkReceiveBuffer) ReceiveUnordered(x *Xgress, payload *Payload, m
 	} else {
 		x.dataPlane.GetMetrics().MarkDuplicatePayload()
 	}
+
+	buffer.queueNext()
+
 	return true
 }
 
-func (buffer *LinkReceiveBuffer) PeekHead() *Payload {
+func (buffer *LinkReceiveBuffer) queueNext() {
 	if val := buffer.tree.LeftValue(); val != nil {
 		payload := val.(*Payload)
 		if payload.Sequence == buffer.sequence+1 {
-			return payload
+			select {
+			case buffer.txQueue <- payload:
+				buffer.tree.Remove(payload.Sequence)
+				buffer.sequence = payload.Sequence
+			default:
+			}
 		}
 	}
-	return nil
 }
 
-func (buffer *LinkReceiveBuffer) Remove(payload *Payload) {
-	buffer.tree.Remove(payload.Sequence)
-	buffer.sequence = payload.Sequence
-}
+func (buffer *LinkReceiveBuffer) NextPayload(closeNotify <-chan struct{}) *Payload {
+	select {
+	case payload := <-buffer.txQueue:
+		return payload
+	default:
+	}
 
-func (buffer *LinkReceiveBuffer) getLastBufferSizeSent() uint32 {
-	return atomic.LoadUint32(&buffer.lastBufferSizeSent)
-}
+	buffer.Lock()
+	buffer.queueNext()
+	buffer.Unlock()
 
-func (buffer *LinkReceiveBuffer) Inspect(x *Xgress) *RecvBufferDetail {
-	timeout := time.After(100 * time.Millisecond)
-	inspectEvent := &receiveBufferInspectEvent{
-		buffer:         buffer,
-		notifyComplete: make(chan *RecvBufferDetail, 1),
+	select {
+	case payload := <-buffer.txQueue:
+		return payload
+	case <-closeNotify:
 	}
 
-	if x.dataPlane.GetPayloadIngester().inspect(inspectEvent, timeout) {
-		select {
-		case result := <-inspectEvent.notifyComplete:
-			return result
-		case <-timeout:
-		}
+	// closed, check if there's anything pending in the queue
+	select {
+	case payload := <-buffer.txQueue:
+		return payload
+	default:
+		return nil
 	}
-
-	return buffer.inspectIncomplete()
 }
 
-func (buffer *LinkReceiveBuffer) inspectComplete() *RecvBufferDetail {
+func (buffer *LinkReceiveBuffer) Inspect() *RecvBufferDetail {
+	buffer.Lock()
+	defer buffer.Unlock()
+
 	nextPayload := "none"
 	if head := buffer.tree.LeftValue(); head != nil {
 		payload := head.(*Payload)
@@ -117,31 +131,9 @@ func (buffer *LinkReceiveBuffer) inspectComplete() *RecvBufferDetail {
 	return &RecvBufferDetail{
 		Size:           buffer.Size(),
 		PayloadCount:   uint32(buffer.tree.Size()),
-		LastSizeSent:   buffer.getLastBufferSizeSent(),
 		Sequence:       buffer.sequence,
 		MaxSequence:    buffer.maxSequence,
 		NextPayload:    nextPayload,
 		AcquiredSafely: true,
 	}
 }
-
-func (buffer *LinkReceiveBuffer) inspectIncomplete() *RecvBufferDetail {
-	return &RecvBufferDetail{
-		Size:           buffer.Size(),
-		LastSizeSent:   buffer.getLastBufferSizeSent(),
-		Sequence:       buffer.sequence,
-		MaxSequence:    buffer.maxSequence,
-		NextPayload:    "unsafe to check",
-		AcquiredSafely: false,
-	}
-}
-
-type receiveBufferInspectEvent struct {
-	buffer         *LinkReceiveBuffer
-	notifyComplete chan *RecvBufferDetail
-}
-
-func (self *receiveBufferInspectEvent) handle() {
-	result := self.buffer.inspectComplete()
-	self.notifyComplete <- result
-}

xgress/payload_ingester.go

@@ -1,60 +1,56 @@
 package xgress
 
-import "time"
-
-type payloadEntry struct {
-	payload *Payload
-	x       *Xgress
-}
+import (
+	"fmt"
+	"github.com/michaelquigley/pfxlog"
+	"github.com/openziti/foundation/v2/goroutines"
+	"github.com/sirupsen/logrus"
+	"runtime/debug"
+	"time"
+)
 
 type PayloadIngester struct {
-	payloadIngest         chan *payloadEntry
-	payloadSendReq        chan *Xgress
-	receiveBufferInspects chan *receiveBufferInspectEvent
-	closeNotify           <-chan struct{}
+	pool goroutines.Pool
 }
 
 func NewPayloadIngester(closeNotify <-chan struct{}) *PayloadIngester {
-	pi := &PayloadIngester{
-		payloadIngest:         make(chan *payloadEntry, 16),
-		payloadSendReq:        make(chan *Xgress, 16),
-		receiveBufferInspects: make(chan *receiveBufferInspectEvent, 4),
-		closeNotify:           closeNotify,
+	return NewPayloadIngesterWithConfig(1, closeNotify)
+}
+
+func NewPayloadIngesterWithConfig(maxWorkers uint32, closeNotify <-chan struct{}) *PayloadIngester {
+	if maxWorkers < 1 {
+		maxWorkers = 1
+	}
+	poolConfig := goroutines.PoolConfig{
+		QueueSize:   uint32(64),
+		MinWorkers:  1,
+		MaxWorkers:  maxWorkers,
+		IdleTime:    30 * time.Second,
+		CloseNotify: closeNotify,
+		PanicHandler: func(err interface{}) {
+			pfxlog.Logger().WithField(logrus.ErrorKey, err).WithField("backtrace", string(debug.Stack())).Error("panic during payload ingest")
+		},
+		WorkerFunction: payloadIngesterWorker,
 	}
 
-	go pi.run()
+	pool, err := goroutines.NewPool(poolConfig)
+	if err != nil {
+		panic(fmt.Errorf("error creating payload ingester handler pool (%w)", err))
+	}
+
+	pi := &PayloadIngester{
+		pool: pool,
+	}
 
 	return pi
 }
 
-func (self *PayloadIngester) inspect(evt *receiveBufferInspectEvent, timeout <-chan time.Time) bool {
-	select {
-	case self.receiveBufferInspects <- evt:
-		return true
-	case <-self.closeNotify:
-	case <-timeout:
-	}
-	return false
+func payloadIngesterWorker(_ uint32, f func()) {
+	f()
 }
 
 func (self *PayloadIngester) ingest(payload *Payload, x *Xgress) {
-	self.payloadIngest <- &payloadEntry{
-		payload: payload,
-		x:       x,
-	}
-}
-
-func (self *PayloadIngester) run() {
-	for {
-		select {
-		case payloadEntry := <-self.payloadIngest:
-			payloadEntry.x.acceptPayload(payloadEntry.payload)
-		case x := <-self.payloadSendReq:
-			x.queueSends()
-		case evt := <-self.receiveBufferInspects:
-			evt.handle()
-		case <-self.closeNotify:
-			return
-		}
-	}
+	_ = self.pool.Queue(func() {
+		x.acceptPayload(payload)
+	})
 }