FunctionScheduler.h

Go to the documentation of this file.

/*
 * Copyright 2015-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 <folly/Function.h>
#include <folly/Range.h>
#include <folly/hash/Hash.h>
#include <chrono>
#include <condition_variable>
#include <mutex>
#include <thread>
#include <unordered_map>
#include <vector>

namespace folly {

class FunctionScheduler {
 public:
  FunctionScheduler();
  ~FunctionScheduler();

  void setSteady(bool steady) {
    steady_ = steady;
  }

  /*
   * Parameters to control the function interval.
   *
   * If isPoisson is true, then use std::poisson_distribution to pick the
   * interval between each invocation of the function.
   *
   * If isPoisson os false, then always use fixed the interval specified to
   * addFunction().
   */
  struct LatencyDistribution {
    bool isPoisson;
    double poissonMean;

    LatencyDistribution(bool poisson, double mean)
        : isPoisson(poisson), poissonMean(mean) {}
  };

  void addFunction(
      Function<void()>&& cb,
      std::chrono::milliseconds interval,
      StringPiece nameID = StringPiece(),
      std::chrono::milliseconds startDelay = std::chrono::milliseconds(0));

  /*
   * Add a new function to the FunctionScheduler with a specified
   * LatencyDistribution
   */
  void addFunction(
      Function<void()>&& cb,
      std::chrono::milliseconds interval,
      const LatencyDistribution& latencyDistr,
      StringPiece nameID = StringPiece(),
      std::chrono::milliseconds startDelay = std::chrono::milliseconds(0));

  void addFunctionOnce(
      Function<void()>&& cb,
      StringPiece nameID = StringPiece(),
      std::chrono::milliseconds startDelay = std::chrono::milliseconds(0));

  void addFunctionUniformDistribution(
      Function<void()>&& cb,
      std::chrono::milliseconds minInterval,
      std::chrono::milliseconds maxInterval,
      StringPiece nameID,
      std::chrono::milliseconds startDelay);

  void addFunctionConsistentDelay(
      Function<void()>&& cb,
      std::chrono::milliseconds interval,
      StringPiece nameID = StringPiece(),
      std::chrono::milliseconds startDelay = std::chrono::milliseconds(0));

  using IntervalDistributionFunc = Function<std::chrono::milliseconds()>;
  using NextRunTimeFunc = Function<std::chrono::steady_clock::time_point(
      std::chrono::steady_clock::time_point,
      std::chrono::steady_clock::time_point)>;

  void addFunctionGenericDistribution(
      Function<void()>&& cb,
      IntervalDistributionFunc&& intervalFunc,
      const std::string& nameID,
      const std::string& intervalDescr,
      std::chrono::milliseconds startDelay);

  void addFunctionGenericNextRunTimeFunctor(
      Function<void()>&& cb,
      NextRunTimeFunc&& fn,
      const std::string& nameID,
      const std::string& intervalDescr,
      std::chrono::milliseconds startDelay);

  bool cancelFunction(StringPiece nameID);
  bool cancelFunctionAndWait(StringPiece nameID);

  void cancelAllFunctions();
  void cancelAllFunctionsAndWait();

  bool resetFunctionTimer(StringPiece nameID);

  bool start();

  bool shutdown();

  void setThreadName(StringPiece threadName);

 private:
  struct RepeatFunc {
    Function<void()> cb;
    NextRunTimeFunc nextRunTimeFunc;
    std::chrono::steady_clock::time_point nextRunTime;
    std::string name;
    std::chrono::milliseconds startDelay;
    std::string intervalDescr;
    bool runOnce;

    RepeatFunc(
        Function<void()>&& cback,
        IntervalDistributionFunc&& intervalFn,
        const std::string& nameID,
        const std::string& intervalDistDescription,
        std::chrono::milliseconds delay,
        bool once)
        : RepeatFunc(
              std::move(cback),
              getNextRunTimeFunc(std::move(intervalFn)),
              nameID,
              intervalDistDescription,
              delay,
              once) {}

    RepeatFunc(
        Function<void()>&& cback,
        NextRunTimeFunc&& nextRunTimeFn,
        const std::string& nameID,
        const std::string& intervalDistDescription,
        std::chrono::milliseconds delay,
        bool once)
        : cb(std::move(cback)),
          nextRunTimeFunc(std::move(nextRunTimeFn)),
          nextRunTime(),
          name(nameID),
          startDelay(delay),
          intervalDescr(intervalDistDescription),
          runOnce(once) {}

    static NextRunTimeFunc getNextRunTimeFunc(
        IntervalDistributionFunc&& intervalFn) {
      return [intervalFn = std::move(intervalFn)](
                 std::chrono::steady_clock::time_point /* curNextRunTime */,
                 std::chrono::steady_clock::time_point curTime) mutable {
        return curTime + intervalFn();
      };
    }

    std::chrono::steady_clock::time_point getNextRunTime() const {
      return nextRunTime;
    }
    void setNextRunTimeStrict(std::chrono::steady_clock::time_point curTime) {
      nextRunTime = nextRunTimeFunc(nextRunTime, curTime);
    }
    void setNextRunTimeSteady() {
      nextRunTime = nextRunTimeFunc(nextRunTime, nextRunTime);
    }
    void resetNextRunTime(std::chrono::steady_clock::time_point curTime) {
      nextRunTime = curTime + startDelay;
    }
    void cancel() {
      // Simply reset cb to an empty function.
      cb = {};
    }
    bool isValid() const {
      return bool(cb);
    }
  };

  struct RunTimeOrder {
    bool operator()(
        const std::unique_ptr<RepeatFunc>& f1,
        const std::unique_ptr<RepeatFunc>& f2) const {
      return f1->getNextRunTime() > f2->getNextRunTime();
    }
  };

  typedef std::vector<std::unique_ptr<RepeatFunc>> FunctionHeap;
  typedef std::unordered_map<StringPiece, RepeatFunc*, Hash> FunctionMap;

  void run();
  void runOneFunction(
      std::unique_lock<std::mutex>& lock,
      std::chrono::steady_clock::time_point now);
  void cancelFunction(const std::unique_lock<std::mutex>& lock, RepeatFunc* it);
  void addFunctionToHeap(
      const std::unique_lock<std::mutex>& lock,
      std::unique_ptr<RepeatFunc> func);

  template <typename RepeatFuncNextRunTimeFunc>
  void addFunctionToHeapChecked(
      Function<void()>&& cb,
      RepeatFuncNextRunTimeFunc&& fn,
      const std::string& nameID,
      const std::string& intervalDescr,
      std::chrono::milliseconds startDelay,
      bool runOnce);

  void addFunctionInternal(
      Function<void()>&& cb,
      NextRunTimeFunc&& fn,
      const std::string& nameID,
      const std::string& intervalDescr,
      std::chrono::milliseconds startDelay,
      bool runOnce);
  void addFunctionInternal(
      Function<void()>&& cb,
      IntervalDistributionFunc&& fn,
      const std::string& nameID,
      const std::string& intervalDescr,
      std::chrono::milliseconds startDelay,
      bool runOnce);

  // Return true if the current function is being canceled
  bool cancelAllFunctionsWithLock(std::unique_lock<std::mutex>& lock);
  bool cancelFunctionWithLock(
      std::unique_lock<std::mutex>& lock,
      StringPiece nameID);

  std::thread thread_;

  // Mutex to protect our member variables.
  std::mutex mutex_;
  bool running_{false};

  // The functions to run.
  // This is a heap, ordered by next run time.
  FunctionHeap functions_;
  FunctionMap functionsMap_;
  RunTimeOrder fnCmp_;

  // The function currently being invoked by the running thread.
  // This is null when the running thread is idle
  RepeatFunc* currentFunction_{nullptr};

  // Condition variable that is signalled whenever a new function is added
  // or when the FunctionScheduler is stopped.
  std::condition_variable runningCondvar_;

  std::string threadName_;
  bool steady_{false};
  bool cancellingCurrentFunction_{false};
};

} // namespace folly