You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by to...@apache.org on 2016/03/22 19:44:16 UTC
incubator-kudu git commit: Add per tablet write RPC throttling
Repository: incubator-kudu
Updated Branches:
refs/heads/master 85fd60dea -> f8a677b1f
Add per tablet write RPC throttling
This commit adds per tablet write RPC throttling, a simple way to
provide some degree of isolation in multi-user environment.
Token bucket is used for throttling, and three config options are
added to specify throttling behavior:
1. tablet_throttler_rpc_per_sec
Maximum write RPC allowed per second,
default 0, which means no limit.
2. tablet_throttler_bytes_per_sec
Maximum write RPC request bytes allowed per second,
defualt 0, which means no limit.
3. tablet_throttler_burst_factor
Burst factor, user can accumulate tokens and use them in a
short time period to archive burst throughput.
By default throttling is not enabled.
The assumptions are:
1. Data ingestion rates(write RPC rates) are mostly stable,
e.g. log and time series.
2. We reserve a stable share of resoures for data ingestion per
TS, e.g. 20000 writes per TS per sec.
3. Each TS can hold roughly same number of tablets/leaders,
e.g. 120 tablets and 40 leaders. We would like write RPC spread
evenly to those 40 leaders, each have 20000 / 40 = 500 capacity.
This capacity also become a simple indicator for user and cluster
admin to do provision and planning:
A workload require 200K op/s write throughput, so the table should
have 200K / 500 = 400 tablets.
The cluster have capacity for 4000 tablets, and currently there are
3000 tablets running, so we can only allow 1000 new tablets or more
nodes is needed.
Change-Id: I33cb6934d27b883a783682cef1e0723100637d45
Reviewed-on: http://gerrit.cloudera.org:8080/2327
Tested-by: Kudu Jenkins
Reviewed-by: Todd Lipcon <to...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kudu/commit/f8a677b1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kudu/tree/f8a677b1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kudu/diff/f8a677b1
Branch: refs/heads/master
Commit: f8a677b1fea197bb88672fb944e89b39b1b80829
Parents: 85fd60d
Author: Binglin Chang <de...@gmail.com>
Authored: Mon Feb 22 14:17:05 2016 +0800
Committer: Todd Lipcon <to...@apache.org>
Committed: Tue Mar 22 18:43:49 2016 +0000
----------------------------------------------------------------------
src/kudu/integration-tests/CMakeLists.txt | 1 +
.../integration-tests/write_throttling-itest.cc | 136 +++++++++++++++++++
src/kudu/tablet/CMakeLists.txt | 1 +
src/kudu/tablet/tablet.cc | 30 ++++
src/kudu/tablet/tablet.h | 7 +
src/kudu/tablet/tablet_throttle-test.cc | 66 +++++++++
src/kudu/tserver/tablet_service.cc | 13 +-
src/kudu/tserver/tserver.proto | 3 +
src/kudu/util/CMakeLists.txt | 2 +
src/kudu/util/throttler-test.cc | 77 +++++++++++
src/kudu/util/throttler.cc | 64 +++++++++
src/kudu/util/throttler.h | 65 +++++++++
12 files changed, 464 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f8a677b1/src/kudu/integration-tests/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/CMakeLists.txt b/src/kudu/integration-tests/CMakeLists.txt
index 42944a6..381ba39 100644
--- a/src/kudu/integration-tests/CMakeLists.txt
+++ b/src/kudu/integration-tests/CMakeLists.txt
@@ -61,6 +61,7 @@ ADD_KUDU_TEST(remote_bootstrap-itest)
ADD_KUDU_TEST(tablet_replacement-itest)
ADD_KUDU_TEST(create-table-itest)
ADD_KUDU_TEST(fuzz-itest)
+ADD_KUDU_TEST(write_throttling-itest)
# Some tests have additional dependencies
set(KUDU_TEST_LINK_LIBS kudu_client kudu_tools_util ${KUDU_TEST_LINK_LIBS})
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f8a677b1/src/kudu/integration-tests/write_throttling-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/write_throttling-itest.cc b/src/kudu/integration-tests/write_throttling-itest.cc
new file mode 100644
index 0000000..e84ab5f
--- /dev/null
+++ b/src/kudu/integration-tests/write_throttling-itest.cc
@@ -0,0 +1,136 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 <gflags/gflags.h>
+#include <glog/logging.h>
+#include <string>
+
+#include "kudu/client/client.h"
+#include "kudu/integration-tests/external_mini_cluster-itest-base.h"
+#include "kudu/util/monotime.h"
+#include "kudu/util/test_util.h"
+
+using std::string;
+using std::vector;
+
+DEFINE_int32(throttling_test_time, 3,
+ "Number of seconds to run write throttling test");
+
+namespace kudu {
+namespace tablet {
+
+using client::KuduClient;
+using client::KuduClientBuilder;
+using client::KuduColumnSchema;
+using client::KuduInsert;
+using client::KuduSchema;
+using client::KuduSchemaBuilder;
+using client::KuduSession;
+using client::KuduTable;
+using client::KuduTableCreator;
+
+class WriteThrottlingTest : public ExternalMiniClusterITestBase {
+ protected:
+ WriteThrottlingTest() {
+ KuduSchemaBuilder b;
+ b.AddColumn("key")->Type(KuduColumnSchema::INT64)->NotNull()->PrimaryKey();
+ b.AddColumn("string_val")->Type(KuduColumnSchema::STRING)->NotNull();
+ CHECK_OK(b.Build(&schema_));
+ }
+
+ void CreateTable() {
+ gscoped_ptr<KuduTableCreator> table_creator(client_->NewTableCreator());
+ ASSERT_OK(table_creator->table_name(kTableName)
+ .schema(&schema_)
+ .num_replicas(1)
+ .Create());
+ ASSERT_OK(client_->OpenTable(kTableName, &table_));
+ }
+
+ protected:
+ static const char* const kTableName;
+
+ KuduSchema schema_;
+ client::sp::shared_ptr<KuduTable> table_;
+};
+
+const char* const WriteThrottlingTest::kTableName = "write-throttling-test-tbl";
+
+TEST_F(WriteThrottlingTest, ThrottleWriteRpcPerSec) {
+ // Normally a single thread session can write at 500 qps,
+ // so we throttle at 100 qps.
+ const int TARGET_QPS = 100;
+ vector<string> ts_flags = { "--tablet_throttler_rpc_per_sec=100",
+ "--tablet_throttler_bytes_per_sec=0" };
+ NO_FATALS(StartCluster(ts_flags));
+ NO_FATALS(CreateTable());
+ string string_val = string(10, 'a');
+ client::sp::shared_ptr<KuduSession> session = client_->NewSession();
+ session->SetTimeoutMillis(5000);
+ ASSERT_OK(session->SetFlushMode(KuduSession::AUTO_FLUSH_SYNC));
+ for (int t = 0; t < FLAGS_throttling_test_time; t++) {
+ MonoTime begin = MonoTime::Now(MonoTime::FINE);
+ for (int i = 0; i < TARGET_QPS; i++) {
+ gscoped_ptr<KuduInsert> insert(table_->NewInsert());
+ KuduPartialRow* row = insert->mutable_row();
+ CHECK_OK(row->SetInt64("key", t * TARGET_QPS + i));
+ CHECK_OK(row->SetString("string_val", string_val));
+ CHECK_OK(session->Apply(insert.release()));
+ }
+ MonoTime end = MonoTime::Now(MonoTime::FINE);
+ MonoDelta delta = end.GetDeltaSince(begin);
+ double qps = TARGET_QPS / delta.ToSeconds();
+ LOG(INFO) << "Iteration " << t << " qps: " << qps;
+ ASSERT_LE(qps, TARGET_QPS * 1.2f);
+ }
+}
+
+TEST_F(WriteThrottlingTest, ThrottleWriteBytesPerSec) {
+ const int TARGET_QPS = 100;
+ const int BYTES_PER_RPC = 3000;
+ // Normally a single thread session can write at 500 qps,
+ // so we throttle at 100 qps * 3000 byte = 300000 byte/s.
+ // 8 byte key + 2992 byte string_val, plus other insignificant parts,
+ // total bytes is still about 3000+ byte per RPC.
+ vector<string> ts_flags = { "--tablet_throttler_rpc_per_sec=0",
+ "--tablet_throttler_bytes_per_sec=300000" };
+ NO_FATALS(StartCluster(ts_flags));
+ NO_FATALS(CreateTable());
+ string string_val = string(BYTES_PER_RPC - 8, 'a');
+ client::sp::shared_ptr<KuduSession> session = client_->NewSession();
+ session->SetTimeoutMillis(5000);
+ ASSERT_OK(session->SetFlushMode(KuduSession::AUTO_FLUSH_SYNC));
+ for (int t = 0; t < FLAGS_throttling_test_time; t++) {
+ MonoTime begin = MonoTime::Now(MonoTime::FINE);
+ for (int i = 0; i < TARGET_QPS; i++) {
+ gscoped_ptr<KuduInsert> insert(table_->NewInsert());
+ KuduPartialRow* row = insert->mutable_row();
+ CHECK_OK(row->SetInt64("key", t * TARGET_QPS + i));
+ CHECK_OK(row->SetString("string_val", string_val));
+ CHECK_OK(session->Apply(insert.release()));
+ }
+ MonoTime end = MonoTime::Now(MonoTime::FINE);
+ MonoDelta delta = end.GetDeltaSince(begin);
+ double qps = TARGET_QPS / delta.ToSeconds();
+ double bps = TARGET_QPS * BYTES_PER_RPC / delta.ToSeconds();
+ LOG(INFO) << "Iteration " << t << " qps: " << qps << " " << bps << " byte/s";
+ ASSERT_LE(bps, TARGET_QPS * BYTES_PER_RPC * 1.2f);
+ }
+}
+
+} // namespace tablet
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f8a677b1/src/kudu/tablet/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/CMakeLists.txt b/src/kudu/tablet/CMakeLists.txt
index 4b6c6e7..e7e94c9 100644
--- a/src/kudu/tablet/CMakeLists.txt
+++ b/src/kudu/tablet/CMakeLists.txt
@@ -111,6 +111,7 @@ ADD_KUDU_TEST(major_delta_compaction-test)
ADD_KUDU_TEST(transactions/transaction_tracker-test)
ADD_KUDU_TEST(tablet_peer-test)
ADD_KUDU_TEST(tablet_random_access-test)
+ADD_KUDU_TEST(tablet_throttle-test)
ADD_KUDU_TEST(tablet_mm_ops-test)
# Some tests don't have dependencies on other tablet stuff
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f8a677b1/src/kudu/tablet/tablet.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet.cc b/src/kudu/tablet/tablet.cc
index eae5bb1..dcf33aa 100644
--- a/src/kudu/tablet/tablet.cc
+++ b/src/kudu/tablet/tablet.cc
@@ -98,6 +98,22 @@ DEFINE_double(fault_crash_before_flush_tablet_meta_after_flush_mrs, 0.0,
TAG_FLAG(fault_crash_before_flush_tablet_meta_after_flush_mrs, unsafe);
+DEFINE_int64(tablet_throttler_rpc_per_sec, 0,
+ "Maximum write RPC rate (op/s) allowed for a tablet, write RPC exceeding this "
+ "limit will be throttled. 0 means no limit.");
+TAG_FLAG(tablet_throttler_rpc_per_sec, experimental);
+
+DEFINE_int64(tablet_throttler_bytes_per_sec, 0,
+ "Maximum write RPC IO rate (byte/s) allowed for a tablet, write RPC exceeding "
+ "this limit will be throttled. 0 means no limit.");
+TAG_FLAG(tablet_throttler_bytes_per_sec, experimental);
+
+DEFINE_double(tablet_throttler_burst_factor, 1.0f,
+ "Burst factor for write RPC throttling. The maximum rate the throttler "
+ "allows within a token refill period (100ms) equals burst factor multiply "
+ "base rate.");
+TAG_FLAG(tablet_throttler_burst_factor, experimental);
+
METRIC_DEFINE_entity(tablet);
METRIC_DEFINE_gauge_size(tablet, memrowset_size, "MemRowSet Memory Usage",
kudu::MetricUnit::kBytes,
@@ -176,6 +192,13 @@ Tablet::Tablet(const scoped_refptr<TabletMetadata>& metadata,
metric_entity_, Bind(&Tablet::EstimateOnDiskSize, Unretained(this)))
->AutoDetach(&metric_detacher_);
}
+
+ if (FLAGS_tablet_throttler_rpc_per_sec > 0 || FLAGS_tablet_throttler_bytes_per_sec > 0) {
+ throttler_.reset(new Throttler(MonoTime::Now(MonoTime::FINE),
+ FLAGS_tablet_throttler_rpc_per_sec,
+ FLAGS_tablet_throttler_bytes_per_sec,
+ FLAGS_tablet_throttler_burst_factor));
+ }
}
Tablet::~Tablet() {
@@ -837,6 +860,13 @@ int32_t Tablet::CurrentMrsIdForTests() const {
return components_->memrowset->mrs_id();
}
+bool Tablet::ShouldThrottleAllow(int64_t bytes) {
+ if (!throttler_) {
+ return true;
+ }
+ return throttler_->Take(MonoTime::Now(MonoTime::FINE), 1, bytes);
+}
+
////////////////////////////////////////////////////////////
// CompactRowSetsOp
////////////////////////////////////////////////////////////
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f8a677b1/src/kudu/tablet/tablet.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet.h b/src/kudu/tablet/tablet.h
index aa31b81..c46aa22 100644
--- a/src/kudu/tablet/tablet.h
+++ b/src/kudu/tablet/tablet.h
@@ -40,6 +40,7 @@
#include "kudu/util/semaphore.h"
#include "kudu/util/slice.h"
#include "kudu/util/status.h"
+#include "kudu/util/throttler.h"
namespace kudu {
@@ -370,6 +371,10 @@ class Tablet {
// Returns a reference to this tablet's memory tracker.
const std::shared_ptr<MemTracker>& mem_tracker() const { return mem_tracker_; }
+ // Throttle a RPC with 'bytes' request size.
+ // Return true if this RPC is allowed.
+ bool ShouldThrottleAllow(int64_t bytes);
+
static const char* kDMSMemTrackerId;
private:
friend class Iterator;
@@ -526,6 +531,8 @@ class Tablet {
gscoped_ptr<TabletMetrics> metrics_;
FunctionGaugeDetacher metric_detacher_;
+ std::unique_ptr<Throttler> throttler_;
+
int64_t next_mrs_id_;
// A pointer to the server's clock.
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f8a677b1/src/kudu/tablet/tablet_throttle-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet_throttle-test.cc b/src/kudu/tablet/tablet_throttle-test.cc
new file mode 100644
index 0000000..c150dcc
--- /dev/null
+++ b/src/kudu/tablet/tablet_throttle-test.cc
@@ -0,0 +1,66 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 <gflags/gflags.h>
+#include <glog/logging.h>
+
+#include "kudu/tablet/tablet.h"
+#include "kudu/tablet/tablet-test-util.h"
+#include "kudu/util/test_macros.h"
+
+DECLARE_int64(tablet_throttler_rpc_per_sec);
+
+DECLARE_int64(tablet_throttler_bytes_per_sec);
+
+namespace kudu {
+namespace tablet {
+
+class TestTabletThrottle : public KuduTabletTest {
+ public:
+ TestTabletThrottle()
+ : KuduTabletTest(CreateBaseSchema()) {
+ }
+
+ virtual void SetUp() OVERRIDE {
+ FLAGS_tablet_throttler_rpc_per_sec = 100;
+ FLAGS_tablet_throttler_bytes_per_sec = 1000 * 1000;
+ KuduTabletTest::SetUp();
+ }
+
+ private:
+ Schema CreateBaseSchema() {
+ return Schema({ ColumnSchema("key", INT32),
+ ColumnSchema("c1", INT32) }, 1);
+ }
+};
+
+TEST_F(TestTabletThrottle, TestThrottle) {
+ std::shared_ptr<Tablet> t = this->tablet();
+ // Make sure token bucket filled up
+ SleepFor(MonoDelta::FromMilliseconds(200));
+ // Since we cannot change time here, it's expected
+ // that the following 11 throttle call should complete
+ // within 100ms, or the test will fail.
+ for (int i = 0; i < 10; i++) {
+ ASSERT_TRUE(t->ShouldThrottleAllow(1));
+ }
+ ASSERT_FALSE(t->ShouldThrottleAllow(1));
+}
+
+} // namespace tablet
+} // namespace kudu
+
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f8a677b1/src/kudu/tserver/tablet_service.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/tablet_service.cc b/src/kudu/tserver/tablet_service.cc
index 5a34927..044b59f 100644
--- a/src/kudu/tserver/tablet_service.cc
+++ b/src/kudu/tserver/tablet_service.cc
@@ -242,7 +242,8 @@ static void SetupErrorAndRespond(TabletServerErrorPB* error,
TabletServerErrorPB::Code code,
rpc::RpcContext* context) {
// Generic "service unavailable" errors will cause the client to retry later.
- if (code == TabletServerErrorPB::UNKNOWN_ERROR && s.IsServiceUnavailable()) {
+ if ((code == TabletServerErrorPB::UNKNOWN_ERROR ||
+ code == TabletServerErrorPB::THROTTLED) && s.IsServiceUnavailable()) {
context->RespondRpcFailure(rpc::ErrorStatusPB::ERROR_SERVER_TOO_BUSY, s);
return;
}
@@ -684,6 +685,16 @@ void TabletServiceImpl::Write(const WriteRequestPB* req,
return;
}
+ uint64_t bytes = req->row_operations().rows().size() +
+ req->row_operations().indirect_data().size();
+ if (!tablet->ShouldThrottleAllow(bytes)) {
+ SetupErrorAndRespond(resp->mutable_error(),
+ Status::ServiceUnavailable("Rejecting Write request: throttled"),
+ TabletServerErrorPB::THROTTLED,
+ context);
+ return;
+ }
+
// Check for memory pressure; don't bother doing any additional work if we've
// exceeded the limit.
double capacity_pct;
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f8a677b1/src/kudu/tserver/tserver.proto
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/tserver.proto b/src/kudu/tserver/tserver.proto
index eee0752..93850a5 100644
--- a/src/kudu/tserver/tserver.proto
+++ b/src/kudu/tserver/tserver.proto
@@ -88,6 +88,9 @@ message TabletServerErrorPB {
// The requested operation is already inprogress, e.g. RemoteBootstrap.
ALREADY_INPROGRESS = 18;
+
+ // The request is throttled.
+ THROTTLED = 19;
}
// The error code.
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f8a677b1/src/kudu/util/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/kudu/util/CMakeLists.txt b/src/kudu/util/CMakeLists.txt
index 241f6f6..c0749b1 100644
--- a/src/kudu/util/CMakeLists.txt
+++ b/src/kudu/util/CMakeLists.txt
@@ -156,6 +156,7 @@ set(UTIL_SRCS
threadlocal.cc
threadpool.cc
thread_restrictions.cc
+ throttler.cc
trace.cc
user.cc
url-coding.cc
@@ -314,6 +315,7 @@ ADD_KUDU_TEST(subprocess-test)
ADD_KUDU_TEST(sync_point-test)
ADD_KUDU_TEST(thread-test)
ADD_KUDU_TEST(threadpool-test)
+ADD_KUDU_TEST(throttler-test)
ADD_KUDU_TEST(trace-test)
ADD_KUDU_TEST(url-coding-test)
ADD_KUDU_TEST(user-test)
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f8a677b1/src/kudu/util/throttler-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/util/throttler-test.cc b/src/kudu/util/throttler-test.cc
new file mode 100644
index 0000000..f2c47b4
--- /dev/null
+++ b/src/kudu/util/throttler-test.cc
@@ -0,0 +1,77 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 "kudu/util/throttler.h"
+
+#include <algorithm>
+#include <cmath>
+#include <cstring>
+
+#include "kudu/util/test_util.h"
+
+namespace kudu {
+
+class ThrottlerTest : public KuduTest {
+};
+
+TEST_F(ThrottlerTest, TestOpThrottle) {
+ // Check operation rate throttling
+ MonoTime now = MonoTime::Now(MonoTime::FINE);
+ Throttler t0(now, 1000, 1000*1000, 1);
+ // Fill up bucket
+ now.AddDelta(MonoDelta::FromMilliseconds(2000));
+ // Check throttle behavior for 1 second.
+ for (int p = 0; p < 10; p++) {
+ for (int i = 0; i < 100; i++) {
+ ASSERT_TRUE(t0.Take(now, 1, 1));
+ }
+ ASSERT_FALSE(t0.Take(now, 1, 1));
+ now.AddDelta(MonoDelta::FromMilliseconds(100));
+ }
+}
+
+TEST_F(ThrottlerTest, TestIOThrottle) {
+ // Check operation rate throttling
+ MonoTime now = MonoTime::Now(MonoTime::FINE);
+ Throttler t0(now, 50000, 1000*1000, 1);
+ // Fill up bucket
+ now.AddDelta(MonoDelta::FromMilliseconds(2000));
+ // Check throttle behavior for 1 second.
+ for (int p = 0; p < 10; p++) {
+ for (int i = 0; i < 100; i++) {
+ ASSERT_TRUE(t0.Take(now, 1, 1000));
+ }
+ ASSERT_FALSE(t0.Take(now, 1, 1000));
+ now.AddDelta(MonoDelta::FromMilliseconds(100));
+ }
+}
+
+TEST_F(ThrottlerTest, TestBurst) {
+ // Check IO rate throttling
+ MonoTime now = MonoTime::Now(MonoTime::FINE);
+ Throttler t0(now, 2000, 1000*1000, 5);
+ // Fill up bucket
+ now.AddDelta(MonoDelta::FromMilliseconds(2000));
+ for (int i = 0; i < 100; i++) {
+ now.AddDelta(MonoDelta::FromMilliseconds(1));
+ ASSERT_TRUE(t0.Take(now, 1, 5000));
+ }
+ ASSERT_TRUE(t0.Take(now, 1, 100000));
+ ASSERT_FALSE(t0.Take(now, 1, 1));
+}
+
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f8a677b1/src/kudu/util/throttler.cc
----------------------------------------------------------------------
diff --git a/src/kudu/util/throttler.cc b/src/kudu/util/throttler.cc
new file mode 100644
index 0000000..0eb4e9c
--- /dev/null
+++ b/src/kudu/util/throttler.cc
@@ -0,0 +1,64 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 "kudu/util/throttler.h"
+
+namespace kudu {
+
+Throttler::Throttler(MonoTime now, uint64_t op_rate, uint64_t byte_rate, double burst_factor) :
+ next_refill_(now) {
+ op_refill_ = op_rate / (MonoTime::kMicrosecondsPerSecond / kRefillPeriodMicros);
+ op_token_ = 0;
+ op_token_max_ = static_cast<uint64_t>(op_refill_ * burst_factor);
+ byte_refill_ = byte_rate / (MonoTime::kMicrosecondsPerSecond / kRefillPeriodMicros);
+ byte_token_ = 0;
+ byte_token_max_ = static_cast<uint64_t>(byte_refill_ * burst_factor);
+}
+
+bool Throttler::Take(MonoTime now, uint64_t op, uint64_t byte) {
+ if (op_refill_ == 0 && byte_refill_ == 0) {
+ return true;
+ }
+ boost::lock_guard<simple_spinlock> lock(lock_);
+ Refill(now);
+ if ((op_refill_ == 0 || op <= op_token_) &&
+ (byte_refill_ == 0 || byte <= byte_token_)) {
+ if (op_refill_ > 0) {
+ op_token_ -= op;
+ }
+ if (byte_refill_ > 0) {
+ byte_token_ -= byte;
+ }
+ return true;
+ }
+ return false;
+}
+
+void Throttler::Refill(MonoTime now) {
+ int64_t d = now.GetDeltaSince(next_refill_).ToMicroseconds();
+ if (d < 0) {
+ return;
+ }
+ uint64_t num_period = d / kRefillPeriodMicros + 1;
+ next_refill_.AddDelta(MonoDelta::FromMicroseconds(num_period * kRefillPeriodMicros));
+ op_token_ += num_period * op_refill_;
+ op_token_ = std::min(op_token_, op_token_max_);
+ byte_token_ += num_period * byte_refill_;
+ byte_token_ = std::min(byte_token_, byte_token_max_);
+}
+
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f8a677b1/src/kudu/util/throttler.h
----------------------------------------------------------------------
diff --git a/src/kudu/util/throttler.h b/src/kudu/util/throttler.h
new file mode 100644
index 0000000..239778a
--- /dev/null
+++ b/src/kudu/util/throttler.h
@@ -0,0 +1,65 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+#ifndef KUDU_UTIL_THROTTLER_H
+#define KUDU_UTIL_THROTTLER_H
+
+#include <algorithm>
+
+#include <boost/thread/locks.hpp>
+
+#include "kudu/gutil/macros.h"
+#include "kudu/util/locks.h"
+#include "kudu/util/monotime.h"
+
+namespace kudu {
+
+// A throttler to throttle both operation/s and IO byte/s.
+class Throttler {
+ public:
+ // Refill period is 100ms.
+ enum {
+ kRefillPeriodMicros = 100000
+ };
+
+ // Construct a throttler with max operation per second, max IO bytes per second
+ // and burst factor (burst_rate = rate * burst), burst rate means maximum
+ // throughput within one refill period.
+ // Set op_per_sec to 0 to disable operation throttling.
+ // Set byte_per_sec to 0 to disable IO bytes throttling.
+ Throttler(MonoTime now, uint64_t op_per_sec, uint64_t byte_per_sec, double burst_factor);
+
+ // Throttle an "operation group" by taking 'op' operation tokens and 'byte' byte tokens.
+ // Return true if there are enough tokens, and operation is allowed.
+ // Return false if there are not enough tokens, and operation is throttled.
+ bool Take(MonoTime now, uint64_t op, uint64_t byte);
+
+ private:
+ void Refill(MonoTime now);
+
+ MonoTime next_refill_;
+ uint64_t op_refill_;
+ uint64_t op_token_;
+ uint64_t op_token_max_;
+ uint64_t byte_refill_;
+ uint64_t byte_token_;
+ uint64_t byte_token_max_;
+ simple_spinlock lock_;
+};
+
+} // namespace kudu
+
+#endif