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

Shutdown ack handler on link close. Remove unnecessary timers

Browse Source
This commit is contained in:
Paul Lorenz
2021-01-18 11:29:36 -05:00
parent 60f7439a33
commit 5f9b28cfc1
8 changed files with 44 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
router/handler_link/accepter.go
+4 -2
View File
@@ -21,12 +21,14 @@ func NewChannelAccepter(c xgress.CtrlChannel, f *forwarder.Forwarder, fo *forwar
}
func (self *channelAccepter) AcceptChannel(xlink xlink.Xlink, ch channel2.Channel, trackLatency bool) error {
closeNotify := make(chan struct{})
ch.SetLogicalName("l/" + xlink.Id().Token)
ch.SetUserData(xlink.Id().Token)
ch.AddCloseHandler(newCloseHandler(xlink, self.ctrl, self.forwarder))
ch.AddCloseHandler(newCloseHandler(xlink, self.ctrl, self.forwarder, closeNotify))
ch.AddErrorHandler(newErrorHandler(xlink, self.ctrl))
ch.AddReceiveHandler(newPayloadHandler(xlink, self.ctrl, self.forwarder))
ch.AddReceiveHandler(newQueuingAckHandler(xlink, self.ctrl, self.forwarder))
ch.AddReceiveHandler(newQueuingAckHandler(xlink, self.ctrl, self.forwarder, closeNotify))
ch.AddReceiveHandler(&channel2.LatencyHandler{})
ch.AddPeekHandler(metrics2.NewChannelPeekHandler(xlink.Id().Token, self.forwarder.MetricsRegistry()))
ch.AddPeekHandler(trace.NewChannelPeekHandler(xlink.Id(), ch, self.forwarder.TraceController(), trace.NewChannelSink(self.ctrl.Channel())))
router/handler_link/ack.go
-6
View File
@@ -22,15 +22,12 @@ import (
"github.com/openziti/fabric/router/xgress"
"github.com/openziti/fabric/router/xlink"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/metrics"
"time"
)
type ackHandler struct {
link xlink.Xlink
ctrl xgress.CtrlChannel
forwarder *forwarder.Forwarder
timer metrics.Timer
}
func newAckHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *forwarder.Forwarder) *ackHandler {
@@ -38,7 +35,6 @@ func newAckHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *forward
link: link,
ctrl: ctrl,
forwarder: forwarder,
timer: forwarder.MetricsRegistry().Timer("xgress.ack.handle_time"),
}
}
@@ -47,7 +43,6 @@ func (self *ackHandler) ContentType() int32 {
}
func (self *ackHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
start := time.Now()
log := pfxlog.ContextLogger(ch.Label())
if ack, err := xgress.UnmarshallAcknowledgement(msg); err == nil {
@@ -57,5 +52,4 @@ func (self *ackHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel
} else {
log.Errorf("unexpected error (%v)", err)
}
self.timer.UpdateSince(start)
}
router/handler_link/close.go
+19 -5
View File
@@ -24,19 +24,33 @@ import (
"github.com/openziti/fabric/router/xgress"
"github.com/openziti/fabric/router/xlink"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/util/concurrenz"
)
type closeHandler struct {
link xlink.Xlink
ctrl xgress.CtrlChannel
forwarder *forwarder.Forwarder
link xlink.Xlink
ctrl xgress.CtrlChannel
forwarder *forwarder.Forwarder
closeNotify chan struct{}
closed concurrenz.AtomicBoolean
}
func newCloseHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *forwarder.Forwarder) *closeHandler {
return &closeHandler{link: link, ctrl: ctrl, forwarder: forwarder}
func newCloseHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *forwarder.Forwarder, closeNotify chan struct{}) *closeHandler {
return &closeHandler{
link: link,
ctrl: ctrl,
forwarder: forwarder,
closeNotify: closeNotify,
}
}
func (self *closeHandler) HandleClose(ch channel2.Channel) {
defer func() {
if self.closed.CompareAndSwap(false, true) {
close(self.closeNotify)
}
}()
log := pfxlog.ContextLogger(ch.Label())
log.Info("link closed")
router/handler_link/payload.go
-6
View File
@@ -22,15 +22,12 @@ import (
"github.com/openziti/fabric/router/xgress"
"github.com/openziti/fabric/router/xlink"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/metrics"
"time"
)
type payloadHandler struct {
link xlink.Xlink
ctrl xgress.CtrlChannel
forwarder *forwarder.Forwarder
timer metrics.Timer
}
func newPayloadHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *forwarder.Forwarder) *payloadHandler {
@@ -38,7 +35,6 @@ func newPayloadHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *for
link: link,
ctrl: ctrl,
forwarder: forwarder,
timer: forwarder.MetricsRegistry().Timer("xgress.payload.handle_time"),
}
}
@@ -47,7 +43,6 @@ func (self *payloadHandler) ContentType() int32 {
}
func (self *payloadHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
start := time.Now()
log := pfxlog.ContextLogger(ch.Label())
payload, err := xgress.UnmarshallPayload(msg)
@@ -61,5 +56,4 @@ func (self *payloadHandler) HandleReceive(msg *channel2.Message, ch channel2.Cha
} else {
log.Errorf("unexpected error (%v)", err)
}
self.timer.UpdateSince(start)
}
router/handler_link/queue_ack.go
+21 -19
View File
@@ -23,31 +23,29 @@ import (
"github.com/openziti/fabric/router/xgress"
"github.com/openziti/fabric/router/xlink"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/metrics"
"sync/atomic"
"time"
)
type queuingAckHandler struct {
link xlink.Xlink
ctrl xgress.CtrlChannel
forwarder *forwarder.Forwarder
timer metrics.Timer
acks *deque.Deque
ackIngest chan *xgress.Acknowledgement
ackForward chan *xgress.Acknowledgement
acksQueueSize int64
closeNotify <-chan struct{}
}
func newQueuingAckHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *forwarder.Forwarder) *queuingAckHandler {
func newQueuingAckHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *forwarder.Forwarder, closeNotify <-chan struct{}) *queuingAckHandler {
result := &queuingAckHandler{
link: link,
ctrl: ctrl,
forwarder: forwarder,
timer: forwarder.MetricsRegistry().Timer("xgress.ack.handle_time"),
acks: deque.New(),
ackIngest: make(chan *xgress.Acknowledgement, 16),
ackForward: make(chan *xgress.Acknowledgement, 1),
link: link,
ctrl: ctrl,
forwarder: forwarder,
acks: deque.New(),
ackIngest: make(chan *xgress.Acknowledgement, 16),
ackForward: make(chan *xgress.Acknowledgement, 1),
closeNotify: closeNotify,
}
go result.ackIngester()
@@ -65,7 +63,6 @@ func (self *queuingAckHandler) ContentType() int32 {
}
func (self *queuingAckHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
start := time.Now()
log := pfxlog.ContextLogger(ch.Label())
if ack, err := xgress.UnmarshallAcknowledgement(msg); err == nil {
@@ -73,7 +70,6 @@ func (self *queuingAckHandler) HandleReceive(msg *channel2.Message, ch channel2.
} else {
log.Errorf("unexpected error (%v)", err)
}
self.timer.UpdateSince(start)
}
func (self *queuingAckHandler) ackIngester() {
@@ -89,6 +85,8 @@ func (self *queuingAckHandler) ackIngester() {
select {
case ack := <-self.ackIngest:
self.acks.PushBack(ack)
case <-self.closeNotify:
return
}
} else {
select {
@@ -96,6 +94,8 @@ func (self *queuingAckHandler) ackIngester() {
self.acks.PushBack(ack)
case self.ackForward <- next:
next = nil
case <-self.closeNotify:
return
}
}
atomic.StoreInt64(&self.acksQueueSize, int64(self.acks.Len()))
@@ -104,12 +104,14 @@ func (self *queuingAckHandler) ackIngester() {
func (self *queuingAckHandler) ackForwarder() {
logger := pfxlog.Logger()
for ack := range self.ackForward {
now := time.Now()
if err := self.forwarder.ForwardAcknowledgement(xgress.Address(self.link.Id().Token), ack); err != nil {
logger.WithError(err).Debugf("unable to forward acknowledgement (%v)", err)
} else {
self.timer.UpdateSince(now)
for {
select {
case ack := <-self.ackForward:
if err := self.forwarder.ForwardAcknowledgement(xgress.Address(self.link.Id().Token), ack); err != nil {
logger.WithError(err).Debugf("unable to forward acknowledgement (%v)", err)
}
case <-self.closeNotify:
return
}
}
}
router/xgress/acker.go
-3
View File
@@ -5,7 +5,6 @@ import (
"github.com/michaelquigley/pfxlog"
"github.com/openziti/foundation/metrics"
"sync/atomic"
"time"
)
var acker *Acker
@@ -89,12 +88,10 @@ func (acker *Acker) ackSender() {
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:
router/xgress/metrics.go
-6
View File
@@ -18,9 +18,6 @@ var localRecvBufferSizeBytesHistogram metrics.Histogram
var localRecvBufferSizeMsgsHistogram metrics.Histogram
var remoteRecvBufferSizeHistogram metrics.Histogram
var payloadWriteTimer metrics.Timer
var ackWriteTimer metrics.Timer
var payloadBufferTimer metrics.Timer
var payloadRelayTimer metrics.Timer
var duplicateAcksMeter metrics.Meter
var txWindowSize metrics.Histogram
@@ -42,9 +39,6 @@ func InitMetrics(registry metrics.UsageRegistry) {
localRecvBufferSizeMsgsHistogram = registry.Histogram("xgress.local.rx_buffer_msgs_size")
remoteRecvBufferSizeHistogram = registry.Histogram("xgress.remote.rx_buffer_size")
payloadWriteTimer = registry.Timer("xgress.tx_write_time")
ackWriteTimer = registry.Timer("xgress.ack_write_time")
payloadBufferTimer = registry.Timer("xgress.payload_buffer_time")
payloadRelayTimer = registry.Timer("xgress.payload_relay_time")
duplicateAcksMeter = registry.Meter("xgress.ack_duplicates")
txWindowSize = registry.Histogram("xgress.tx_window_size")
router/xgress/xgress.go
-4
View File
@@ -256,16 +256,12 @@ func (self *Xgress) payloadIngester(payload *Payload) {
go self.rx()
}
start := time.Now()
if !self.Options.RandomDrops || rand.Int31n(self.Options.Drop1InN) != 1 {
self.PayloadReceived(payload)
} else {
pfxlog.ContextLogger(self.Label()).WithFields(payload.GetLoggerFields()).Error("drop!")
}
next := time.Now()
payloadBufferTimer.Update(next.Sub(start))
self.queueSends()
payloadRelayTimer.UpdateSince(next)
}
func (self *Xgress) queueSends() {