folly/experimental/FunctionScheduler.h

   1 /*
   2  * Copyright 2016 Facebook, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #pragma once
  18
  19 #include <folly/Function.h>
  20 #include <folly/Range.h>
  21 #include <chrono>
  22 #include <condition_variable>
  23 #include <mutex>
  24 #include <thread>
  25 #include <vector>
  26
  27 namespace folly {
  28
  29 /**
  30  * Schedules any number of functions to run at various intervals. E.g.,
  31  *
  32  *   FunctionScheduler fs;
  33  *
  34  *   fs.addFunction([&] { LOG(INFO) << "tick..."; }, seconds(1), "ticker");
  35  *   fs.addFunction(std::bind(&TestClass::doStuff, this), minutes(5), "stuff");
  36  *   fs.start();
  37  *   ........
  38  *   fs.cancelFunction("ticker");
  39  *   fs.addFunction([&] { LOG(INFO) << "tock..."; }, minutes(3), "tocker");
  40  *   ........
  41  *   fs.shutdown();
  42  *
  43  *
  44  * Note: the class uses only one thread - if you want to use more than one
  45  *       thread use multiple FunctionScheduler objects
  46  *
  47  * start() schedules the functions, while shutdown() terminates further
  48  * scheduling.
  49  */
  50 class FunctionScheduler {
  51  public:
  52   FunctionScheduler();
  53   ~FunctionScheduler();
  54
  55   /**
  56    * By default steady is false, meaning schedules may lag behind overtime.
  57    * This could be due to long running tasks or time drift because of randomness
  58    * in thread wakeup time.
  59    * By setting steady to true, FunctionScheduler will attempt to catch up.
  60    * i.e. more like a cronjob
  61    *
  62    * NOTE: it's only safe to set this before calling start()
  63    */
  64   void setSteady(bool steady) { steady_ = steady; }
  65
  66   /*
  67    * Parameters to control the function interval.
  68    *
  69    * If isPoisson is true, then use std::poisson_distribution to pick the
  70    * interval between each invocation of the function.
  71    *
  72    * If isPoisson os false, then always use fixed the interval specified to
  73    * addFunction().
  74    */
  75   struct LatencyDistribution {
  76     bool isPoisson;
  77     double poissonMean;
  78
  79     LatencyDistribution(bool poisson, double mean)
  80       : isPoisson(poisson),
  81         poissonMean(mean) {
  82     }
  83   };
  84
  85   /**
  86    * Adds a new function to the FunctionScheduler.
  87    *
  88    * Functions will not be run until start() is called.  When start() is
  89    * called, each function will be run after its specified startDelay.
  90    * Functions may also be added after start() has been called, in which case
  91    * startDelay is still honored.
  92    *
  93    * Throws an exception on error.  In particular, each function must have a
  94    * unique name--two functions cannot be added with the same name.
  95    */
  96   void addFunction(Function<void()>&& cb,
  97                    std::chrono::milliseconds interval,
  98                    StringPiece nameID = StringPiece(),
  99                    std::chrono::milliseconds startDelay =
 100                      std::chrono::milliseconds(0));
 101
 102   /*
 103    * Add a new function to the FunctionScheduler with a specified
 104    * LatencyDistribution
 105    */
 106   void addFunction(
 107       Function<void()>&& cb,
 108       std::chrono::milliseconds interval,
 109       const LatencyDistribution& latencyDistr,
 110       StringPiece nameID = StringPiece(),
 111       std::chrono::milliseconds startDelay = std::chrono::milliseconds(0));
 112
 113   /**
 114     * Add a new function to the FunctionScheduler with the time
 115     * interval being distributed uniformly within the given interval
 116     * [minInterval, maxInterval].
 117     */
 118   void addFunctionUniformDistribution(Function<void()>&& cb,
 119                                       std::chrono::milliseconds minInterval,
 120                                       std::chrono::milliseconds maxInterval,
 121                                       StringPiece nameID,
 122                                       std::chrono::milliseconds startDelay);
 123
 124   /**
 125    * A type alias for function that is called to determine the time
 126    * interval for the next scheduled run.
 127    */
 128   using IntervalDistributionFunc = Function<std::chrono::milliseconds()>;
 129
 130   /**
 131    * Add a new function to the FunctionScheduler. The scheduling interval
 132    * is determined by the interval distribution functor, which is called
 133    * every time the next function execution is scheduled. This allows
 134    * for supporting custom interval distribution algorithms in addition
 135    * to built in constant interval; and Poisson and jitter distributions
 136    * (@see FunctionScheduler::addFunction and
 137    * @see FunctionScheduler::addFunctionJitterInterval).
 138    */
 139   void addFunctionGenericDistribution(
 140       Function<void()>&& cb,
 141       IntervalDistributionFunc&& intervalFunc,
 142       const std::string& nameID,
 143       const std::string& intervalDescr,
 144       std::chrono::milliseconds startDelay);
 145
 146   /**
 147    * Cancels the function with the specified name, so it will no longer be run.
 148    *
 149    * Returns false if no function exists with the specified name.
 150    */
 151   bool cancelFunction(StringPiece nameID);
 152
 153   /**
 154    * All functions registered will be canceled.
 155    */
 156   void cancelAllFunctions();
 157
 158   /**
 159    * Resets the specified function's timer.
 160    * When resetFunctionTimer is called, the specified function's timer will
 161    * be reset with the same parameters it was passed initially, including
 162    * its startDelay. If the startDelay was 0, the function will be invoked
 163    * immediately.
 164    *
 165    * Returns false if no function exists with the specified name.
 166    */
 167   bool resetFunctionTimer(StringPiece nameID);
 168
 169   /**
 170    * Starts the scheduler.
 171    *
 172    * Returns false if the scheduler was already running.
 173    */
 174   bool start();
 175
 176   /**
 177    * Stops the FunctionScheduler.
 178    *
 179    * It may be restarted later by calling start() again.
 180    */
 181   void shutdown();
 182
 183   /**
 184    * Set the name of the worker thread.
 185    */
 186   void setThreadName(StringPiece threadName);
 187
 188  private:
 189   struct RepeatFunc {
 190     Function<void()> cb;
 191     IntervalDistributionFunc intervalFunc;
 192     std::chrono::steady_clock::time_point nextRunTime;
 193     std::string name;
 194     std::chrono::milliseconds startDelay;
 195     std::string intervalDescr;
 196
 197     RepeatFunc(Function<void()>&& cback,
 198                IntervalDistributionFunc&& intervalFn,
 199                const std::string& nameID,
 200                const std::string& intervalDistDescription,
 201                std::chrono::milliseconds delay)
 202         : cb(std::move(cback)),
 203           intervalFunc(std::move(intervalFn)),
 204           nextRunTime(),
 205           name(nameID),
 206           startDelay(delay),
 207           intervalDescr(intervalDistDescription) {}
 208
 209     std::chrono::steady_clock::time_point getNextRunTime() const {
 210       return nextRunTime;
 211     }
 212     void setNextRunTimeStrict(std::chrono::steady_clock::time_point curTime) {
 213       nextRunTime = curTime + intervalFunc();
 214     }
 215     void setNextRunTimeSteady() { nextRunTime += intervalFunc(); }
 216     void resetNextRunTime(std::chrono::steady_clock::time_point curTime) {
 217       nextRunTime = curTime + startDelay;
 218     }
 219     void cancel() {
 220       // Simply reset cb to an empty function.
 221       cb = {};
 222     }
 223     bool isValid() const { return bool(cb); }
 224   };
 225
 226   struct RunTimeOrder {
 227     bool operator()(const RepeatFunc& f1, const RepeatFunc& f2) const {
 228       return f1.getNextRunTime() > f2.getNextRunTime();
 229     }
 230   };
 231
 232   typedef std::vector<RepeatFunc> FunctionHeap;
 233
 234   void run();
 235   void runOneFunction(std::unique_lock<std::mutex>& lock,
 236                       std::chrono::steady_clock::time_point now);
 237   void cancelFunction(const std::unique_lock<std::mutex>& lock,
 238                       FunctionHeap::iterator it);
 239   void addFunctionToHeap(const std::unique_lock<std::mutex>& lock,
 240                          RepeatFunc&& func);
 241
 242   std::thread thread_;
 243
 244   // Mutex to protect our member variables.
 245   std::mutex mutex_;
 246   bool running_{false};
 247
 248   // The functions to run.
 249   // This is a heap, ordered by next run time.
 250   FunctionHeap functions_;
 251   RunTimeOrder fnCmp_;
 252
 253   // The function currently being invoked by the running thread.
 254   // This is null when the running thread is idle
 255   RepeatFunc* currentFunction_{nullptr};
 256
 257   // Condition variable that is signalled whenever a new function is added
 258   // or when the FunctionScheduler is stopped.
 259   std::condition_variable runningCondvar_;
 260
 261   std::string threadName_;
 262   bool steady_{false};
 263 };
 264
 265 }