[VOL-5291] On demand statistics for ONU and OLT
Change-Id: I4850bb0f0d2235122cb0c1bcf835b3672bb34436
Signed-off-by: Akash Reddy Kankanala <akash.kankanala@radisys.com>
diff --git a/vendor/google.golang.org/grpc/idle.go b/vendor/google.golang.org/grpc/idle.go
new file mode 100644
index 0000000..dc3dc72
--- /dev/null
+++ b/vendor/google.golang.org/grpc/idle.go
@@ -0,0 +1,287 @@
+/*
+ *
+ * Copyright 2023 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package grpc
+
+import (
+ "fmt"
+ "math"
+ "sync"
+ "sync/atomic"
+ "time"
+)
+
+// For overriding in unit tests.
+var timeAfterFunc = func(d time.Duration, f func()) *time.Timer {
+ return time.AfterFunc(d, f)
+}
+
+// idlenessEnforcer is the functionality provided by grpc.ClientConn to enter
+// and exit from idle mode.
+type idlenessEnforcer interface {
+ exitIdleMode() error
+ enterIdleMode() error
+}
+
+// idlenessManager defines the functionality required to track RPC activity on a
+// channel.
+type idlenessManager interface {
+ onCallBegin() error
+ onCallEnd()
+ close()
+}
+
+type noopIdlenessManager struct{}
+
+func (noopIdlenessManager) onCallBegin() error { return nil }
+func (noopIdlenessManager) onCallEnd() {}
+func (noopIdlenessManager) close() {}
+
+// idlenessManagerImpl implements the idlenessManager interface. It uses atomic
+// operations to synchronize access to shared state and a mutex to guarantee
+// mutual exclusion in a critical section.
+type idlenessManagerImpl struct {
+ // State accessed atomically.
+ lastCallEndTime int64 // Unix timestamp in nanos; time when the most recent RPC completed.
+ activeCallsCount int32 // Count of active RPCs; -math.MaxInt32 means channel is idle or is trying to get there.
+ activeSinceLastTimerCheck int32 // Boolean; True if there was an RPC since the last timer callback.
+ closed int32 // Boolean; True when the manager is closed.
+
+ // Can be accessed without atomics or mutex since these are set at creation
+ // time and read-only after that.
+ enforcer idlenessEnforcer // Functionality provided by grpc.ClientConn.
+ timeout int64 // Idle timeout duration nanos stored as an int64.
+
+ // idleMu is used to guarantee mutual exclusion in two scenarios:
+ // - Opposing intentions:
+ // - a: Idle timeout has fired and handleIdleTimeout() is trying to put
+ // the channel in idle mode because the channel has been inactive.
+ // - b: At the same time an RPC is made on the channel, and onCallBegin()
+ // is trying to prevent the channel from going idle.
+ // - Competing intentions:
+ // - The channel is in idle mode and there are multiple RPCs starting at
+ // the same time, all trying to move the channel out of idle. Only one
+ // of them should succeed in doing so, while the other RPCs should
+ // piggyback on the first one and be successfully handled.
+ idleMu sync.RWMutex
+ actuallyIdle bool
+ timer *time.Timer
+}
+
+// newIdlenessManager creates a new idleness manager implementation for the
+// given idle timeout.
+func newIdlenessManager(enforcer idlenessEnforcer, idleTimeout time.Duration) idlenessManager {
+ if idleTimeout == 0 {
+ return noopIdlenessManager{}
+ }
+
+ i := &idlenessManagerImpl{
+ enforcer: enforcer,
+ timeout: int64(idleTimeout),
+ }
+ i.timer = timeAfterFunc(idleTimeout, i.handleIdleTimeout)
+ return i
+}
+
+// resetIdleTimer resets the idle timer to the given duration. This method
+// should only be called from the timer callback.
+func (i *idlenessManagerImpl) resetIdleTimer(d time.Duration) {
+ i.idleMu.Lock()
+ defer i.idleMu.Unlock()
+
+ if i.timer == nil {
+ // Only close sets timer to nil. We are done.
+ return
+ }
+
+ // It is safe to ignore the return value from Reset() because this method is
+ // only ever called from the timer callback, which means the timer has
+ // already fired.
+ i.timer.Reset(d)
+}
+
+// handleIdleTimeout is the timer callback that is invoked upon expiry of the
+// configured idle timeout. The channel is considered inactive if there are no
+// ongoing calls and no RPC activity since the last time the timer fired.
+func (i *idlenessManagerImpl) handleIdleTimeout() {
+ if i.isClosed() {
+ return
+ }
+
+ if atomic.LoadInt32(&i.activeCallsCount) > 0 {
+ i.resetIdleTimer(time.Duration(i.timeout))
+ return
+ }
+
+ // There has been activity on the channel since we last got here. Reset the
+ // timer and return.
+ if atomic.LoadInt32(&i.activeSinceLastTimerCheck) == 1 {
+ // Set the timer to fire after a duration of idle timeout, calculated
+ // from the time the most recent RPC completed.
+ atomic.StoreInt32(&i.activeSinceLastTimerCheck, 0)
+ i.resetIdleTimer(time.Duration(atomic.LoadInt64(&i.lastCallEndTime) + i.timeout - time.Now().UnixNano()))
+ return
+ }
+
+ // This CAS operation is extremely likely to succeed given that there has
+ // been no activity since the last time we were here. Setting the
+ // activeCallsCount to -math.MaxInt32 indicates to onCallBegin() that the
+ // channel is either in idle mode or is trying to get there.
+ if !atomic.CompareAndSwapInt32(&i.activeCallsCount, 0, -math.MaxInt32) {
+ // This CAS operation can fail if an RPC started after we checked for
+ // activity at the top of this method, or one was ongoing from before
+ // the last time we were here. In both case, reset the timer and return.
+ i.resetIdleTimer(time.Duration(i.timeout))
+ return
+ }
+
+ // Now that we've set the active calls count to -math.MaxInt32, it's time to
+ // actually move to idle mode.
+ if i.tryEnterIdleMode() {
+ // Successfully entered idle mode. No timer needed until we exit idle.
+ return
+ }
+
+ // Failed to enter idle mode due to a concurrent RPC that kept the channel
+ // active, or because of an error from the channel. Undo the attempt to
+ // enter idle, and reset the timer to try again later.
+ atomic.AddInt32(&i.activeCallsCount, math.MaxInt32)
+ i.resetIdleTimer(time.Duration(i.timeout))
+}
+
+// tryEnterIdleMode instructs the channel to enter idle mode. But before
+// that, it performs a last minute check to ensure that no new RPC has come in,
+// making the channel active.
+//
+// Return value indicates whether or not the channel moved to idle mode.
+//
+// Holds idleMu which ensures mutual exclusion with exitIdleMode.
+func (i *idlenessManagerImpl) tryEnterIdleMode() bool {
+ i.idleMu.Lock()
+ defer i.idleMu.Unlock()
+
+ if atomic.LoadInt32(&i.activeCallsCount) != -math.MaxInt32 {
+ // We raced and lost to a new RPC. Very rare, but stop entering idle.
+ return false
+ }
+ if atomic.LoadInt32(&i.activeSinceLastTimerCheck) == 1 {
+ // An very short RPC could have come in (and also finished) after we
+ // checked for calls count and activity in handleIdleTimeout(), but
+ // before the CAS operation. So, we need to check for activity again.
+ return false
+ }
+
+ // No new RPCs have come in since we last set the active calls count value
+ // -math.MaxInt32 in the timer callback. And since we have the lock, it is
+ // safe to enter idle mode now.
+ if err := i.enforcer.enterIdleMode(); err != nil {
+ logger.Errorf("Failed to enter idle mode: %v", err)
+ return false
+ }
+
+ // Successfully entered idle mode.
+ i.actuallyIdle = true
+ return true
+}
+
+// onCallBegin is invoked at the start of every RPC.
+func (i *idlenessManagerImpl) onCallBegin() error {
+ if i.isClosed() {
+ return nil
+ }
+
+ if atomic.AddInt32(&i.activeCallsCount, 1) > 0 {
+ // Channel is not idle now. Set the activity bit and allow the call.
+ atomic.StoreInt32(&i.activeSinceLastTimerCheck, 1)
+ return nil
+ }
+
+ // Channel is either in idle mode or is in the process of moving to idle
+ // mode. Attempt to exit idle mode to allow this RPC.
+ if err := i.exitIdleMode(); err != nil {
+ // Undo the increment to calls count, and return an error causing the
+ // RPC to fail.
+ atomic.AddInt32(&i.activeCallsCount, -1)
+ return err
+ }
+
+ atomic.StoreInt32(&i.activeSinceLastTimerCheck, 1)
+ return nil
+}
+
+// exitIdleMode instructs the channel to exit idle mode.
+//
+// Holds idleMu which ensures mutual exclusion with tryEnterIdleMode.
+func (i *idlenessManagerImpl) exitIdleMode() error {
+ i.idleMu.Lock()
+ defer i.idleMu.Unlock()
+
+ if !i.actuallyIdle {
+ // This can happen in two scenarios:
+ // - handleIdleTimeout() set the calls count to -math.MaxInt32 and called
+ // tryEnterIdleMode(). But before the latter could grab the lock, an RPC
+ // came in and onCallBegin() noticed that the calls count is negative.
+ // - Channel is in idle mode, and multiple new RPCs come in at the same
+ // time, all of them notice a negative calls count in onCallBegin and get
+ // here. The first one to get the lock would got the channel to exit idle.
+ //
+ // Either way, nothing to do here.
+ return nil
+ }
+
+ if err := i.enforcer.exitIdleMode(); err != nil {
+ return fmt.Errorf("channel failed to exit idle mode: %v", err)
+ }
+
+ // Undo the idle entry process. This also respects any new RPC attempts.
+ atomic.AddInt32(&i.activeCallsCount, math.MaxInt32)
+ i.actuallyIdle = false
+
+ // Start a new timer to fire after the configured idle timeout.
+ i.timer = timeAfterFunc(time.Duration(i.timeout), i.handleIdleTimeout)
+ return nil
+}
+
+// onCallEnd is invoked at the end of every RPC.
+func (i *idlenessManagerImpl) onCallEnd() {
+ if i.isClosed() {
+ return
+ }
+
+ // Record the time at which the most recent call finished.
+ atomic.StoreInt64(&i.lastCallEndTime, time.Now().UnixNano())
+
+ // Decrement the active calls count. This count can temporarily go negative
+ // when the timer callback is in the process of moving the channel to idle
+ // mode, but one or more RPCs come in and complete before the timer callback
+ // can get done with the process of moving to idle mode.
+ atomic.AddInt32(&i.activeCallsCount, -1)
+}
+
+func (i *idlenessManagerImpl) isClosed() bool {
+ return atomic.LoadInt32(&i.closed) == 1
+}
+
+func (i *idlenessManagerImpl) close() {
+ atomic.StoreInt32(&i.closed, 1)
+
+ i.idleMu.Lock()
+ i.timer.Stop()
+ i.timer = nil
+ i.idleMu.Unlock()
+}