EventBase.h

Go to the documentation of this file.

/*
 * Copyright 2014-present Facebook, Inc.
 *
 * 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.
 */
#pragma once

#include <atomic>
#include <cerrno>
#include <cmath>
#include <cstdlib>
#include <functional>
#include <list>
#include <memory>
#include <mutex>
#include <queue>
#include <set>
#include <stack>
#include <unordered_map>
#include <unordered_set>
#include <utility>

#include <boost/intrusive/list.hpp>
#include <glog/logging.h>

#include <folly/Executor.h>
#include <folly/Function.h>
#include <folly/Portability.h>
#include <folly/ScopeGuard.h>
#include <folly/executors/DrivableExecutor.h>
#include <folly/executors/IOExecutor.h>
#include <folly/executors/ScheduledExecutor.h>
#include <folly/executors/SequencedExecutor.h>
#include <folly/experimental/ExecutionObserver.h>
#include <folly/io/async/AsyncTimeout.h>
#include <folly/io/async/HHWheelTimer.h>
#include <folly/io/async/Request.h>
#include <folly/io/async/TimeoutManager.h>
#include <folly/portability/Event.h>
#include <folly/synchronization/CallOnce.h>

namespace folly {

using Cob = Func; // defined in folly/Executor.h
template <typename MessageT>
class NotificationQueue;

namespace detail {
class EventBaseLocalBase;

class EventBaseLocalBaseBase {
 public:
  virtual void onEventBaseDestruction(EventBase& evb) = 0;
  virtual ~EventBaseLocalBaseBase() = default;
};
} // namespace detail
template <typename T>
class EventBaseLocal;

class EventBaseObserver {
 public:
  virtual ~EventBaseObserver() = default;

  virtual uint32_t getSampleRate() const = 0;

  virtual void loopSample(int64_t busyTime, int64_t idleTime) = 0;
};

// Helper class that sets and retrieves the EventBase associated with a given
// request via RequestContext. See Request.h for that mechanism.
class RequestEventBase : public RequestData {
 public:
  static EventBase* get() {
    auto data = dynamic_cast<RequestEventBase*>(
        RequestContext::get()->getContextData(kContextDataName));
    if (!data) {
      return nullptr;
    }
    return data->eb_;
  }

  static void set(EventBase* eb) {
    RequestContext::get()->setContextData(
        kContextDataName,
        std::unique_ptr<RequestEventBase>(new RequestEventBase(eb)));
  }

  bool hasCallback() override {
    return false;
  }

