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