text: #include "common/event/dispatcher_impl.h"
  
  #include <chrono>
  #include <cstdint>
  #include <functional>
  #include <string>
  #include <vector>
  
  #include "envoy/api/api.h"
  #include "envoy/network/listen_socket.h"
  #include "envoy/network/listener.h"
  
  #include "common/buffer/buffer_impl.h"
  #include "common/common/lock_guard.h"
  #include "common/common/thread.h"
  #include "common/event/file_event_impl.h"
  #include "common/event/signal_impl.h"
  #include "common/filesystem/watcher_impl.h"
  #include "common/network/connection_impl.h"
  #include "common/network/dns_impl.h"
  #include "common/network/listener_impl.h"
  
  #include "event2/event.h"
  
  namespace Envoy {
  namespace Event {
  
  DispatcherImpl::DispatcherImpl(TimeSystem& time_system, Api::Api& api)
      : DispatcherImpl(time_system, Buffer::WatermarkFactoryPtr{new Buffer::WatermarkBufferFactory},
                       api) {
    // The dispatcher won't work as expected if libevent hasn't been configured to use threads.
    RELEASE_ASSERT(Libevent::Global::initialized(), "");
  }
  
  DispatcherImpl::DispatcherImpl(TimeSystem& time_system, Buffer::WatermarkFactoryPtr&& factory,
                                 Api::Api& api)
      : api_(api), time_system_(time_system), buffer_factory_(std::move(factory)),
        base_(event_base_new()), scheduler_(time_system_.createScheduler(base_)),
        deferred_delete_timer_(createTimer([this]() -> void { clearDeferredDeleteList(); })),
        post_timer_(createTimer([this]() -> void { runPostCallbacks(); })),
        current_to_delete_(&to_delete_1_) {
    RELEASE_ASSERT(Libevent::Global::initialized(), "");
  }
  
  DispatcherImpl::~DispatcherImpl() {}
  
  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);
  
    // Swap the current deletion vector so that if we do deferred delete while we are deleting, we
    // use the other vector. We will get another callback to delete that vector.
    if (current_to_delete_ == &to_delete_1_) {
      current_to_delete_ = &to_delete_2_;
    } else {
      current_to_delete_ = &to_delete_1_;
    }
  
    deferred_deleting_ = true;
  
    // Calling clear() on the vector does not specify which order destructors run in. We want to
    // destroy in FIFO order so just do it manually. This required 2 passes over the vector which is
    // not optimal but can be cleaned up later if needed.
    for (size_t i = 0; i < num_to_delete; i++) {
      (*to_delete)[i].reset();
    }
  
    to_delete->clear();
    deferred_deleting_ = false;
  }
  
  Network::ConnectionPtr
  DispatcherImpl::createServerConnection(Network::ConnectionSocketPtr&& socket,
                                         Network::TransportSocketPtr&& transport_socket) {
    ASSERT(isThreadSafe());
    return std::make_unique<Network::ConnectionImpl>(*this, std::move(socket),
                                                     std::move(transport_socket), true);
  }
  
  Network::ClientConnectionPtr
  DispatcherImpl::createClientConnection(Network::Address::InstanceConstSharedPtr address,
                                         Network::Address::InstanceConstSharedPtr source_address,
                                         Network::TransportSocketPtr&& transport_socket,
                                         const Network::ConnectionSocket::OptionsSharedPtr& options) {
    ASSERT(isThreadSafe());
    return std::make_unique<Network::ClientConnectionImpl>(*this, address, source_address,
                                                           std::move(transport_socket), options);
  }
  
  Network::DnsResolverSharedPtr DispatcherImpl::createDnsResolver(
      const std::vector<Network::Address::InstanceConstSharedPtr>& resolvers) {
    ASSERT(isThreadSafe());
    return Network::DnsResolverSharedPtr{new Network::DnsResolverImpl(*this, resolvers)};
  }
  
  FileEventPtr DispatcherImpl::createFileEvent(int fd, FileReadyCb cb, FileTriggerType trigger,
                                               uint32_t events) {
    ASSERT(isThreadSafe());
    return FileEventPtr{new FileEventImpl(*this, fd, cb, trigger, events)};
  }
  
  Filesystem::WatcherPtr DispatcherImpl::createFilesystemWatcher() {
    ASSERT(isThreadSafe());
    return Filesystem::WatcherPtr{new Filesystem::WatcherImpl(*this)};
  }
  
  Network::ListenerPtr
  DispatcherImpl::createListener(Network::Socket& socket, Network::ListenerCallbacks& cb,
                                 bool bind_to_port, bool hand_off_restored_destination_connections) {
    ASSERT(isThreadSafe());
    return Network::ListenerPtr{new Network::ListenerImpl(*this, socket, cb, bind_to_port,
                                                          hand_off_restored_destination_connections)};
  }
  
  TimerPtr DispatcherImpl::createTimer(TimerCb cb) {
    ASSERT(isThreadSafe());
    return scheduler_->createTimer(cb);
  }
  
  void DispatcherImpl::deferredDelete(DeferredDeletablePtr&& to_delete) {
    ASSERT(isThreadSafe());
    current_to_delete_->emplace_back(std::move(to_delete));
    ENVOY_LOG(trace, "item added to deferred deletion list (size={})", current_to_delete_->size());
    if (1 == current_to_delete_->size()) {
      deferred_delete_timer_->enableTimer(std::chrono::milliseconds(0));
    }
  }
  
  void DispatcherImpl::exit() { event_base_loopexit(base_.get(), nullptr); }
  
  SignalEventPtr DispatcherImpl::listenForSignal(int signal_num, SignalCb cb) {
    ASSERT(isThreadSafe());
    return SignalEventPtr{new SignalEventImpl(*this, signal_num, cb)};
  }
  
  void DispatcherImpl::post(std::function<void()> callback) {
    bool do_post;
    {
      Thread::LockGuard lock(post_lock_);
      do_post = post_callbacks_.empty();
      post_callbacks_.push_back(callback);
    }
  
    if (do_post) {
      post_timer_->enableTimer(std::chrono::milliseconds(0));
    }
  }
  
  void DispatcherImpl::run(RunType type) {
    run_tid_ = api_.threadFactory().currentThreadId();
  
    // Flush all post callbacks before we run the event loop. We do this because there are post
    // callbacks that have to get run before the initial event loop starts running. libevent does
    // not guarantee that events are run in any particular order. So even if we post() and call
    // event_base_once() before some other event, the other event might get called first.
    runPostCallbacks();
  
    event_base_loop(base_.get(), type == RunType::NonBlock ? EVLOOP_NONBLOCK : 0);
  }
  
  void DispatcherImpl::runPostCallbacks() {
    while (true) {
      // It is important that this declaration is inside the body of the loop so that the callback is
      // destructed while post_lock_ is not held. If callback is declared outside the loop and reused
      // for each iteration, the previous iteration's callback is destructed when callback is
      // re-assigned, which happens while holding the lock. This can lead to a deadlock (via
      // recursive mutex acquisition) if destroying the callback runs a destructor, which through some
      // callstack calls post() on this dispatcher.
      std::function<void()> callback;
      {
        Thread::LockGuard lock(post_lock_);
        if (post_callbacks_.empty()) {
          return;
        }
        callback = post_callbacks_.front();
        post_callbacks_.pop_front();
      }
      callback();
    }
  }
  
  } // namespace Event
  } // namespace Envoy
  