folly/experimental/FunctionScheduler.cpp

   1 /*
   2  * Copyright 2015-present 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 #include <folly/experimental/FunctionScheduler.h>
  18
  19 #include <random>
  20
  21 #include <folly/Conv.h>
  22 #include <folly/Random.h>
  23 #include <folly/String.h>
  24 #include <folly/system/ThreadName.h>
  25
  26 using std::chrono::milliseconds;
  27 using std::chrono::steady_clock;
  28
  29 namespace folly {
  30
  31 namespace {
  32
  33 struct ConstIntervalFunctor {
  34   const milliseconds constInterval;
  35
  36   explicit ConstIntervalFunctor(milliseconds interval)
  37       : constInterval(interval) {
  38     if (interval < milliseconds::zero()) {
  39       throw std::invalid_argument(
  40           "FunctionScheduler: "
  41           "time interval must be non-negative");
  42     }
  43   }
  44
  45   milliseconds operator()() const { return constInterval; }
  46 };
  47
  48 struct PoissonDistributionFunctor {
  49   std::default_random_engine generator;
  50   std::poisson_distribution<int> poissonRandom;
  51
  52   explicit PoissonDistributionFunctor(double meanPoissonMs)
  53       : poissonRandom(meanPoissonMs) {
  54     if (meanPoissonMs < 0.0) {
  55       throw std::invalid_argument(
  56           "FunctionScheduler: "
  57           "Poisson mean interval must be non-negative");
  58     }
  59   }
  60
  61   milliseconds operator()() { return milliseconds(poissonRandom(generator)); }
  62 };
  63
  64 struct UniformDistributionFunctor {
  65   std::default_random_engine generator;
  66   std::uniform_int_distribution<milliseconds::rep> dist;
  67
  68   UniformDistributionFunctor(milliseconds minInterval, milliseconds maxInterval)
  69       : generator(Random::rand32()),
  70         dist(minInterval.count(), maxInterval.count()) {
  71     if (minInterval > maxInterval) {
  72       throw std::invalid_argument(
  73           "FunctionScheduler: "
  74           "min time interval must be less or equal than max interval");
  75     }
  76     if (minInterval < milliseconds::zero()) {
  77       throw std::invalid_argument(
  78           "FunctionScheduler: "
  79           "time interval must be non-negative");
  80     }
  81   }
  82
  83   milliseconds operator()() { return milliseconds(dist(generator)); }
  84 };
  85
  86 } // namespace
  87
  88 FunctionScheduler::FunctionScheduler() {}
  89
  90 FunctionScheduler::~FunctionScheduler() {
  91   // make sure to stop the thread (if running)
  92   shutdown();
  93 }
  94
  95 void FunctionScheduler::addFunction(Function<void()>&& cb,
  96                                     milliseconds interval,
  97                                     StringPiece nameID,
  98                                     milliseconds startDelay) {
  99   addFunctionInternal(
 100       std::move(cb),
 101       ConstIntervalFunctor(interval),
 102       nameID.str(),
 103       to<std::string>(interval.count(), "ms"),
 104       startDelay,
 105       false /*runOnce*/);
 106 }
 107
 108 void FunctionScheduler::addFunction(Function<void()>&& cb,
 109                                     milliseconds interval,
 110                                     const LatencyDistribution& latencyDistr,
 111                                     StringPiece nameID,
 112                                     milliseconds startDelay) {
 113   if (latencyDistr.isPoisson) {
 114     addFunctionInternal(
 115         std::move(cb),
 116         PoissonDistributionFunctor(latencyDistr.poissonMean),
 117         nameID.str(),
 118         to<std::string>(latencyDistr.poissonMean, "ms (Poisson mean)"),
 119         startDelay,
 120         false /*runOnce*/);
 121   } else {
 122     addFunction(std::move(cb), interval, nameID, startDelay);
 123   }
 124 }
 125
 126 void FunctionScheduler::addFunctionOnce(
 127     Function<void()>&& cb,
 128     StringPiece nameID,
 129     milliseconds startDelay) {
 130   addFunctionInternal(
 131       std::move(cb),
 132       ConstIntervalFunctor(milliseconds::zero()),
 133       nameID.str(),
 134       "once",
 135       startDelay,
 136       true /*runOnce*/);
 137 }
 138
 139 void FunctionScheduler::addFunctionUniformDistribution(
 140     Function<void()>&& cb,
 141     milliseconds minInterval,
 142     milliseconds maxInterval,
 143     StringPiece nameID,
 144     milliseconds startDelay) {
 145   addFunctionInternal(
 146       std::move(cb),
 147       UniformDistributionFunctor(minInterval, maxInterval),
 148       nameID.str(),
 149       to<std::string>(
 150           "[", minInterval.count(), " , ", maxInterval.count(), "] ms"),
 151       startDelay,
 152       false /*runOnce*/);
 153 }
 154
 155 void FunctionScheduler::addFunctionGenericDistribution(
 156     Function<void()>&& cb,
 157     IntervalDistributionFunc&& intervalFunc,
 158     const std::string& nameID,
 159     const std::string& intervalDescr,
 160     milliseconds startDelay) {
 161   addFunctionInternal(
 162       std::move(cb),
 163       std::move(intervalFunc),
 164       nameID,
 165       intervalDescr,
 166       startDelay,
 167       false /*runOnce*/);
 168 }
 169
 170 void FunctionScheduler::addFunctionInternal(
 171     Function<void()>&& cb,
 172     IntervalDistributionFunc&& intervalFunc,
 173     const std::string& nameID,
 174     const std::string& intervalDescr,
 175     milliseconds startDelay,
 176     bool runOnce) {
 177   if (!cb) {
 178     throw std::invalid_argument(
 179         "FunctionScheduler: Scheduled function must be set");
 180   }
 181   if (!intervalFunc) {
 182     throw std::invalid_argument(
 183         "FunctionScheduler: interval distribution function must be set");
 184   }
 185   if (startDelay < milliseconds::zero()) {
 186     throw std::invalid_argument(
 187         "FunctionScheduler: start delay must be non-negative");
 188   }
 189
 190   std::unique_lock<std::mutex> l(mutex_);
 191   auto it = functionsMap_.find(nameID);
 192   // check if the nameID is unique
 193   if (it != functionsMap_.end() && it->second->isValid()) {
 194     throw std::invalid_argument(to<std::string>(
 195         "FunctionScheduler: a function named \"", nameID, "\" already exists"));
 196   }
 197
 198   if (currentFunction_ && currentFunction_->name == nameID) {
 199     throw std::invalid_argument(to<std::string>(
 200         "FunctionScheduler: a function named \"", nameID, "\" already exists"));
 201   }
 202
 203   addFunctionToHeap(
 204       l,
 205       std::make_unique<RepeatFunc>(
 206           std::move(cb),
 207           std::move(intervalFunc),
 208           nameID,
 209           intervalDescr,
 210           startDelay,
 211           runOnce));
 212 }
 213
 214 bool FunctionScheduler::cancelFunctionWithLock(
 215     std::unique_lock<std::mutex>& lock,
 216     StringPiece nameID) {
 217   CHECK_EQ(lock.owns_lock(), true);
 218   if (currentFunction_ && currentFunction_->name == nameID) {
 219     functionsMap_.erase(currentFunction_->name);
 220     // This function is currently being run. Clear currentFunction_
 221     // The running thread will see this and won't reschedule the function.
 222     currentFunction_ = nullptr;
 223     cancellingCurrentFunction_ = true;
 224     return true;
 225   }
 226   return false;
 227 }
 228
 229 bool FunctionScheduler::cancelFunction(StringPiece nameID) {
 230   std::unique_lock<std::mutex> l(mutex_);
 231   if (cancelFunctionWithLock(l, nameID)) {
 232     return true;
 233   }
 234   auto it = functionsMap_.find(nameID);
 235   if (it != functionsMap_.end() && it->second->isValid()) {
 236     cancelFunction(l, it->second);
 237     return true;
 238   }
 239
 240   return false;
 241 }
 242
 243 bool FunctionScheduler::cancelFunctionAndWait(StringPiece nameID) {
 244   std::unique_lock<std::mutex> l(mutex_);
 245
 246   if (cancelFunctionWithLock(l, nameID)) {
 247     runningCondvar_.wait(l, [this]() { return !cancellingCurrentFunction_; });
 248     return true;
 249   }
 250
 251   auto it = functionsMap_.find(nameID);
 252     if (it != functionsMap_.end() && it->second->isValid()) {
 253       cancelFunction(l, it->second);
 254       return true;
 255     }
 256   return false;
 257 }
 258
 259 void FunctionScheduler::cancelFunction(const std::unique_lock<std::mutex>& l,
 260                                       RepeatFunc* it) {
 261   // This function should only be called with mutex_ already locked.
 262   DCHECK(l.mutex() == &mutex_);
 263   DCHECK(l.owns_lock());
 264   functionsMap_.erase(it->name);
 265   it->cancel();
 266 }
 267
 268 bool FunctionScheduler::cancelAllFunctionsWithLock(
 269     std::unique_lock<std::mutex>& lock) {
 270   CHECK_EQ(lock.owns_lock(), true);
 271   functions_.clear();
 272   functionsMap_.clear();
 273   if (currentFunction_) {
 274     cancellingCurrentFunction_ = true;
 275   }
 276   currentFunction_ = nullptr;
 277   return cancellingCurrentFunction_;
 278 }
 279
 280 void FunctionScheduler::cancelAllFunctions() {
 281   std::unique_lock<std::mutex> l(mutex_);
 282   cancelAllFunctionsWithLock(l);
 283 }
 284
 285 void FunctionScheduler::cancelAllFunctionsAndWait() {
 286   std::unique_lock<std::mutex> l(mutex_);
 287   if (cancelAllFunctionsWithLock(l)) {
 288     runningCondvar_.wait(l, [this]() { return !cancellingCurrentFunction_; });
 289   }
 290 }
 291
 292 bool FunctionScheduler::resetFunctionTimer(StringPiece nameID) {
 293   std::unique_lock<std::mutex> l(mutex_);
 294   if (currentFunction_ && currentFunction_->name == nameID) {
 295     if (cancellingCurrentFunction_ || currentFunction_->runOnce) {
 296       return false;
 297     }
 298     currentFunction_->resetNextRunTime(steady_clock::now());
 299     return true;
 300   }
 301
 302   // Since __adjust_heap() isn't a part of the standard API, there's no way to
 303   // fix the heap ordering if we adjust the key (nextRunTime) for the existing
 304   // RepeatFunc. Instead, we just cancel it and add an identical object.
 305   auto it = functionsMap_.find(nameID);
 306
 307   if (it != functionsMap_.end() && it->second->isValid()) {
 308     auto funcCopy = std::make_unique<RepeatFunc>(std::move(*(it->second)));
 309     it->second->cancel();
 310     // This will take care of making sure that functionsMap_[it->first] =
 311     // funcCopy.
 312     addFunctionToHeap(l, std::move(funcCopy));
 313     return true;
 314   }
 315   return false;
 316 }
 317
 318 bool FunctionScheduler::start() {
 319   std::unique_lock<std::mutex> l(mutex_);
 320   if (running_) {
 321     return false;
 322   }
 323
 324   VLOG(1) << "Starting FunctionScheduler with " << functions_.size()
 325           << " functions.";
 326   auto now = steady_clock::now();
 327   // Reset the next run time. for all functions.
 328   // note: this is needed since one can shutdown() and start() again
 329   for (const auto& f : functions_) {
 330     f->resetNextRunTime(now);
 331     VLOG(1) << "   - func: " << (f->name.empty() ? "(anon)" : f->name.c_str())
 332             << ", period = " << f->intervalDescr
 333             << ", delay = " << f->startDelay.count() << "ms";
 334   }
 335   std::make_heap(functions_.begin(), functions_.end(), fnCmp_);
 336
 337   thread_ = std::thread([&] { this->run(); });
 338   running_ = true;
 339
 340   return true;
 341 }
 342
 343 bool FunctionScheduler::shutdown() {
 344   {
 345     std::lock_guard<std::mutex> g(mutex_);
 346     if (!running_) {
 347       return false;
 348     }
 349
 350     running_ = false;
 351     runningCondvar_.notify_one();
 352   }
 353   thread_.join();
 354   return true;
 355 }
 356
 357 void FunctionScheduler::run() {
 358   std::unique_lock<std::mutex> lock(mutex_);
 359
 360   if (!threadName_.empty()) {
 361     folly::setThreadName(threadName_);
 362   }
 363
 364   while (running_) {
 365     // If we have nothing to run, wait until a function is added or until we
 366     // are stopped.
 367     if (functions_.empty()) {
 368       runningCondvar_.wait(lock);
 369       continue;
 370     }
 371
 372     auto now = steady_clock::now();
 373
 374     // Move the next function to run to the end of functions_
 375     std::pop_heap(functions_.begin(), functions_.end(), fnCmp_);
 376
 377     // Check to see if the function was cancelled.
 378     // If so, just remove it and continue around the loop.
 379     if (!functions_.back()->isValid()) {
 380       functions_.pop_back();
 381       continue;
 382     }
 383
 384     auto sleepTime = functions_.back()->getNextRunTime() - now;
 385     if (sleepTime < milliseconds::zero()) {
 386       // We need to run this function now
 387       runOneFunction(lock, now);
 388       runningCondvar_.notify_all();
 389     } else {
 390       // Re-add the function to the heap, and wait until we actually
 391       // need to run it.
 392       std::push_heap(functions_.begin(), functions_.end(), fnCmp_);
 393       runningCondvar_.wait_for(lock, sleepTime);
 394     }
 395   }
 396 }
 397
 398 void FunctionScheduler::runOneFunction(std::unique_lock<std::mutex>& lock,
 399                                        steady_clock::time_point now) {
 400   DCHECK(lock.mutex() == &mutex_);
 401   DCHECK(lock.owns_lock());
 402
 403   // The function to run will be at the end of functions_ already.
 404   //
 405   // Fully remove it from functions_ now.
 406   // We need to release mutex_ while we invoke this function, and we need to
 407   // maintain the heap property on functions_ while mutex_ is unlocked.
 408   auto func = std::move(functions_.back());
 409   functions_.pop_back();
 410   if (!func->cb) {
 411     VLOG(5) << func->name << "function has been canceled while waiting";
 412     return;
 413   }
 414   currentFunction_ = func.get();
 415   // Update the function's next run time.
 416   if (steady_) {
 417     // This allows scheduler to catch up
 418     func->setNextRunTimeSteady();
 419   } else {
 420     // Note that we set nextRunTime based on the current time where we started
 421     // the function call, rather than the time when the function finishes.
 422     // This ensures that we call the function once every time interval, as
 423     // opposed to waiting time interval seconds between calls.  (These can be
 424     // different if the function takes a significant amount of time to run.)
 425     func->setNextRunTimeStrict(now);
 426   }
 427
 428   // Release the lock while we invoke the user's function
 429   lock.unlock();
 430
 431   // Invoke the function
 432   try {
 433     VLOG(5) << "Now running " << func->name;
 434     func->cb();
 435   } catch (const std::exception& ex) {
 436     LOG(ERROR) << "Error running the scheduled function <"
 437       << func->name << ">: " << exceptionStr(ex);
 438   }
 439
 440   // Re-acquire the lock
 441   lock.lock();
 442
 443   if (!currentFunction_) {
 444     // The function was cancelled while we were running it.
 445     // We shouldn't reschedule it;
 446     cancellingCurrentFunction_ = false;
 447     return;
 448   }
 449   if (currentFunction_->runOnce) {
 450     // Don't reschedule if the function only needed to run once.
 451     functionsMap_.erase(currentFunction_->name);
 452     currentFunction_ = nullptr;
 453     return;
 454   }
 455
 456   // Re-insert the function into our functions_ heap.
 457   // We only maintain the heap property while running_ is set.  (running_ may
 458   // have been cleared while we were invoking the user's function.)
 459   functions_.push_back(std::move(func));
 460
 461   // Clear currentFunction_
 462   currentFunction_ = nullptr;
 463
 464   if (running_) {
 465     std::push_heap(functions_.begin(), functions_.end(), fnCmp_);
 466   }
 467 }
 468
 469 void FunctionScheduler::addFunctionToHeap(
 470     const std::unique_lock<std::mutex>& lock,
 471     std::unique_ptr<RepeatFunc> func) {
 472   // This function should only be called with mutex_ already locked.
 473   DCHECK(lock.mutex() == &mutex_);
 474   DCHECK(lock.owns_lock());
 475
 476   functions_.push_back(std::move(func));
 477   functionsMap_[functions_.back()->name] = functions_.back().get();
 478   if (running_) {
 479     functions_.back()->resetNextRunTime(steady_clock::now());
 480     std::push_heap(functions_.begin(), functions_.end(), fnCmp_);
 481     // Signal the running thread to wake up and see if it needs to change
 482     // its current scheduling decision.
 483     runningCondvar_.notify_one();
 484   }
 485 }
 486
 487 void FunctionScheduler::setThreadName(StringPiece threadName) {
 488   std::unique_lock<std::mutex> l(mutex_);
 489   threadName_ = threadName.str();
 490 }
 491
 492 } // namespace folly