overload: create scaled timers via the dispatcher

include/envoy/api/api.h

@@ -30,6 +30,18 @@ class Api {
    */
   virtual Event::DispatcherPtr allocateDispatcher(const std::string& name) PURE;
 
+  /**
+   * Allocate a dispatcher.
+   * @param name the identity name for a dispatcher, e.g. "worker_2" or "main_thread".
+   *             This name will appear in per-handler/worker statistics, such as
+   *             "server.worker_2.watchdog_miss".
+   * @param scaled_timer_factory the factory to use when creating the scaled timer manager.
+   * @return Event::DispatcherPtr which is owned by the caller.
+   */
+  virtual Event::DispatcherPtr
+  allocateDispatcher(const std::string& name,
+                     const Event::ScaledRangeTimerManagerFactory& scaled_timer_factory) PURE;
+
   /**
    * Allocate a dispatcher.
    * @param name the identity name for a dispatcher, e.g. "worker_2" or "main_thread".

include/envoy/event/BUILD

@@ -19,11 +19,12 @@ envoy_cc_library(
     deps = [
         ":deferred_deletable",
         ":file_event_interface",
+        ":scaled_range_timer_manager_interface",
         ":schedulable_cb_interface",
         ":signal_interface",
+        ":timer_interface",
         "//include/envoy/common:scope_tracker_interface",
         "//include/envoy/common:time_interface",
-        "//include/envoy/event:timer_interface",
         "//include/envoy/filesystem:watcher_interface",
         "//include/envoy/network:connection_handler_interface",
         "//include/envoy/network:connection_interface",

include/envoy/event/dispatcher.h

@@ -9,6 +9,7 @@
 #include "envoy/common/scope_tracker.h"
 #include "envoy/common/time.h"
 #include "envoy/event/file_event.h"
+#include "envoy/event/scaled_range_timer_manager.h"
 #include "envoy/event/schedulable_cb.h"
 #include "envoy/event/signal.h"
 #include "envoy/event/timer.h"
@@ -77,6 +78,21 @@ class DispatcherBase {
    */
   virtual Event::TimerPtr createTimer(TimerCb cb) PURE;
 
+  /**
+   * Allocates a scaled timer. @see Timer for docs on how to use the timer.
+   * @param timer_type the type of timer to create.
+   * @param cb supplies the callback to invoke when the timer fires.
+   */
+  virtual Event::TimerPtr createScaledTimer(Event::ScaledRangeTimerManager::TimerType timer_type,
+                                            TimerCb cb) PURE;
+
+  /**
+   * Allocates a scaled timer. @see Timer for docs on how to use the timer.
+   * @param minimum the rule for computing the minimum value of the timer.
+   * @param cb supplies the callback to invoke when the timer fires.
+   */
+  virtual Event::TimerPtr createScaledTimer(ScaledTimerMinimum minimum, TimerCb cb) PURE;
+
   /**
    * Allocates a schedulable callback. @see SchedulableCallback for docs on how to use the wrapped
    * callback.

include/envoy/event/scaled_range_timer_manager.h

@@ -18,6 +18,17 @@ namespace Event {
  */
 class ScaledRangeTimerManager {
 public:
+  enum class TimerType {
+    // Timers created with this type will never be scaled. This should only be used for testing.
+    UnscaledRealTimerForTest,
+    // The amount of time an HTTP connection to a downstream client can remain idle (no streams).
+    // This corresponds to the HTTP_DOWNSTREAM_CONNECTION_IDLE TimerType in overload.proto.
+    HttpDownstreamIdleConnectionTimeout,
+    // The amount of time an HTTP stream from a downstream client can remain idle. This corresponds
+    // to the HTTP_DOWNSTREAM_STREAM_IDLE TimerType in overload.proto.
+    HttpDownstreamIdleStreamTimeout,
+  };
+
   virtual ~ScaledRangeTimerManager() = default;
 
   /**
@@ -28,6 +39,12 @@ class ScaledRangeTimerManager {
    */
   virtual TimerPtr createTimer(ScaledTimerMinimum minimum, TimerCb callback) PURE;
 
+  /**
+   * Creates a new timer backed by the manager using the provided timer type to
+   * determine the minimum.
+   */
+  virtual TimerPtr createTimer(TimerType timer_type, TimerCb callback) PURE;
+
   /**
    * Sets the scale factor for all timers created through this manager. The value should be between
    * 0 and 1, inclusive. The scale factor affects the amount of time timers spend in their target
@@ -41,5 +58,8 @@ class ScaledRangeTimerManager {
 
 using ScaledRangeTimerManagerPtr = std::unique_ptr<ScaledRangeTimerManager>;
 
+class Dispatcher;
+using ScaledRangeTimerManagerFactory = std::function<ScaledRangeTimerManagerPtr(Dispatcher&)>;
+
 } // namespace Event
 } // namespace Envoy

include/envoy/event/scaled_timer_minimum.h

@@ -13,26 +13,31 @@ namespace Event {
  * Describes a minimum timer value that is equal to a scale factor applied to the maximum.
  */
 struct ScaledMinimum {
-  explicit ScaledMinimum(UnitFloat scale_factor) : scale_factor_(scale_factor) {}
+  explicit constexpr ScaledMinimum(UnitFloat scale_factor) : scale_factor_(scale_factor) {}
+  inline bool operator==(const ScaledMinimum& other) const {
+    return other.scale_factor_.value() == scale_factor_.value();
+  }
   const UnitFloat scale_factor_;
 };
 
 /**
  * Describes a minimum timer value that is an absolute duration.
  */
 struct AbsoluteMinimum {
-  explicit AbsoluteMinimum(std::chrono::milliseconds value) : value_(value) {}
+  explicit constexpr AbsoluteMinimum(std::chrono::milliseconds value) : value_(value) {}
+  inline bool operator==(const AbsoluteMinimum& other) const { return other.value_ == value_; }
   const std::chrono::milliseconds value_;
 };
 
 /**
  * Class that describes how to compute a minimum timeout given a maximum timeout value. It wraps
  * ScaledMinimum and AbsoluteMinimum and provides a single computeMinimum() method.
  */
-class ScaledTimerMinimum : private absl::variant<ScaledMinimum, AbsoluteMinimum> {
+class ScaledTimerMinimum {
 public:
-  // Use base class constructor.
-  using absl::variant<ScaledMinimum, AbsoluteMinimum>::variant;
+  // Forward arguments to impl_'s constructor.
+  constexpr ScaledTimerMinimum(AbsoluteMinimum arg) : impl_(arg) {}
+  constexpr ScaledTimerMinimum(ScaledMinimum arg) : impl_(arg) {}
 
   // Computes the minimum value for a given maximum timeout. If this object was constructed with a
   // - ScaledMinimum value:
@@ -51,9 +56,13 @@ class ScaledTimerMinimum : private absl::variant<ScaledMinimum, AbsoluteMinimum>
       }
       const std::chrono::milliseconds value_;
     };
-    return absl::visit<Visitor, const absl::variant<ScaledMinimum, AbsoluteMinimum>&>(
-        Visitor(maximum), *this);
+    return absl::visit(Visitor(maximum), impl_);
   }
+
+  inline bool operator==(const ScaledTimerMinimum& other) const { return impl_ == other.impl_; }
+
+private:
+  absl::variant<ScaledMinimum, AbsoluteMinimum> impl_;
 };
 
 } // namespace Event

