Commit be171a18 authored by Adam Simpkins's avatar Adam Simpkins Committed by Praveen Kumar Ramakrishnan

update FunctionScheduler to use std::chrono::steady_clock

Summary:
Use std::chrono::steady_clock instead of clock_gettime(CLOCK_MONOTONIC).
In particular this fixes the build on Mac OS X, which doesn't have
CLOCK_MONOTONIC.

This also updates the code to use steady_clock::time_point correctly, instead
of using a raw milliseconds value for time since the epoch.

Test Plan:
Included unit tests, which were copied over from the legacy internal Facebook
(non-folly) version of this code.

Reviewed By: ldbrandy@fb.com

Subscribers: jwatzman, doug, fbcode-common-diffs@, net-systems@, exa, folly-diffs@, yfeldblum, chalfant

FB internal diff: D2051557

Signature: t1:2051557:1431019654:ee76cfcf8318cc3d8a8d1522b3fc97f08831ecf4
parent e3eecdae
...@@ -19,26 +19,9 @@ ...@@ -19,26 +19,9 @@
#include <folly/Conv.h> #include <folly/Conv.h>
#include <folly/String.h> #include <folly/String.h>
#ifdef _POSIX_MONOTONIC_CLOCK
#define FOLLY_TIME_MONOTONIC_CLOCK CLOCK_MONOTONIC
#else
#define FOLLY_TIME_MONOTONIC_CLOCK CLOCK_REALTIME
#endif
using namespace std; using namespace std;
using std::chrono::seconds;
using std::chrono::milliseconds; using std::chrono::milliseconds;
using std::chrono::steady_clock;
static milliseconds nowInMS() {
struct timespec ts /*= void*/;
if (clock_gettime(FOLLY_TIME_MONOTONIC_CLOCK, &ts)) {
// Only possible failures are EFAULT or EINVAL, both practically
// impossible. But an assert can't hurt.
assert(false);
}
return milliseconds(
static_cast<int64_t>(ts.tv_sec * 1000.0 + ts.tv_nsec / 1000000.0 + 0.5));
}
namespace folly { namespace folly {
...@@ -91,7 +74,7 @@ void FunctionScheduler::addFunctionInternal(const std::function<void()>& cb, ...@@ -91,7 +74,7 @@ void FunctionScheduler::addFunctionInternal(const std::function<void()>& cb,
functions_.emplace_back(cb, interval, nameID.str(), startDelay, functions_.emplace_back(cb, interval, nameID.str(), startDelay,
latencyDistr.isPoisson, latencyDistr.poissonMean); latencyDistr.isPoisson, latencyDistr.poissonMean);
if (running_) { if (running_) {
functions_.back().setNextRunTime(nowInMS() + startDelay); functions_.back().setNextRunTime(steady_clock::now() + startDelay);
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.
...@@ -154,7 +137,7 @@ bool FunctionScheduler::start() { ...@@ -154,7 +137,7 @@ bool FunctionScheduler::start() {
VLOG(1) << "Starting FunctionScheduler with " << functions_.size() VLOG(1) << "Starting FunctionScheduler with " << functions_.size()
<< " functions."; << " functions.";
milliseconds now(nowInMS()); auto now = steady_clock::now();
// 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_) {
...@@ -198,7 +181,7 @@ void FunctionScheduler::run() { ...@@ -198,7 +181,7 @@ void FunctionScheduler::run() {
continue; continue;
} }
milliseconds now(nowInMS()); auto now = steady_clock::now();
// Move the next function to run to the end of functions_ // Move the next function to run to the end of functions_
std::pop_heap(functions_.begin(), functions_.end(), fnCmp_); std::pop_heap(functions_.begin(), functions_.end(), fnCmp_);
...@@ -224,7 +207,7 @@ void FunctionScheduler::run() { ...@@ -224,7 +207,7 @@ void FunctionScheduler::run() {
} }
void FunctionScheduler::runOneFunction(std::unique_lock<std::mutex>& lock, void FunctionScheduler::runOneFunction(std::unique_lock<std::mutex>& lock,
std::chrono::milliseconds now) { steady_clock::time_point now) {
DCHECK(lock.mutex() == &mutex_); DCHECK(lock.mutex() == &mutex_);
DCHECK(lock.owns_lock()); DCHECK(lock.owns_lock());
......
...@@ -135,7 +135,7 @@ class FunctionScheduler { ...@@ -135,7 +135,7 @@ class FunctionScheduler {
struct RepeatFunc { struct RepeatFunc {
std::function<void()> cb; std::function<void()> cb;
std::chrono::milliseconds timeInterval; std::chrono::milliseconds timeInterval;
std::chrono::milliseconds lastRunTime; std::chrono::steady_clock::time_point lastRunTime;
std::string name; std::string name;
std::chrono::milliseconds startDelay; std::chrono::milliseconds startDelay;
bool isPoissonDistr; bool isPoissonDistr;
...@@ -150,17 +150,17 @@ class FunctionScheduler { ...@@ -150,17 +150,17 @@ class FunctionScheduler {
double meanPoisson = 1.0) double meanPoisson = 1.0)
: cb(cback), : cb(cback),
timeInterval(interval), timeInterval(interval),
lastRunTime(0), lastRunTime(),
name(nameID), name(nameID),
startDelay(delay), startDelay(delay),
isPoissonDistr(poisson), isPoissonDistr(poisson),
poisson_random(meanPoisson) { poisson_random(meanPoisson) {
} }
std::chrono::milliseconds getNextRunTime() const { std::chrono::steady_clock::time_point getNextRunTime() const {
return lastRunTime + timeInterval; return lastRunTime + timeInterval;
} }
void setNextRunTime(std::chrono::milliseconds time) { void setNextRunTime(std::chrono::steady_clock::time_point time) {
lastRunTime = time - timeInterval; lastRunTime = time - timeInterval;
} }
void setTimeIntervalPoissonDistr() { void setTimeIntervalPoissonDistr() {
...@@ -185,7 +185,7 @@ class FunctionScheduler { ...@@ -185,7 +185,7 @@ class FunctionScheduler {
void run(); void run();
void runOneFunction(std::unique_lock<std::mutex>& lock, void runOneFunction(std::unique_lock<std::mutex>& lock,
std::chrono::milliseconds 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);
......
/*
* Copyright 2015 Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/FunctionScheduler.h>
#include <atomic>
#include <gtest/gtest.h>
using namespace folly;
using std::chrono::milliseconds;
namespace {
/*
* Helper functions for controlling how long this test takes.
*
* Using larger intervals here will make the tests less flaky when run on
* heavily loaded systems. However, this will also make the tests take longer
* to run.
*/
static const auto timeFactor = std::chrono::milliseconds(100);
std::chrono::milliseconds testInterval(int n) {
return n * timeFactor;
}
void delay(int n) {
std::chrono::microseconds usec(n * timeFactor);
usleep(usec.count());
}
} // unnamed namespace
TEST(FunctionScheduler, SimpleAdd) {
int total = 0;
FunctionScheduler fs;
fs.addFunction([&] { total += 2; }, testInterval(2), "add2");
fs.start();
delay(1);
EXPECT_EQ(2, total);
fs.shutdown();
delay(2);
EXPECT_EQ(2, total);
}
TEST(FunctionScheduler, AddCancel) {
int total = 0;
FunctionScheduler fs;
fs.addFunction([&] { total += 2; }, testInterval(2), "add2");
fs.start();
delay(1);
EXPECT_EQ(2, total);
delay(2);
EXPECT_EQ(4, total);
EXPECT_TRUE(fs.cancelFunction("add2"));
EXPECT_FALSE(fs.cancelFunction("NO SUCH FUNC"));
delay(2);
EXPECT_EQ(4, total);
fs.addFunction([&] { total += 1; }, testInterval(2), "add2");
EXPECT_FALSE(fs.start()); // already running
delay(1);
EXPECT_EQ(5, total);
delay(2);
EXPECT_EQ(6, total);
fs.shutdown();
}
TEST(FunctionScheduler, AddCancel2) {
int total = 0;
FunctionScheduler fs;
// Test adds and cancels while the scheduler is stopped
EXPECT_FALSE(fs.cancelFunction("add2"));
fs.addFunction([&] { total += 1; }, testInterval(2), "add2");
EXPECT_TRUE(fs.cancelFunction("add2"));
EXPECT_FALSE(fs.cancelFunction("add2"));
fs.addFunction([&] { total += 2; }, testInterval(2), "add2");
fs.addFunction([&] { total += 3; }, testInterval(3), "add3");
EXPECT_EQ(0, total);
fs.start();
delay(1);
EXPECT_EQ(5, total);
// Cancel add2 while the scheduler is running
EXPECT_TRUE(fs.cancelFunction("add2"));
EXPECT_FALSE(fs.cancelFunction("add2"));
EXPECT_FALSE(fs.cancelFunction("bogus"));
delay(3);
EXPECT_EQ(8, total);
EXPECT_TRUE(fs.cancelFunction("add3"));
// Test a function that cancels itself
int selfCancelCount = 0;
fs.addFunction(
[&] {
++selfCancelCount;
if (selfCancelCount > 2) {
fs.cancelFunction("selfCancel");
}
},
testInterval(1), "selfCancel", testInterval(1));
delay(4);
EXPECT_EQ(3, selfCancelCount);
EXPECT_FALSE(fs.cancelFunction("selfCancel"));
// Test a function that schedules another function
int adderCount = 0;
int fn2Count = 0;
auto fn2 = [&] { ++fn2Count; };
auto fnAdder = [&] {
++adderCount;
if (adderCount == 2) {
fs.addFunction(fn2, testInterval(3), "fn2", testInterval(2));
}
};
fs.addFunction(fnAdder, testInterval(4), "adder");
// t0: adder fires
delay(1); // t1
EXPECT_EQ(1, adderCount);
EXPECT_EQ(0, fn2Count);
// t4: adder fires, schedules fn2
delay(4); // t5
EXPECT_EQ(2, adderCount);
EXPECT_EQ(0, fn2Count);
// t6: fn2 fires
delay(2); // t7
EXPECT_EQ(2, adderCount);
EXPECT_EQ(1, fn2Count);
// t8: adder fires
// t9: fn2 fires
delay(3); // t10
EXPECT_EQ(3, adderCount);
EXPECT_EQ(2, fn2Count);
EXPECT_TRUE(fs.cancelFunction("fn2"));
EXPECT_TRUE(fs.cancelFunction("adder"));
delay(5); // t10
EXPECT_EQ(3, adderCount);
EXPECT_EQ(2, fn2Count);
EXPECT_EQ(8, total);
EXPECT_EQ(3, selfCancelCount);
}
TEST(FunctionScheduler, AddMultiple) {
int total = 0;
FunctionScheduler fs;
fs.addFunction([&] { total += 2; }, testInterval(2), "add2");
fs.addFunction([&] { total += 3; }, testInterval(3), "add3");
EXPECT_THROW(fs.addFunction([&] { total += 2; }, testInterval(2), "add2"),
std::invalid_argument); // function name already exists
fs.start();
delay(1);
EXPECT_EQ(5, total);
delay(4);
EXPECT_EQ(12, total);
EXPECT_TRUE(fs.cancelFunction("add2"));
delay(2);
EXPECT_EQ(15, total);
fs.shutdown();
delay(3);
EXPECT_EQ(15, total);
fs.shutdown();
}
TEST(FunctionScheduler, AddAfterStart) {
int total = 0;
FunctionScheduler fs;
fs.addFunction([&] { total += 2; }, testInterval(2), "add2");
fs.addFunction([&] { total += 3; }, testInterval(2), "add3");
fs.start();
delay(3);
EXPECT_EQ(10, total);
fs.addFunction([&] { total += 2; }, testInterval(3), "add22");
delay(2);
EXPECT_EQ(17, total);
}
TEST(FunctionScheduler, ShutdownStart) {
int total = 0;
FunctionScheduler fs;
fs.addFunction([&] { total += 2; }, testInterval(2), "add2");
fs.start();
delay(1);
fs.shutdown();
fs.start();
delay(1);
EXPECT_EQ(4, total);
EXPECT_FALSE(fs.cancelFunction("add3")); // non existing
delay(2);
EXPECT_EQ(6, total);
}
TEST(FunctionScheduler, AddInvalid) {
int total = 0;
FunctionScheduler fs;
// interval may not be negative
EXPECT_THROW(fs.addFunction([&] { total += 2; }, testInterval(-1), "add2"),
std::invalid_argument);
EXPECT_FALSE(fs.cancelFunction("addNoFunc"));
}
TEST(FunctionScheduler, NoFunctions) {
FunctionScheduler fs;
EXPECT_TRUE(fs.start());
fs.shutdown();
FunctionScheduler fs2;
fs2.shutdown();
}
TEST(FunctionScheduler, AddWhileRunning) {
int total = 0;
FunctionScheduler fs;
fs.start();
delay(1);
fs.addFunction([&] { total += 2; }, testInterval(2), "add2");
// The function should be invoked nearly immediately when we add it
// and the FunctionScheduler is already running
usleep(50000);
EXPECT_EQ(2, total);
delay(2);
EXPECT_EQ(4, total);
}
TEST(FunctionScheduler, NoShutdown) {
int total = 0;
{
FunctionScheduler fs;
fs.addFunction([&] { total += 2; }, testInterval(1), "add2");
fs.start();
usleep(50000);
EXPECT_EQ(2, total);
}
// Destroyed the FunctionScheduler without calling shutdown.
// Everything should have been cleaned up, and the function will no longer
// get called.
delay(2);
EXPECT_EQ(2, total);
}
TEST(FunctionScheduler, StartDelay) {
int total = 0;
FunctionScheduler fs;
fs.addFunction([&] { total += 2; }, testInterval(2), "add2",
testInterval(2));
fs.addFunction([&] { total += 3; }, testInterval(3), "add3",
testInterval(2));
EXPECT_THROW(fs.addFunction([&] { total += 2; }, testInterval(3),
"addX", testInterval(-1)),
std::invalid_argument);
fs.start();
delay(1); // t1
EXPECT_EQ(0, total);
// t2 : add2 total=2
// t2 : add3 total=5
delay(2); // t3
EXPECT_EQ(5, total);
// t4 : add2: total=7
// t5 : add3: total=10
// t6 : add2: total=12
delay(4); // t7
EXPECT_EQ(12, total);
fs.cancelFunction("add2");
// t8 : add3: total=15
delay(2); // t9
EXPECT_EQ(15, total);
fs.shutdown();
delay(3);
EXPECT_EQ(15, total);
fs.shutdown();
}
TEST(FunctionScheduler, NoSteadyCatchup) {
std::atomic<int> ticks(0);
FunctionScheduler fs;
// fs.setSteady(false); is the default
fs.addFunction([&ticks] {
if (++ticks == 2) {
std::this_thread::sleep_for(
std::chrono::milliseconds(200));
}
},
milliseconds(5));
fs.start();
std::this_thread::sleep_for(std::chrono::milliseconds(500));
// no steady catch up means we'd tick once for 200ms, then remaining
// 300ms / 5 = 60 times
EXPECT_LE(ticks.load(), 61);
}
TEST(FunctionScheduler, SteadyCatchup) {
std::atomic<int> ticks(0);
FunctionScheduler fs;
fs.setSteady(true);
fs.addFunction([&ticks] {
if (++ticks == 2) {
std::this_thread::sleep_for(
std::chrono::milliseconds(200));
}
},
milliseconds(5));
fs.start();
std::this_thread::sleep_for(std::chrono::milliseconds(500));
// tick every 5ms. Despite tick == 2 is slow, later ticks should be fast
// enough to catch back up to schedule
EXPECT_NEAR(100, ticks.load(), 10);
}
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