 private:
  explicit RequestEventBase(EventBase* eb) : eb_(eb) {}
  EventBase* eb_;
  static constexpr const char* kContextDataName{"EventBase"};
};

class VirtualEventBase;

class EventBase : private boost::noncopyable,
                  public TimeoutManager,
                  public DrivableExecutor,
                  public IOExecutor,
                  public SequencedExecutor,
                  public ScheduledExecutor {
 public:
  using Func = folly::Function<void()>;

  class LoopCallback
      : public boost::intrusive::list_base_hook<
            boost::intrusive::link_mode<boost::intrusive::auto_unlink>> {
   public:
    virtual ~LoopCallback() = default;

    virtual void runLoopCallback() noexcept = 0;
    void cancelLoopCallback() {
      context_.reset();
      unlink();
    }

    bool isLoopCallbackScheduled() const {
      return is_linked();
    }

   private:
    typedef boost::intrusive::
        list<LoopCallback, boost::intrusive::constant_time_size<false>>
            List;

    // EventBase needs access to LoopCallbackList (and therefore to hook_)
    friend class EventBase;
    friend class VirtualEventBase;
    std::shared_ptr<RequestContext> context_;
  };

  class FunctionLoopCallback : public LoopCallback {
   public:
    explicit FunctionLoopCallback(Func&& function)
        : function_(std::move(function)) {}

    void runLoopCallback() noexcept override {
      function_();
      delete this;
    }

   private:
    Func function_;
  };

  // Like FunctionLoopCallback, but saves one allocation. Use with caution.
  //
  // The caller is responsible for maintaining the lifetime of this callback
  // until after the point at which the contained function is called.
  class StackFunctionLoopCallback : public LoopCallback {
   public:
    explicit StackFunctionLoopCallback(Func&& function)
        : function_(std::move(function)) {}
    void runLoopCallback() noexcept override {
      Func(std::move(function_))();
    }

   private:
    Func function_;
  };

  EventBase() : EventBase(true) {}

  explicit EventBase(bool enableTimeMeasurement);

  explicit EventBase(event_base* evb, bool enableTimeMeasurement = true);
  ~EventBase() override;

  bool loop();

  [[deprecated("This should only be used in legacy unit tests")]] bool
  loopIgnoreKeepAlive();

  bool loopOnce(int flags = 0);

  void loopForever();

  void terminateLoopSoon();

  void runInLoop(LoopCallback* callback, bool thisIteration = false);

  void runInLoop(Func c, bool thisIteration = false);

  void runOnDestruction(LoopCallback* callback);

  void runBeforeLoop(LoopCallback* callback);

  template <typename T>
  bool runInEventBaseThread(void (*fn)(T*), T* arg);

  bool runInEventBaseThread(Func fn);

  /*
   * Like runInEventBaseThread, but the caller waits for the callback to be
   * executed.
   */
  template <typename T>
  bool runInEventBaseThreadAndWait(void (*fn)(T*), T* arg);

  /*
   * Like runInEventBaseThread, but the caller waits for the callback to be
   * executed.
   */
  bool runInEventBaseThreadAndWait(Func fn);

  /*
   * Like runInEventBaseThreadAndWait, except if the caller is already in the
   * event base thread, the functor is simply run inline.
   */
  template <typename T>
  bool runImmediatelyOrRunInEventBaseThreadAndWait(void (*fn)(T*), T* arg);

  /*
   * Like runInEventBaseThreadAndWait, except if the caller is already in the
   * event base thread, the functor is simply run inline.
   */
  bool runImmediatelyOrRunInEventBaseThreadAndWait(Func fn);

  void setMaxLatency(std::chrono::microseconds maxLatency, Func maxLatencyCob) {
    assert(enableTimeMeasurement_);
    maxLatency_ = maxLatency;
    maxLatencyCob_ = std::move(maxLatencyCob);
  }

  void setLoadAvgMsec(std::chrono::milliseconds ms);

  void resetLoadAvg(double value = 0.0);

  double getAvgLoopTime() const {
    assert(enableTimeMeasurement_);
    return avgLoopTime_.get();
  }

  bool isRunning() const {
    return loopThread_.load(std::memory_order_relaxed) != std::thread::id();
  }

  void waitUntilRunning();

  size_t getNotificationQueueSize() const;

  void setMaxReadAtOnce(uint32_t maxAtOnce);

  bool isInEventBaseThread() const {
    auto tid = loopThread_.load(std::memory_order_relaxed);
    return tid == std::thread::id() || tid == std::this_thread::get_id();
  }

  bool inRunningEventBaseThread() const {
    return loopThread_.load(std::memory_order_relaxed) ==
        std::this_thread::get_id();
  }

  void checkIsInEventBaseThread() const;
  void dcheckIsInEventBaseThread() const {
    if (kIsDebug) {
      checkIsInEventBaseThread();
    }
  }

  HHWheelTimer& timer() {
    if (!wheelTimer_) {
      wheelTimer_ = HHWheelTimer::newTimer(this);
    }
    return *wheelTimer_.get();
  }

  // --------- interface to underlying libevent base ------------
  // Avoid using these functions if possible.  These functions are not
  // guaranteed to always be present if we ever provide alternative EventBase
  // implementations that do not use libevent internally.
  event_base* getLibeventBase() const {
    return evb_;
  }
  static const char* getLibeventVersion();
  static const char* getLibeventMethod();

  void bumpHandlingTime() final;

  class SmoothLoopTime {
   public:
    explicit SmoothLoopTime(std::chrono::microseconds timeInterval)
        : expCoeff_(-1.0 / timeInterval.count()), value_(0.0) {
      VLOG(11) << "expCoeff_ " << expCoeff_ << " " << __PRETTY_FUNCTION__;
    }

    void setTimeInterval(std::chrono::microseconds timeInterval);
    void reset(double value = 0.0);

    void addSample(
        std::chrono::microseconds total,
        std::chrono::microseconds busy);

    double get() const {
      // Add the outstanding buffered times linearly, to avoid
      // expensive exponentiation
      auto lcoeff = buffer_time_.count() * -expCoeff_;
      return value_ * (1.0 - lcoeff) + lcoeff * busy_buffer_.count();
    }

    void dampen(double factor) {
      value_ *= factor;
    }

   private:
    double expCoeff_;
    double value_;
    std::chrono::microseconds buffer_time_{0};
    std::chrono::microseconds busy_buffer_{0};
    std::size_t buffer_cnt_{0};
    static constexpr std::chrono::milliseconds buffer_interval_{10};
  };

  void setObserver(const std::shared_ptr<EventBaseObserver>& observer) {
    assert(enableTimeMeasurement_);
    observer_ = observer;
  }

  const std::shared_ptr<EventBaseObserver>& getObserver() {
    return observer_;
  }

  void setExecutionObserver(ExecutionObserver* observer) {
    executionObserver_ = observer;
  }

  ExecutionObserver* getExecutionObserver() {
    return executionObserver_;
  }

  void setName(const std::string& name);

  const std::string& getName();

  void add(Cob fn) override {
    // runInEventBaseThread() takes a const&,
    // so no point in doing std::move here.
    runInEventBaseThread(std::move(fn));
  }

  void drive() override {
    ++loopKeepAliveCount_;
    SCOPE_EXIT {
      --loopKeepAliveCount_;
    };
    loopOnce();
  }

  // Implements the ScheduledExecutor interface
  void scheduleAt(Func&& fn, TimePoint const& timeout) override;

  // TimeoutManager
  void attachTimeoutManager(
      AsyncTimeout* obj,
      TimeoutManager::InternalEnum internal) final;

  void detachTimeoutManager(AsyncTimeout* obj) final;

  bool scheduleTimeout(AsyncTimeout* obj, TimeoutManager::timeout_type timeout)
      final;

  void cancelTimeout(AsyncTimeout* obj) final;

  bool isInTimeoutManagerThread() final {
    return isInEventBaseThread();
  }

  // Returns a VirtualEventBase attached to this EventBase. Can be used to
  // pass to APIs which expect VirtualEventBase. This VirtualEventBase will be
  // destroyed together with the EventBase.
  //
  // Any number of VirtualEventBases instances may be independently constructed,
  // which are backed by this EventBase. This method should be only used if you
  // don't need to manage the life time of the VirtualEventBase used.
  folly::VirtualEventBase& getVirtualEventBase();

  EventBase* getEventBase() override;

 protected:
  bool keepAliveAcquire() override {
    if (inRunningEventBaseThread()) {
      loopKeepAliveCount_++;
    } else {
      loopKeepAliveCountAtomic_.fetch_add(1, std::memory_order_relaxed);
    }
    return true;
  }

  void keepAliveRelease() override {
    if (!inRunningEventBaseThread()) {
      return add([this] { loopKeepAliveCount_--; });
    }
    loopKeepAliveCount_--;
  }

 private:
  void applyLoopKeepAlive();

  ssize_t loopKeepAliveCount();

  /*
   * Helper function that tells us whether we have already handled
   * some event/timeout/callback in this loop iteration.
   */
  bool nothingHandledYet() const noexcept;

  typedef LoopCallback::List LoopCallbackList;
  class FunctionRunner;

  bool loopBody(int flags = 0, bool ignoreKeepAlive = false);

  // executes any callbacks queued by runInLoop(); returns false if none found
  bool runLoopCallbacks();

  void initNotificationQueue();

  // should only be accessed through public getter
  HHWheelTimer::UniquePtr wheelTimer_;

  LoopCallbackList loopCallbacks_;
  LoopCallbackList runBeforeLoopCallbacks_;
  LoopCallbackList onDestructionCallbacks_;

  // This will be null most of the time, but point to currentCallbacks
  // if we are in the middle of running loop callbacks, such that
  // runInLoop(..., true) will always run in the current loop
  // iteration.
  LoopCallbackList* runOnceCallbacks_;

  // stop_ is set by terminateLoopSoon() and is used by the main loop
  // to determine if it should exit
  std::atomic<bool> stop_;

  // The ID of the thread running the main loop.
  // std::thread::id{} if loop is not running.
  std::atomic<std::thread::id> loopThread_;

  // pointer to underlying event_base class doing the heavy lifting
  event_base* evb_;

  // A notification queue for runInEventBaseThread() to use
  // to send function requests to the EventBase thread.
  std::unique_ptr<NotificationQueue<Func>> queue_;
  std::unique_ptr<FunctionRunner> fnRunner_;
  ssize_t loopKeepAliveCount_{0};
  std::atomic<ssize_t> loopKeepAliveCountAtomic_{0};
  bool loopKeepAliveActive_{false};

  // limit for latency in microseconds (0 disables)
  std::chrono::microseconds maxLatency_;

  // exponentially-smoothed average loop time for latency-limiting
  SmoothLoopTime avgLoopTime_;

  // smoothed loop time used to invoke latency callbacks; differs from
  // avgLoopTime_ in that it's scaled down after triggering a callback
  // to reduce spamminess
  SmoothLoopTime maxLatencyLoopTime_;

  // callback called when latency limit is exceeded
  Func maxLatencyCob_;

  // Enables/disables time measurements in loopBody(). if disabled, the
  // following functionality that relies on time-measurement, will not
  // be supported: avg loop time, observer and max latency.
  const bool enableTimeMeasurement_;

  // Wrap-around loop counter to detect beginning of each loop
  std::size_t nextLoopCnt_;
  std::size_t latestLoopCnt_;
  std::chrono::steady_clock::time_point startWork_;
  // Prevent undefined behavior from invoking event_base_loop() reentrantly.
  // This is needed since many projects use libevent-1.4, which lacks commit
  // b557b175c00dc462c1fce25f6e7dd67121d2c001 from
  // https://github.com/libevent/libevent/.
  bool invokingLoop_{false};

  // Observer to export counters
  std::shared_ptr<EventBaseObserver> observer_;
  uint32_t observerSampleCount_;

  // EventHandler's execution observer.
  ExecutionObserver* executionObserver_;

  // Name of the thread running this EventBase
  std::string name_;

  // allow runOnDestruction() to be called from any threads
  std::mutex onDestructionCallbacksMutex_;

  // see EventBaseLocal
  friend class detail::EventBaseLocalBase;
  template <typename T>
  friend class EventBaseLocal;
  std::unordered_map<std::size_t, std::shared_ptr<void>> localStorage_;
  std::unordered_set<detail::EventBaseLocalBaseBase*> localStorageToDtor_;

  folly::once_flag virtualEventBaseInitFlag_;
  std::unique_ptr<VirtualEventBase> virtualEventBase_;
};

template <typename T>
bool EventBase::runInEventBaseThread(void (*fn)(T*), T* arg) {
  return runInEventBaseThread([=] { fn(arg); });
}

template <typename T>
bool EventBase::runInEventBaseThreadAndWait(void (*fn)(T*), T* arg) {
  return runInEventBaseThreadAndWait([=] { fn(arg); });
}

template <typename T>
bool EventBase::runImmediatelyOrRunInEventBaseThreadAndWait(
    void (*fn)(T*),
    T* arg) {
  return runImmediatelyOrRunInEventBaseThreadAndWait([=] { fn(arg); });
}

} // namespace folly