kudu-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From wdberke...@apache.org
Subject [2/5] kudu git commit: KUDU-2289: Tablet deletion should be throttled
Date Thu, 05 Apr 2018 21:34:42 GMT
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 <adar@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 <wdberkeley@apache.org>
Authored: Wed Mar 7 08:52:59 2018 -0800
Committer: Will Berkeley <wdberkeley@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.
 //


Mime
View raw message