folly/experimental/FunctionScheduler.cpp

   1 /*
   2  * Copyright 2017 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     // TODO: This moves out of RepeatFunc object while folly:Function can
 296     // potentially be executed. This might be unsafe.
 297     auto funcPtrCopy = std::make_unique<RepeatFunc>(std::move(*currentFunction_));
 298     // This function is currently being run. Clear currentFunction_
 299     // to avoid rescheduling it, and add the function again to honor the
 300     // startDelay.
 301     currentFunction_ = nullptr;
 302     addFunctionToHeap(l, std::move(funcPtrCopy));
 303     return true;
 304   }
 305
 306   // Since __adjust_heap() isn't a part of the standard API, there's no way to
 307   // fix the heap ordering if we adjust the key (nextRunTime) for the existing
 308   // RepeatFunc. Instead, we just cancel it and add an identical object.
 309   auto it = functionsMap_.find(nameID);
 310
 311   if (it != functionsMap_.end() && it->second->isValid()) {
 312     auto funcCopy = std::make_unique<RepeatFunc>(std::move(*(it->second)));
 313     it->second->cancel();
 314     // This will take care of making sure that functionsMap_[it->first] =
 315     // funcCopy.
 316     addFunctionToHeap(l, std::move(funcCopy));
 317     return true;
 318   }
 319   return false;
 320 }
 321
 322 bool FunctionScheduler::start() {
 323   std::unique_lock<std::mutex> l(mutex_);
 324   if (running_) {
 325     return false;
 326   }
 327
 328   VLOG(1) << "Starting FunctionScheduler with " << functions_.size()
 329           << " functions.";
 330   auto now = steady_clock::now();
 331   // Reset the next run time. for all functions.
 332   // note: this is needed since one can shutdown() and start() again
 333   for (const auto& f : functions_) {
 334     f->resetNextRunTime(now);
 335     VLOG(1) << "   - func: " << (f->name.empty() ? "(anon)" : f->name.c_str())
 336             << ", period = " << f->intervalDescr
 337             << ", delay = " << f->startDelay.count() << "ms";
 338   }
 339   std::make_heap(functions_.begin(), functions_.end(), fnCmp_);
 340
 341   thread_ = std::thread([&] { this->run(); });
 342   running_ = true;
 343
 344   return true;
 345 }
 346
 347 bool FunctionScheduler::shutdown() {
 348   {
 349     std::lock_guard<std::mutex> g(mutex_);
 350     if (!running_) {
 351       return false;
 352     }
 353
 354     running_ = false;
 355     runningCondvar_.notify_one();
 356   }
 357   thread_.join();
 358   return true;
 359 }
 360
 361 void FunctionScheduler::run() {
 362   std::unique_lock<std::mutex> lock(mutex_);
 363
 364   if (!threadName_.empty()) {
 365     folly::setThreadName(threadName_);
 366   }
 367
 368   while (running_) {
 369     // If we have nothing to run, wait until a function is added or until we
 370     // are stopped.
 371     if (functions_.empty()) {
 372       runningCondvar_.wait(lock);
 373       continue;
 374     }
 375
 376     auto now = steady_clock::now();
 377
 378     // Move the next function to run to the end of functions_
 379     std::pop_heap(functions_.begin(), functions_.end(), fnCmp_);
 380
 381     // Check to see if the function was cancelled.
 382     // If so, just remove it and continue around the loop.
 383     if (!functions_.back()->isValid()) {
 384       functions_.pop_back();
 385       continue;
 386     }
 387
 388     auto sleepTime = functions_.back()->getNextRunTime() - now;
 389     if (sleepTime < milliseconds::zero()) {
 390       // We need to run this function now
 391       runOneFunction(lock, now);
 392       runningCondvar_.notify_all();
 393     } else {
 394       // Re-add the function to the heap, and wait until we actually
 395       // need to run it.
 396       std::push_heap(functions_.begin(), functions_.end(), fnCmp_);
 397       runningCondvar_.wait_for(lock, sleepTime);
 398     }
 399   }
 400 }
 401
 402 void FunctionScheduler::runOneFunction(std::unique_lock<std::mutex>& lock,
 403                                        steady_clock::time_point now) {
 404   DCHECK(lock.mutex() == &mutex_);
 405   DCHECK(lock.owns_lock());
 406
 407   // The function to run will be at the end of functions_ already.
 408   //
 409   // Fully remove it from functions_ now.
 410   // We need to release mutex_ while we invoke this function, and we need to
 411   // maintain the heap property on functions_ while mutex_ is unlocked.
 412   auto func = std::move(functions_.back());
 413   functions_.pop_back();
 414   if (!func->cb) {
 415     VLOG(5) << func->name << "function has been canceled while waiting";
 416     return;
 417   }
 418   currentFunction_ = func.get();
 419   // Update the function's next run time.
 420   if (steady_) {
 421     // This allows scheduler to catch up
 422     func->setNextRunTimeSteady();
 423   } else {
 424     // Note that we set nextRunTime based on the current time where we started
 425     // the function call, rather than the time when the function finishes.
 426     // This ensures that we call the function once every time interval, as
 427     // opposed to waiting time interval seconds between calls.  (These can be
 428     // different if the function takes a significant amount of time to run.)
 429     func->setNextRunTimeStrict(now);
 430   }
 431
 432   // Release the lock while we invoke the user's function
 433   lock.unlock();
 434
 435   // Invoke the function
 436   try {
 437     VLOG(5) << "Now running " << func->name;
 438     func->cb();
 439   } catch (const std::exception& ex) {
 440     LOG(ERROR) << "Error running the scheduled function <"
 441       << func->name << ">: " << exceptionStr(ex);
 442   }
 443
 444   // Re-acquire the lock
 445   lock.lock();
 446
 447   if (!currentFunction_) {
 448     // The function was cancelled while we were running it.
 449     // We shouldn't reschedule it;
 450     cancellingCurrentFunction_ = false;
 451     return;
 452   }
 453   if (currentFunction_->runOnce) {
 454     // Don't reschedule if the function only needed to run once.
 455     functionsMap_.erase(currentFunction_->name);
 456     currentFunction_ = nullptr;
 457     return;
 458   }
 459
 460   // Re-insert the function into our functions_ heap.
 461   // We only maintain the heap property while running_ is set.  (running_ may
 462   // have been cleared while we were invoking the user's function.)
 463   functions_.push_back(std::move(func));
 464
 465   // Clear currentFunction_
 466   currentFunction_ = nullptr;
 467
 468   if (running_) {
 469     std::push_heap(functions_.begin(), functions_.end(), fnCmp_);
 470   }
 471 }
 472
 473 void FunctionScheduler::addFunctionToHeap(
 474     const std::unique_lock<std::mutex>& lock,
 475     std::unique_ptr<RepeatFunc> func) {
 476   // This function should only be called with mutex_ already locked.
 477   DCHECK(lock.mutex() == &mutex_);
 478   DCHECK(lock.owns_lock());
 479
 480   functions_.push_back(std::move(func));
 481   functionsMap_[functions_.back()->name] = functions_.back().get();
 482   if (running_) {
 483     functions_.back()->resetNextRunTime(steady_clock::now());
 484     std::push_heap(functions_.begin(), functions_.end(), fnCmp_);
 485     // Signal the running thread to wake up and see if it needs to change
 486     // its current scheduling decision.
 487     runningCondvar_.notify_one();
 488   }
 489 }
 490
 491 void FunctionScheduler::setThreadName(StringPiece threadName) {
 492   std::unique_lock<std::mutex> l(mutex_);
 493   threadName_ = threadName.str();
 494 }
 495
 496 } // namespace folly