From 9e0c884820c5ac931bf35cbcd8838b357de8924f Mon Sep 17 00:00:00 2001 From: Ondrej Fabry Date: Tue, 20 Sep 2022 21:28:10 +0200 Subject: [PATCH 1/2] Disable default reply timeout This commit allows disabling reply timeout by setting it to 0 and sets this as a default value. Signed-off-by: Ondrej Fabry --- api/api.go | 5 +++-- core/channel.go | 12 +++++++++--- core/connection.go | 2 +- 3 files changed, 13 insertions(+), 6 deletions(-) diff --git a/api/api.go b/api/api.go index c4f069fa..9b1d4bbc 100644 --- a/api/api.go +++ b/api/api.go @@ -86,8 +86,9 @@ type Channel interface { // buffer is full, the notifications will not be delivered into it. SubscribeNotification(notifChan chan Message, event Message) (SubscriptionCtx, error) - // SetReplyTimeout sets the timeout for replies from VPP. It represents the maximum time the API waits for a reply - // from VPP before returning an error. + // SetReplyTimeout sets the timeout for replies from VPP. It represents the maximum time the client waits for a reply + // from VPP before returning a timeout error. Setting the reply timeout to 0 disables it. The initial reply timeout is + //set to the value of core.DefaultReplyTimeout. SetReplyTimeout(timeout time.Duration) // CheckCompatibility checks the compatiblity for the given messages. diff --git a/core/channel.go b/core/channel.go index b82dab61..93370797 100644 --- a/core/channel.go +++ b/core/channel.go @@ -259,6 +259,8 @@ func (sub *subscriptionCtx) Unsubscribe() error { return fmt.Errorf("subscription for %q not found", sub.event.GetMessageName()) } +const maxInt64 = 1<<63 - 1 + // receiveReplyInternal receives a reply from the reply channel into the provided msg structure. func (ch *Channel) receiveReplyInternal(msg api.Message, expSeqNum uint16) (lastReplyReceived bool, err error) { if msg == nil { @@ -276,7 +278,11 @@ func (ch *Channel) receiveReplyInternal(msg api.Message, expSeqNum uint16) (last } } - timer := time.NewTimer(ch.replyTimeout) + timeout := ch.replyTimeout + if timeout <= 0 { + timeout = maxInt64 + } + timer := time.NewTimer(timeout) for { select { // blocks until a reply comes to ReplyChan or until timeout expires @@ -295,8 +301,8 @@ func (ch *Channel) receiveReplyInternal(msg api.Message, expSeqNum uint16) (last log.WithFields(logrus.Fields{ "expSeqNum": expSeqNum, "channel": ch.id, - }).Debugf("timeout (%v) waiting for reply: %s", ch.replyTimeout, msg.GetMessageName()) - err = fmt.Errorf("no reply received within the timeout period %s", ch.replyTimeout) + }).Debugf("timeout (%v) waiting for reply: %s", timeout, msg.GetMessageName()) + err = fmt.Errorf("no reply received within the timeout period %s", timeout) return false, err } } diff --git a/core/connection.go b/core/connection.go index 5607de33..6d1a6435 100644 --- a/core/connection.go +++ b/core/connection.go @@ -47,7 +47,7 @@ var ( HealthCheckProbeInterval = time.Second // default health check probe interval HealthCheckReplyTimeout = time.Millisecond * 250 // timeout for reply to a health check probe HealthCheckThreshold = 2 // number of failed health checks until the error is reported - DefaultReplyTimeout = time.Second // default timeout for replies from VPP + DefaultReplyTimeout = time.Duration(0) // default timeout for replies from VPP is disabled ) // ConnectionState represents the current state of the connection to VPP. From a0d563981d95cb87e8feb68be6471136d056dab0 Mon Sep 17 00:00:00 2001 From: Ondrej Fabry Date: Mon, 3 Oct 2022 17:57:43 +0200 Subject: [PATCH 2/2] Add warning log for slow replies Signed-off-by: Ondrej Fabry --- core/channel.go | 13 ++++++++++--- core/connection.go | 9 ++++++--- 2 files changed, 16 insertions(+), 6 deletions(-) diff --git a/core/channel.go b/core/channel.go index 93370797..cc0c76ba 100644 --- a/core/channel.go +++ b/core/channel.go @@ -278,11 +278,13 @@ func (ch *Channel) receiveReplyInternal(msg api.Message, expSeqNum uint16) (last } } + slowReplyDur := WarnSlowReplyDuration timeout := ch.replyTimeout if timeout <= 0 { timeout = maxInt64 } - timer := time.NewTimer(timeout) + timeoutTimer := time.NewTimer(timeout) + slowTimer := time.NewTimer(slowReplyDur) for { select { // blocks until a reply comes to ReplyChan or until timeout expires @@ -296,8 +298,13 @@ func (ch *Channel) receiveReplyInternal(msg api.Message, expSeqNum uint16) (last continue } return lastReplyReceived, err - - case <-timer.C: + case <-slowTimer.C: + log.WithFields(logrus.Fields{ + "expSeqNum": expSeqNum, + "channel": ch.id, + }).Warnf("reply is taking too long (>%v): %v ", slowReplyDur, msg.GetMessageName()) + continue + case <-timeoutTimer.C: log.WithFields(logrus.Fields{ "expSeqNum": expSeqNum, "channel": ch.id, diff --git a/core/connection.go b/core/connection.go index 6d1a6435..701b979e 100644 --- a/core/connection.go +++ b/core/connection.go @@ -25,11 +25,10 @@ import ( logger "github.com/sirupsen/logrus" - "go.fd.io/govpp/core/genericpool" - "go.fd.io/govpp/adapter" "go.fd.io/govpp/api" "go.fd.io/govpp/codec" + "go.fd.io/govpp/core/genericpool" ) const ( @@ -47,7 +46,11 @@ var ( HealthCheckProbeInterval = time.Second // default health check probe interval HealthCheckReplyTimeout = time.Millisecond * 250 // timeout for reply to a health check probe HealthCheckThreshold = 2 // number of failed health checks until the error is reported - DefaultReplyTimeout = time.Duration(0) // default timeout for replies from VPP is disabled +) + +var ( + DefaultReplyTimeout = time.Duration(0) // default timeout for replies from VPP is disabled + WarnSlowReplyDuration = time.Second * 1 // duration of slow replies after which a warning is printed ) // ConnectionState represents the current state of the connection to VPP.