: DispatcherImpl(name, api, time_system, {}) {}

    : DispatcherImpl(

          name, api, time_system,

          [](Dispatcher& dispatcher) {

            return std::make_unique<ScaledRangeTimerManagerImpl>(dispatcher);

          },

          watermark_factory) {}

    : DispatcherImpl(name, api.threadFactory(), api.timeSource(), api.fileSystem(), time_system,

                     scaled_timer_factory,

                     watermark_factory != nullptr

                         ? watermark_factory

                         : std::make_shared<Buffer::WatermarkBufferFactory>(

                               api.bootstrap().overload_manager().buffer_factory_config())) {}

    : name_(name), thread_factory_(thread_factory), time_source_(time_source),

      file_system_(file_system), buffer_factory_(watermark_factory),

      scheduler_(time_system.createScheduler(base_scheduler_, base_scheduler_)),

          base_scheduler_.createSchedulableCallback([this]() -> void { runThreadLocalDelete(); })),

      deferred_delete_cb_(base_scheduler_.createSchedulableCallback(

          [this]() -> void { clearDeferredDeleteList(); })),

      post_cb_(base_scheduler_.createSchedulableCallback([this]() -> void { runPostCallbacks(); })),

      current_to_delete_(&to_delete_1_), scaled_timer_manager_(scaled_timer_factory(*this)) {

  ASSERT(!name_.empty());

  FatalErrorHandler::registerFatalErrorHandler(*this);

  updateApproximateMonotonicTimeInternal();

  base_scheduler_.registerOnCheckCallback(

      std::bind(&DispatcherImpl::updateApproximateMonotonicTime, this));

}

DispatcherImpl::~DispatcherImpl() {

  ENVOY_LOG(debug, "destroying dispatcher {}", name_);

  FatalErrorHandler::removeFatalErrorHandler(*this);

}

                                      std::chrono::milliseconds min_touch_interval) {

  ASSERT(!watchdog_registration_, "Each dispatcher can have at most one registered watchdog.");

  watchdog_registration_ =

      std::make_unique<WatchdogRegistration>(watchdog, *scheduler_, min_touch_interval, *this);

}

                                     const absl::optional<std::string>& prefix) {

  const std::string effective_prefix = prefix.has_value() ? *prefix : absl::StrCat(name_, ".");

  post([this, &scope, effective_prefix] {

    stats_prefix_ = effective_prefix + "dispatcher";

    stats_ = std::make_unique<DispatcherStats>(

        DispatcherStats{ALL_DISPATCHER_STATS(POOL_HISTOGRAM_PREFIX(scope, stats_prefix_ + "."))});

    base_scheduler_.initializeStats(stats_.get());

    ENVOY_LOG(debug, "running {} on thread {}", stats_prefix_, run_tid_.debugString());

  });

}

void DispatcherImpl::clearDeferredDeleteList() {

  ASSERT(isThreadSafe());

  std::vector<DeferredDeletablePtr>* to_delete = current_to_delete_;

  size_t num_to_delete = to_delete->size();

  if (deferred_deleting_ || !num_to_delete) {

    return;

  }

  ENVOY_LOG(trace, "clearing deferred deletion list (size={})", num_to_delete);

  if (current_to_delete_ == &to_delete_1_) {

    current_to_delete_ = &to_delete_2_;

  } else {

    current_to_delete_ = &to_delete_1_;

  }

  touchWatchdog();

  deferred_deleting_ = true;

  for (size_t i = 0; i < num_to_delete; i++) {

    (*to_delete)[i].reset();

  }

  to_delete->clear();

  deferred_deleting_ = false;

}

                                       StreamInfo::StreamInfo& stream_info) {

  ASSERT(isThreadSafe());

  return std::make_unique<Network::ServerConnectionImpl>(*this, std::move(socket),

                                                         std::move(transport_socket), stream_info);

}

    const Network::TransportSocketOptionsConstSharedPtr& transport_options) {

  ASSERT(isThreadSafe());

  auto* factory = Config::Utility::getFactoryByName<Network::ClientConnectionFactory>(

      std::string(address->addressType()));

#if defined(__linux__)

  if (source_address && source_address->networkNamespace().has_value()) {

#endif

  return factory->createClientConnection(*this, address, source_address,

                                         std::move(transport_socket), options, transport_options);

}

                                             uint32_t events) {

  ASSERT(isThreadSafe());

  return FileEventPtr{new FileEventImpl(

      *this, fd,

      [this, cb](uint32_t events) {

        touchWatchdog();

        return cb(events);

      },

      trigger, events)};

}

Filesystem::WatcherPtr DispatcherImpl::createFilesystemWatcher() {

  ASSERT(isThreadSafe());

  return Filesystem::WatcherPtr{new Filesystem::WatcherImpl(*this, file_system_)};

}

TimerPtr DispatcherImpl::createTimer(TimerCb cb) {

  ASSERT(isThreadSafe());

  return createTimerInternal(cb);

}

TimerPtr DispatcherImpl::createScaledTimer(ScaledTimerType 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]() {

    touchWatchdog();

    cb();

  });

}

TimerPtr DispatcherImpl::createTimerInternal(TimerCb cb) {

  return scheduler_->createTimer(

      [this, cb]() {

        touchWatchdog();

        cb();

      },

      *this);

}

