You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by wd...@apache.org on 2018/04/05 21:34:42 UTC
[2/5] kudu git commit: KUDU-2289: Tablet deletion should be throttled
KUDU-2289: Tablet deletion should be throttled
When a table is deleted, the master eagerly sends DeleteTablet
requests for every replica of every tablet. Since DeleteTablet
can be IO-heavy and DeleteTablet is run by service threads,
deleting tables could harm other concurrent workloads.
This changes DeleteTablet to run on a threadpool. The number
of threads is capped by --num_tablets_to_delete_simultaneously,
which default to the number of data dirs, a proxy for the
number of disks. This should help throttle tablet deletions,
both preventing them from monopolizing service threads and
limiting their IO.
Change-Id: I3819bf8a3acf8ea03a76cc6cacd92d85bb114998
Reviewed-on: http://gerrit.cloudera.org:8080/9551
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Tested-by: Kudu Jenkins
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/165688f8
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/165688f8
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/165688f8
Branch: refs/heads/master
Commit: 165688f83bbb00a41090e909ae4e17c463fa75e4
Parents: 7e35aee
Author: Will Berkeley <wd...@apache.org>
Authored: Wed Mar 7 08:52:59 2018 -0800
Committer: Will Berkeley <wd...@gmail.com>
Committed: Thu Apr 5 19:52:34 2018 +0000
----------------------------------------------------------------------
src/kudu/tserver/tablet_service.cc | 22 ++---
src/kudu/tserver/ts_tablet_manager.cc | 140 +++++++++++++++++++++++------
src/kudu/tserver/ts_tablet_manager.h | 27 ++++--
src/kudu/util/threadpool.h | 2 +-
4 files changed, 147 insertions(+), 44 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/165688f8/src/kudu/tserver/tablet_service.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/tablet_service.cc b/src/kudu/tserver/tablet_service.cc
index 3854f5c..2f4d285 100644
--- a/src/kudu/tserver/tablet_service.cc
+++ b/src/kudu/tserver/tablet_service.cc
@@ -805,16 +805,18 @@ void TabletServiceAdminImpl::DeleteTablet(const DeleteTabletRequestPB* req,
if (req->has_cas_config_opid_index_less_or_equal()) {
cas_config_opid_index_less_or_equal = req->cas_config_opid_index_less_or_equal();
}
- TabletServerErrorPB::Code error_code;
- Status s = server_->tablet_manager()->DeleteTablet(req->tablet_id(),
- delete_type,
- cas_config_opid_index_less_or_equal,
- &error_code);
- if (PREDICT_FALSE(!s.ok())) {
- HandleErrorResponse(req, resp, context, error_code, s);
- return;
- }
- context->RespondSuccess();
+
+ auto response_callback = [context, req, resp](const Status& s, TabletServerErrorPB::Code code) {
+ if (PREDICT_FALSE(!s.ok())) {
+ HandleErrorResponse(req, resp, context, code, s);
+ return;
+ }
+ context->RespondSuccess();
+ };
+ server_->tablet_manager()->DeleteTabletAsync(req->tablet_id(),
+ delete_type,
+ cas_config_opid_index_less_or_equal,
+ response_callback);
}
void TabletServiceImpl::Write(const WriteRequestPB* req,
http://git-wip-us.apache.org/repos/asf/kudu/blob/165688f8/src/kudu/tserver/ts_tablet_manager.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/ts_tablet_manager.cc b/src/kudu/tserver/ts_tablet_manager.cc
index f9341af..034d6ab 100644
--- a/src/kudu/tserver/ts_tablet_manager.cc
+++ b/src/kudu/tserver/ts_tablet_manager.cc
@@ -84,6 +84,13 @@ DEFINE_int32(num_tablets_to_open_simultaneously, 0,
"may make sense to manually tune this.");
TAG_FLAG(num_tablets_to_open_simultaneously, advanced);
+DEFINE_int32(num_tablets_to_delete_simultaneously, 0,
+ "Number of threads available to delete tablets. If this is set to 0 (the "
+ "default), then the number of delete threads will be set based on the number "
+ "of data directories. If the data directories are on some very fast storage "
+ "device such as SSD or a RAID array, it may make sense to manually tune this.");
+TAG_FLAG(num_tablets_to_delete_simultaneously, advanced);
+
DEFINE_int32(tablet_start_warn_threshold_ms, 500,
"If a tablet takes more than this number of millis to start, issue "
"a warning with a trace.");
@@ -119,6 +126,10 @@ DEFINE_int32(tablet_state_walk_min_period_ms, 1000,
"tablet map to update tablet state counts.");
TAG_FLAG(tablet_state_walk_min_period_ms, advanced);
+DEFINE_int32(delete_tablet_inject_latency_ms, 0,
+ "Amount of delay in milliseconds to inject into delete tablet operations.");
+TAG_FLAG(delete_tablet_inject_latency_ms, unsafe);
+
DECLARE_bool(raft_prepare_replacement_before_eviction);
METRIC_DEFINE_gauge_int32(server, tablets_num_not_initialized,
@@ -247,6 +258,41 @@ TSTabletManager::TSTabletManager(TabletServer* server)
->AutoDetach(&metric_detacher_);
}
+// Base class for Runnables submitted against TSTabletManager threadpools whose
+// whose callback must fire, for example if the callback responds to an RPC.
+class TabletManagerRunnable : public Runnable {
+public:
+ TabletManagerRunnable(TSTabletManager* ts_tablet_manager,
+ std::function<void(const Status&, TabletServerErrorPB::Code)> cb)
+ : ts_tablet_manager_(ts_tablet_manager),
+ cb_(std::move(cb)) {
+ }
+
+ virtual ~TabletManagerRunnable() {
+ // If the Runnable is destroyed without the Run() method being invoked, we
+ // must invoke the user callback ourselves in order to free request
+ // resources. This may happen when the ThreadPool is shut down while the
+ // Runnable is enqueued.
+ if (!cb_invoked_) {
+ cb_(Status::ServiceUnavailable("Tablet server shutting down"),
+ TabletServerErrorPB::THROTTLED);
+ }
+ }
+
+ // Disable automatic invocation of the callback by the destructor.
+ // Does not disable invocation of the callback by Run().
+ void DisableCallback() {
+ cb_invoked_ = true;
+ }
+
+protected:
+ TSTabletManager* const ts_tablet_manager_;
+ const std::function<void(const Status&, TabletServerErrorPB::Code)> cb_;
+ bool cb_invoked_ = false;
+
+ DISALLOW_COPY_AND_ASSIGN(TabletManagerRunnable);
+};
+
TSTabletManager::~TSTabletManager() {
}
@@ -261,7 +307,7 @@ Status TSTabletManager::Init() {
.set_max_threads(FLAGS_num_tablets_to_copy_simultaneously)
.Build(&tablet_copy_pool_));
- // Start the threadpool we'll use to open tablets.
+ // Start the threadpools we'll use to open and delete tablets.
// This has to be done in Init() instead of the constructor, since the
// FsManager isn't initialized until this point.
int max_open_threads = FLAGS_num_tablets_to_open_simultaneously;
@@ -272,6 +318,14 @@ Status TSTabletManager::Init() {
RETURN_NOT_OK(ThreadPoolBuilder("tablet-open")
.set_max_threads(max_open_threads)
.Build(&open_tablet_pool_));
+ int max_delete_threads = FLAGS_num_tablets_to_delete_simultaneously;
+ if (max_delete_threads == 0) {
+ // Default to the number of disks.
+ max_delete_threads = fs_manager_->GetDataRootDirs().size();
+ }
+ RETURN_NOT_OK(ThreadPoolBuilder("tablet-delete")
+ .set_max_threads(max_delete_threads)
+ .Build(&delete_tablet_pool_));
// Search for tablets in the metadata dir.
vector<string> tablet_ids;
@@ -417,42 +471,22 @@ Status TSTabletManager::CheckLeaderTermNotLower(const string& tablet_id,
}
// Tablet Copy runnable that will run on a ThreadPool.
-class TabletCopyRunnable : public Runnable {
+class TabletCopyRunnable : public TabletManagerRunnable {
public:
TabletCopyRunnable(TSTabletManager* ts_tablet_manager,
const StartTabletCopyRequestPB* req,
std::function<void(const Status&, TabletServerErrorPB::Code)> cb)
- : ts_tablet_manager_(ts_tablet_manager),
- req_(req),
- cb_(std::move(cb)) {
+ : TabletManagerRunnable(ts_tablet_manager, cb),
+ req_(req) {
}
- virtual ~TabletCopyRunnable() {
- // If the Runnable is destroyed without the Run() method being invoked, we
- // must invoke the user callback ourselves in order to free request
- // resources. This may happen when the ThreadPool is shut down while the
- // Runnable is enqueued.
- if (!cb_invoked_) {
- cb_(Status::ServiceUnavailable("Tablet server shutting down"),
- TabletServerErrorPB::THROTTLED);
- }
- }
-
- virtual void Run() override {
+ void Run() override {
ts_tablet_manager_->RunTabletCopy(req_, cb_);
cb_invoked_ = true;
}
- // Disable automatic invocation of the callback by the destructor.
- void DisableCallback() {
- cb_invoked_ = true;
- }
-
private:
- TSTabletManager* const ts_tablet_manager_;
const StartTabletCopyRequestPB* const req_;
- const std::function<void(const Status&, TabletServerErrorPB::Code)> cb_;
- bool cb_invoked_ = false;
DISALLOW_COPY_AND_ASSIGN(TabletCopyRunnable);
};
@@ -774,6 +808,53 @@ Status TSTabletManager::BeginReplicaStateTransition(
return Status::OK();
}
+// Delete Tablet runnable that will run on a ThreadPool.
+class DeleteTabletRunnable : public TabletManagerRunnable {
+public:
+ DeleteTabletRunnable(TSTabletManager* ts_tablet_manager,
+ const std::string& tablet_id,
+ tablet::TabletDataState delete_type,
+ const boost::optional<int64_t>& cas_config_index,
+ std::function<void(const Status&, TabletServerErrorPB::Code)> cb)
+ : TabletManagerRunnable(ts_tablet_manager, cb),
+ tablet_id_(tablet_id),
+ delete_type_(delete_type),
+ cas_config_index_(cas_config_index) {
+ }
+
+ void Run() override {
+ TabletServerErrorPB::Code code = TabletServerErrorPB::UNKNOWN_ERROR;
+ Status s = ts_tablet_manager_->DeleteTablet(tablet_id_, delete_type_, cas_config_index_, &code);
+ cb_(s, code);
+ cb_invoked_ = true;
+ }
+
+private:
+ const string tablet_id_;
+ const tablet::TabletDataState delete_type_;
+ const boost::optional<int64_t> cas_config_index_;
+
+ DISALLOW_COPY_AND_ASSIGN(DeleteTabletRunnable);
+};
+
+void TSTabletManager::DeleteTabletAsync(
+ const std::string& tablet_id,
+ tablet::TabletDataState delete_type,
+ const boost::optional<int64_t>& cas_config_index,
+ std::function<void(const Status&, TabletServerErrorPB::Code)> cb) {
+ auto runnable = std::make_shared<DeleteTabletRunnable>(this, tablet_id, delete_type,
+ cas_config_index, cb);
+ Status s = delete_tablet_pool_->Submit(runnable);
+ if (PREDICT_TRUE(s.ok())) {
+ return;
+ }
+
+ // Threadpool submission failed, so we'll invoke the callback ourselves.
+ runnable->DisableCallback();
+ cb(s, s.IsServiceUnavailable() ? TabletServerErrorPB::THROTTLED :
+ TabletServerErrorPB::UNKNOWN_ERROR);
+}
+
Status TSTabletManager::DeleteTablet(
const string& tablet_id,
TabletDataState delete_type,
@@ -789,6 +870,12 @@ Status TSTabletManager::DeleteTablet(
TRACE("Deleting tablet $0", tablet_id);
+ if (PREDICT_FALSE(FLAGS_delete_tablet_inject_latency_ms > 0)) {
+ LOG(WARNING) << "Injecting " << FLAGS_delete_tablet_inject_latency_ms
+ << "ms of latency into DeleteTablet";
+ SleepFor(MonoDelta::FromMilliseconds(FLAGS_delete_tablet_inject_latency_ms));
+ }
+
scoped_refptr<TabletReplica> replica;
scoped_refptr<TransitionInProgressDeleter> deleter;
RETURN_NOT_OK(BeginReplicaStateTransition(tablet_id, "deleting tablet", &replica,
@@ -1044,6 +1131,9 @@ void TSTabletManager::Shutdown() {
// Shut down the bootstrap pool, so no new tablets are registered after this point.
open_tablet_pool_->Shutdown();
+ // Shut down the delete pool, so no new tablets are deleted after this point.
+ delete_tablet_pool_->Shutdown();
+
// Take a snapshot of the replicas list -- that way we don't have to hold
// on to the lock while shutting them down, which might cause a lock
// inversion. (see KUDU-308 for example).
http://git-wip-us.apache.org/repos/asf/kudu/blob/165688f8/src/kudu/tserver/ts_tablet_manager.h
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/ts_tablet_manager.h b/src/kudu/tserver/ts_tablet_manager.h
index a02a13c..742b5a4 100644
--- a/src/kudu/tserver/ts_tablet_manager.h
+++ b/src/kudu/tserver/ts_tablet_manager.h
@@ -122,14 +122,22 @@ class TSTabletManager : public tserver::TabletReplicaLookupIf {
consensus::RaftConfigPB config,
scoped_refptr<tablet::TabletReplica>* replica);
- // Delete the specified tablet.
- // 'delete_type' must be one of TABLET_DATA_DELETED or TABLET_DATA_TOMBSTONED
- // or else returns Status::IllegalArgument.
- // 'cas_config_index' is optionally specified to enable an
- // atomic DeleteTablet operation that only occurs if the latest committed
- // Raft config change op has an opid_index equal to or less than the specified
- // value. If not, 'error_code' is set to CAS_FAILED and a non-OK Status is
- // returned.
+ // Delete the specified tablet asynchronously with callback 'cb'.
+ // - If the async task cannot be started, 'cb' will be called with
+ // Status::ServiceUnavailable and TabletServerErrorPB::THROTTLED.
+ // - 'delete_type' must be one of TABLET_DATA_DELETED or TABLET_DATA_TOMBSTONED.
+ // - 'cas_config_index' is optionally specified to enable an
+ // atomic DeleteTablet operation that only occurs if the latest committed
+ // Raft config change op has an opid_index equal to or less than the specified
+ // value. If not, the callback is called with a non-OK Status and error code
+ // CAS_FAILED.
+ void DeleteTabletAsync(const std::string& tablet_id,
+ tablet::TabletDataState delete_type,
+ const boost::optional<int64_t>& cas_config_index,
+ std::function<void(const Status&, TabletServerErrorPB::Code)> cb);
+
+ // Delete the specified tablet synchronously.
+ // See DeleteTabletAsync() for more information.
Status DeleteTablet(const std::string& tablet_id,
tablet::TabletDataState delete_type,
const boost::optional<int64_t>& cas_config_index,
@@ -361,6 +369,9 @@ class TSTabletManager : public tserver::TabletReplicaLookupIf {
// Thread pool used to open the tablets async, whether bootstrap is required or not.
gscoped_ptr<ThreadPool> open_tablet_pool_;
+ // Thread pool used to delete tablets asynchronously.
+ gscoped_ptr<ThreadPool> delete_tablet_pool_;
+
FunctionGaugeDetacher metric_detacher_;
DISALLOW_COPY_AND_ASSIGN(TSTabletManager);
http://git-wip-us.apache.org/repos/asf/kudu/blob/165688f8/src/kudu/util/threadpool.h
----------------------------------------------------------------------
diff --git a/src/kudu/util/threadpool.h b/src/kudu/util/threadpool.h
index f1a7a37..1557486 100644
--- a/src/kudu/util/threadpool.h
+++ b/src/kudu/util/threadpool.h
@@ -88,7 +88,7 @@ struct ThreadPoolMetrics {
// The TraceMetrics implementation relies on the number of distinct counter
// names being small. Thus, if the thread pool name itself is dynamically
// generated, the default behavior described above would result in an
-// unbounded number of distinct cunter names. The 'trace_metric_prefix'
+// unbounded number of distinct counter names. The 'trace_metric_prefix'
// setting can be used to override the prefix used in generating the trace
// metric names.
//