You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by ad...@apache.org on 2017/09/29 20:31:29 UTC
kudu git commit: periodic: prevent runaway callback loops
Repository: kudu
Updated Branches:
refs/heads/master 0340a4376 -> 8f6ef48ea
periodic: prevent runaway callback loops
One of the key semantics of PeriodicTimer is that Stop() doesn't wait for
an outstanding scheduled callback to finish. It returns immediately and
when the scheduled callback runs, it'll see that Stop() was called and exit.
However, if Start() is called after Stop() but before the callback runs,
both "callback loops" will remain alive. The existing locking will ensure
that their periods converge and that the functor is only invoked once per
period, but it's still an unnecessary resource drain.
This patch addresses this by assigning a "generation" to each callback loop.
When a callback runs, it'll exit if its generation is older than the timer's
current generation. That means an additional N scheduled callbacks for N
Stop/Start call pairs, but after one period all but one of the callbacks
will detect the generational change and exit.
All of this wouldn't be necessary if Stop() could cancel the underlying
callback, but Messenger::ScheduleOnReactor() doesn't support cancelation,
and it'd be far more work to add that than to employ this workaround.
While making this change I decided to replace the various atomic primitives
with basic ones protected by the existing spinlock; I found it easier to
grok the critical sections this way.
Change-Id: I37bfd547db2a6c58d15d228979c88b9b871f72c5
Reviewed-on: http://gerrit.cloudera.org:8080/8116
Tested-by: Kudu Jenkins
Reviewed-by: Alexey Serbin <as...@cloudera.com>
Reviewed-by: Todd Lipcon <to...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/8f6ef48e
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/8f6ef48e
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/8f6ef48e
Branch: refs/heads/master
Commit: 8f6ef48ea6cb178e5e5b3c88edf5bfc221d9354a
Parents: 0340a43
Author: Adar Dembo <ad...@cloudera.com>
Authored: Wed Sep 20 17:48:00 2017 -0700
Committer: Adar Dembo <ad...@cloudera.com>
Committed: Fri Sep 29 20:30:39 2017 +0000
----------------------------------------------------------------------
src/kudu/rpc/periodic-test.cc | 60 ++++++++++++++++++++++++++++++--------
src/kudu/rpc/periodic.cc | 58 +++++++++++++++++++++++++++---------
src/kudu/rpc/periodic.h | 56 ++++++++++++++++++++++++++++-------
src/kudu/rpc/reactor.cc | 7 +++--
4 files changed, 143 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/8f6ef48e/src/kudu/rpc/periodic-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/periodic-test.cc b/src/kudu/rpc/periodic-test.cc
index 39e2e40..d0d2b99 100644
--- a/src/kudu/rpc/periodic-test.cc
+++ b/src/kudu/rpc/periodic-test.cc
@@ -34,8 +34,17 @@ using std::shared_ptr;
namespace kudu {
namespace rpc {
-class PeriodicTimerTest : public KuduTest,
- public ::testing::WithParamInterface<double> {
+class PeriodicTimerTest : public KuduTest {
+ public:
+ PeriodicTimerTest()
+ : period_ms_(200) {}
+
+ protected:
+ const int64_t period_ms_;
+};
+
+class JitteredPeriodicTimerTest : public PeriodicTimerTest,
+ public ::testing::WithParamInterface<double> {
public:
// In TSAN builds it takes a long time to de-schedule a thread. Also,
// the actual time that thread spends sleeping in SleepFor() scenarios
@@ -44,13 +53,12 @@ class PeriodicTimerTest : public KuduTest,
// is observed even under substantial load. Otherwise it would be necessary
// to introduce additional logic to verify that the actual timings satisfy
// the implicit constraints of the test scenarios below.
- PeriodicTimerTest()
- : period_ms_(200),
- counter_(0) {
+ JitteredPeriodicTimerTest()
+ : counter_(0) {
}
virtual void SetUp() override {
- KuduTest::SetUp();
+ PeriodicTimerTest::SetUp();
MessengerBuilder builder("test");
ASSERT_OK(builder.Build(&messenger_));
@@ -70,17 +78,16 @@ class PeriodicTimerTest : public KuduTest,
}
protected:
- const int64_t period_ms_;
atomic<int64_t> counter_;
shared_ptr<Messenger> messenger_;
shared_ptr<PeriodicTimer> timer_;
};
INSTANTIATE_TEST_CASE_P(AllJitterModes,
- PeriodicTimerTest,
+ JitteredPeriodicTimerTest,
::testing::Values(0.0, 0.25));
-TEST_P(PeriodicTimerTest, TestStartStop) {
+TEST_P(JitteredPeriodicTimerTest, TestStartStop) {
// Before the timer starts, the counter's value should not change.
SleepFor(MonoDelta::FromMilliseconds(period_ms_ * 2));
ASSERT_EQ(0, counter_);
@@ -102,7 +109,7 @@ TEST_P(PeriodicTimerTest, TestStartStop) {
counter_ == v + 1);
}
-TEST_P(PeriodicTimerTest, TestReset) {
+TEST_P(JitteredPeriodicTimerTest, TestReset) {
timer_->Start();
MonoTime start_time = MonoTime::Now();
@@ -119,7 +126,7 @@ TEST_P(PeriodicTimerTest, TestReset) {
}
}
-TEST_P(PeriodicTimerTest, TestResetWithDelta) {
+TEST_P(JitteredPeriodicTimerTest, TestResetWithDelta) {
timer_->Start();
timer_->Snooze(MonoDelta::FromMilliseconds(period_ms_ * 5));
@@ -133,7 +140,7 @@ TEST_P(PeriodicTimerTest, TestResetWithDelta) {
});
}
-TEST_P(PeriodicTimerTest, TestStartWithDelta) {
+TEST_P(JitteredPeriodicTimerTest, TestStartWithDelta) {
timer_->Start(MonoDelta::FromMilliseconds(period_ms_ * 5));
// One period later, the counter still hasn't incremented...
@@ -146,5 +153,34 @@ TEST_P(PeriodicTimerTest, TestStartWithDelta) {
});
}
+TEST_F(PeriodicTimerTest, TestCallbackRestartsTimer) {
+ const int64_t kPeriods = 10;
+
+ shared_ptr<Messenger> messenger;
+ ASSERT_OK(MessengerBuilder("test").Build(&messenger));
+
+ // Create a timer that restarts itself from within its functor.
+ shared_ptr<PeriodicTimer> timer = PeriodicTimer::Create(
+ messenger,
+ [&] {
+ timer->Stop();
+ timer->Start();
+ },
+ MonoDelta::FromMilliseconds(period_ms_),
+ 0.0); // jittering would just complicate the measurements below
+
+ // Run the timer for a fixed amount of time.
+ timer->Start();
+ SleepFor(MonoDelta::FromMilliseconds(period_ms_ * kPeriods));
+ timer->Stop();
+
+ // Although the timer is restarted by its functor, its overall period should
+ // remain more or less the same (since the period expired just as the functor
+ // ran). As such, we should see no more than three callbacks per period:
+ // one to start scheduling the callback loop, one when it fires, and one more
+ // after it has been replaced by a new callback loop.
+ ASSERT_LE(timer->NumCallbacksForTests(), kPeriods * 3);
+}
+
} // namespace rpc
} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/8f6ef48e/src/kudu/rpc/periodic.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/periodic.cc b/src/kudu/rpc/periodic.cc
index e3bcf71..f4b33db 100644
--- a/src/kudu/rpc/periodic.cc
+++ b/src/kudu/rpc/periodic.cc
@@ -25,7 +25,6 @@
#include <glog/logging.h>
#include "kudu/rpc/messenger.h"
-#include "kudu/util/atomic.h"
#include "kudu/util/monotime.h"
#include "kudu/util/random.h"
#include "kudu/util/random_util.h"
@@ -56,6 +55,8 @@ PeriodicTimer::PeriodicTimer(
period_(period),
jitter_pct_(jitter_pct),
rng_(GetRandomSeed32()),
+ current_callback_generation_(0),
+ num_callbacks_for_tests_(0),
started_(false) {
DCHECK_GE(jitter_pct_, 0);
DCHECK_LE(jitter_pct_, 1);
@@ -66,22 +67,34 @@ PeriodicTimer::~PeriodicTimer() {
}
void PeriodicTimer::Start(boost::optional<MonoDelta> next_task_delta) {
- if (!started_.CompareAndSwap(false, true)) {
- Snooze(std::move(next_task_delta));
- Callback();
+ std::unique_lock<simple_spinlock> l(lock_);
+ if (!started_) {
+ started_ = true;
+ SnoozeUnlocked(std::move(next_task_delta));
+ int new_callback_generation = ++current_callback_generation_;
+
+ // Invoke Callback() with the lock released.
+ l.unlock();
+ Callback(new_callback_generation);
}
}
void PeriodicTimer::Stop() {
- started_.Store(false);
+ std::lock_guard<simple_spinlock> l(lock_);
+ started_ = false;
}
void PeriodicTimer::Snooze(boost::optional<MonoDelta> next_task_delta) {
- if (!started_.Load()) {
+ std::lock_guard<simple_spinlock> l(lock_);
+ SnoozeUnlocked(std::move(next_task_delta));
+}
+
+void PeriodicTimer::SnoozeUnlocked(boost::optional<MonoDelta> next_task_delta) {
+ DCHECK(lock_.is_locked());
+ if (!started_) {
return;
}
- std::lock_guard<simple_spinlock> l(lock_);
if (!next_task_delta) {
// Given jitter percentage J and period P, this yields a delay somewhere
// between (1-J)*P and (1+J)*P.
@@ -101,11 +114,12 @@ MonoDelta PeriodicTimer::GetMinimumPeriod() {
period_.ToMilliseconds());
}
-void PeriodicTimer::Callback() {
- if (!started_.Load()) {
- return;
- }
+int64_t PeriodicTimer::NumCallbacksForTests() const {
+ std::lock_guard<simple_spinlock> l(lock_);
+ return num_callbacks_for_tests_;
+}
+void PeriodicTimer::Callback(int64_t my_callback_generation) {
// To simplify the implementation, a timer may have only one outstanding
// callback scheduled at a time. This means that once the callback is
// scheduled, the timer's task cannot run any earlier than whenever the
@@ -125,6 +139,24 @@ void PeriodicTimer::Callback() {
bool run_task = false;
{
std::lock_guard<simple_spinlock> l(lock_);
+ num_callbacks_for_tests_++;
+
+ // If the timer was stopped, exit.
+ if (!started_) {
+ return;
+ }
+
+ // If there's a new callback loop in town, exit.
+ //
+ // We could check again just before calling Messenger::ScheduleOnReactor()
+ // (in case someone else restarted the timer while the functor ran, or in
+ // case the functor itself restarted the timer), but there's no real reason
+ // to do so: the very next iteration of this callback loop will wind up here
+ // and exit.
+ if (current_callback_generation_ > my_callback_generation) {
+ return;
+ }
+
MonoTime now = MonoTime::Now();
if (now < next_task_time_) {
// It's not yet time to run the task. Reduce the scheduled delay if
@@ -144,13 +176,13 @@ void PeriodicTimer::Callback() {
// Capture a weak_ptr reference into the submitted functor so that we can
// safely handle the functor outliving its timer.
weak_ptr<PeriodicTimer> w = shared_from_this();
- messenger_->ScheduleOnReactor([w](const Status& s) {
+ messenger_->ScheduleOnReactor([w, my_callback_generation](const Status& s) {
if (!s.ok()) {
// The reactor was shut down.
return;
}
if (auto timer = w.lock()) {
- timer->Callback();
+ timer->Callback(my_callback_generation);
}
}, delay);
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/8f6ef48e/src/kudu/rpc/periodic.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/periodic.h b/src/kudu/rpc/periodic.h
index ada5020..f5b1e77 100644
--- a/src/kudu/rpc/periodic.h
+++ b/src/kudu/rpc/periodic.h
@@ -17,13 +17,14 @@
#pragma once
+#include <cstdint>
#include <functional>
#include <memory>
#include <boost/optional/optional.hpp>
+#include <gtest/gtest_prod.h>
#include "kudu/gutil/macros.h"
-#include "kudu/util/atomic.h"
#include "kudu/util/locks.h"
#include "kudu/util/make_shared.h"
#include "kudu/util/monotime.h"
@@ -36,9 +37,21 @@ class Messenger;
// Repeatedly runs a task on a fixed period.
//
-// PeriodicTimer is built using Messenger::ScheduleOnReactor() as its
-// scheduling primitive. This is merely for convenience; it could also be built
-// on libev, a hashed wheel timer, or something equivalent.
+// PeriodicTimer's periodicity is maintained via tail recursive calls to
+// Messenger::ScheduleOnReactor(). Every time the scheduled callback is
+// invoked, it checks the current time, updates some internal bookkeeping,
+// runs the user's task if the time is right, and makes another call to
+// Messenger::ScheduleOnReactor() to run itself again in the future. This
+// looping behavior is called a "callback loop".
+//
+// Every time Stop() and then Start() (or just Start(), if this is the first
+// such call) are invoked, PeriodicTimer will kick off a new callback loop. If
+// there was an old loop, it remains intact until its scheduled callback runs,
+// at which point it will detect that a new loop was created and exit.
+//
+// The use of Messenger::ScheduleOnReactor() is merely for convenience;
+// PeriodicTimer could also be built on libev, a hashed wheel timer, o
+// something equivalent.
//
// PeriodicTimers have shared ownership, but that's largely an implementation
// detail to support asynchronous stopping. Users can treat them as exclusively
@@ -108,6 +121,8 @@ class PeriodicTimer : public std::enable_shared_from_this<PeriodicTimer> {
void Stop();
private:
+ FRIEND_TEST(PeriodicTimerTest, TestCallbackRestartsTimer);
+
PeriodicTimer(std::shared_ptr<Messenger> messenger,
RunTaskFunctor functor,
MonoDelta period,
@@ -118,7 +133,17 @@ class PeriodicTimer : public std::enable_shared_from_this<PeriodicTimer> {
MonoDelta GetMinimumPeriod();
// Called by Messenger::ScheduleOnReactor when the timer fires.
- void Callback();
+ // 'my_callback_generation' is the callback generation assigned to this loop
+ // when it was constructed.
+ void Callback(int64_t my_callback_generation);
+
+ // Like Snooze() but must be called with 'lock_' held.
+ void SnoozeUnlocked(boost::optional<MonoDelta> next_task_delta = boost::none);
+
+ // Returns the number of times that Callback() has been called by this timer.
+ //
+ // Should only be used for tests!
+ int64_t NumCallbacksForTests() const;
// Schedules invocations of Callback() in the future.
std::shared_ptr<Messenger> messenger_;
@@ -132,18 +157,29 @@ class PeriodicTimer : public std::enable_shared_from_this<PeriodicTimer> {
// User-specified jitter percentage.
const double jitter_pct_;
- // Protects 'rng_' and 'next_task_time_'.
- simple_spinlock lock_;
+ // Protects all mutable state below.
+ mutable simple_spinlock lock_;
// PRNG used when generating jitter.
Random rng_;
- // Whether the timer is running or not.
- AtomicBool started_;
-
// The next time at which the task's functor should be run.
MonoTime next_task_time_;
+ // The most recent callback generation.
+ //
+ // When started, a callback loop is assigned a generation, which it remembers
+ // for its entire lifespan. If 'current_callback_generation_' exceeds the
+ // loop's assigned generation, that means another loop has been created and
+ // the (now old) loop should exit.
+ int64_t current_callback_generation_;
+
+ // The number of times that Callback() has been invoked.
+ int64_t num_callbacks_for_tests_;
+
+ // Whether the timer is running or not.
+ bool started_;
+
ALLOW_MAKE_SHARED(PeriodicTimer);
DISALLOW_COPY_AND_ASSIGN(PeriodicTimer);
http://git-wip-us.apache.org/repos/asf/kudu/blob/8f6ef48e/src/kudu/rpc/reactor.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/reactor.cc b/src/kudu/rpc/reactor.cc
index a7b71db..8eb096f 100644
--- a/src/kudu/rpc/reactor.cc
+++ b/src/kudu/rpc/reactor.cc
@@ -98,7 +98,6 @@ METRIC_DEFINE_histogram(server, reactor_active_latency_us,
kudu::MetricUnit::kMicroseconds,
"TODO", 1000000, 2);
-
namespace kudu {
namespace rpc {
@@ -138,8 +137,10 @@ ReactorThread::ReactorThread(Reactor *reactor, const MessengerBuilder& bld)
total_server_conns_cnt_(0) {
if (bld.metric_entity_) {
- invoke_us_histogram_ = METRIC_reactor_active_latency_us.Instantiate(bld.metric_entity_);
- load_percent_histogram_ = METRIC_reactor_load_percent.Instantiate(bld.metric_entity_);
+ invoke_us_histogram_ =
+ METRIC_reactor_active_latency_us.Instantiate(bld.metric_entity_);
+ load_percent_histogram_ =
+ METRIC_reactor_load_percent.Instantiate(bld.metric_entity_);
}
}