Commit 3e6ccd5c authored by Eugene Pekurovsky's avatar Eugene Pekurovsky Committed by facebook-github-bot-4

folly::FunctionScheduler: Adding support for uniform interval distribution

Summary: 1) Added uniform interval distribution functionality.
2) Added a generic API for custom interval distribution algorithms.
3) Fixed an issue with removing a canceled function.
4) Did some code cleanup along the way.

Reviewed By: @​kaanb

Differential Revision: D2339911
parent 6699f91a
...@@ -15,18 +15,77 @@ ...@@ -15,18 +15,77 @@
*/ */
#include <folly/experimental/FunctionScheduler.h> #include <folly/experimental/FunctionScheduler.h>
#include <folly/ThreadName.h>
#include <random>
#include <folly/Conv.h> #include <folly/Conv.h>
#include <folly/Random.h>
#include <folly/String.h> #include <folly/String.h>
#include <folly/ThreadName.h>
using namespace std;
using std::chrono::milliseconds; using std::chrono::milliseconds;
using std::chrono::steady_clock; using std::chrono::steady_clock;
namespace folly { namespace folly {
FunctionScheduler::FunctionScheduler() { namespace {
}
struct ConstIntervalFunctor {
const milliseconds constInterval;
explicit ConstIntervalFunctor(milliseconds interval)
: constInterval(interval) {
if (interval < milliseconds::zero()) {
throw std::invalid_argument(
"FunctionScheduler: "
"time interval must be non-negative");
}
}
milliseconds operator()() const { return constInterval; }
};
struct PoissonDistributionFunctor {
std::default_random_engine generator;
std::poisson_distribution<int> poissonRandom;
explicit PoissonDistributionFunctor(double meanPoissonMs)
: poissonRandom(meanPoissonMs) {
if (meanPoissonMs < 0.0) {
throw std::invalid_argument(
"FunctionScheduler: "
"Poisson mean interval must be non-negative");
}
}
milliseconds operator()() { return milliseconds(poissonRandom(generator)); }
};
struct UniformDistributionFunctor {
std::default_random_engine generator;
std::uniform_int_distribution<> dist;
UniformDistributionFunctor(milliseconds minInterval, milliseconds maxInterval)
: generator(Random::rand32()),
dist(minInterval.count(), maxInterval.count()) {
if (minInterval > maxInterval) {
throw std::invalid_argument(
"FunctionScheduler: "
"min time interval must be less or equal than max interval");
}
if (minInterval < milliseconds::zero()) {
throw std::invalid_argument(
"FunctionScheduler: "
"time interval must be non-negative");
}
}
milliseconds operator()() { return milliseconds(dist(generator)); }
};
} // anonymous namespace
FunctionScheduler::FunctionScheduler() {}
FunctionScheduler::~FunctionScheduler() { FunctionScheduler::~FunctionScheduler() {
// make sure to stop the thread (if running) // make sure to stop the thread (if running)
...@@ -37,8 +96,12 @@ void FunctionScheduler::addFunction(const std::function<void()>& cb, ...@@ -37,8 +96,12 @@ void FunctionScheduler::addFunction(const std::function<void()>& cb,
milliseconds interval, milliseconds interval,
StringPiece nameID, StringPiece nameID,
milliseconds startDelay) { milliseconds startDelay) {
LatencyDistribution latencyDistr(false, 0.0); addFunctionGenericDistribution(
addFunction(cb, interval, latencyDistr, nameID, startDelay); cb,
IntervalDistributionFunc(ConstIntervalFunctor(interval)),
nameID.str(),
to<std::string>(interval.count(), "ms"),
startDelay);
} }
void FunctionScheduler::addFunction(const std::function<void()>& cb, void FunctionScheduler::addFunction(const std::function<void()>& cb,
...@@ -46,34 +109,72 @@ void FunctionScheduler::addFunction(const std::function<void()>& cb, ...@@ -46,34 +109,72 @@ void FunctionScheduler::addFunction(const std::function<void()>& cb,
const LatencyDistribution& latencyDistr, const LatencyDistribution& latencyDistr,
StringPiece nameID, StringPiece nameID,
milliseconds startDelay) { milliseconds startDelay) {
if (interval < milliseconds::zero()) { if (latencyDistr.isPoisson) {
throw std::invalid_argument("FunctionScheduler: " addFunctionGenericDistribution(
"time interval must be non-negative"); cb,
IntervalDistributionFunc(
PoissonDistributionFunctor(latencyDistr.poissonMean)),
nameID.str(),
to<std::string>(latencyDistr.poissonMean, "ms (Poisson mean)"),
startDelay);
} else {
addFunction(cb, interval, nameID, startDelay);
}
}
void FunctionScheduler::addFunctionUniformDistribution(
const std::function<void()>& cb,
milliseconds minInterval,
milliseconds maxInterval,
StringPiece nameID,
milliseconds startDelay) {
addFunctionGenericDistribution(
cb,
IntervalDistributionFunc(
UniformDistributionFunctor(minInterval, maxInterval)),
nameID.str(),
to<std::string>(
"[", minInterval.count(), " , ", maxInterval.count(), "] ms"),
startDelay);
}
void FunctionScheduler::addFunctionGenericDistribution(
const std::function<void()>& cb,
const IntervalDistributionFunc& intervalFunc,
const std::string& nameID,
const std::string& intervalDescr,
milliseconds startDelay) {
if (!cb) {
throw std::invalid_argument(
"FunctionScheduler: Scheduled function must be set");
}
if (!intervalFunc) {
throw std::invalid_argument(
"FunctionScheduler: interval distribution function must be set");
} }
if (startDelay < milliseconds::zero()) { if (startDelay < milliseconds::zero()) {
throw std::invalid_argument("FunctionScheduler: " throw std::invalid_argument(
"start delay must be non-negative"); "FunctionScheduler: start delay must be non-negative");
} }
std::lock_guard<std::mutex> l(mutex_); std::lock_guard<std::mutex> l(mutex_);
// check if the nameID is unique // check if the nameID is unique
for (const auto& f : functions_) { for (const auto& f : functions_) {
if (f.isValid() && f.name == nameID) { if (f.isValid() && f.name == nameID) {
throw std::invalid_argument(to<string>( throw std::invalid_argument(
"FunctionScheduler: a function named \"", nameID, to<std::string>("FunctionScheduler: a function named \"",
"\" already exists")); nameID,
"\" already exists"));
} }
} }
if (currentFunction_ && currentFunction_->name == nameID) { if (currentFunction_ && currentFunction_->name == nameID) {
throw std::invalid_argument(to<string>( throw std::invalid_argument(to<std::string>(
"FunctionScheduler: a function named \"", nameID, "FunctionScheduler: a function named \"", nameID, "\" already exists"));
"\" already exists"));
} }
functions_.emplace_back(cb, interval, nameID.str(), startDelay, functions_.emplace_back(cb, intervalFunc, nameID, intervalDescr, startDelay);
latencyDistr.isPoisson, latencyDistr.poissonMean);
if (running_) { if (running_) {
functions_.back().setNextRunTime(steady_clock::now() + startDelay); functions_.back().resetNextRunTime(steady_clock::now());
std::push_heap(functions_.begin(), functions_.end(), fnCmp_); std::push_heap(functions_.begin(), functions_.end(), fnCmp_);
// Signal the running thread to wake up and see if it needs to change it's // Signal the running thread to wake up and see if it needs to change it's
// current scheduling decision. // current scheduling decision.
...@@ -140,11 +241,10 @@ bool FunctionScheduler::start() { ...@@ -140,11 +241,10 @@ bool FunctionScheduler::start() {
// Reset the next run time. for all functions. // Reset the next run time. for all functions.
// note: this is needed since one can shutdown() and start() again // note: this is needed since one can shutdown() and start() again
for (auto& f : functions_) { for (auto& f : functions_) {
f.setNextRunTime(now + f.startDelay); f.resetNextRunTime(now);
VLOG(1) << " - func: " VLOG(1) << " - func: " << (f.name.empty() ? "(anon)" : f.name.c_str())
<< (f.name.empty() ? "(anon)" : f.name.c_str()) << ", period = " << f.intervalDescr
<< ", period = " << f.timeInterval.count() << ", delay = " << f.startDelay.count() << "ms";
<< "ms, delay = " << f.startDelay.count() << "ms";
} }
std::make_heap(functions_.begin(), functions_.end(), fnCmp_); std::make_heap(functions_.begin(), functions_.end(), fnCmp_);
...@@ -217,19 +317,23 @@ void FunctionScheduler::runOneFunction(std::unique_lock<std::mutex>& lock, ...@@ -217,19 +317,23 @@ void FunctionScheduler::runOneFunction(std::unique_lock<std::mutex>& lock,
// maintain the heap property on functions_ while mutex_ is unlocked. // maintain the heap property on functions_ while mutex_ is unlocked.
RepeatFunc func(std::move(functions_.back())); RepeatFunc func(std::move(functions_.back()));
functions_.pop_back(); functions_.pop_back();
if (!func.cb) {
VLOG(5) << func.name << "function has been canceled while waiting";
return;
}
currentFunction_ = &func; currentFunction_ = &func;
// Update the function's run time, and re-insert it into the heap. // Update the function's next run time.
if (steady_) { if (steady_) {
// This allows scheduler to catch up // This allows scheduler to catch up
func.lastRunTime += func.timeInterval; func.setNextRunTimeSteady();
} else { } else {
// Note that we adjust lastRunTime to the current time where we started the // Note that we set nextRunTime based on the current time where we started
// function call, rather than the time when the function finishes. // the function call, rather than the time when the function finishes.
// This ensures that we call the function once every time interval, as // This ensures that we call the function once every time interval, as
// opposed to waiting time interval seconds between calls. (These can be // opposed to waiting time interval seconds between calls. (These can be
// different if the function takes a significant amount of time to run.) // different if the function takes a significant amount of time to run.)
func.lastRunTime = now; func.setNextRunTimeStrict(now);
} }
// Release the lock while we invoke the user's function // Release the lock while we invoke the user's function
...@@ -259,9 +363,6 @@ void FunctionScheduler::runOneFunction(std::unique_lock<std::mutex>& lock, ...@@ -259,9 +363,6 @@ void FunctionScheduler::runOneFunction(std::unique_lock<std::mutex>& lock,
// Re-insert the function into our functions_ heap. // Re-insert the function into our functions_ heap.
// We only maintain the heap property while running_ is set. (running_ may // We only maintain the heap property while running_ is set. (running_ may
// have been cleared while we were invoking the user's function.) // have been cleared while we were invoking the user's function.)
if (func.isPoissonDistr) {
func.setTimeIntervalPoissonDistr();
}
functions_.push_back(std::move(func)); functions_.push_back(std::move(func));
if (running_) { if (running_) {
std::push_heap(functions_.begin(), functions_.end(), fnCmp_); std::push_heap(functions_.begin(), functions_.end(), fnCmp_);
......
...@@ -23,7 +23,6 @@ ...@@ -23,7 +23,6 @@
#include <mutex> #include <mutex>
#include <thread> #include <thread>
#include <vector> #include <vector>
#include <random>
namespace folly { namespace folly {
...@@ -104,12 +103,45 @@ class FunctionScheduler { ...@@ -104,12 +103,45 @@ class FunctionScheduler {
* Add a new function to the FunctionScheduler with a specified * Add a new function to the FunctionScheduler with a specified
* LatencyDistribution * LatencyDistribution
*/ */
void addFunction(const std::function<void()>& cb, void addFunction(
std::chrono::milliseconds interval, const std::function<void()>& cb,
const LatencyDistribution& latencyDistr, std::chrono::milliseconds interval,
StringPiece nameID = StringPiece(), const LatencyDistribution& latencyDistr,
std::chrono::milliseconds startDelay = StringPiece nameID = StringPiece(),
std::chrono::milliseconds(0)); std::chrono::milliseconds startDelay = std::chrono::milliseconds(0));
/**
* Add a new function to the FunctionScheduler with the time
* interval being distributed uniformly within the given interval
* [minInterval, maxInterval].
*/
void addFunctionUniformDistribution(const std::function<void()>& cb,
std::chrono::milliseconds minInterval,
std::chrono::milliseconds maxInterval,
StringPiece nameID,
std::chrono::milliseconds startDelay);
/**
* A type alias for function that is called to determine the time
* interval for the next scheduled run.
*/
using IntervalDistributionFunc = std::function<std::chrono::milliseconds()>;
/**
* Add a new function to the FunctionScheduler. The scheduling interval
* is determined by the interval distribution functor, which is called
* every time the next function execution is scheduled. This allows
* for supporting custom interval distribution algorithms in addition
* to built in constant interval; and Poisson and jitter distributions
* (@see FunctionScheduler::addFunction and
* @see FunctionScheduler::addFunctionJitterInterval).
*/
void addFunctionGenericDistribution(
const std::function<void()>& cb,
const IntervalDistributionFunc& intervalFunc,
const std::string& nameID,
const std::string& intervalDescr,
std::chrono::milliseconds startDelay);
/** /**
* Cancels the function with the specified name, so it will no longer be run. * Cancels the function with the specified name, so it will no longer be run.
...@@ -142,63 +174,56 @@ class FunctionScheduler { ...@@ -142,63 +174,56 @@ class FunctionScheduler {
*/ */
void setThreadName(StringPiece threadName); void setThreadName(StringPiece threadName);
private: private:
struct RepeatFunc { struct RepeatFunc {
std::function<void()> cb; std::function<void()> cb;
std::chrono::milliseconds timeInterval; IntervalDistributionFunc intervalFunc;
std::chrono::steady_clock::time_point lastRunTime; std::chrono::steady_clock::time_point nextRunTime;
std::string name; std::string name;
std::chrono::milliseconds startDelay; std::chrono::milliseconds startDelay;
bool isPoissonDistr; std::string intervalDescr;
std::default_random_engine generator;
std::poisson_distribution<int> poisson_random;
RepeatFunc(const std::function<void()>& cback, RepeatFunc(const std::function<void()>& cback,
std::chrono::milliseconds interval, const IntervalDistributionFunc& intervalFn,
const std::string& nameID, const std::string& nameID,
std::chrono::milliseconds delay, const std::string& intervalDistDescription,
bool poisson = false, std::chrono::milliseconds delay)
double meanPoisson = 1.0) : cb(cback),
: cb(cback), intervalFunc(intervalFn),
timeInterval(interval), nextRunTime(),
lastRunTime(), name(nameID),
name(nameID), startDelay(delay),
startDelay(delay), intervalDescr(intervalDistDescription) {}
isPoissonDistr(poisson),
poisson_random(meanPoisson) {
}
std::chrono::steady_clock::time_point getNextRunTime() const { std::chrono::steady_clock::time_point getNextRunTime() const {
return lastRunTime + timeInterval; return nextRunTime;
} }
void setNextRunTime(std::chrono::steady_clock::time_point time) { void setNextRunTimeStrict(std::chrono::steady_clock::time_point curTime) {
lastRunTime = time - timeInterval; nextRunTime = curTime + intervalFunc();
} }
void setTimeIntervalPoissonDistr() { void setNextRunTimeSteady() { nextRunTime += intervalFunc(); }
if (isPoissonDistr) { void resetNextRunTime(std::chrono::steady_clock::time_point curTime) {
timeInterval = std::chrono::milliseconds(poisson_random(generator)); nextRunTime = curTime + startDelay;
}
} }
void cancel() { void cancel() {
// Simply reset cb to an empty function. // Simply reset cb to an empty function.
cb = std::function<void()>(); cb = std::function<void()>();
} }
bool isValid() const { bool isValid() const { return bool(cb); }
return bool(cb);
}
}; };
struct RunTimeOrder { struct RunTimeOrder {
bool operator()(const RepeatFunc& f1, const RepeatFunc& f2) const { bool operator()(const RepeatFunc& f1, const RepeatFunc& f2) const {
return f1.getNextRunTime() > f2.getNextRunTime(); return f1.getNextRunTime() > f2.getNextRunTime();
} }
}; };
typedef std::vector<RepeatFunc> FunctionHeap; typedef std::vector<RepeatFunc> FunctionHeap;
void run(); void run();
void runOneFunction(std::unique_lock<std::mutex>& lock, void runOneFunction(std::unique_lock<std::mutex>& lock,
std::chrono::steady_clock::time_point now); std::chrono::steady_clock::time_point now);
void cancelFunction(const std::unique_lock<std::mutex> &lock, void cancelFunction(const std::unique_lock<std::mutex>& lock,
FunctionHeap::iterator it); FunctionHeap::iterator it);
std::thread thread_; std::thread thread_;
......
...@@ -15,7 +15,11 @@ ...@@ -15,7 +15,11 @@
*/ */
#include <folly/experimental/FunctionScheduler.h> #include <folly/experimental/FunctionScheduler.h>
#include <algorithm>
#include <atomic> #include <atomic>
#include <cassert>
#include <random>
#include <folly/Random.h>
#include <gtest/gtest.h> #include <gtest/gtest.h>
using namespace folly; using namespace folly;
...@@ -31,8 +35,12 @@ namespace { ...@@ -31,8 +35,12 @@ namespace {
* to run. * to run.
*/ */
static const auto timeFactor = std::chrono::milliseconds(100); static const auto timeFactor = std::chrono::milliseconds(100);
std::chrono::milliseconds testInterval(int n) { std::chrono::milliseconds testInterval(int n) { return n * timeFactor; }
return n * timeFactor; int getTicksWithinRange(int n, int min, int max) {
assert(min <= max);
n = std::max(min, n);
n = std::min(max, n);
return n;
} }
void delay(int n) { void delay(int n) {
std::chrono::microseconds usec(n * timeFactor); std::chrono::microseconds usec(n * timeFactor);
...@@ -321,3 +329,86 @@ TEST(FunctionScheduler, SteadyCatchup) { ...@@ -321,3 +329,86 @@ TEST(FunctionScheduler, SteadyCatchup) {
// enough to catch back up to schedule // enough to catch back up to schedule
EXPECT_NEAR(100, ticks.load(), 10); EXPECT_NEAR(100, ticks.load(), 10);
} }
TEST(FunctionScheduler, UniformDistribution) {
int total = 0;
const int kTicks = 2;
std::chrono::milliseconds minInterval =
testInterval(kTicks) - (timeFactor / 5);
std::chrono::milliseconds maxInterval =
testInterval(kTicks) + (timeFactor / 5);
FunctionScheduler fs;
fs.addFunctionUniformDistribution([&] { total += 2; },
minInterval,
maxInterval,
"UniformDistribution",
std::chrono::milliseconds(0));
fs.start();
delay(1);
EXPECT_EQ(2, total);
delay(kTicks);
EXPECT_EQ(4, total);
delay(kTicks);
EXPECT_EQ(6, total);
fs.shutdown();
delay(2);
EXPECT_EQ(6, total);
}
TEST(FunctionScheduler, ExponentialBackoff) {
int total = 0;
int expectedInterval = 0;
int nextInterval = 2;
FunctionScheduler fs;
fs.addFunctionGenericDistribution(
[&] { total += 2; },
[&expectedInterval, nextInterval]() mutable {
expectedInterval = nextInterval;
nextInterval *= nextInterval;
return testInterval(expectedInterval);
},
"ExponentialBackoff",
"2^n * 100ms",
std::chrono::milliseconds(0));
fs.start();
delay(1);
EXPECT_EQ(2, total);
delay(expectedInterval);
EXPECT_EQ(4, total);
delay(expectedInterval);
EXPECT_EQ(6, total);
fs.shutdown();
delay(2);
EXPECT_EQ(6, total);
}
TEST(FunctionScheduler, GammaIntervalDistribution) {
int total = 0;
int expectedInterval = 0;
FunctionScheduler fs;
std::default_random_engine generator(folly::Random::rand32());
// The alpha and beta arguments are selected, somewhat randomly, to be 2.0.
// These values do not matter much in this test, as we are not testing the
// std::gamma_distribution itself...
std::gamma_distribution<double> gamma(2.0, 2.0);
fs.addFunctionGenericDistribution(
[&] { total += 2; },
[&expectedInterval, generator, gamma]() mutable {
expectedInterval =
getTicksWithinRange(static_cast<int>(gamma(generator)), 2, 10);
return testInterval(expectedInterval);
},
"GammaDistribution",
"gamma(2.0,2.0)*100ms",
std::chrono::milliseconds(0));
fs.start();
delay(1);
EXPECT_EQ(2, total);
delay(expectedInterval);
EXPECT_EQ(4, total);
delay(expectedInterval);
EXPECT_EQ(6, total);
fs.shutdown();
delay(2);
EXPECT_EQ(6, total);
}
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment