gsadmin/ziti
1
0
Fork 0
mirror of https://github.com/openziti/ziti.git synced 2026-10-14 08:10:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Shutdown payload ingester, acker and retransmitter go-routines on router shutdown

Browse Source
This commit is contained in:
Paul Lorenz
2020-12-15 13:02:30 -05:00
parent ea4d1de71b
commit 1640669e13
6 changed files with 68 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
router/forwarder/forwarder.go
+4 -4
View File
@@ -49,7 +49,7 @@ type XgressDestination interface {
Label() string
}
func NewForwarder(metricsRegistry metrics.UsageRegistry, options *Options) *Forwarder {
func NewForwarder(metricsRegistry metrics.UsageRegistry, options *Options, closeNotify <-chan struct{}) *Forwarder {
forwarder := &Forwarder{
sessions: newSessionTable(),
destinations: newDestinationTable(),
@@ -58,9 +58,9 @@ func NewForwarder(metricsRegistry metrics.UsageRegistry, options *Options) *Forw
Options: options,
}
xgress.InitPayloadIngester()
xgress.InitAcker(forwarder, metricsRegistry)
xgress.InitRetransmitter(forwarder, metricsRegistry)
xgress.InitPayloadIngester(closeNotify)
xgress.InitAcker(forwarder, metricsRegistry, closeNotify)
xgress.InitRetransmitter(forwarder, metricsRegistry, closeNotify)
return forwarder
}
router/router.go
+5 -2
View File
@@ -81,11 +81,14 @@ func Create(config *Config, versionProvider common.VersionProvider) *Router {
metricsRegistry := metrics.NewUsageRegistryFromConfig(metricsConfig)
xgress.InitMetrics(metricsRegistry)
closeNotify := make(chan struct{})
fwd := forwarder.NewForwarder(metricsRegistry, config.Forwarder, closeNotify)
return &Router{
config: config,
forwarder: forwarder.NewForwarder(metricsRegistry, config.Forwarder),
forwarder: fwd,
metricsRegistry: metricsRegistry,
shutdownC: make(chan struct{}),
shutdownC: closeNotify,
eventDispatcher: eventDispatcher,
versionProvider: versionProvider,
}
router/xgress/acker.go
+26 -15
View File
@@ -10,8 +10,8 @@ import (
var acker *Acker
func InitAcker(forwarder PayloadBufferForwarder, metrics metrics.Registry) {
acker = NewAcker(forwarder, metrics)
func InitAcker(forwarder PayloadBufferForwarder, metrics metrics.Registry, closeNotify <-chan struct{}) {
acker = NewAcker(forwarder, metrics, closeNotify)
}
type ackEntry struct {
@@ -25,14 +25,16 @@ type Acker struct {
ackIngest chan *ackEntry
ackSend chan *ackEntry
acksQueueSize int64
closeNotify <-chan struct{}
}
func NewAcker(forwarder PayloadBufferForwarder, metrics metrics.Registry) *Acker {
func NewAcker(forwarder PayloadBufferForwarder, metrics metrics.Registry, closeNotify <-chan struct{}) *Acker {
result := &Acker{
forwarder: forwarder,
acks: deque.New(),
ackIngest: make(chan *ackEntry, 16),
ackSend: make(chan *ackEntry, 1),
forwarder: forwarder,
acks: deque.New(),
ackIngest: make(chan *ackEntry, 16),
ackSend: make(chan *ackEntry, 1),
closeNotify: closeNotify,
}
go result.ackIngester()
@@ -65,6 +67,8 @@ func (acker *Acker) ackIngester() {
select {
case ack := <-acker.ackIngest:
acker.acks.PushBack(ack)
case <-acker.closeNotify:
return
}
} else {
select {
@@ -72,6 +76,8 @@ func (acker *Acker) ackIngester() {
acker.acks.PushBack(ack)
case acker.ackSend <- next:
next = nil
case <-acker.closeNotify:
return
}
}
atomic.StoreInt64(&acker.acksQueueSize, int64(acker.acks.Len()))
@@ -80,14 +86,19 @@ func (acker *Acker) ackIngester() {
func (acker *Acker) ackSender() {
logger := pfxlog.Logger()
for nextAck := range acker.ackSend {
now := time.Now()
if err := acker.forwarder.ForwardAcknowledgement(nextAck.Address, nextAck.Acknowledgement); err != nil {
logger.WithError(err).Debugf("unexpected error while sending ack from %v", nextAck.Address)
ackFailures.Mark(1)
} else {
ackWriteTimer.UpdateSince(now)
ackTxMeter.Mark(1)
for {
select {
case nextAck := <-acker.ackSend:
now := time.Now()
if err := acker.forwarder.ForwardAcknowledgement(nextAck.Address, nextAck.Acknowledgement); err != nil {
logger.WithError(err).Debugf("unexpected error while sending ack from %v", nextAck.Address)
ackFailures.Mark(1)
} else {
ackWriteTimer.UpdateSince(now)
ackTxMeter.Mark(1)
}
case <-acker.closeNotify:
return
}
}
}
router/xgress/ordering_test.go
+3 -2
View File
@@ -52,9 +52,10 @@ func (n noopForwarder) ForwardAcknowledgement(srcAddr Address, acknowledgement *
func Test_Ordering(t *testing.T) {
metricsRegistry := metrics.NewUsageRegistry("test", map[string]string{}, time.Minute, noopMetricsHandler{})
InitPayloadIngester()
closeNotify := make(chan struct{})
InitPayloadIngester(closeNotify)
InitMetrics(metricsRegistry)
InitAcker(&noopForwarder{}, metricsRegistry)
InitAcker(&noopForwarder{}, metricsRegistry, closeNotify)
conn := &testConn{
ch: make(chan uint64, 1),
router/xgress/payload_ingester.go
+7 -3
View File
@@ -2,8 +2,8 @@ package xgress
var payloadIngester *PayloadIngester
func InitPayloadIngester() {
payloadIngester = NewPayloadIngester()
func InitPayloadIngester(closeNotify <-chan struct{}) {
payloadIngester = NewPayloadIngester(closeNotify)
}
type payloadEntry struct {
@@ -14,12 +14,14 @@ type payloadEntry struct {
type PayloadIngester struct {
payloadIngest chan *payloadEntry
payloadSendReq chan *Xgress
closeNotify <-chan struct{}
}
func NewPayloadIngester() *PayloadIngester {
func NewPayloadIngester(closeNotify <-chan struct{}) *PayloadIngester {
pi := &PayloadIngester{
payloadIngest: make(chan *payloadEntry, 16),
payloadSendReq: make(chan *Xgress, 16),
closeNotify: closeNotify,
}
go pi.run()
@@ -41,6 +43,8 @@ func (payloadIngester *PayloadIngester) run() {
payloadEntry.x.payloadIngester(payloadEntry.payload)
case x := <-payloadIngester.payloadSendReq:
x.queueSends()
case <-payloadIngester.closeNotify:
return
}
}
}
router/xgress/retransmitter.go
+23 -12
View File
@@ -8,8 +8,8 @@ import (
var retransmitter *Retransmitter
func InitRetransmitter(forwarder PayloadBufferForwarder, metrics metrics.Registry) {
retransmitter = NewRetransmitter(forwarder, metrics)
func InitRetransmitter(forwarder PayloadBufferForwarder, metrics metrics.Registry, closeNotify <-chan struct{}) {
retransmitter = NewRetransmitter(forwarder, metrics, closeNotify)
}
type Retransmitter struct {
@@ -19,13 +19,15 @@ type Retransmitter struct {
retransmitIngest chan *txPayload
retransmitSend chan *txPayload
retransmitsQueueSize int64
closeNotify <-chan struct{}
}
func NewRetransmitter(forwarder PayloadBufferForwarder, metrics metrics.Registry) *Retransmitter {
func NewRetransmitter(forwarder PayloadBufferForwarder, metrics metrics.Registry, closeNotify <-chan struct{}) *Retransmitter {
ctrl := &Retransmitter{
forwarder: forwarder,
retransmitIngest: make(chan *txPayload, 16),
retransmitSend: make(chan *txPayload, 1),
closeNotify: closeNotify,
}
go ctrl.retransmitIngester()
@@ -107,6 +109,8 @@ func (retransmitter *Retransmitter) retransmitIngester() {
select {
case retransmit := <-retransmitter.retransmitIngest:
retransmitter.acceptRetransmit(retransmit)
case <-retransmitter.closeNotify:
return
}
} else {
select {
@@ -114,6 +118,8 @@ func (retransmitter *Retransmitter) retransmitIngester() {
retransmitter.acceptRetransmit(retransmit)
case retransmitter.retransmitSend <- next:
next = nil
case <-retransmitter.closeNotify:
return
}
}
}
@@ -129,16 +135,21 @@ func (retransmitter *Retransmitter) acceptRetransmit(txp *txPayload) {
func (retransmitter *Retransmitter) retransmitSender() {
logger := pfxlog.Logger()
for retransmit := range retransmitter.retransmitSend {
if !retransmit.isAcked() {
if err := retransmitter.forwarder.ForwardPayload(retransmit.x.address, retransmit.payload); err != nil {
logger.WithError(err).Errorf("unexpected error while retransmitting payload from %v", retransmit.x.address)
retransmissionFailures.Mark(1)
} else {
retransmit.markSent()
retransmissions.Mark(1)
for {
select {
case retransmit := <-retransmitter.retransmitSend:
if !retransmit.isAcked() {
if err := retransmitter.forwarder.ForwardPayload(retransmit.x.address, retransmit.payload); err != nil {
logger.WithError(err).Errorf("unexpected error while retransmitting payload from %v", retransmit.x.address)
retransmissionFailures.Mark(1)
} else {
retransmit.markSent()
retransmissions.Mark(1)
}
retransmit.dequeued()
}
retransmit.dequeued()
case <-retransmitter.closeNotify:
return
}
}
}