void DispatcherImpl::deferredDelete(DeferredDeletablePtr&& to_delete) {

  ASSERT(isThreadSafe());

  if (to_delete != nullptr) {

    to_delete->deleteIsPending();

    current_to_delete_->emplace_back(std::move(to_delete));

    ENVOY_LOG(trace, "item added to deferred deletion list (size={})", current_to_delete_->size());

    if (current_to_delete_->size() == 1) {

      deferred_delete_cb_->scheduleCallbackCurrentIteration();

    }

  }

}

void DispatcherImpl::exit() { base_scheduler_.loopExit(); }

SignalEventPtr DispatcherImpl::listenForSignal(signal_t signal_num, SignalCb cb) {

  ASSERT(isThreadSafe());

  return SignalEventPtr{new SignalEventImpl(*this, signal_num, cb)};

}

void DispatcherImpl::post(PostCb callback) {

  bool do_post;

  {

    Thread::LockGuard lock(post_lock_);

    do_post = post_callbacks_.empty();

    post_callbacks_.push_back(std::move(callback));

  }

  if (do_post) {

    post_cb_->scheduleCallbackCurrentIteration();

  }

}

void DispatcherImpl::deleteInDispatcherThread(DispatcherThreadDeletableConstPtr deletable) {

  bool need_schedule;

  {

    Thread::LockGuard lock(thread_local_deletable_lock_);

    need_schedule = deletables_in_dispatcher_thread_.empty();

    deletables_in_dispatcher_thread_.emplace_back(std::move(deletable));

  }

  if (need_schedule) {

    thread_local_delete_cb_->scheduleCallbackCurrentIteration();

  }

}

void DispatcherImpl::run(RunType type) {

  run_tid_ = thread_factory_.currentThreadId();

  runPostCallbacks();

  base_scheduler_.run(type);

}

MonotonicTime DispatcherImpl::approximateMonotonicTime() const {

  return approximate_monotonic_time_;

}

void DispatcherImpl::shutdown() {

  ASSERT(isThreadSafe());

  auto deferred_deletables_size = current_to_delete_->size();

  std::list<std::function<void()>>::size_type post_callbacks_size;

  {

    Thread::LockGuard lock(post_lock_);

    post_callbacks_size = post_callbacks_.size();

  }

  std::list<DispatcherThreadDeletableConstPtr> local_deletables;

  {

    Thread::LockGuard lock(thread_local_deletable_lock_);

    local_deletables = std::move(deletables_in_dispatcher_thread_);

  }

  auto thread_local_deletables_size = local_deletables.size();

  while (!local_deletables.empty()) {

    local_deletables.pop_front();

  }

  ASSERT(!shutdown_called_);

  shutdown_called_ = true;

  ENVOY_LOG(

      trace,

      "{} destroyed {} thread local objects. Peek {} deferred deletables, {} post callbacks. ",

      __FUNCTION__, deferred_deletables_size, post_callbacks_size, thread_local_deletables_size);

}

void DispatcherImpl::updateApproximateMonotonicTime() { updateApproximateMonotonicTimeInternal(); }

void DispatcherImpl::updateApproximateMonotonicTimeInternal() {

  approximate_monotonic_time_ = time_source_.monotonicTime();

}

void DispatcherImpl::runThreadLocalDelete() {

  std::list<DispatcherThreadDeletableConstPtr> to_be_delete;

  {

    Thread::LockGuard lock(thread_local_deletable_lock_);

    to_be_delete = std::move(deletables_in_dispatcher_thread_);

    ASSERT(deletables_in_dispatcher_thread_.empty());

  }

  while (!to_be_delete.empty()) {

    touchWatchdog();

    to_be_delete.pop_front();

  }

}

void DispatcherImpl::runPostCallbacks() {

  clearDeferredDeleteList();

  std::list<PostCb> callbacks;

  {

    Thread::LockGuard lock(post_lock_);

    callbacks = std::move(post_callbacks_);

    ASSERT(post_callbacks_.empty());

  }

  while (!callbacks.empty()) {

    touchWatchdog();

    callbacks.front()();

    callbacks.pop_front();

  }

}

void DispatcherImpl::onFatalError(std::ostream& os) const {

  if (isThreadSafe()) {

    for (auto iter = tracked_object_stack_.rbegin(); iter != tracked_object_stack_.rend(); ++iter) {

      (*iter)->dumpState(os);

    }

  }

}

    const FatalAction::FatalActionPtrList& actions) const {

  if (run_tid_.isEmpty() || (run_tid_ != thread_factory_.currentThreadId())) {

    return;

  }

  for (const auto& action : actions) {

    action->run(tracked_object_stack_);

  }

}

void DispatcherImpl::touchWatchdog() {

  if (watchdog_registration_) {

    watchdog_registration_->touchWatchdog();

  }

}

void DispatcherImpl::pushTrackedObject(const ScopeTrackedObject* object) {

  ASSERT(isThreadSafe());

  ASSERT(object != nullptr);

  tracked_object_stack_.push_back(object);

  ASSERT(tracked_object_stack_.size() <= ExpectedMaxTrackedObjectStackDepth);

}

void DispatcherImpl::popTrackedObject(const ScopeTrackedObject* expected_object) {

  ASSERT(isThreadSafe());

  ASSERT(expected_object != nullptr);

  RELEASE_ASSERT(!tracked_object_stack_.empty(), "Tracked Object Stack is empty, nothing to pop!");

  const ScopeTrackedObject* top = tracked_object_stack_.back();

  tracked_object_stack_.pop_back();

  ASSERT(top == expected_object,

         "Popped the top of the tracked object stack, but it wasn't the expected object!");

}