include/envoy/server/overload/BUILD

@@ -23,6 +23,7 @@ envoy_cc_library(
     name = "thread_local_overload_state",
     hdrs = ["thread_local_overload_state.h"],
     deps = [
+        "//include/envoy/event:scaled_range_timer_manager_interface",
         "//include/envoy/event:scaled_timer_minimum",
         "//include/envoy/event:timer_interface",
         "//include/envoy/thread_local:thread_local_object",

include/envoy/server/overload/overload_manager.h

@@ -4,6 +4,7 @@
 
 #include "envoy/common/pure.h"
 #include "envoy/event/dispatcher.h"
+#include "envoy/event/scaled_range_timer_manager.h"
 #include "envoy/server/overload/thread_local_overload_state.h"
 
 #include "common/singleton/const_singleton.h"
@@ -69,6 +70,11 @@ class OverloadManager {
    * an alternative to registering a callback for overload action state changes.
    */
   virtual ThreadLocalOverloadState& getThreadLocalOverloadState() PURE;
+
+  /**
+   * Get a factory for constructing scaled timer managers that respond to overload state.
+   */
+  virtual Event::ScaledRangeTimerManagerFactory scaledTimerFactory() PURE;
 };
 
 } // namespace Server

include/envoy/server/overload/thread_local_overload_state.h

@@ -4,6 +4,7 @@
 #include <string>
 
 #include "envoy/common/pure.h"
+#include "envoy/event/scaled_range_timer_manager.h"
 #include "envoy/event/scaled_timer_minimum.h"
 #include "envoy/event/timer.h"
 #include "envoy/thread_local/thread_local_object.h"
@@ -40,32 +41,13 @@ class OverloadActionState {
  */
 using OverloadActionCb = std::function<void(OverloadActionState)>;
 
-enum class OverloadTimerType {
-  // Timers created with this type will never be scaled. This should only be used for testing.
-  UnscaledRealTimerForTest,
-  // The amount of time an HTTP connection to a downstream client can remain idle (no streams). This
-  // corresponds to the HTTP_DOWNSTREAM_CONNECTION_IDLE TimerType in overload.proto.
-  HttpDownstreamIdleConnectionTimeout,
-  // The amount of time an HTTP stream from a downstream client can remain idle. This corresponds to
-  // the HTTP_DOWNSTREAM_STREAM_IDLE TimerType in overload.proto.
-  HttpDownstreamIdleStreamTimeout,
-};
-
 /**
  * Thread-local copy of the state of each configured overload action.
  */
 class ThreadLocalOverloadState : public ThreadLocal::ThreadLocalObject {
 public:
   // Get a thread-local reference to the value for the given action key.
   virtual const OverloadActionState& getState(const std::string& action) PURE;
-
-  // Get a scaled timer whose minimum corresponds to the configured value for the given timer type.
-  virtual Event::TimerPtr createScaledTimer(OverloadTimerType timer_type,
-                                            Event::TimerCb callback) PURE;
-
-  // Get a scaled timer whose minimum is determined by the given scaling rule.
-  virtual Event::TimerPtr createScaledTimer(Event::ScaledTimerMinimum minimum,
-                                            Event::TimerCb callback) PURE;
 };
 
 } // namespace Server

source/common/api/api_impl.cc

@@ -18,12 +18,20 @@ Impl::Impl(Thread::ThreadFactory& thread_factory, Stats::Store& store,
       process_context_(process_context), watermark_factory_(std::move(watermark_factory)) {}
 
 Event::DispatcherPtr Impl::allocateDispatcher(const std::string& name) {
-  return std::make_unique<Event::DispatcherImpl>(name, *this, time_system_, watermark_factory_);
+  return std::make_unique<Event::DispatcherImpl>(name, *this, time_system_, nullptr,
+                                                 watermark_factory_);
+}
+Event::DispatcherPtr
+Impl::allocateDispatcher(const std::string& name,
+                         const Event::ScaledRangeTimerManagerFactory& scaled_timer_factory) {
+  return std::make_unique<Event::DispatcherImpl>(name, *this, time_system_, scaled_timer_factory,
+                                                 watermark_factory_);
 }
 
 Event::DispatcherPtr Impl::allocateDispatcher(const std::string& name,
                                               Buffer::WatermarkFactoryPtr&& factory) {
-  return std::make_unique<Event::DispatcherImpl>(name, *this, time_system_, std::move(factory));
+  return std::make_unique<Event::DispatcherImpl>(name, *this, time_system_, nullptr,
+                                                 std::move(factory));
 }
 
 } // namespace Api

source/common/api/api_impl.h

@@ -24,6 +24,9 @@ class Impl : public Api {
 
   // Api::Api
   Event::DispatcherPtr allocateDispatcher(const std::string& name) override;
+  Event::DispatcherPtr
+  allocateDispatcher(const std::string& name,
+                     const Event::ScaledRangeTimerManagerFactory& scaled_timer_factory) override;
   Event::DispatcherPtr allocateDispatcher(const std::string& name,
                                           Buffer::WatermarkFactoryPtr&& watermark_factory) override;
   Thread::ThreadFactory& threadFactory() override { return thread_factory_; }

source/common/event/BUILD

@@ -34,6 +34,7 @@ envoy_cc_library(
         ":libevent_scheduler_lib",
         ":real_time_system_lib",
         ":signal_lib",
+        ":scaled_range_timer_manager_lib",
         "//include/envoy/common:scope_tracker_interface",
         "//include/envoy/common:time_interface",
         "//include/envoy/event:signal_interface",

source/common/event/dispatcher_impl.cc

@@ -15,6 +15,7 @@
 #include "common/common/thread.h"
 #include "common/event/file_event_impl.h"
 #include "common/event/libevent_scheduler.h"
+#include "common/event/scaled_range_timer_manager_impl.h"
 #include "common/event/signal_impl.h"
 #include "common/event/timer_impl.h"
 #include "common/filesystem/watcher_impl.h"
@@ -37,17 +38,22 @@
 namespace Envoy {
 namespace Event {
 
-DispatcherImpl::DispatcherImpl(const std::string& name, Api::Api& api,
-                               Event::TimeSystem& time_system,
-                               const Buffer::WatermarkFactorySharedPtr& factory)
+DispatcherImpl::DispatcherImpl(
+    const std::string& name, Api::Api& api, Event::TimeSystem& time_system,
+    const std::function<ScaledRangeTimerManagerPtr(Dispatcher&)>& scaled_timer_factory,
+    const Buffer::WatermarkFactorySharedPtr& watermark_factory)
     : name_(name), api_(api),
-      buffer_factory_(factory != nullptr ? factory
-                                         : std::make_shared<Buffer::WatermarkBufferFactory>()),
+      buffer_factory_(watermark_factory != nullptr
+                          ? watermark_factory
+                          : std::make_shared<Buffer::WatermarkBufferFactory>()),
       scheduler_(time_system.createScheduler(base_scheduler_, base_scheduler_)),
       deferred_delete_cb_(base_scheduler_.createSchedulableCallback(
           [this]() -> void { clearDeferredDeleteList(); })),
       post_cb_(base_scheduler_.createSchedulableCallback([this]() -> void { runPostCallbacks(); })),
-      current_to_delete_(&to_delete_1_) {
+      current_to_delete_(&to_delete_1_),
+      scaled_timer_manager_(scaled_timer_factory != nullptr
+                                ? scaled_timer_factory(*this)
+                                : std::make_unique<ScaledRangeTimerManagerImpl>(*this)) {
   ASSERT(!name_.empty());
   FatalErrorHandler::registerFatalErrorHandler(*this);
   updateApproximateMonotonicTimeInternal();
@@ -190,6 +196,17 @@ TimerPtr DispatcherImpl::createTimer(TimerCb cb) {
   return createTimerInternal(cb);
 }
 
+TimerPtr DispatcherImpl::createScaledTimer(ScaledRangeTimerManager::TimerType timer_type,
+                                           TimerCb cb) {
+  ASSERT(isThreadSafe());
+  return scaled_timer_manager_->createTimer(timer_type, std::move(cb));
+}
+
+TimerPtr DispatcherImpl::createScaledTimer(ScaledTimerMinimum minimum, TimerCb cb) {
+  ASSERT(isThreadSafe());
+  return scaled_timer_manager_->createTimer(minimum, std::move(cb));
+}
+
 Event::SchedulableCallbackPtr DispatcherImpl::createSchedulableCallback(std::function<void()> cb) {
   ASSERT(isThreadSafe());
   return base_scheduler_.createSchedulableCallback([this, cb]() {

source/common/event/dispatcher_impl.h

@@ -31,7 +31,8 @@ class DispatcherImpl : Logger::Loggable<Logger::Id::main>,
                        public FatalErrorHandlerInterface {
 public:
   DispatcherImpl(const std::string& name, Api::Api& api, Event::TimeSystem& time_system,
-                 const Buffer::WatermarkFactorySharedPtr& factory = nullptr);
+                 const ScaledRangeTimerManagerFactory& scaled_timer_factory = nullptr,
+                 const Buffer::WatermarkFactorySharedPtr& watermark_factory = nullptr);
   ~DispatcherImpl() override;
 
   /**
@@ -67,6 +68,9 @@ class DispatcherImpl : Logger::Loggable<Logger::Id::main>,
   Network::UdpListenerPtr createUdpListener(Network::SocketSharedPtr socket,
                                             Network::UdpListenerCallbacks& cb) override;
   TimerPtr createTimer(TimerCb cb) override;
+  TimerPtr createScaledTimer(ScaledRangeTimerManager::TimerType timer_type, TimerCb cb) override;
+  TimerPtr createScaledTimer(ScaledTimerMinimum minimum, TimerCb cb) override;
+
   Event::SchedulableCallbackPtr createSchedulableCallback(std::function<void()> cb) override;
   void deferredDelete(DeferredDeletablePtr&& to_delete) override;
   void exit() override;
@@ -154,6 +158,7 @@ class DispatcherImpl : Logger::Loggable<Logger::Id::main>,
   bool deferred_deleting_{};
   MonotonicTime approximate_monotonic_time_;
   WatchdogRegistrationPtr watchdog_registration_;
+  const ScaledRangeTimerManagerPtr scaled_timer_manager_;
 };
 
 } // namespace Event

source/common/event/scaled_range_timer_manager_impl.cc

@@ -137,15 +137,28 @@ class ScaledRangeTimerManagerImpl::RangeTimerImpl final : public Timer {
   const ScopeTrackedObject* scope_;
 };
 
-ScaledRangeTimerManagerImpl::ScaledRangeTimerManagerImpl(Dispatcher& dispatcher)
-    : dispatcher_(dispatcher), scale_factor_(1.0) {}
+ScaledRangeTimerManagerImpl::ScaledRangeTimerManagerImpl(
+    Dispatcher& dispatcher, const TimerTypeMapConstSharedPtr& timer_minimums)
+    : dispatcher_(dispatcher),
+      timer_minimums_(timer_minimums != nullptr ? timer_minimums
+                                                : std::make_shared<TimerTypeMap>()),
+      scale_factor_(1.0) {}
 
 ScaledRangeTimerManagerImpl::~ScaledRangeTimerManagerImpl() {
   // Scaled timers created by the manager shouldn't outlive it. This is
   // necessary but not sufficient to guarantee that.
   ASSERT(queues_.empty());
 }
 
+TimerPtr ScaledRangeTimerManagerImpl::createTimer(TimerType timer_type, TimerCb callback) {
+  const auto minimum_it = timer_minimums_->find(timer_type);
+  const Event::ScaledTimerMinimum minimum =
+      minimum_it != timer_minimums_->end()
+          ? minimum_it->second
+          : Event::ScaledTimerMinimum(Event::ScaledMinimum(UnitFloat::max()));
+  return createTimer(minimum, std::move(callback));
+}
+
 TimerPtr ScaledRangeTimerManagerImpl::createTimer(ScaledTimerMinimum minimum, TimerCb callback) {
   return std::make_unique<RangeTimerImpl>(minimum, callback, *this);
 }