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 2017/06/27 23:03:41 UTC
[1/5] kudu git commit: kserver: introduce KuduServer class
Repository: kudu
Updated Branches:
refs/heads/master 57a07ae72 -> cbd2fb673
kserver: introduce KuduServer class
The kserver module (with the KuduServer class) is an abstraction that sits
between ServerBase and Master/TabletServer. The idea is for it to
encapsulate all Kudu-specific knowledge that should be shared by Masters and
Tablet Servers without polluting ServerBase, which has no Kudu-specific
knowledge and could be reused by any C++ application.
This patch introduces the new class (just lifecycle methods for now) and
wires it into Master/TabletServer. There's no new functionality.
Change-Id: I41ffd09ce46d9d744aa25343cec77b7a33b88563
Reviewed-on: http://gerrit.cloudera.org:8080/7239
Reviewed-by: Todd Lipcon <to...@apache.org>
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/1a7630e8
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/1a7630e8
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/1a7630e8
Branch: refs/heads/master
Commit: 1a7630e84715480e93e2d1caae34865f517ef831
Parents: 57a07ae
Author: Adar Dembo <ad...@cloudera.com>
Authored: Tue Jun 20 13:50:58 2017 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Tue Jun 27 22:45:54 2017 +0000
----------------------------------------------------------------------
CMakeLists.txt | 1 +
src/kudu/kserver/CMakeLists.txt | 30 ++++++++++++++++
src/kudu/kserver/kserver.cc | 55 ++++++++++++++++++++++++++++++
src/kudu/kserver/kserver.h | 62 ++++++++++++++++++++++++++++++++++
src/kudu/master/CMakeLists.txt | 1 +
src/kudu/master/master.cc | 14 ++++----
src/kudu/master/master.h | 13 ++++---
src/kudu/server/server_base.h | 7 ++--
src/kudu/tserver/CMakeLists.txt | 21 ++++++------
src/kudu/tserver/tablet_server.cc | 16 ++++-----
src/kudu/tserver/tablet_server.h | 15 ++++----
11 files changed, 192 insertions(+), 43 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/1a7630e8/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/CMakeLists.txt b/CMakeLists.txt
index ad0c47b..f58a6df 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -1128,6 +1128,7 @@ add_subdirectory(src/kudu/fs)
# Google util libraries borrowed from supersonic, tcmalloc, Chromium, etc.
add_subdirectory(src/kudu/gutil)
add_subdirectory(src/kudu/integration-tests)
+add_subdirectory(src/kudu/kserver)
add_subdirectory(src/kudu/master)
add_subdirectory(src/kudu/rpc)
add_subdirectory(src/kudu/security)
http://git-wip-us.apache.org/repos/asf/kudu/blob/1a7630e8/src/kudu/kserver/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/kudu/kserver/CMakeLists.txt b/src/kudu/kserver/CMakeLists.txt
new file mode 100644
index 0000000..a6f3738
--- /dev/null
+++ b/src/kudu/kserver/CMakeLists.txt
@@ -0,0 +1,30 @@
+# 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.
+
+#########################################
+# kserver
+#########################################
+
+set(KSERVER_SRCS
+ kserver.cc
+)
+
+add_library(kserver ${KSERVER_SRCS})
+target_link_libraries(kserver
+ gutil
+ kudu_util
+ server_process)
http://git-wip-us.apache.org/repos/asf/kudu/blob/1a7630e8/src/kudu/kserver/kserver.cc
----------------------------------------------------------------------
diff --git a/src/kudu/kserver/kserver.cc b/src/kudu/kserver/kserver.cc
new file mode 100644
index 0000000..20268b4
--- /dev/null
+++ b/src/kudu/kserver/kserver.cc
@@ -0,0 +1,55 @@
+// 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/kserver/kserver.h"
+
+#include <string>
+#include <utility>
+
+#include "kudu/server/server_base_options.h"
+#include "kudu/util/status.h"
+
+using std::string;
+
+namespace kudu {
+
+using server::ServerBaseOptions;
+
+namespace kserver {
+
+KuduServer::KuduServer(string name,
+ const ServerBaseOptions& options,
+ const string& metric_namespace)
+ : ServerBase(std::move(name), options, metric_namespace) {
+}
+
+Status KuduServer::Init() {
+ RETURN_NOT_OK(ServerBase::Init());
+ return Status::OK();
+}
+
+Status KuduServer::Start() {
+ RETURN_NOT_OK(ServerBase::Start());
+ return Status::OK();
+}
+
+void KuduServer::Shutdown() {
+ ServerBase::Shutdown();
+}
+
+} // namespace kserver
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/1a7630e8/src/kudu/kserver/kserver.h
----------------------------------------------------------------------
diff --git a/src/kudu/kserver/kserver.h b/src/kudu/kserver/kserver.h
new file mode 100644
index 0000000..1f37886
--- /dev/null
+++ b/src/kudu/kserver/kserver.h
@@ -0,0 +1,62 @@
+// 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.
+
+#pragma once
+
+#include <string>
+
+#include "kudu/gutil/macros.h"
+#include "kudu/server/server_base.h"
+
+namespace kudu {
+class Status;
+
+namespace server {
+struct ServerBaseOptions;
+}
+
+namespace kserver {
+
+// Kudu server instance.
+//
+// While ServerBase is a generic C++ "server" class, KuduServer includes
+// functionality that is specific to Kudu (i.e. pertains to the hosting of
+// tablets, to Raft consensus, etc.).
+class KuduServer : public server::ServerBase {
+ public:
+ // Constructs a new KuduServer instance and performs all no-fail member
+ // initializations.
+ KuduServer(std::string name,
+ const server::ServerBaseOptions& options,
+ const std::string& metric_namespace);
+
+ // Finalizes the initialization of a KuduServer by performing any member
+ // initializations that may fail.
+ virtual Status Init() override;
+
+ // Exposes an initialized KuduServer to the network or filesystem.
+ virtual Status Start() override;
+
+ // Shuts down a KuduServer instance.
+ virtual void Shutdown() override;
+
+ private:
+ DISALLOW_COPY_AND_ASSIGN(KuduServer);
+};
+
+} // namespace kserver
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/1a7630e8/src/kudu/master/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/kudu/master/CMakeLists.txt b/src/kudu/master/CMakeLists.txt
index e504d8a..029e5c1 100644
--- a/src/kudu/master/CMakeLists.txt
+++ b/src/kudu/master/CMakeLists.txt
@@ -49,6 +49,7 @@ add_library(master ${MASTER_SRCS})
target_link_libraries(master
gutil
krpc
+ kserver
kudu_common
kudu_util
master_proto
http://git-wip-us.apache.org/repos/asf/kudu/blob/1a7630e8/src/kudu/master/master.cc
----------------------------------------------------------------------
diff --git a/src/kudu/master/master.cc b/src/kudu/master/master.cc
index 2d883f6..fb72256 100644
--- a/src/kudu/master/master.cc
+++ b/src/kudu/master/master.cc
@@ -82,7 +82,7 @@ namespace kudu {
namespace master {
Master::Master(const MasterOptions& opts)
- : ServerBase("Master", opts, "kudu.master"),
+ : KuduServer("Master", opts, "kudu.master"),
state_(kStopped),
ts_manager_(new TSManager()),
catalog_manager_(new CatalogManager(this)),
@@ -110,7 +110,7 @@ Status Master::Init() {
RETURN_NOT_OK(ThreadPoolBuilder("init").set_max_threads(1).Build(&init_pool_));
- RETURN_NOT_OK(ServerBase::Init());
+ RETURN_NOT_OK(KuduServer::Init());
if (web_server_) {
RETURN_NOT_OK(path_handlers_->Register(web_server_.get()));
@@ -148,10 +148,10 @@ Status Master::StartAsync() {
gscoped_ptr<ServiceIf> tablet_copy_service(new TabletCopyServiceImpl(
this, catalog_manager_.get()));
- RETURN_NOT_OK(ServerBase::RegisterService(std::move(impl)));
- RETURN_NOT_OK(ServerBase::RegisterService(std::move(consensus_service)));
- RETURN_NOT_OK(ServerBase::RegisterService(std::move(tablet_copy_service)));
- RETURN_NOT_OK(ServerBase::Start());
+ RETURN_NOT_OK(RegisterService(std::move(impl)));
+ RETURN_NOT_OK(RegisterService(std::move(consensus_service)));
+ RETURN_NOT_OK(RegisterService(std::move(tablet_copy_service)));
+ RETURN_NOT_OK(KuduServer::Start());
// Now that we've bound, construct our ServerRegistrationPB.
RETURN_NOT_OK(InitMasterRegistration());
@@ -212,7 +212,7 @@ void Master::Shutdown() {
string name = ToString();
LOG(INFO) << name << " shutting down...";
maintenance_manager_->Shutdown();
- ServerBase::Shutdown();
+ KuduServer::Shutdown();
catalog_manager_->Shutdown();
LOG(INFO) << name << " shutdown complete.";
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/1a7630e8/src/kudu/master/master.h
----------------------------------------------------------------------
diff --git a/src/kudu/master/master.h b/src/kudu/master/master.h
index 1976d77..66132a1 100644
--- a/src/kudu/master/master.h
+++ b/src/kudu/master/master.h
@@ -25,9 +25,9 @@
#include "kudu/common/wire_protocol.h"
#include "kudu/gutil/gscoped_ptr.h"
#include "kudu/gutil/macros.h"
-#include "kudu/master/master_options.h"
+#include "kudu/kserver/kserver.h"
#include "kudu/master/master.pb.h"
-#include "kudu/server/server_base.h"
+#include "kudu/master/master_options.h"
#include "kudu/util/metrics.h"
#include "kudu/util/promise.h"
#include "kudu/util/status.h"
@@ -56,7 +56,7 @@ class MasterCertAuthority;
class MasterPathHandlers;
class TSManager;
-class Master : public server::ServerBase {
+class Master : public kserver::KuduServer {
public:
static const uint16_t kDefaultPort = 7051;
static const uint16_t kDefaultWebPort = 8051;
@@ -64,8 +64,9 @@ class Master : public server::ServerBase {
explicit Master(const MasterOptions& opts);
~Master();
- Status Init();
- Status Start();
+ virtual Status Init() override;
+ virtual Status Start() override;
+ virtual void Shutdown() override;
Status StartAsync();
Status WaitForCatalogManagerInit() const;
@@ -76,8 +77,6 @@ class Master : public server::ServerBase {
Status WaitUntilCatalogManagerIsLeaderAndReadyForTests(const MonoDelta& timeout)
WARN_UNUSED_RESULT;
- void Shutdown();
-
std::string ToString() const;
MasterCertAuthority* cert_authority() { return cert_authority_.get(); }
http://git-wip-us.apache.org/repos/asf/kudu/blob/1a7630e8/src/kudu/server/server_base.h
----------------------------------------------------------------------
diff --git a/src/kudu/server/server_base.h b/src/kudu/server/server_base.h
index 951c3c3..69fd650 100644
--- a/src/kudu/server/server_base.h
+++ b/src/kudu/server/server_base.h
@@ -119,10 +119,11 @@ class ServerBase {
const std::string& metric_namespace);
virtual ~ServerBase();
- Status Init();
+ virtual Status Init();
+ virtual Status Start();
+ virtual void Shutdown();
+
Status RegisterService(gscoped_ptr<rpc::ServiceIf> rpc_impl);
- Status Start();
- void Shutdown();
void LogUnauthorizedAccess(rpc::RpcContext* rpc) const;
http://git-wip-us.apache.org/repos/asf/kudu/blob/1a7630e8/src/kudu/tserver/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/CMakeLists.txt b/src/kudu/tserver/CMakeLists.txt
index 8cf2ea6..512988e 100644
--- a/src/kudu/tserver/CMakeLists.txt
+++ b/src/kudu/tserver/CMakeLists.txt
@@ -119,21 +119,22 @@ set(TSERVER_SRCS
add_library(tserver ${TSERVER_SRCS})
target_link_libraries(tserver
- protobuf
- tablet_copy_proto
- tserver_proto
- tserver_admin_proto
- tserver_service_proto
- master_proto
- consensus_proto
- log_proto
- log
consensus
+ consensus_proto
krpc
+ kserver
+ log
+ log_proto
+ master_proto
+ protobuf
security
server_common
server_process
- tablet)
+ tablet
+ tablet_copy_proto
+ tserver_proto
+ tserver_admin_proto
+ tserver_service_proto)
#########################################
# kudu-tserver
http://git-wip-us.apache.org/repos/asf/kudu/blob/1a7630e8/src/kudu/tserver/tablet_server.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/tablet_server.cc b/src/kudu/tserver/tablet_server.cc
index 800ff10..6c91480 100644
--- a/src/kudu/tserver/tablet_server.cc
+++ b/src/kudu/tserver/tablet_server.cc
@@ -42,7 +42,7 @@ namespace kudu {
namespace tserver {
TabletServer::TabletServer(const TabletServerOptions& opts)
- : ServerBase("TabletServer", opts, "kudu.tabletserver"),
+ : KuduServer("TabletServer", opts, "kudu.tabletserver"),
initted_(false),
fail_heartbeats_for_tests_(false),
opts_(opts),
@@ -82,7 +82,7 @@ Status TabletServer::Init() {
// our heartbeat thread will loop until successfully connecting.
RETURN_NOT_OK(ValidateMasterAddressResolution());
- RETURN_NOT_OK(ServerBase::Init());
+ RETURN_NOT_OK(KuduServer::Init());
if (web_server_) {
RETURN_NOT_OK(path_handlers_->Register(web_server_.get()));
}
@@ -112,11 +112,11 @@ Status TabletServer::Start() {
gscoped_ptr<ServiceIf> tablet_copy_service(new TabletCopyServiceImpl(
this, tablet_manager_.get()));
- RETURN_NOT_OK(ServerBase::RegisterService(std::move(ts_service)));
- RETURN_NOT_OK(ServerBase::RegisterService(std::move(admin_service)));
- RETURN_NOT_OK(ServerBase::RegisterService(std::move(consensus_service)));
- RETURN_NOT_OK(ServerBase::RegisterService(std::move(tablet_copy_service)));
- RETURN_NOT_OK(ServerBase::Start());
+ RETURN_NOT_OK(RegisterService(std::move(ts_service)));
+ RETURN_NOT_OK(RegisterService(std::move(admin_service)));
+ RETURN_NOT_OK(RegisterService(std::move(consensus_service)));
+ RETURN_NOT_OK(RegisterService(std::move(tablet_copy_service)));
+ RETURN_NOT_OK(KuduServer::Start());
RETURN_NOT_OK(heartbeater_->Start());
RETURN_NOT_OK(maintenance_manager_->Init(fs_manager_->uuid()));
@@ -131,7 +131,7 @@ void TabletServer::Shutdown() {
LOG(INFO) << "TabletServer shutting down...";
maintenance_manager_->Shutdown();
WARN_NOT_OK(heartbeater_->Stop(), "Failed to stop TS Heartbeat thread");
- ServerBase::Shutdown();
+ KuduServer::Shutdown();
tablet_manager_->Shutdown();
LOG(INFO) << "TabletServer shut down complete. Bye!";
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/1a7630e8/src/kudu/tserver/tablet_server.h
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/tablet_server.h b/src/kudu/tserver/tablet_server.h
index e353c6d..40a1c39 100644
--- a/src/kudu/tserver/tablet_server.h
+++ b/src/kudu/tserver/tablet_server.h
@@ -24,8 +24,7 @@
#include "kudu/gutil/atomicops.h"
#include "kudu/gutil/gscoped_ptr.h"
#include "kudu/gutil/macros.h"
-#include "kudu/server/server_base.h"
-#include "kudu/server/webserver_options.h"
+#include "kudu/kserver/kserver.h"
#include "kudu/tserver/tablet_server_options.h"
#include "kudu/tserver/tserver.pb.h"
#include "kudu/util/net/net_util.h"
@@ -43,10 +42,10 @@ class ScannerManager;
class TabletServerPathHandlers;
class TSTabletManager;
-class TabletServer : public server::ServerBase {
+class TabletServer : public kserver::KuduServer {
public:
- // TODO: move this out of this header, since clients want to use this
- // constant as well.
+ // TODO(unknown): move this out of this header, since clients want to use
+ // this constant as well.
static const uint16_t kDefaultPort = 7050;
static const uint16_t kDefaultWebPort = 8050;
@@ -58,13 +57,13 @@ class TabletServer : public server::ServerBase {
// Some initialization tasks are asynchronous, such as the bootstrapping
// of tablets. Caller can block, waiting for the initialization to fully
// complete by calling WaitInited().
- Status Init();
+ virtual Status Init() override;
// Waits for the tablet server to complete the initialization.
Status WaitInited();
- Status Start();
- void Shutdown();
+ virtual Status Start() override;
+ virtual void Shutdown() override;
std::string ToString() const;
[3/5] kudu git commit: server: move apply_pool into KuduServer
Posted by to...@apache.org.
server: move apply_pool into KuduServer
The server-wide apply_pool was instantiated in two places: SysCatalogTable
(for the master) and TsTabletManager (for the tserver). This commit moves it
into KuduServer and unifies the instantiation. Note: the apply pool
semantics have not changed.
Some interesting side effects:
1. The master will now generate apply pool metrics.
2. The apply pool is shut down a little bit later in server shutdown than it
was before, though I don't see any issues with this.
Change-Id: Ie7ffc886269aa6531517a52fef29c4408a925aed
Reviewed-on: http://gerrit.cloudera.org:8080/6984
Tested-by: Kudu Jenkins
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/74f99e40
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/74f99e40
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/74f99e40
Branch: refs/heads/master
Commit: 74f99e40da6ccab0b9042e7f38a50176db1d8899
Parents: 4c5c512
Author: Adar Dembo <ad...@cloudera.com>
Authored: Mon May 22 12:27:04 2017 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Tue Jun 27 22:53:23 2017 +0000
----------------------------------------------------------------------
src/kudu/kserver/kserver.cc | 47 ++++++++++++++++++++++++++++++
src/kudu/kserver/kserver.h | 8 +++++
src/kudu/master/catalog_manager.cc | 1 -
src/kudu/master/sys_catalog.cc | 9 ++----
src/kudu/master/sys_catalog.h | 5 +---
src/kudu/tserver/tablet_server.cc | 2 +-
src/kudu/tserver/ts_tablet_manager.cc | 44 +++-------------------------
src/kudu/tserver/ts_tablet_manager.h | 8 +----
8 files changed, 65 insertions(+), 59 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/74f99e40/src/kudu/kserver/kserver.cc
----------------------------------------------------------------------
diff --git a/src/kudu/kserver/kserver.cc b/src/kudu/kserver/kserver.cc
index 20268b4..855eded 100644
--- a/src/kudu/kserver/kserver.cc
+++ b/src/kudu/kserver/kserver.cc
@@ -21,7 +21,9 @@
#include <utility>
#include "kudu/server/server_base_options.h"
+#include "kudu/util/metrics.h"
#include "kudu/util/status.h"
+#include "kudu/util/threadpool.h"
using std::string;
@@ -31,6 +33,28 @@ using server::ServerBaseOptions;
namespace kserver {
+METRIC_DEFINE_histogram(server, op_apply_queue_length, "Operation Apply Queue Length",
+ MetricUnit::kTasks,
+ "Number of operations waiting to be applied to the tablet. "
+ "High queue lengths indicate that the server is unable to process "
+ "operations as fast as they are being written to the WAL.",
+ 10000, 2);
+
+METRIC_DEFINE_histogram(server, op_apply_queue_time, "Operation Apply Queue Time",
+ MetricUnit::kMicroseconds,
+ "Time that operations spent waiting in the apply queue before being "
+ "processed. High queue times indicate that the server is unable to "
+ "process operations as fast as they are being written to the WAL.",
+ 10000000, 2);
+
+METRIC_DEFINE_histogram(server, op_apply_run_time, "Operation Apply Run Time",
+ MetricUnit::kMicroseconds,
+ "Time that operations spent being applied to the tablet. "
+ "High values may indicate that the server is under-provisioned or "
+ "that operations consist of very large batches.",
+ 10000000, 2);
+
+
KuduServer::KuduServer(string name,
const ServerBaseOptions& options,
const string& metric_namespace)
@@ -39,6 +63,16 @@ KuduServer::KuduServer(string name,
Status KuduServer::Init() {
RETURN_NOT_OK(ServerBase::Init());
+
+ RETURN_NOT_OK(ThreadPoolBuilder("apply")
+ .Build(&tablet_apply_pool_));
+ tablet_apply_pool_->SetQueueLengthHistogram(
+ METRIC_op_apply_queue_length.Instantiate(metric_entity_));
+ tablet_apply_pool_->SetQueueTimeMicrosHistogram(
+ METRIC_op_apply_queue_time.Instantiate(metric_entity_));
+ tablet_apply_pool_->SetRunTimeMicrosHistogram(
+ METRIC_op_apply_run_time.Instantiate(metric_entity_));
+
return Status::OK();
}
@@ -48,6 +82,19 @@ Status KuduServer::Start() {
}
void KuduServer::Shutdown() {
+ // Shut down the messenger early, waiting for any reactor threads to finish
+ // running. This ensures that any ref-counted objects inside closures run by
+ // reactor threads will be destroyed before we shut down server-wide thread
+ // pools below, which is important because those objects may own tokens
+ // belonging to the pools.
+ //
+ // Note: prior to this call, it is assumed that any incoming RPCs deferred
+ // from reactor threads have already been cleaned up.
+ messenger_->Shutdown();
+
+ if (tablet_apply_pool_) {
+ tablet_apply_pool_->Shutdown();
+ }
ServerBase::Shutdown();
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/74f99e40/src/kudu/kserver/kserver.h
----------------------------------------------------------------------
diff --git a/src/kudu/kserver/kserver.h b/src/kudu/kserver/kserver.h
index 1f37886..56d62dd 100644
--- a/src/kudu/kserver/kserver.h
+++ b/src/kudu/kserver/kserver.h
@@ -19,11 +19,13 @@
#include <string>
+#include "kudu/gutil/gscoped_ptr.h"
#include "kudu/gutil/macros.h"
#include "kudu/server/server_base.h"
namespace kudu {
class Status;
+class ThreadPool;
namespace server {
struct ServerBaseOptions;
@@ -54,7 +56,13 @@ class KuduServer : public server::ServerBase {
// Shuts down a KuduServer instance.
virtual void Shutdown() override;
+ ThreadPool* tablet_apply_pool() const { return tablet_apply_pool_.get(); }
+
private:
+
+ // Thread pool for applying transactions, shared between all tablets.
+ gscoped_ptr<ThreadPool> tablet_apply_pool_;
+
DISALLOW_COPY_AND_ASSIGN(KuduServer);
};
http://git-wip-us.apache.org/repos/asf/kudu/blob/74f99e40/src/kudu/master/catalog_manager.cc
----------------------------------------------------------------------
diff --git a/src/kudu/master/catalog_manager.cc b/src/kudu/master/catalog_manager.cc
index de632f9..6a1a471 100644
--- a/src/kudu/master/catalog_manager.cc
+++ b/src/kudu/master/catalog_manager.cc
@@ -1085,7 +1085,6 @@ Status CatalogManager::InitSysCatalogAsync(bool is_first_run) {
std::lock_guard<LockType> l(lock_);
unique_ptr<SysCatalogTable> new_catalog(
new SysCatalogTable(master_,
- master_->metric_registry(),
Bind(&CatalogManager::ElectedAsLeaderCb,
Unretained(this))));
if (is_first_run) {
http://git-wip-us.apache.org/repos/asf/kudu/blob/74f99e40/src/kudu/master/sys_catalog.cc
----------------------------------------------------------------------
diff --git a/src/kudu/master/sys_catalog.cc b/src/kudu/master/sys_catalog.cc
index bf7ae7a..2c251dc 100644
--- a/src/kudu/master/sys_catalog.cc
+++ b/src/kudu/master/sys_catalog.cc
@@ -96,7 +96,6 @@ const char* const SysCatalogTable::kSysCertAuthorityEntryId =
const char* const SysCatalogTable::kInjectedFailureStatusMsg =
"INJECTED FAILURE";
-
namespace {
// Return true if the two PBs are equal.
@@ -116,12 +115,11 @@ bool ArePBsEqual(const google::protobuf::Message& prev_pb,
} // anonymous namespace
-SysCatalogTable::SysCatalogTable(Master* master, MetricRegistry* metrics,
+SysCatalogTable::SysCatalogTable(Master* master,
ElectedLeaderCallback leader_cb)
- : metric_registry_(metrics),
+ : metric_registry_(master->metric_registry()),
master_(master),
leader_cb_(std::move(leader_cb)) {
- CHECK_OK(ThreadPoolBuilder("apply").Build(&apply_pool_));
}
SysCatalogTable::~SysCatalogTable() {
@@ -131,7 +129,6 @@ void SysCatalogTable::Shutdown() {
if (tablet_replica_) {
tablet_replica_->Shutdown();
}
- apply_pool_->Shutdown();
}
Status SysCatalogTable::Load(FsManager *fs_manager) {
@@ -310,7 +307,7 @@ Status SysCatalogTable::SetupTablet(const scoped_refptr<tablet::TabletMetadata>&
tablet_replica_.reset(new TabletReplica(
metadata,
local_peer_pb_,
- apply_pool_.get(),
+ master_->tablet_apply_pool(),
Bind(&SysCatalogTable::SysCatalogStateChanged, Unretained(this), metadata->tablet_id())));
consensus::ConsensusBootstrapInfo consensus_info;
http://git-wip-us.apache.org/repos/asf/kudu/blob/74f99e40/src/kudu/master/sys_catalog.h
----------------------------------------------------------------------
diff --git a/src/kudu/master/sys_catalog.h b/src/kudu/master/sys_catalog.h
index 1f306b6..36456f3 100644
--- a/src/kudu/master/sys_catalog.h
+++ b/src/kudu/master/sys_catalog.h
@@ -108,8 +108,7 @@ class SysCatalogTable {
// the consensus configuration's progress, any long running tasks (e.g., scanning
// tablets) should be performed asynchronously (by, e.g., submitting
// them to a to a separate threadpool).
- SysCatalogTable(Master* master, MetricRegistry* metrics,
- ElectedLeaderCallback leader_cb);
+ SysCatalogTable(Master* master, ElectedLeaderCallback leader_cb);
~SysCatalogTable();
@@ -248,8 +247,6 @@ class SysCatalogTable {
MetricRegistry* metric_registry_;
- gscoped_ptr<ThreadPool> apply_pool_;
-
scoped_refptr<tablet::TabletReplica> tablet_replica_;
Master* master_;
http://git-wip-us.apache.org/repos/asf/kudu/blob/74f99e40/src/kudu/tserver/tablet_server.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/tablet_server.cc b/src/kudu/tserver/tablet_server.cc
index 658ecfb..bf98057 100644
--- a/src/kudu/tserver/tablet_server.cc
+++ b/src/kudu/tserver/tablet_server.cc
@@ -46,7 +46,7 @@ TabletServer::TabletServer(const TabletServerOptions& opts)
initted_(false),
fail_heartbeats_for_tests_(false),
opts_(opts),
- tablet_manager_(new TSTabletManager(fs_manager_.get(), this, metric_registry())),
+ tablet_manager_(new TSTabletManager(this)),
scanner_manager_(new ScannerManager(metric_entity())),
path_handlers_(new TabletServerPathHandlers(this)),
maintenance_manager_(new MaintenanceManager(MaintenanceManager::kDefaultOptions)) {
http://git-wip-us.apache.org/repos/asf/kudu/blob/74f99e40/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 1553fca..866374b 100644
--- a/src/kudu/tserver/ts_tablet_manager.cc
+++ b/src/kudu/tserver/ts_tablet_manager.cc
@@ -54,7 +54,6 @@
#include "kudu/util/env_util.h"
#include "kudu/util/fault_injection.h"
#include "kudu/util/flag_tags.h"
-#include "kudu/util/metrics.h"
#include "kudu/util/pb_util.h"
#include "kudu/util/stopwatch.h"
#include "kudu/util/trace.h"
@@ -104,29 +103,7 @@ TAG_FLAG(fault_crash_after_tc_files_fetched, unsafe);
namespace kudu {
namespace tserver {
-METRIC_DEFINE_histogram(server, op_apply_queue_length, "Operation Apply Queue Length",
- MetricUnit::kTasks,
- "Number of operations waiting to be applied to the tablet. "
- "High queue lengths indicate that the server is unable to process "
- "operations as fast as they are being written to the WAL.",
- 10000, 2);
-
-METRIC_DEFINE_histogram(server, op_apply_queue_time, "Operation Apply Queue Time",
- MetricUnit::kMicroseconds,
- "Time that operations spent waiting in the apply queue before being "
- "processed. High queue times indicate that the server is unable to "
- "process operations as fast as they are being written to the WAL.",
- 10000000, 2);
-
-METRIC_DEFINE_histogram(server, op_apply_run_time, "Operation Apply Run Time",
- MetricUnit::kMicroseconds,
- "Time that operations spent being applied to the tablet. "
- "High values may indicate that the server is under-provisioned or "
- "that operations consist of very large batches.",
- 10000000, 2);
-
using consensus::ConsensusMetadata;
-using consensus::ConsensusStatePB;
using consensus::OpId;
using consensus::RaftConfigPB;
using consensus::RaftPeerPB;
@@ -150,21 +127,11 @@ using tablet::TabletMetadata;
using tablet::TabletReplica;
using tserver::TabletCopyClient;
-TSTabletManager::TSTabletManager(FsManager* fs_manager,
- TabletServer* server,
- MetricRegistry* metric_registry)
- : fs_manager_(fs_manager),
+TSTabletManager::TSTabletManager(TabletServer* server)
+ : fs_manager_(server->fs_manager()),
server_(server),
- metric_registry_(metric_registry),
+ metric_registry_(server->metric_registry()),
state_(MANAGER_INITIALIZING) {
-
- CHECK_OK(ThreadPoolBuilder("apply").Build(&apply_pool_));
- apply_pool_->SetQueueLengthHistogram(
- METRIC_op_apply_queue_length.Instantiate(server_->metric_entity()));
- apply_pool_->SetQueueTimeMicrosHistogram(
- METRIC_op_apply_queue_time.Instantiate(server_->metric_entity()));
- apply_pool_->SetRunTimeMicrosHistogram(
- METRIC_op_apply_run_time.Instantiate(server_->metric_entity()));
}
TSTabletManager::~TSTabletManager() {
@@ -594,7 +561,7 @@ scoped_refptr<TabletReplica> TSTabletManager::CreateAndRegisterTabletReplica(
scoped_refptr<TabletReplica> replica(
new TabletReplica(meta,
local_peer_pb_,
- apply_pool_.get(),
+ server_->tablet_apply_pool(),
Bind(&TSTabletManager::MarkTabletDirty,
Unretained(this),
meta->tablet_id())));
@@ -877,9 +844,6 @@ void TSTabletManager::Shutdown() {
replica->Shutdown();
}
- // Shut down the apply pool.
- apply_pool_->Shutdown();
-
{
std::lock_guard<rw_spinlock> l(lock_);
// We don't expect anyone else to be modifying the map after we start the
http://git-wip-us.apache.org/repos/asf/kudu/blob/74f99e40/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 13fd87f..66f1b4d 100644
--- a/src/kudu/tserver/ts_tablet_manager.h
+++ b/src/kudu/tserver/ts_tablet_manager.h
@@ -78,10 +78,7 @@ class TransitionInProgressDeleter;
class TSTabletManager : public tserver::TabletReplicaLookupIf {
public:
// Construct the tablet manager.
- // 'fs_manager' must remain valid until this object is destructed.
- TSTabletManager(FsManager* fs_manager,
- TabletServer* server,
- MetricRegistry* metric_registry);
+ explicit TSTabletManager(TabletServer* server);
virtual ~TSTabletManager();
@@ -315,9 +312,6 @@ 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 for apply transactions, shared between all tablets.
- gscoped_ptr<ThreadPool> apply_pool_;
-
DISALLOW_COPY_AND_ASSIGN(TSTabletManager);
};
[5/5] kudu git commit: threadpool: new test for pools with no
max_threads
Posted by to...@apache.org.
threadpool: new test for pools with no max_threads
This test ensures that a pool created with effectively no max_threads works
as expected. That is:
1. Tokenless tasks should trigger the creation of a new thread.
2. Serial token-based tasks can create new threads, but only up to the
number of tokens submitted against.
I intend to use this "feature" to consolidate some Raft thread pools where a
num_cpus upper bound may be undesirable (i.e. where tasks submitted to the
pools may result in blocking IO).
Change-Id: I37d0bd0a05098bcc1a81ec9f1ac33e74e98781e4
Reviewed-on: http://gerrit.cloudera.org:8080/6945
Tested-by: Kudu Jenkins
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/cbd2fb67
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/cbd2fb67
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/cbd2fb67
Branch: refs/heads/master
Commit: cbd2fb6730ec4f2fabc97fc60757c4ce2203d7b4
Parents: 428a34b
Author: Adar Dembo <ad...@cloudera.com>
Authored: Sun May 21 20:54:19 2017 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Tue Jun 27 22:55:10 2017 +0000
----------------------------------------------------------------------
src/kudu/util/threadpool-test.cc | 82 +++++++++++++++++++++++++++--------
src/kudu/util/threadpool.h | 1 +
2 files changed, 66 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/cbd2fb67/src/kudu/util/threadpool-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/util/threadpool-test.cc b/src/kudu/util/threadpool-test.cc
index 0beb962..a66dfde 100644
--- a/src/kudu/util/threadpool-test.cc
+++ b/src/kudu/util/threadpool-test.cc
@@ -30,6 +30,7 @@
#include "kudu/gutil/atomicops.h"
#include "kudu/gutil/bind.h"
#include "kudu/gutil/strings/substitute.h"
+#include "kudu/gutil/sysinfo.h"
#include "kudu/util/barrier.h"
#include "kudu/util/countdown_latch.h"
#include "kudu/util/locks.h"
@@ -157,6 +158,10 @@ class SlowTask : public Runnable {
latch_->Wait();
}
+ static shared_ptr<Runnable> NewSlowTask(CountDownLatch* latch) {
+ return std::make_shared<SlowTask>(latch);
+ }
+
private:
CountDownLatch* latch_;
};
@@ -171,13 +176,16 @@ TEST_F(ThreadPoolTest, TestThreadPoolWithNoMinimum) {
ASSERT_TRUE(pool_->num_threads_ == 0);
// We get up to 3 threads when submitting work.
CountDownLatch latch(1);
- ASSERT_OK(pool_->Submit(shared_ptr<Runnable>(new SlowTask(&latch))));
- ASSERT_OK(pool_->Submit(shared_ptr<Runnable>(new SlowTask(&latch))));
+ auto cleanup = MakeScopedCleanup([&]() {
+ latch.CountDown();
+ });
+ ASSERT_OK(pool_->Submit(SlowTask::NewSlowTask(&latch)));
+ ASSERT_OK(pool_->Submit(SlowTask::NewSlowTask(&latch)));
ASSERT_EQ(2, pool_->num_threads_);
- ASSERT_OK(pool_->Submit(shared_ptr<Runnable>(new SlowTask(&latch))));
+ ASSERT_OK(pool_->Submit(SlowTask::NewSlowTask(&latch)));
ASSERT_EQ(3, pool_->num_threads_);
// The 4th piece of work gets queued.
- ASSERT_OK(pool_->Submit(shared_ptr<Runnable>(new SlowTask(&latch))));
+ ASSERT_OK(pool_->Submit(SlowTask::NewSlowTask(&latch)));
ASSERT_EQ(3, pool_->num_threads_);
// Finish all work
latch.CountDown();
@@ -187,6 +195,46 @@ TEST_F(ThreadPoolTest, TestThreadPoolWithNoMinimum) {
ASSERT_EQ(0, pool_->num_threads_);
}
+TEST_F(ThreadPoolTest, TestThreadPoolWithNoMaxThreads) {
+ // By default a threadpool's max_threads is set to the number of CPUs, so
+ // this test submits more tasks than that to ensure that the number of CPUs
+ // isn't some kind of upper bound.
+ const int kNumCPUs = base::NumCPUs();
+
+ // Build a threadpool with no limit on the maximum number of threads.
+ ASSERT_OK(RebuildPoolWithBuilder(ThreadPoolBuilder(kDefaultPoolName)
+ .set_max_threads(std::numeric_limits<int>::max())));
+ CountDownLatch latch(1);
+ auto cleanup_latch = MakeScopedCleanup([&]() {
+ latch.CountDown();
+ });
+
+ // Submit tokenless tasks. Each should create a new thread.
+ for (int i = 0; i < kNumCPUs * 2; i++) {
+ ASSERT_OK(pool_->Submit(SlowTask::NewSlowTask(&latch)));
+ }
+ ASSERT_EQ((kNumCPUs * 2), pool_->num_threads_);
+
+ // Submit tasks on two tokens. Only two threads should be created.
+ unique_ptr<ThreadPoolToken> t1 = pool_->NewToken(ThreadPool::ExecutionMode::SERIAL);
+ unique_ptr<ThreadPoolToken> t2 = pool_->NewToken(ThreadPool::ExecutionMode::SERIAL);
+ for (int i = 0; i < kNumCPUs * 2; i++) {
+ ThreadPoolToken* t = (i % 2 == 0) ? t1.get() : t2.get();
+ ASSERT_OK(t->Submit(SlowTask::NewSlowTask(&latch)));
+ }
+ ASSERT_EQ((kNumCPUs * 2) + 2, pool_->num_threads_);
+
+ // Submit more tokenless tasks. Each should create a new thread.
+ for (int i = 0; i < kNumCPUs; i++) {
+ ASSERT_OK(pool_->Submit(SlowTask::NewSlowTask(&latch)));
+ }
+ ASSERT_EQ((kNumCPUs * 3) + 2, pool_->num_threads_);
+
+ latch.CountDown();
+ pool_->Wait();
+ pool_->Shutdown();
+}
+
// Regression test for a bug where a task is submitted exactly
// as a thread is about to exit. Previously this could hang forever.
TEST_F(ThreadPoolTest, TestRace) {
@@ -219,16 +267,16 @@ TEST_F(ThreadPoolTest, TestVariableSizeThreadPool) {
ASSERT_EQ(1, pool_->num_threads_);
// We get up to 4 threads when submitting work.
CountDownLatch latch(1);
- ASSERT_OK(pool_->Submit(shared_ptr<Runnable>(new SlowTask(&latch))));
+ ASSERT_OK(pool_->Submit(SlowTask::NewSlowTask(&latch)));
ASSERT_EQ(1, pool_->num_threads_);
- ASSERT_OK(pool_->Submit(shared_ptr<Runnable>(new SlowTask(&latch))));
+ ASSERT_OK(pool_->Submit(SlowTask::NewSlowTask(&latch)));
ASSERT_EQ(2, pool_->num_threads_);
- ASSERT_OK(pool_->Submit(shared_ptr<Runnable>(new SlowTask(&latch))));
+ ASSERT_OK(pool_->Submit(SlowTask::NewSlowTask(&latch)));
ASSERT_EQ(3, pool_->num_threads_);
- ASSERT_OK(pool_->Submit(shared_ptr<Runnable>(new SlowTask(&latch))));
+ ASSERT_OK(pool_->Submit(SlowTask::NewSlowTask(&latch)));
ASSERT_EQ(4, pool_->num_threads_);
// The 5th piece of work gets queued.
- ASSERT_OK(pool_->Submit(shared_ptr<Runnable>(new SlowTask(&latch))));
+ ASSERT_OK(pool_->Submit(SlowTask::NewSlowTask(&latch)));
ASSERT_EQ(4, pool_->num_threads_);
// Finish all work
latch.CountDown();
@@ -247,9 +295,9 @@ TEST_F(ThreadPoolTest, TestMaxQueueSize) {
CountDownLatch latch(1);
// We will be able to submit two tasks: one for max_threads == 1 and one for
// max_queue_size == 1.
- ASSERT_OK(pool_->Submit(shared_ptr<Runnable>(new SlowTask(&latch))));
- ASSERT_OK(pool_->Submit(shared_ptr<Runnable>(new SlowTask(&latch))));
- Status s = pool_->Submit(shared_ptr<Runnable>(new SlowTask(&latch)));
+ ASSERT_OK(pool_->Submit(SlowTask::NewSlowTask(&latch)));
+ ASSERT_OK(pool_->Submit(SlowTask::NewSlowTask(&latch)));
+ Status s = pool_->Submit(SlowTask::NewSlowTask(&latch));
CHECK(s.IsServiceUnavailable()) << "Expected failure due to queue blowout:" << s.ToString();
latch.CountDown();
pool_->Wait();
@@ -266,9 +314,9 @@ TEST_F(ThreadPoolTest, TestZeroQueueSize) {
CountDownLatch latch(1);
for (int i = 0; i < kMaxThreads; i++) {
- ASSERT_OK(pool_->Submit(shared_ptr<Runnable>(new SlowTask(&latch))));
+ ASSERT_OK(pool_->Submit(SlowTask::NewSlowTask(&latch)));
}
- Status s = pool_->Submit(shared_ptr<Runnable>(new SlowTask(&latch)));
+ Status s = pool_->Submit(SlowTask::NewSlowTask(&latch));
ASSERT_TRUE(s.IsServiceUnavailable()) << s.ToString();
ASSERT_STR_CONTAINS(s.ToString(), "Thread pool is at capacity");
latch.CountDown();
@@ -629,9 +677,9 @@ TEST_P(ThreadPoolTestTokenTypes, TestTokenSubmissionsAdhereToMaxQueueSize) {
});
// We will be able to submit two tasks: one for max_threads == 1 and one for
// max_queue_size == 1.
- ASSERT_OK(t->Submit(shared_ptr<Runnable>(new SlowTask(&latch))));
- ASSERT_OK(t->Submit(shared_ptr<Runnable>(new SlowTask(&latch))));
- Status s = t->Submit(shared_ptr<Runnable>(new SlowTask(&latch)));
+ ASSERT_OK(t->Submit(SlowTask::NewSlowTask(&latch)));
+ ASSERT_OK(t->Submit(SlowTask::NewSlowTask(&latch)));
+ Status s = t->Submit(SlowTask::NewSlowTask(&latch));
ASSERT_TRUE(s.IsServiceUnavailable());
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/cbd2fb67/src/kudu/util/threadpool.h
----------------------------------------------------------------------
diff --git a/src/kudu/util/threadpool.h b/src/kudu/util/threadpool.h
index e15c991..5abe0b4 100644
--- a/src/kudu/util/threadpool.h
+++ b/src/kudu/util/threadpool.h
@@ -217,6 +217,7 @@ class ThreadPool {
void SetRunTimeMicrosHistogram(const scoped_refptr<Histogram>& hist);
private:
+ FRIEND_TEST(ThreadPoolTest, TestThreadPoolWithNoMaxThreads);
FRIEND_TEST(ThreadPoolTest, TestThreadPoolWithNoMinimum);
FRIEND_TEST(ThreadPoolTest, TestVariableSizeThreadPool);
[2/5] kudu git commit: KUDU-1863: improve overall safety of graceful
server shutdown
Posted by to...@apache.org.
KUDU-1863: improve overall safety of graceful server shutdown
What exactly is a "safe" graceful shutdown? I define it like this:
1. Stop handling network traffic (i.e. RPCs and HTTP requests):
a. Future incoming RPCs are responded to with an error.
b. Future outgoing RPCs are immediately rejected (i.e. their callbacks
run on the issuing threads with an error status).
c. Queued but not yet handled incoming RPCs are responded to with an
error.
d. A "black hole" is set up for anyone who tries to respond to a deferred
incoming RPC (i.e. an RPC that was thunked from an RPC thread and may
be waiting for external stimuli).
e. Outgoing RPCs awaiting a response are marked as failed and their
callbacks are run on RPC threads with an error status.
2. Wait for any RPC/HTTP-related threads to finish. They may be processing
an incoming RPC, processing the response from an outgoing RPC, or dealing
with a failed RPC from step 2e.
3. Shut down RPC-using subsystems. Each one is responsible for cleaning up
its deferred incoming RPCs. If any responses are sent, they will be black
holed as per step 1d.
4. Shut down remaining server state.
By ordering step 3 after steps 1 and 2, we guarantee that there are no
reactor threads modifying state belonging to the subsystems that we are
trying to shut down. Unfortunately, adhering to this order is very
difficult today: we have to choose between steps 1a and 1b or destroying the
RPC subsystem altogether, which causes problems in step 3. Moreover, it's
not possible to stop handling HTTP requests without tearing down state,
which leads to failures in some subsystems.
So for now, we'll settle for a modified form of the above:
1. Stop accepting new RPCs.
2. Shut down RPC-using subsystems.
3. Destroy all RPC RPC state, waiting for outstanding RPC threads to finish.
4. Shut down the rest of the server.
This patch implements this modified form of a safe shutdown:
- During shutdown, servers first disable the acceptance of new RPCs and stop
the webserver.
- Servers now shut down master and tserver-specific subsystems before
generic subsystems.
- Shutting down generic subsystems first explicitly waits for reactor
threads to finish via a new synchronous Messenger::Shutdown.
The bulk of the patch is to Messenger::Shutdown and friends. While there, I
also reduced lock acquisition to just critical sections (fixing KUDU-1863 in
the process), extended the life of tls_context_ to Messenger destruction,
and changed Shutdown to be callable with registered services.
Change-Id: Ibbda29dd471e4dc1a8cb4f472cc456ccdb1e48cc
Reviewed-on: http://gerrit.cloudera.org:8080/7183
Tested-by: Kudu Jenkins
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/4c5c5126
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/4c5c5126
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/4c5c5126
Branch: refs/heads/master
Commit: 4c5c51264a89c309c4c8927052a233aa3ba2aa43
Parents: 1a7630e
Author: Adar Dembo <ad...@cloudera.com>
Authored: Wed Jun 14 13:12:02 2017 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Tue Jun 27 22:53:02 2017 +0000
----------------------------------------------------------------------
src/kudu/master/master.cc | 9 +++-
src/kudu/rpc/messenger.cc | 90 ++++++++++++++++++++++------------
src/kudu/rpc/messenger.h | 42 +++++++++++-----
src/kudu/rpc/reactor.cc | 14 ++++--
src/kudu/rpc/reactor.h | 20 ++++----
src/kudu/server/rpc_server.cc | 2 +-
src/kudu/server/server_base.cc | 20 ++++++--
src/kudu/server/server_base.h | 28 +++++++++++
src/kudu/tserver/tablet_server.cc | 14 ++++--
9 files changed, 175 insertions(+), 64 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/4c5c5126/src/kudu/master/master.cc
----------------------------------------------------------------------
diff --git a/src/kudu/master/master.cc b/src/kudu/master/master.cc
index fb72256..5fe077c 100644
--- a/src/kudu/master/master.cc
+++ b/src/kudu/master/master.cc
@@ -211,9 +211,16 @@ void Master::Shutdown() {
if (state_ == kRunning) {
string name = ToString();
LOG(INFO) << name << " shutting down...";
+
+ // 1. Stop accepting new RPCs.
+ UnregisterAllServices();
+
+ // 2. Shut down the master's subsystems.
maintenance_manager_->Shutdown();
- KuduServer::Shutdown();
catalog_manager_->Shutdown();
+
+ // 3. Shut down generic subsystems.
+ KuduServer::Shutdown();
LOG(INFO) << name << " shutdown complete.";
}
state_ = kStopped;
http://git-wip-us.apache.org/repos/asf/kudu/blob/4c5c5126/src/kudu/rpc/messenger.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/messenger.cc b/src/kudu/rpc/messenger.cc
index 8415fd2..fcf46a9 100644
--- a/src/kudu/rpc/messenger.cc
+++ b/src/kudu/rpc/messenger.cc
@@ -322,7 +322,13 @@ Status MessengerBuilder::Build(shared_ptr<Messenger> *msgr) {
// See comment on Messenger::retain_self_ member.
void Messenger::AllExternalReferencesDropped() {
- Shutdown();
+ // The last external ref may have been dropped in the context of a task
+ // running on a reactor thread. If that's the case, a SYNC shutdown here
+ // would deadlock.
+ //
+ // If a SYNC shutdown is desired, Shutdown() should be called explicitly.
+ ShutdownInternal(ShutdownMode::ASYNC);
+
CHECK(retain_self_.get());
// If we have no more external references, then we no longer
// need to retain ourself. We'll destruct as soon as all our
@@ -332,23 +338,41 @@ void Messenger::AllExternalReferencesDropped() {
}
void Messenger::Shutdown() {
+ ShutdownInternal(ShutdownMode::SYNC);
+}
+
+void Messenger::ShutdownInternal(ShutdownMode mode) {
+ if (mode == ShutdownMode::SYNC) {
+ ThreadRestrictions::AssertWaitAllowed();
+ }
+
// Since we're shutting down, it's OK to block.
+ //
+ // TODO(adar): this ought to be removed (i.e. if ASYNC, waiting should be
+ // forbidden, and if SYNC, we already asserted above), but that's not
+ // possible while shutting down thread and acceptor pools still involves
+ // joining threads.
ThreadRestrictions::ScopedAllowWait allow_wait;
- std::lock_guard<percpu_rwlock> guard(lock_);
- if (closing_) {
- return;
- }
- VLOG(1) << "shutting down messenger " << name_;
- closing_ = true;
+ acceptor_vec_t pools_to_shutdown;
+ RpcServicesMap services_to_release;
+ {
+ std::lock_guard<percpu_rwlock> guard(lock_);
+ if (closing_) {
+ return;
+ }
+ VLOG(1) << "shutting down messenger " << name_;
+ closing_ = true;
- DCHECK(rpc_services_.empty()) << "Unregister RPC services before shutting down Messenger";
- rpc_services_.clear();
+ services_to_release = std::move(rpc_services_);
+ pools_to_shutdown = std::move(acceptor_pools_);
+ }
- for (const shared_ptr<AcceptorPool>& acceptor_pool : acceptor_pools_) {
- acceptor_pool->Shutdown();
+ // Destroy state outside of the lock.
+ services_to_release.clear();
+ for (const auto& p : pools_to_shutdown) {
+ p->Shutdown();
}
- acceptor_pools_.clear();
// Need to shut down negotiation pool before the reactors, since the
// reactors close the Connection sockets, and may race against the negotiation
@@ -357,9 +381,8 @@ void Messenger::Shutdown() {
server_negotiation_pool_->Shutdown();
for (Reactor* reactor : reactors_) {
- reactor->Shutdown();
+ reactor->Shutdown(mode);
}
- tls_context_.reset();
}
Status Messenger::AddAcceptorPool(const Sockaddr &accept_addr,
@@ -398,21 +421,27 @@ Status Messenger::RegisterService(const string& service_name,
}
}
-Status Messenger::UnregisterAllServices() {
- std::lock_guard<percpu_rwlock> guard(lock_);
- rpc_services_.clear();
- return Status::OK();
+void Messenger::UnregisterAllServices() {
+ RpcServicesMap to_release;
+ {
+ std::lock_guard<percpu_rwlock> guard(lock_);
+ to_release = std::move(rpc_services_);
+ }
+ // Release the map outside of the lock.
}
-// Unregister an RpcService.
Status Messenger::UnregisterService(const string& service_name) {
- std::lock_guard<percpu_rwlock> guard(lock_);
- if (rpc_services_.erase(service_name)) {
- return Status::OK();
- } else {
- return Status::ServiceUnavailable(Substitute("service $0 not registered on $1",
- service_name, name_));
+ scoped_refptr<RpcService> to_release;
+ {
+ std::lock_guard<percpu_rwlock> guard(lock_);
+ to_release = EraseKeyReturnValuePtr(&rpc_services_, service_name);
+ if (!to_release) {
+ return Status::ServiceUnavailable(Substitute(
+ "service $0 not registered on $1", service_name, name_));
+ }
}
+ // Release the service outside of the lock.
+ return Status::OK();
}
void Messenger::QueueOutboundCall(const shared_ptr<OutboundCall> &call) {
@@ -519,13 +548,14 @@ void Messenger::ScheduleOnReactor(const boost::function<void(const Status&)>& fu
}
const scoped_refptr<RpcService> Messenger::rpc_service(const string& service_name) const {
- std::lock_guard<percpu_rwlock> guard(lock_);
scoped_refptr<RpcService> service;
- if (FindCopy(rpc_services_, service_name, &service)) {
- return service;
- } else {
- return scoped_refptr<RpcService>(nullptr);
+ {
+ std::lock_guard<percpu_rwlock> guard(lock_);
+ if (!FindCopy(rpc_services_, service_name, &service)) {
+ return scoped_refptr<RpcService>(nullptr);
+ }
}
+ return service;
}
ThreadPool* Messenger::negotiation_pool(Connection::Direction dir) {
http://git-wip-us.apache.org/repos/asf/kudu/blob/4c5c5126/src/kudu/rpc/messenger.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/messenger.h b/src/kudu/rpc/messenger.h
index 2ff5951..73e02fa 100644
--- a/src/kudu/rpc/messenger.h
+++ b/src/kudu/rpc/messenger.h
@@ -149,6 +149,7 @@ class Messenger {
friend class MessengerBuilder;
friend class Proxy;
friend class Reactor;
+ friend class ReactorThread;
typedef std::vector<std::shared_ptr<AcceptorPool> > acceptor_vec_t;
typedef std::unordered_map<std::string, scoped_refptr<RpcService> > RpcServicesMap;
@@ -156,7 +157,10 @@ class Messenger {
~Messenger();
- // Stop all communication and prevent further use.
+ // Stops all communication and prevents further use. If called explicitly,
+ // also waits for outstanding tasks running on reactor threads to finish,
+ // which means it may not be called from a reactor task.
+ //
// It's not required to call this -- dropping the shared_ptr provided
// from MessengerBuilder::Build will automatically call this method.
void Shutdown();
@@ -178,13 +182,18 @@ class Messenger {
std::shared_ptr<AcceptorPool>* pool);
// Register a new RpcService to handle inbound requests.
+ //
+ // Returns an error if a service with the same name is already registered.
Status RegisterService(const std::string& service_name,
const scoped_refptr<RpcService>& service);
- // Unregister currently-registered RpcService.
+ // Unregister an RpcService by name.
+ //
+ // Returns an error if no service with this name can be found.
Status UnregisterService(const std::string& service_name);
- Status UnregisterAllServices();
+ // Unregisters all RPC services.
+ void UnregisterAllServices();
// Queue a call for transmission. This will pick the appropriate reactor,
// and enqueue a task on that reactor to assign and send the call.
@@ -259,6 +268,15 @@ class Messenger {
void RunTimeoutThread();
void UpdateCurTime();
+ // Shuts down the messenger.
+ //
+ // Depending on 'mode', may or may not wait on any outstanding reactor tasks.
+ enum class ShutdownMode {
+ SYNC,
+ ASYNC,
+ };
+ void ShutdownInternal(ShutdownMode mode);
+
// Called by external-facing shared_ptr when the user no longer holds
// any references. See 'retain_self_' for more info.
void AllExternalReferencesDropped();
@@ -315,7 +333,7 @@ class Messenger {
// | |
// v
// Messenger <------------ shared_ptr[2] --- Reactor
- // ^ | ----------- bare pointer --> Reactor
+ // ^ | ------------- bare pointer --> Reactor
// \__/
// shared_ptr[2]
// (retain_self_)
@@ -327,10 +345,11 @@ class Messenger {
//
// The teardown sequence is as follows:
// Option 1): User calls "Shutdown()" explicitly:
- // - Messenger::Shutdown tells Reactors to shut down
- // - When each reactor thread finishes, it drops its shared_ptr[2]
+ // - Messenger::Shutdown tells Reactors to shut down.
+ // - When each reactor thread finishes, it drops its shared_ptr[2].
// - the Messenger::retain_self instance remains, keeping the Messenger
// alive.
+ // - Before returning, Messenger::Shutdown waits for Reactors to shut down.
// - The user eventually drops its shared_ptr[1], which calls
// Messenger::AllExternalReferencesDropped. This drops retain_self_
// and results in object destruction.
@@ -342,11 +361,12 @@ class Messenger {
// - When the last Reactor thread dies, there will be no more shared_ptr[1] references
// and the Messenger will be destroyed.
//
- // The main goal of all of this confusion is that the reactor threads need to be able
- // to shut down asynchronously, and we need to keep the Messenger alive until they
- // do so. So, handing out a normal shared_ptr to users would force the Messenger
- // destructor to Join() the reactor threads, which causes a problem if the user
- // tries to destruct the Messenger from within a Reactor thread itself.
+ // The main goal of all of this confusion is that when using option 2, the
+ // reactor threads need to be able to shut down asynchronously, and we need
+ // to keep the Messenger alive until they do so. If normal shared_ptrs were
+ // handed out to users, the Messenger destructor may be forced to Join() the
+ // reactor threads, which deadlocks if the user destructs the Messenger from
+ // within a Reactor thread itself.
std::shared_ptr<Messenger> retain_self_;
DISALLOW_COPY_AND_ASSIGN(Messenger);
http://git-wip-us.apache.org/repos/asf/kudu/blob/4c5c5126/src/kudu/rpc/reactor.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/reactor.cc b/src/kudu/rpc/reactor.cc
index df4a661..33746c5 100644
--- a/src/kudu/rpc/reactor.cc
+++ b/src/kudu/rpc/reactor.cc
@@ -129,11 +129,17 @@ Status ReactorThread::Init() {
return kudu::Thread::Create("reactor", "rpc reactor", &ReactorThread::RunThread, this, &thread_);
}
-void ReactorThread::Shutdown() {
+void ReactorThread::Shutdown(Messenger::ShutdownMode mode) {
CHECK(reactor_->closing()) << "Should be called after setting closing_ flag";
VLOG(1) << name() << ": shutting down Reactor thread.";
WakeThread();
+
+ if (mode == Messenger::ShutdownMode::SYNC) {
+ // Join() will return a bad status if asked to join on the currently
+ // running thread.
+ CHECK_OK(ThreadJoiner(thread_.get()).Join());
+ }
}
void ReactorThread::ShutdownInternal() {
@@ -588,7 +594,7 @@ Status Reactor::Init() {
return thread_.Init();
}
-void Reactor::Shutdown() {
+void Reactor::Shutdown(Messenger::ShutdownMode mode) {
{
std::lock_guard<LockType> l(lock_);
if (closing_) {
@@ -597,7 +603,7 @@ void Reactor::Shutdown() {
closing_ = true;
}
- thread_.Shutdown();
+ thread_.Shutdown(mode);
// Abort all pending tasks. No new tasks can get scheduled after this
// because ScheduleReactorTask() tests the closing_ flag set above.
@@ -610,7 +616,7 @@ void Reactor::Shutdown() {
}
Reactor::~Reactor() {
- Shutdown();
+ Shutdown(Messenger::ShutdownMode::ASYNC);
}
const std::string& Reactor::name() const {
http://git-wip-us.apache.org/repos/asf/kudu/blob/4c5c5126/src/kudu/rpc/reactor.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/reactor.h b/src/kudu/rpc/reactor.h
index 37eedd4..e9d131b 100644
--- a/src/kudu/rpc/reactor.h
+++ b/src/kudu/rpc/reactor.h
@@ -17,8 +17,7 @@
#ifndef KUDU_RPC_REACTOR_H
#define KUDU_RPC_REACTOR_H
-#include <stdint.h>
-
+#include <cstdint>
#include <list>
#include <map>
#include <memory>
@@ -32,12 +31,13 @@
#include "kudu/gutil/ref_counted.h"
#include "kudu/rpc/connection.h"
+#include "kudu/rpc/messenger.h"
#include "kudu/rpc/transfer.h"
-#include "kudu/util/thread.h"
#include "kudu/util/locks.h"
#include "kudu/util/monotime.h"
#include "kudu/util/net/socket.h"
#include "kudu/util/status.h"
+#include "kudu/util/thread.h"
namespace kudu {
@@ -49,8 +49,6 @@ typedef std::list<scoped_refptr<Connection>> conn_list_t;
class DumpRunningRpcsRequestPB;
class DumpRunningRpcsResponsePB;
-class Messenger;
-class MessengerBuilder;
class Reactor;
enum class CredentialsPolicy;
@@ -148,10 +146,11 @@ class ReactorThread {
Status DumpRunningRpcs(const DumpRunningRpcsRequestPB& req,
DumpRunningRpcsResponsePB* resp);
- // Block until the Reactor thread is shut down
+ // Shuts down a reactor thread, optionally waiting for it to exit.
+ // Reactor::Shutdown() must have been called already.
//
- // This must be called from another thread.
- void Shutdown();
+ // If mode == SYNC, may not be called from the reactor thread itself.
+ void Shutdown(Messenger::ShutdownMode mode);
// This method is thread-safe.
void WakeThread();
@@ -290,8 +289,9 @@ class Reactor {
const MessengerBuilder &bld);
Status Init();
- // Block until the Reactor is shut down
- void Shutdown();
+ // Shuts down the reactor and its corresponding thread, optionally waiting
+ // until the thread has exited.
+ void Shutdown(Messenger::ShutdownMode mode);
~Reactor();
http://git-wip-us.apache.org/repos/asf/kudu/blob/4c5c5126/src/kudu/server/rpc_server.cc
----------------------------------------------------------------------
diff --git a/src/kudu/server/rpc_server.cc b/src/kudu/server/rpc_server.cc
index 1f8c542..a4069f0 100644
--- a/src/kudu/server/rpc_server.cc
+++ b/src/kudu/server/rpc_server.cc
@@ -195,7 +195,7 @@ void RpcServer::Shutdown() {
acceptor_pools_.clear();
if (messenger_) {
- WARN_NOT_OK(messenger_->UnregisterAllServices(), "Unable to unregister our services");
+ messenger_->UnregisterAllServices();
}
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/4c5c5126/src/kudu/server/server_base.cc
----------------------------------------------------------------------
diff --git a/src/kudu/server/server_base.cc b/src/kudu/server/server_base.cc
index eba74ae..7ffbf46 100644
--- a/src/kudu/server/server_base.cc
+++ b/src/kudu/server/server_base.cc
@@ -472,6 +472,18 @@ Status ServerBase::Start() {
}
void ServerBase::Shutdown() {
+ // First, stop accepting incoming requests and wait for any outstanding
+ // requests to finish processing.
+ //
+ // Note: prior to Messenger::Shutdown, it is assumed that any incoming RPCs
+ // deferred from reactor threads have already been cleaned up.
+ if (web_server_) {
+ web_server_->Stop();
+ }
+ rpc_server_->Shutdown();
+ messenger_->Shutdown();
+
+ // Next, shut down remaining server components.
stop_background_threads_latch_.CountDown();
if (metrics_logging_thread_) {
metrics_logging_thread_->Join();
@@ -479,10 +491,10 @@ void ServerBase::Shutdown() {
if (excess_log_deleter_thread_) {
excess_log_deleter_thread_->Join();
}
- if (web_server_) {
- web_server_->Stop();
- }
- rpc_server_->Shutdown();
+}
+
+void ServerBase::UnregisterAllServices() {
+ messenger_->UnregisterAllServices();
}
} // namespace server
http://git-wip-us.apache.org/repos/asf/kudu/blob/4c5c5126/src/kudu/server/server_base.h
----------------------------------------------------------------------
diff --git a/src/kudu/server/server_base.h b/src/kudu/server/server_base.h
index 69fd650..b0095db 100644
--- a/src/kudu/server/server_base.h
+++ b/src/kudu/server/server_base.h
@@ -120,11 +120,39 @@ class ServerBase {
virtual ~ServerBase();
virtual Status Init();
+
+ // Starts the server, including activating its RPC and HTTP endpoints such
+ // that incoming requests get processed.
virtual Status Start();
+
+ // Shuts down the server.
virtual void Shutdown();
+ // Registers a new RPC service. Once Start() is called, the server will
+ // process and dispatch incoming RPCs belonging to this service.
Status RegisterService(gscoped_ptr<rpc::ServiceIf> rpc_impl);
+ // Unregisters all RPC services. After this function returns, any subsequent
+ // incoming RPCs will be rejected.
+ //
+ // When shutting down, this function should be called before shutting down
+ // higher-level subsystems. For example:
+ // 1. ServerBase::UnregisterAllServices()
+ // 2. <shut down other subsystems>
+ // 3. ServerBase::Shutdown()
+ //
+ // TODO(adar): this should also wait on all outstanding RPCs to finish via
+ // Messenger::Shutdown, but doing that causes too many other shutdown-related
+ // issues. Here are a few that I observed:
+ // - tserver heartbeater threads access acceptor pool socket state.
+ // - Shutting down TabletReplicas invokes RPC callbacks for aborted
+ // transactions, but Messenger::Shutdown has already destroyed too much
+ // necessary RPC state.
+ //
+ // TODO(adar): this should also shutdown the webserver, but it isn't safe to
+ // do that before before shutting down the tserver heartbeater.
+ void UnregisterAllServices();
+
void LogUnauthorizedAccess(rpc::RpcContext* rpc) const;
const std::string name_;
http://git-wip-us.apache.org/repos/asf/kudu/blob/4c5c5126/src/kudu/tserver/tablet_server.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/tablet_server.cc b/src/kudu/tserver/tablet_server.cc
index 6c91480..658ecfb 100644
--- a/src/kudu/tserver/tablet_server.cc
+++ b/src/kudu/tserver/tablet_server.cc
@@ -128,12 +128,20 @@ Status TabletServer::Start() {
void TabletServer::Shutdown() {
if (initted_) {
- LOG(INFO) << "TabletServer shutting down...";
+ string name = ToString();
+ LOG(INFO) << name << " shutting down...";
+
+ // 1. Stop accepting new RPCs.
+ UnregisterAllServices();
+
+ // 2. Shut down the tserver's subsystems.
maintenance_manager_->Shutdown();
WARN_NOT_OK(heartbeater_->Stop(), "Failed to stop TS Heartbeat thread");
- KuduServer::Shutdown();
tablet_manager_->Shutdown();
- LOG(INFO) << "TabletServer shut down complete. Bye!";
+
+ // 3. Shut down generic subsystems.
+ KuduServer::Shutdown();
+ LOG(INFO) << name << " shutdown complete.";
}
}
[4/5] kudu git commit: threadpool: token-based task sequencing
Posted by to...@apache.org.
threadpool: token-based task sequencing
This patch adds task sequencing to util/threadpool. Task sequencing allows
M contexts to share a single pool with N threads while also ensuring that
the pool executes tasks belonging to each context in order. Previously this
was only achievable via "one singleton pool per context", which grossly
inflated the total number of threads and overall state.
The new logic is implemented in ThreadPoolToken. Tasks submitted via tokens
are logically grouped together. They may also be sequenced, depending on the
token's execution mode. Tokens expose Wait() variants which will block on
the task group. Tokens may also be shut down before being destroyed, which
has semantics equivalent to shutting down the pool itself (i.e. tasks may no
longer be submitted with these tokens).
Some notes:
- I evaluated two other implementations. In one, tokens had an implicit
lifecycle that was automatically managed by the threadpool. While simpler
for clients, the threadpool was more inefficient with more allocations and
deallocations in each submission. In the other, token submission was built
using regular task submission. This afforded a certain separation between
the "core" of the threadpool and the token logic, but it complicated
locking and tracking of queued tasks.
- I tried to keep submission (whether token-based or tokenless) fast. Just
the change from std::list to std::deque for the main queue ought to
improve performance of tokenless submissions. The next bottleneck is
likely to be lock contention on the global threadpool lock.
Change-Id: If46dc34212027b6ea5dbc2ead7c7af8be57f2c70
Reviewed-on: http://gerrit.cloudera.org:8080/6874
Tested-by: Kudu Jenkins
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/428a34ba
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/428a34ba
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/428a34ba
Branch: refs/heads/master
Commit: 428a34ba0aecfb81178d821732feac694a9b35c1
Parents: 74f99e4
Author: Adar Dembo <ad...@cloudera.com>
Authored: Thu May 11 20:58:52 2017 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Tue Jun 27 22:54:56 2017 +0000
----------------------------------------------------------------------
src/kudu/util/debug-util.h | 12 +-
src/kudu/util/threadpool-test.cc | 399 +++++++++++++++++++++++++++++++++-
src/kudu/util/threadpool.cc | 387 +++++++++++++++++++++++++++++----
src/kudu/util/threadpool.h | 274 +++++++++++++++++++----
4 files changed, 986 insertions(+), 86 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/428a34ba/src/kudu/util/debug-util.h
----------------------------------------------------------------------
diff --git a/src/kudu/util/debug-util.h b/src/kudu/util/debug-util.h
index 2617520..fc12445 100644
--- a/src/kudu/util/debug-util.h
+++ b/src/kudu/util/debug-util.h
@@ -94,19 +94,29 @@ void HexStackTraceToString(char* buf, size_t size);
// Requires external synchronization.
class StackTrace {
public:
+
+ // Constructs a new (uncollected) stack trace.
StackTrace()
: num_frames_(0) {
}
+ // Resets the stack trace to an uncollected state.
void Reset() {
num_frames_ = 0;
}
+ // Returns true if Collect() (but not Reset()) has been called on this stack trace.
+ bool HasCollected() const {
+ return num_frames_ > 0;
+ }
+
+ // Copies the contents of 's' into this stack trace.
void CopyFrom(const StackTrace& s) {
memcpy(this, &s, sizeof(s));
}
- bool Equals(const StackTrace& s) {
+ // Returns true if the stack trace 's' matches this trace.
+ bool Equals(const StackTrace& s) const {
return s.num_frames_ == num_frames_ &&
strings::memeq(frames_, s.frames_,
num_frames_ * sizeof(frames_[0]));
http://git-wip-us.apache.org/repos/asf/kudu/blob/428a34ba/src/kudu/util/threadpool-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/util/threadpool-test.cc b/src/kudu/util/threadpool-test.cc
index 6bd5826..0beb962 100644
--- a/src/kudu/util/threadpool-test.cc
+++ b/src/kudu/util/threadpool-test.cc
@@ -15,8 +15,13 @@
// specific language governing permissions and limitations
// under the License.
+#include <atomic>
+#include <limits>
#include <memory>
+#include <mutex>
#include <string>
+#include <thread>
+#include <vector>
#include <boost/bind.hpp>
#include <glog/logging.h>
@@ -24,16 +29,27 @@
#include "kudu/gutil/atomicops.h"
#include "kudu/gutil/bind.h"
+#include "kudu/gutil/strings/substitute.h"
+#include "kudu/util/barrier.h"
#include "kudu/util/countdown_latch.h"
+#include "kudu/util/locks.h"
#include "kudu/util/metrics.h"
#include "kudu/util/promise.h"
+#include "kudu/util/random.h"
#include "kudu/util/scoped_cleanup.h"
-#include "kudu/util/threadpool.h"
#include "kudu/util/test_macros.h"
#include "kudu/util/test_util.h"
+#include "kudu/util/threadpool.h"
#include "kudu/util/trace.h"
+using std::atomic;
using std::shared_ptr;
+using std::string;
+using std::thread;
+using std::unique_ptr;
+using std::vector;
+
+using strings::Substitute;
namespace kudu {
@@ -363,5 +379,386 @@ TEST_F(ThreadPoolTest, TestSlowDestructor) {
ASSERT_LT((MonoTime::Now() - start).ToSeconds(), 5);
}
+// For test cases that should run with both kinds of tokens.
+class ThreadPoolTestTokenTypes : public ThreadPoolTest,
+ public testing::WithParamInterface<ThreadPool::ExecutionMode> {};
+
+INSTANTIATE_TEST_CASE_P(Tokens, ThreadPoolTestTokenTypes,
+ ::testing::Values(ThreadPool::ExecutionMode::SERIAL,
+ ThreadPool::ExecutionMode::CONCURRENT));
+
+
+TEST_P(ThreadPoolTestTokenTypes, TestTokenSubmitAndWait) {
+ unique_ptr<ThreadPoolToken> t = pool_->NewToken(GetParam());
+ int i = 0;
+ ASSERT_OK(t->SubmitFunc([&]() {
+ SleepFor(MonoDelta::FromMilliseconds(1));
+ i++;
+ }));
+ t->Wait();
+ ASSERT_EQ(1, i);
+}
+
+TEST_F(ThreadPoolTest, TestTokenSubmitsProcessedSerially) {
+ unique_ptr<ThreadPoolToken> t = pool_->NewToken(ThreadPool::ExecutionMode::SERIAL);
+ Random r(SeedRandom());
+ string result;
+ for (char c = 'a'; c < 'f'; c++) {
+ // Sleep a little first so that there's a higher chance of out-of-order
+ // appends if the submissions did execute in parallel.
+ int sleep_ms = r.Next() % 5;
+ ASSERT_OK(t->SubmitFunc([&result, c, sleep_ms]() {
+ SleepFor(MonoDelta::FromMilliseconds(sleep_ms));
+ result += c;
+ }));
+ }
+ t->Wait();
+ ASSERT_EQ("abcde", result);
+}
+
+TEST_P(ThreadPoolTestTokenTypes, TestTokenSubmitsProcessedConcurrently) {
+ const int kNumTokens = 5;
+ ASSERT_OK(RebuildPoolWithBuilder(ThreadPoolBuilder(kDefaultPoolName)
+ .set_max_threads(kNumTokens)));
+ vector<unique_ptr<ThreadPoolToken>> tokens;
+
+ // A violation to the tested invariant would yield a deadlock, so let's set
+ // up an alarm to bail us out.
+ alarm(60);
+ auto cleanup = MakeScopedCleanup([&] {
+ alarm(0); // Disable alarm on test exit.
+ });
+ shared_ptr<Barrier> b = std::make_shared<Barrier>(kNumTokens + 1);
+ for (int i = 0; i < kNumTokens; i++) {
+ tokens.emplace_back(pool_->NewToken(GetParam()));
+ ASSERT_OK(tokens.back()->SubmitFunc([b]() {
+ b->Wait();
+ }));
+ }
+
+ // This will deadlock if the above tasks weren't all running concurrently.
+ b->Wait();
+}
+
+TEST_F(ThreadPoolTest, TestTokenSubmitsNonSequential) {
+ const int kNumSubmissions = 5;
+ ASSERT_OK(RebuildPoolWithBuilder(ThreadPoolBuilder(kDefaultPoolName)
+ .set_max_threads(kNumSubmissions)));
+
+ // A violation to the tested invariant would yield a deadlock, so let's set
+ // up an alarm to bail us out.
+ alarm(60);
+ auto cleanup = MakeScopedCleanup([&] {
+ alarm(0); // Disable alarm on test exit.
+ });
+ shared_ptr<Barrier> b = std::make_shared<Barrier>(kNumSubmissions + 1);
+ unique_ptr<ThreadPoolToken> t = pool_->NewToken(ThreadPool::ExecutionMode::CONCURRENT);
+ for (int i = 0; i < kNumSubmissions; i++) {
+ ASSERT_OK(t->SubmitFunc([b]() {
+ b->Wait();
+ }));
+ }
+
+ // This will deadlock if the above tasks weren't all running concurrently.
+ b->Wait();
+}
+
+TEST_P(ThreadPoolTestTokenTypes, TestTokenShutdown) {
+ ASSERT_OK(RebuildPoolWithBuilder(ThreadPoolBuilder(kDefaultPoolName)
+ .set_max_threads(4)));
+
+ unique_ptr<ThreadPoolToken> t1(pool_->NewToken(GetParam()));
+ unique_ptr<ThreadPoolToken> t2(pool_->NewToken(GetParam()));
+ CountDownLatch l1(1);
+ CountDownLatch l2(1);
+
+ // A violation to the tested invariant would yield a deadlock, so let's set
+ // up an alarm to bail us out.
+ alarm(60);
+ auto cleanup = MakeScopedCleanup([&] {
+ alarm(0); // Disable alarm on test exit.
+ });
+
+ for (int i = 0; i < 3; i++) {
+ ASSERT_OK(t1->SubmitFunc([&]() {
+ l1.Wait();
+ }));
+ }
+ for (int i = 0; i < 3; i++) {
+ ASSERT_OK(t2->SubmitFunc([&]() {
+ l2.Wait();
+ }));
+ }
+
+ // Unblock all of t1's tasks, but not t2's tasks.
+ l1.CountDown();
+
+ // If this also waited for t2's tasks, it would deadlock.
+ t1->Shutdown();
+
+ // We can no longer submit to t1 but we can still submit to t2.
+ ASSERT_TRUE(t1->SubmitFunc([](){}).IsServiceUnavailable());
+ ASSERT_OK(t2->SubmitFunc([](){}));
+
+ // Unblock t2's tasks.
+ l2.CountDown();
+ t2->Shutdown();
+}
+
+TEST_P(ThreadPoolTestTokenTypes, TestTokenWaitForAll) {
+ const int kNumTokens = 3;
+ const int kNumSubmissions = 20;
+ Random r(SeedRandom());
+ vector<unique_ptr<ThreadPoolToken>> tokens;
+ for (int i = 0; i < kNumTokens; i++) {
+ tokens.emplace_back(pool_->NewToken(GetParam()));
+ }
+
+ atomic<int32_t> v(0);
+ for (int i = 0; i < kNumSubmissions; i++) {
+ // Sleep a little first to raise the likelihood of the test thread
+ // reaching Wait() before the submissions finish.
+ int sleep_ms = r.Next() % 5;
+
+ auto task = [&v, sleep_ms]() {
+ SleepFor(MonoDelta::FromMilliseconds(sleep_ms));
+ v++;
+ };
+
+ // Half of the submissions will be token-less, and half will use a token.
+ if (i % 2 == 0) {
+ ASSERT_OK(pool_->SubmitFunc(task));
+ } else {
+ int token_idx = r.Next() % tokens.size();
+ ASSERT_OK(tokens[token_idx]->SubmitFunc(task));
+ }
+ }
+ pool_->Wait();
+ ASSERT_EQ(kNumSubmissions, v);
+}
+
+TEST_F(ThreadPoolTest, TestFuzz) {
+ const int kNumOperations = 1000;
+ Random r(SeedRandom());
+ vector<unique_ptr<ThreadPoolToken>> tokens;
+
+ for (int i = 0; i < kNumOperations; i++) {
+ // Operation distribution:
+ //
+ // - Submit without a token: 40%
+ // - Submit with a randomly selected token: 35%
+ // - Allocate a new token: 10%
+ // - Wait on a randomly selected token: 7%
+ // - Shutdown a randomly selected token: 4%
+ // - Deallocate a randomly selected token: 2%
+ // - Wait for all submissions: 2%
+ int op = r.Next() % 100;
+ if (op < 40) {
+ // Submit without a token.
+ int sleep_ms = r.Next() % 5;
+ ASSERT_OK(pool_->SubmitFunc([sleep_ms]() {
+ // Sleep a little first to increase task overlap.
+ SleepFor(MonoDelta::FromMilliseconds(sleep_ms));
+ }));
+ } else if (op < 75) {
+ // Submit with a randomly selected token.
+ if (tokens.empty()) {
+ continue;
+ }
+ int sleep_ms = r.Next() % 5;
+ int token_idx = r.Next() % tokens.size();
+ Status s = tokens[token_idx]->SubmitFunc([sleep_ms]() {
+ // Sleep a little first to increase task overlap.
+ SleepFor(MonoDelta::FromMilliseconds(sleep_ms));
+ });
+ ASSERT_TRUE(s.ok() || s.IsServiceUnavailable());
+ } else if (op < 85) {
+ // Allocate a token with a randomly selected policy.
+ ThreadPool::ExecutionMode mode = r.Next() % 2 ?
+ ThreadPool::ExecutionMode::SERIAL :
+ ThreadPool::ExecutionMode::CONCURRENT;
+ tokens.emplace_back(pool_->NewToken(mode));
+ } else if (op < 92) {
+ // Wait on a randomly selected token.
+ if (tokens.empty()) {
+ continue;
+ }
+ int token_idx = r.Next() % tokens.size();
+ tokens[token_idx]->Wait();
+ } else if (op < 96) {
+ // Shutdown a randomly selected token.
+ if (tokens.empty()) {
+ continue;
+ }
+ int token_idx = r.Next() % tokens.size();
+ tokens[token_idx]->Shutdown();
+ } else if (op < 98) {
+ // Deallocate a randomly selected token.
+ if (tokens.empty()) {
+ continue;
+ }
+ auto it = tokens.begin();
+ int token_idx = r.Next() % tokens.size();
+ std::advance(it, token_idx);
+ tokens.erase(it);
+ } else {
+ // Wait on everything.
+ ASSERT_LT(op, 100);
+ ASSERT_GE(op, 98);
+ pool_->Wait();
+ }
+ }
+
+ // Some test runs will shut down the pool before the tokens, and some won't.
+ // Either way should be safe.
+ if (r.Next() % 2 == 0) {
+ pool_->Shutdown();
+ }
+}
+
+TEST_P(ThreadPoolTestTokenTypes, TestTokenSubmissionsAdhereToMaxQueueSize) {
+ ASSERT_OK(RebuildPoolWithBuilder(ThreadPoolBuilder(kDefaultPoolName)
+ .set_min_threads(1)
+ .set_max_threads(1)
+ .set_max_queue_size(1)));
+
+ CountDownLatch latch(1);
+ unique_ptr<ThreadPoolToken> t = pool_->NewToken(GetParam());
+ auto cleanup = MakeScopedCleanup([&]() {
+ latch.CountDown();
+ });
+ // We will be able to submit two tasks: one for max_threads == 1 and one for
+ // max_queue_size == 1.
+ ASSERT_OK(t->Submit(shared_ptr<Runnable>(new SlowTask(&latch))));
+ ASSERT_OK(t->Submit(shared_ptr<Runnable>(new SlowTask(&latch))));
+ Status s = t->Submit(shared_ptr<Runnable>(new SlowTask(&latch)));
+ ASSERT_TRUE(s.IsServiceUnavailable());
+}
+
+TEST_F(ThreadPoolTest, TestTokenConcurrency) {
+ const int kNumTokens = 20;
+ const int kTestRuntimeSecs = 1;
+ const int kCycleThreads = 2;
+ const int kShutdownThreads = 2;
+ const int kWaitThreads = 2;
+ const int kSubmitThreads = 8;
+
+ vector<shared_ptr<ThreadPoolToken>> tokens;
+ Random rng(SeedRandom());
+
+ // Protects 'tokens' and 'rng'.
+ simple_spinlock lock;
+
+ // Fetch a token from 'tokens' at random.
+ auto GetRandomToken = [&]() -> shared_ptr<ThreadPoolToken> {
+ std::lock_guard<simple_spinlock> l(lock);
+ int idx = rng.Uniform(kNumTokens);
+ return tokens[idx];
+ };
+
+ // Preallocate all of the tokens.
+ for (int i = 0; i < kNumTokens; i++) {
+ ThreadPool::ExecutionMode mode;
+ {
+ std::lock_guard<simple_spinlock> l(lock);
+ mode = rng.Next() % 2 ?
+ ThreadPool::ExecutionMode::SERIAL :
+ ThreadPool::ExecutionMode::CONCURRENT;
+ }
+ tokens.emplace_back(pool_->NewToken(mode).release());
+ }
+
+ atomic<int64_t> total_num_tokens_cycled(0);
+ atomic<int64_t> total_num_tokens_shutdown(0);
+ atomic<int64_t> total_num_tokens_waited(0);
+ atomic<int64_t> total_num_tokens_submitted(0);
+
+ CountDownLatch latch(1);
+ vector<thread> threads;
+
+ for (int i = 0; i < kCycleThreads; i++) {
+ // Pick a token at random and replace it.
+ //
+ // The replaced token is only destroyed when the last ref is dropped,
+ // possibly by another thread.
+ threads.emplace_back([&]() {
+ int num_tokens_cycled = 0;
+ while (latch.count()) {
+ {
+ std::lock_guard<simple_spinlock> l(lock);
+ int idx = rng.Uniform(kNumTokens);
+ ThreadPool::ExecutionMode mode = rng.Next() % 2 ?
+ ThreadPool::ExecutionMode::SERIAL :
+ ThreadPool::ExecutionMode::CONCURRENT;
+ tokens[idx] = shared_ptr<ThreadPoolToken>(pool_->NewToken(mode).release());
+ }
+ num_tokens_cycled++;
+
+ // Sleep a bit, otherwise this thread outpaces the other threads and
+ // nothing interesting happens to most tokens.
+ SleepFor(MonoDelta::FromMicroseconds(10));
+ }
+ total_num_tokens_cycled += num_tokens_cycled;
+ });
+ }
+
+ for (int i = 0; i < kShutdownThreads; i++) {
+ // Pick a token at random and shut it down. Submitting a task to a shut
+ // down token will return a ServiceUnavailable error.
+ threads.emplace_back([&]() {
+ int num_tokens_shutdown = 0;
+ while (latch.count()) {
+ GetRandomToken()->Shutdown();
+ num_tokens_shutdown++;
+ }
+ total_num_tokens_shutdown += num_tokens_shutdown;
+ });
+ }
+
+ for (int i = 0; i < kWaitThreads; i++) {
+ // Pick a token at random and wait for any outstanding tasks.
+ threads.emplace_back([&]() {
+ int num_tokens_waited = 0;
+ while (latch.count()) {
+ GetRandomToken()->Wait();
+ num_tokens_waited++;
+ }
+ total_num_tokens_waited += num_tokens_waited;
+ });
+ }
+
+ for (int i = 0; i < kSubmitThreads; i++) {
+ // Pick a token at random and submit a task to it.
+ threads.emplace_back([&]() {
+ int num_tokens_submitted = 0;
+ Random rng(SeedRandom());
+ while (latch.count()) {
+ int sleep_ms = rng.Next() % 5;
+ Status s = GetRandomToken()->SubmitFunc([sleep_ms]() {
+ // Sleep a little first so that tasks are running during other events.
+ SleepFor(MonoDelta::FromMilliseconds(sleep_ms));
+ });
+ CHECK(s.ok() || s.IsServiceUnavailable());
+ num_tokens_submitted++;
+ }
+ total_num_tokens_submitted += num_tokens_submitted;
+ });
+ }
+
+ SleepFor(MonoDelta::FromSeconds(kTestRuntimeSecs));
+ latch.CountDown();
+ for (auto& t : threads) {
+ t.join();
+ }
+
+ LOG(INFO) << Substitute("Tokens cycled ($0 threads): $1",
+ kCycleThreads, total_num_tokens_cycled.load());
+ LOG(INFO) << Substitute("Tokens shutdown ($0 threads): $1",
+ kShutdownThreads, total_num_tokens_shutdown.load());
+ LOG(INFO) << Substitute("Tokens waited ($0 threads): $1",
+ kWaitThreads, total_num_tokens_waited.load());
+ LOG(INFO) << Substitute("Tokens submitted ($0 threads): $1",
+ kSubmitThreads, total_num_tokens_submitted.load());
+}
} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/428a34ba/src/kudu/util/threadpool.cc
----------------------------------------------------------------------
diff --git a/src/kudu/util/threadpool.cc b/src/kudu/util/threadpool.cc
index b3f4ddf..eda052f 100644
--- a/src/kudu/util/threadpool.cc
+++ b/src/kudu/util/threadpool.cc
@@ -17,13 +17,19 @@
#include "kudu/util/threadpool.h"
+#include <algorithm>
+#include <iostream>
+#include <limits>
+#include <memory>
+#include <string>
+
#include <boost/function.hpp>
#include <gflags/gflags.h>
#include <glog/logging.h>
-#include <limits>
-#include <string>
#include "kudu/gutil/callback.h"
+#include "kudu/gutil/macros.h"
+#include "kudu/gutil/map-util.h"
#include "kudu/gutil/stl_util.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/gutil/sysinfo.h"
@@ -34,6 +40,8 @@
namespace kudu {
+using std::shared_ptr;
+using std::unique_ptr;
using strings::Substitute;
////////////////////////////////////////////////////////
@@ -53,6 +61,22 @@ class FunctionRunnable : public Runnable {
};
////////////////////////////////////////////////////////
+// ClosureRunnable
+////////////////////////////////////////////////////////
+
+class ClosureRunnable : public Runnable {
+ public:
+ explicit ClosureRunnable(Closure cl) : cl_(std::move(cl)) {}
+
+ void Run() OVERRIDE {
+ cl_.Run();
+ }
+
+ private:
+ Closure cl_;
+};
+
+////////////////////////////////////////////////////////
// ThreadPoolBuilder
////////////////////////////////////////////////////////
@@ -98,6 +122,178 @@ Status ThreadPoolBuilder::Build(gscoped_ptr<ThreadPool>* pool) const {
}
////////////////////////////////////////////////////////
+// ThreadPoolToken
+////////////////////////////////////////////////////////
+
+ThreadPoolToken::ThreadPoolToken(ThreadPool* pool,
+ ThreadPool::ExecutionMode mode)
+ : mode_(mode),
+ pool_(pool),
+ state_(State::IDLE),
+ not_running_cond_(&pool->lock_),
+ active_threads_(0) {
+}
+
+ThreadPoolToken::~ThreadPoolToken() {
+ Shutdown();
+ pool_->ReleaseToken(this);
+}
+
+Status ThreadPoolToken::SubmitClosure(Closure c) {
+ return Submit(std::make_shared<ClosureRunnable>(std::move(c)));
+}
+
+Status ThreadPoolToken::SubmitFunc(boost::function<void()> f) {
+ return Submit(std::make_shared<FunctionRunnable>(std::move(f)));
+}
+
+Status ThreadPoolToken::Submit(shared_ptr<Runnable> r) {
+ return pool_->DoSubmit(std::move(r), this);
+}
+
+void ThreadPoolToken::Shutdown() {
+ MutexLock unique_lock(pool_->lock_);
+ pool_->CheckNotPoolThreadUnlocked();
+
+ // Clear the queue under the lock, but defer the releasing of the tasks
+ // outside the lock, in case there are concurrent threads wanting to access
+ // the ThreadPool. The task's destructors may acquire locks, etc, so this
+ // also prevents lock inversions.
+ deque<ThreadPool::Task> to_release = std::move(entries_);
+ pool_->total_queued_tasks_ -= to_release.size();
+
+ switch (state()) {
+ case State::IDLE:
+ // There were no tasks outstanding; we can quiesce the token immediately.
+ Transition(State::QUIESCED);
+ break;
+ case State::RUNNING:
+ // There were outstanding tasks. If any are still running, switch to
+ // QUIESCING and wait for them to finish (the worker thread executing
+ // the token's last task will switch the token to QUIESCED). Otherwise,
+ // we can quiesce the token immediately.
+
+ // Note: this is an O(n) operation, but it's expected to be infrequent.
+ // Plus doing it this way (rather than switching to QUIESCING and waiting
+ // for a worker thread to process the queue entry) helps retain state
+ // transition symmetry with ThreadPool::Shutdown.
+ for (auto it = pool_->queue_.begin(); it != pool_->queue_.end();) {
+ if (*it == this) {
+ it = pool_->queue_.erase(it);
+ } else {
+ it++;
+ }
+ }
+
+ if (active_threads_ == 0) {
+ Transition(State::QUIESCED);
+ break;
+ }
+ Transition(State::QUIESCING);
+ FALLTHROUGH_INTENDED;
+ case State::QUIESCING:
+ // The token is already quiescing. Just wait for a worker thread to
+ // switch it to QUIESCED.
+ while (state() != State::QUIESCED) {
+ not_running_cond_.Wait();
+ }
+ break;
+ default:
+ break;
+ }
+
+ // Finally release the queued tasks, outside the lock.
+ unique_lock.Unlock();
+ for (auto& t : to_release) {
+ if (t.trace) {
+ t.trace->Release();
+ }
+ }
+}
+
+void ThreadPoolToken::Wait() {
+ MutexLock unique_lock(pool_->lock_);
+ pool_->CheckNotPoolThreadUnlocked();
+ while (IsActive()) {
+ not_running_cond_.Wait();
+ }
+}
+
+bool ThreadPoolToken::WaitUntil(const MonoTime& until) {
+ return WaitFor(until - MonoTime::Now());
+}
+
+bool ThreadPoolToken::WaitFor(const MonoDelta& delta) {
+ MutexLock unique_lock(pool_->lock_);
+ pool_->CheckNotPoolThreadUnlocked();
+ while (IsActive()) {
+ if (!not_running_cond_.TimedWait(delta)) {
+ return false;
+ }
+ }
+ return true;
+}
+
+void ThreadPoolToken::Transition(State new_state) {
+#ifndef NDEBUG
+ CHECK_NE(state_, new_state);
+
+ switch (state_) {
+ case State::IDLE:
+ CHECK(new_state == State::RUNNING ||
+ new_state == State::QUIESCED);
+ if (new_state == State::RUNNING) {
+ CHECK(!entries_.empty());
+ } else {
+ CHECK(entries_.empty());
+ CHECK_EQ(active_threads_, 0);
+ }
+ break;
+ case State::RUNNING:
+ CHECK(new_state == State::IDLE ||
+ new_state == State::QUIESCING ||
+ new_state == State::QUIESCED);
+ CHECK(entries_.empty());
+ if (new_state == State::QUIESCING) {
+ CHECK_GT(active_threads_, 0);
+ }
+ break;
+ case State::QUIESCING:
+ CHECK(new_state == State::QUIESCED);
+ CHECK_EQ(active_threads_, 0);
+ break;
+ case State::QUIESCED:
+ CHECK(false); // QUIESCED is a terminal state
+ break;
+ default:
+ LOG(FATAL) << "Unknown token state: " << state_;
+ }
+#endif
+
+ // Take actions based on the state we're entering.
+ switch (new_state) {
+ case State::IDLE:
+ case State::QUIESCED:
+ not_running_cond_.Broadcast();
+ break;
+ default:
+ break;
+ }
+
+ state_ = new_state;
+}
+
+const char* ThreadPoolToken::StateToString(State s) {
+ switch (s) {
+ case State::IDLE: return "IDLE"; break;
+ case State::RUNNING: return "RUNNING"; break;
+ case State::QUIESCING: return "QUIESCING"; break;
+ case State::QUIESCED: return "QUIESCED"; break;
+ }
+ return "<cannot reach here>";
+}
+
+////////////////////////////////////////////////////////
// ThreadPool
////////////////////////////////////////////////////////
@@ -113,7 +309,8 @@ ThreadPool::ThreadPool(const ThreadPoolBuilder& builder)
not_empty_(&lock_),
num_threads_(0),
active_threads_(0),
- queue_size_(0) {
+ total_queued_tasks_(0),
+ tokenless_(NewToken(ExecutionMode::CONCURRENT)) {
string prefix = !builder.trace_metric_prefix_.empty() ?
builder.trace_metric_prefix_ : builder.name_;
@@ -127,6 +324,10 @@ ThreadPool::ThreadPool(const ThreadPoolBuilder& builder)
}
ThreadPool::~ThreadPool() {
+ // There should only be one live token: the one used in tokenless submission.
+ CHECK_EQ(1, tokens_.size()) << Substitute(
+ "Threadpool $0 destroyed with $1 allocated tokens",
+ name_, tokens_.size());
Shutdown();
}
@@ -150,43 +351,95 @@ void ThreadPool::Shutdown() {
MutexLock unique_lock(lock_);
CheckNotPoolThreadUnlocked();
+ // Note: this is the same error seen at submission if the pool is at
+ // capacity, so clients can't tell them apart. This isn't really a practical
+ // concern though because shutting down a pool typically requires clients to
+ // be quiesced first, so there's no danger of a client getting confused.
pool_status_ = Status::ServiceUnavailable("The pool has been shut down.");
- // Clear the queue_ member under the lock, but defer the releasing
- // of the entries outside the lock, in case there are concurrent threads
+ // Clear the various queues under the lock, but defer the releasing
+ // of the tasks outside the lock, in case there are concurrent threads
// wanting to access the ThreadPool. The task's destructors may acquire
// locks, etc, so this also prevents lock inversions.
- auto to_release = std::move(queue_);
queue_.clear();
- queue_size_ = 0;
- not_empty_.Broadcast();
+ deque<deque<Task>> to_release;
+ for (auto* t : tokens_) {
+ if (!t->entries_.empty()) {
+ to_release.emplace_back(std::move(t->entries_));
+ }
+ switch (t->state()) {
+ case ThreadPoolToken::State::IDLE:
+ // The token is idle; we can quiesce it immediately.
+ t->Transition(ThreadPoolToken::State::QUIESCED);
+ break;
+ case ThreadPoolToken::State::RUNNING:
+ // The token has tasks associated with it. If they're merely queued
+ // (i.e. there are no active threads), the tasks will have been removed
+ // above and we can quiesce immediately. Otherwise, we need to wait for
+ // the threads to finish.
+ t->Transition(t->active_threads_ > 0 ?
+ ThreadPoolToken::State::QUIESCING :
+ ThreadPoolToken::State::QUIESCED);
+ break;
+ default:
+ break;
+ }
+ }
- // The Runnable doesn't have Abort() so we must wait
- // and hopefully the abort is done outside before calling Shutdown().
+ // The queues are empty. Wake any sleeping worker threads and wait for all
+ // of them to exit. Some worker threads will exit immediately upon waking,
+ // while others will exit after they finish executing an outstanding task.
+ total_queued_tasks_ = 0;
+ not_empty_.Broadcast();
while (num_threads_ > 0) {
no_threads_cond_.Wait();
}
- // Finally release the tasks that were in the queue, outside the lock.
+ // All the threads have exited. Check the state of each token.
+ for (auto* t : tokens_) {
+ DCHECK(t->state() == ThreadPoolToken::State::IDLE ||
+ t->state() == ThreadPoolToken::State::QUIESCED);
+ }
+
+ // Finally release the queued tasks, outside the lock.
unique_lock.Unlock();
- for (QueueEntry& e : to_release) {
- if (e.trace) {
- e.trace->Release();
+ for (auto& token : to_release) {
+ for (auto& t : token) {
+ if (t.trace) {
+ t.trace->Release();
+ }
}
}
}
-Status ThreadPool::SubmitClosure(const Closure& task) {
- // TODO: once all uses of boost::bind-based tasks are dead, implement this
- // in a more straight-forward fashion.
- return SubmitFunc(boost::bind(&Closure::Run, task));
+unique_ptr<ThreadPoolToken> ThreadPool::NewToken(ExecutionMode mode) {
+ MutexLock guard(lock_);
+ unique_ptr<ThreadPoolToken> t(new ThreadPoolToken(this, mode));
+ InsertOrDie(&tokens_, t.get());
+ return t;
+}
+
+void ThreadPool::ReleaseToken(ThreadPoolToken* t) {
+ MutexLock guard(lock_);
+ CHECK(!t->IsActive()) << Substitute("Token with state $0 may not be released",
+ ThreadPoolToken::StateToString(t->state()));
+ CHECK_EQ(1, tokens_.erase(t));
+}
+
+Status ThreadPool::SubmitClosure(Closure c) {
+ return Submit(std::make_shared<ClosureRunnable>(std::move(c)));
}
-Status ThreadPool::SubmitFunc(boost::function<void()> func) {
- return Submit(std::shared_ptr<Runnable>(new FunctionRunnable(std::move(func))));
+Status ThreadPool::SubmitFunc(boost::function<void()> f) {
+ return Submit(std::make_shared<FunctionRunnable>(std::move(f)));
}
-Status ThreadPool::Submit(std::shared_ptr<Runnable> task) {
+Status ThreadPool::Submit(shared_ptr<Runnable> r) {
+ return DoSubmit(std::move(r), tokenless_.get());
+}
+
+Status ThreadPool::DoSubmit(shared_ptr<Runnable> r, ThreadPoolToken* token) {
+ DCHECK(token);
MonoTime submit_time = MonoTime::Now();
MutexLock guard(lock_);
@@ -194,13 +447,17 @@ Status ThreadPool::Submit(std::shared_ptr<Runnable> task) {
return pool_status_;
}
+ if (PREDICT_FALSE(!token->MaySubmitNewTasks())) {
+ return Status::ServiceUnavailable("Thread pool token was shut down");
+ }
+
// Size limit check.
int64_t capacity_remaining = static_cast<int64_t>(max_threads_) - active_threads_ +
- static_cast<int64_t>(max_queue_size_) - queue_size_;
+ static_cast<int64_t>(max_queue_size_) - total_queued_tasks_;
if (capacity_remaining < 1) {
return Status::ServiceUnavailable(
Substitute("Thread pool is at capacity ($0/$1 tasks running, $2/$3 tasks queued)",
- num_threads_, max_threads_, queue_size_, max_queue_size_));
+ num_threads_, max_threads_, total_queued_tasks_, max_queue_size_));
}
// Should we create another thread?
@@ -212,8 +469,10 @@ Status ThreadPool::Submit(std::shared_ptr<Runnable> task) {
// It's also harmless.
//
// Of course, we never create more than max_threads_ threads no matter what.
+ int threads_from_this_submit =
+ token->IsActive() && token->mode() == ExecutionMode::SERIAL ? 0 : 1;
int inactive_threads = num_threads_ - active_threads_;
- int additional_threads = (queue_size_ + 1) - inactive_threads;
+ int additional_threads = (queue_.size() + threads_from_this_submit) - inactive_threads;
if (additional_threads > 0 && num_threads_ < max_threads_) {
Status status = CreateThreadUnlocked();
if (!status.ok()) {
@@ -228,18 +487,29 @@ Status ThreadPool::Submit(std::shared_ptr<Runnable> task) {
}
}
- QueueEntry e;
- e.runnable = std::move(task);
- e.trace = Trace::CurrentTrace();
+ Task task;
+ task.runnable = std::move(r);
+ task.trace = Trace::CurrentTrace();
// Need to AddRef, since the thread which submitted the task may go away,
// and we don't want the trace to be destructed while waiting in the queue.
- if (e.trace) {
- e.trace->AddRef();
+ if (task.trace) {
+ task.trace->AddRef();
}
- e.submit_time = submit_time;
-
- queue_.emplace_back(std::move(e));
- int length_at_submit = queue_size_++;
+ task.submit_time = submit_time;
+
+ // Add the task to the token's queue.
+ ThreadPoolToken::State state = token->state();
+ DCHECK(state == ThreadPoolToken::State::IDLE ||
+ state == ThreadPoolToken::State::RUNNING);
+ token->entries_.emplace_back(std::move(task));
+ if (state == ThreadPoolToken::State::IDLE ||
+ token->mode() == ExecutionMode::CONCURRENT) {
+ queue_.emplace_back(token);
+ if (state == ThreadPoolToken::State::IDLE) {
+ token->Transition(ThreadPoolToken::State::RUNNING);
+ }
+ }
+ int length_at_submit = total_queued_tasks_++;
guard.Unlock();
not_empty_.Signal();
@@ -254,7 +524,7 @@ Status ThreadPool::Submit(std::shared_ptr<Runnable> task) {
void ThreadPool::Wait() {
MutexLock unique_lock(lock_);
CheckNotPoolThreadUnlocked();
- while ((!queue_.empty()) || (active_threads_ > 0)) {
+ while (total_queued_tasks_ > 0 || active_threads_ > 0) {
idle_cond_.Wait();
}
}
@@ -266,7 +536,7 @@ bool ThreadPool::WaitUntil(const MonoTime& until) {
bool ThreadPool::WaitFor(const MonoDelta& delta) {
MutexLock unique_lock(lock_);
CheckNotPoolThreadUnlocked();
- while ((!queue_.empty()) || (active_threads_ > 0)) {
+ while (total_queued_tasks_ > 0 || active_threads_ > 0) {
if (!idle_cond_.TimedWait(delta)) {
return false;
}
@@ -274,7 +544,6 @@ bool ThreadPool::WaitFor(const MonoDelta& delta) {
return true;
}
-
void ThreadPool::SetQueueLengthHistogram(const scoped_refptr<Histogram>& hist) {
queue_length_histogram_ = hist;
}
@@ -317,23 +586,28 @@ void ThreadPool::DispatchThread(bool permanent) {
continue;
}
- // Fetch a pending task
- QueueEntry entry = std::move(queue_.front());
+ // Get the next token and task to execute.
+ ThreadPoolToken* token = queue_.front();
queue_.pop_front();
- queue_size_--;
+ DCHECK_EQ(ThreadPoolToken::State::RUNNING, token->state());
+ DCHECK(!token->entries_.empty());
+ Task task = std::move(token->entries_.front());
+ token->entries_.pop_front();
+ token->active_threads_++;
+ --total_queued_tasks_;
++active_threads_;
unique_lock.Unlock();
// Release the reference which was held by the queued item.
- ADOPT_TRACE(entry.trace);
- if (entry.trace) {
- entry.trace->Release();
+ ADOPT_TRACE(task.trace);
+ if (task.trace) {
+ task.trace->Release();
}
// Update metrics
MonoTime now(MonoTime::Now());
- int64_t queue_time_us = (now - entry.submit_time).ToMicroseconds();
+ int64_t queue_time_us = (now - task.submit_time).ToMicroseconds();
TRACE_COUNTER_INCREMENT(queue_time_trace_metric_name_, queue_time_us);
if (queue_time_us_histogram_) {
queue_time_us_histogram_->Increment(queue_time_us);
@@ -344,7 +618,7 @@ void ThreadPool::DispatchThread(bool permanent) {
MicrosecondsInt64 start_wall_us = GetMonoTimeMicros();
MicrosecondsInt64 start_cpu_us = GetThreadCpuTimeMicros();
- entry.runnable->Run();
+ task.runnable->Run();
int64_t wall_us = GetMonoTimeMicros() - start_wall_us;
int64_t cpu_us = GetThreadCpuTimeMicros() - start_cpu_us;
@@ -361,9 +635,26 @@ void ThreadPool::DispatchThread(bool permanent) {
// objects, and we don't want to block submission of the threadpool.
// In the worst case, the destructor might even try to do something
// with this threadpool, and produce a deadlock.
- entry.runnable.reset();
+ task.runnable.reset();
unique_lock.Lock();
+ // Possible states:
+ // 1. The token was shut down while we ran its task. Transition to QUIESCED.
+ // 2. The token has no more queued tasks. Transition back to IDLE.
+ // 3. The token has more tasks. Requeue it and transition back to RUNNABLE.
+ ThreadPoolToken::State state = token->state();
+ DCHECK(state == ThreadPoolToken::State::RUNNING ||
+ state == ThreadPoolToken::State::QUIESCING);
+ if (--token->active_threads_ == 0) {
+ if (state == ThreadPoolToken::State::QUIESCING) {
+ DCHECK(token->entries_.empty());
+ token->Transition(ThreadPoolToken::State::QUIESCED);
+ } else if (token->entries_.empty()) {
+ token->Transition(ThreadPoolToken::State::IDLE);
+ } else if (token->mode() == ExecutionMode::SERIAL) {
+ queue_.emplace_back(token);
+ }
+ }
if (--active_threads_ == 0) {
idle_cond_.Broadcast();
}
@@ -381,7 +672,7 @@ void ThreadPool::DispatchThread(bool permanent) {
// Sanity check: if we're the last thread exiting, the queue ought to be
// empty. Otherwise it will never get processed.
CHECK(queue_.empty());
- DCHECK_EQ(0, queue_size_);
+ DCHECK_EQ(0, total_queued_tasks_);
}
}
@@ -407,4 +698,8 @@ void ThreadPool::CheckNotPoolThreadUnlocked() {
}
}
+std::ostream& operator<<(std::ostream& o, ThreadPoolToken::State s) {
+ return o << ThreadPoolToken::StateToString(s);
+}
+
} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/428a34ba/src/kudu/util/threadpool.h
----------------------------------------------------------------------
diff --git a/src/kudu/util/threadpool.h b/src/kudu/util/threadpool.h
index 7ac6d17..e15c991 100644
--- a/src/kudu/util/threadpool.h
+++ b/src/kudu/util/threadpool.h
@@ -17,13 +17,14 @@
#ifndef KUDU_UTIL_THREAD_POOL_H
#define KUDU_UTIL_THREAD_POOL_H
-#include <boost/function.hpp>
-#include <gtest/gtest_prod.h>
-#include <list>
+#include <deque>
+#include <iosfwd>
#include <memory>
-#include <unordered_set>
#include <string>
-#include <vector>
+#include <unordered_set>
+
+#include <boost/function.hpp>
+#include <gtest/gtest_prod.h>
#include "kudu/gutil/callback_forward.h"
#include "kudu/gutil/gscoped_ptr.h"
@@ -40,6 +41,7 @@ namespace kudu {
class Histogram;
class Thread;
class ThreadPool;
+class ThreadPoolToken;
class Trace;
class Runnable {
@@ -124,8 +126,25 @@ class ThreadPoolBuilder {
};
// Thread pool with a variable number of threads.
-// The pool can execute a class that implements the Runnable interface, or a
-// boost::function, which can be obtained via boost::bind().
+//
+// Tasks submitted directly to the thread pool enter a FIFO queue and are
+// dispatched to a worker thread when one becomes free. Tasks may also be
+// submitted via ThreadPoolTokens. The token Wait() and Shutdown() functions
+// can then be used to block on logical groups of tasks.
+//
+// A token operates in one of two ExecutionModes, determined at token
+// construction time:
+// 1. SERIAL: submitted tasks are run one at a time.
+// 2. CONCURRENT: submitted tasks may be run in parallel. This isn't unlike
+// tasks submitted without a token, but the logical grouping that tokens
+// impart can be useful when a pool is shared by many contexts (e.g. to
+// safely shut down one context, to derive context-specific metrics, etc.).
+//
+// Tasks submitted without a token or via ExecutionMode::CONCURRENT tokens are
+// processed in FIFO order. On the other hand, ExecutionMode::SERIAL tokens are
+// processed in a round-robin fashion, one task at a time. This prevents them
+// from starving one another. However, tokenless (and CONCURRENT token-based)
+// tasks can starve SERIAL token-based tasks.
//
// Usage Example:
// static void Func(int n) { ... }
@@ -140,7 +159,7 @@ class ThreadPoolBuilder {
// .set_idle_timeout(MonoDelta::FromMilliseconds(2000))
// .Build(&thread_pool));
// thread_pool->Submit(shared_ptr<Runnable>(new Task()));
-// thread_pool->Submit(boost::bind(&Func, 10));
+// thread_pool->SubmitFunc(boost::bind(&Func, 10));
class ThreadPool {
public:
~ThreadPool();
@@ -153,16 +172,16 @@ class ThreadPool {
// require an explicit "abort" notification to exit from the run loop.
void Shutdown();
- // Submit a function using the kudu Closure system.
- Status SubmitClosure(const Closure& task) WARN_UNUSED_RESULT;
+ // Submits a function using the kudu Closure system.
+ Status SubmitClosure(Closure c) WARN_UNUSED_RESULT;
- // Submit a function binded using boost::bind(&FuncName, args...)
- Status SubmitFunc(boost::function<void()> func) WARN_UNUSED_RESULT;
+ // Submits a function bound using boost::bind(&FuncName, args...).
+ Status SubmitFunc(boost::function<void()> f) WARN_UNUSED_RESULT;
- // Submit a Runnable class
- Status Submit(std::shared_ptr<Runnable> task) WARN_UNUSED_RESULT;
+ // Submits a Runnable class.
+ Status Submit(std::shared_ptr<Runnable> r) WARN_UNUSED_RESULT;
- // Wait until all the tasks are completed.
+ // Waits until all the tasks are completed.
void Wait();
// Waits for the pool to reach the idle state, or until 'until' time is reached.
@@ -173,52 +192,66 @@ class ThreadPool {
// Returns true if the pool reached the idle state, false otherwise.
bool WaitFor(const MonoDelta& delta);
- // Return the current number of tasks waiting in the queue.
- // Typically used for metrics.
- int queue_length() const {
- return ANNOTATE_UNPROTECTED_READ(queue_size_);
- }
+ // Allocates a new token for use in token-based task submission. All tokens
+ // must be destroyed before their ThreadPool is destroyed.
+ //
+ // There is no limit on the number of tokens that may be allocated.
+ enum class ExecutionMode {
+ // Tasks submitted via this token will be executed serially.
+ SERIAL,
- // Attach a histogram which measures the queue length seen by tasks when they enter
+ // Tasks submitted via this token may be executed concurrently.
+ CONCURRENT,
+ };
+ std::unique_ptr<ThreadPoolToken> NewToken(ExecutionMode mode);
+
+ // Attaches a histogram which measures the queue length seen by tasks when they enter
// the thread pool's queue.
void SetQueueLengthHistogram(const scoped_refptr<Histogram>& hist);
- // Attach a histogram which measures the amount of time that tasks spend waiting in
+ // Attaches a histogram which measures the amount of time that tasks spend waiting in
// the queue.
void SetQueueTimeMicrosHistogram(const scoped_refptr<Histogram>& hist);
- // Attach a histogram which measures the amount of time that tasks spend running.
+ // Attaches a histogram which measures the amount of time that tasks spend running.
void SetRunTimeMicrosHistogram(const scoped_refptr<Histogram>& hist);
private:
+ FRIEND_TEST(ThreadPoolTest, TestThreadPoolWithNoMinimum);
+ FRIEND_TEST(ThreadPoolTest, TestVariableSizeThreadPool);
+
friend class ThreadPoolBuilder;
+ friend class ThreadPoolToken;
+
+ // Client-provided task to be executed by this pool.
+ struct Task {
+ std::shared_ptr<Runnable> runnable;
+ Trace* trace;
+
+ // Time at which the entry was submitted to the pool.
+ MonoTime submit_time;
+ };
- // Create a new thread pool using a builder.
+ // Creates a new thread pool using a builder.
explicit ThreadPool(const ThreadPoolBuilder& builder);
- // Initialize the thread pool by starting the minimum number of threads.
+ // Initializes the thread pool by starting the minimum number of threads.
Status Init();
// Dispatcher responsible for dequeueing and executing the tasks
void DispatchThread(bool permanent);
- // Create new thread. Required that lock_ is held.
+ // Creates new thread. Required that lock_ is held.
Status CreateThreadUnlocked();
// Aborts if the current thread is a member of this thread pool.
void CheckNotPoolThreadUnlocked();
- private:
- FRIEND_TEST(ThreadPoolTest, TestThreadPoolWithNoMinimum);
- FRIEND_TEST(ThreadPoolTest, TestVariableSizeThreadPool);
-
- struct QueueEntry {
- std::shared_ptr<Runnable> runnable;
- Trace* trace;
+ // Submits a task to be run via token.
+ Status DoSubmit(std::shared_ptr<Runnable> r, ThreadPoolToken* token);
- // Time at which the entry was submitted to the pool.
- MonoTime submit_time;
- };
+ // Releases token 't' and invalidates it.
+ void ReleaseToken(ThreadPoolToken* t);
const std::string name_;
const int min_threads_;
@@ -226,15 +259,53 @@ class ThreadPool {
const int max_queue_size_;
const MonoDelta idle_timeout_;
+ // Overall status of the pool. Set to an error when the pool is shut down.
+ //
+ // Protected by 'lock_'.
Status pool_status_;
+
+ // Synchronizes many of the members of the pool and all of its
+ // condition variables.
Mutex lock_;
+
+ // Condition variable for "pool is idling". Waiters wake up when
+ // active_threads_ reaches zero.
ConditionVariable idle_cond_;
+
+ // Condition variable for "pool has no threads". Waiters wake up when
+ // num_threads_ reaches zero.
ConditionVariable no_threads_cond_;
+
+ // Condition variable for "queue is not empty". Waiters wake up when
+ // a new task is queued.
ConditionVariable not_empty_;
+
+ // Number of threads currently running.
+ //
+ // Protected by lock_.
int num_threads_;
+
+ // Number of threads currently running and executing client tasks.
+ //
+ // Protected by lock_.
int active_threads_;
- int queue_size_;
- std::list<QueueEntry> queue_;
+
+ // Total number of client tasks queued, either directly (queue_) or
+ // indirectly (tokens_).
+ //
+ // Protected by lock_.
+ int total_queued_tasks_;
+
+ // All allocated tokens.
+ //
+ // Protected by lock_.
+ std::unordered_set<ThreadPoolToken*> tokens_;
+
+ // FIFO of tokens from which tasks should be executed. Does not own the
+ // tokens; they are owned by clients and are removed from the FIFO on shutdown.
+ //
+ // Protected by lock_.
+ std::deque<ThreadPoolToken*> queue_;
// Pointers to all running threads. Raw pointers are safe because a Thread
// may only go out of scope after being removed from threads_.
@@ -242,6 +313,9 @@ class ThreadPool {
// Protected by lock_.
std::unordered_set<Thread*> threads_;
+ // ExecutionMode::CONCURRENT token used by the pool for tokenless submission.
+ std::unique_ptr<ThreadPoolToken> tokenless_;
+
scoped_refptr<Histogram> queue_length_histogram_;
scoped_refptr<Histogram> queue_time_us_histogram_;
scoped_refptr<Histogram> run_time_us_histogram_;
@@ -253,5 +327,129 @@ class ThreadPool {
DISALLOW_COPY_AND_ASSIGN(ThreadPool);
};
+// Entry point for token-based task submission and blocking for a particular
+// thread pool. Tokens can only be created via ThreadPool::NewToken().
+//
+// All functions are thread-safe. Mutable members are protected via the
+// ThreadPool's lock.
+class ThreadPoolToken {
+ public:
+ // Destroys the token.
+ //
+ // May be called on a token with outstanding tasks, as Shutdown() will be
+ // called first to take care of them.
+ ~ThreadPoolToken();
+
+ // Submits a function using the kudu Closure system.
+ Status SubmitClosure(Closure c) WARN_UNUSED_RESULT;
+
+ // Submits a function bound using boost::bind(&FuncName, args...).
+ Status SubmitFunc(boost::function<void()> f) WARN_UNUSED_RESULT;
+
+ // Submits a Runnable class.
+ Status Submit(std::shared_ptr<Runnable> r) WARN_UNUSED_RESULT;
+
+ // Marks the token as unusable for future submissions. Any queued tasks not
+ // yet running are destroyed. If tasks are in flight, Shutdown() will wait
+ // on their completion before returning.
+ void Shutdown();
+
+ // Waits until all the tasks submitted via this token are completed.
+ void Wait();
+
+ // Waits for all submissions using this token are complete, or until 'until'
+ // time is reached.
+ //
+ // Returns true if all submissions are complete, false otherwise.
+ bool WaitUntil(const MonoTime& until);
+
+ // Waits for all submissions using this token are complete, or until 'delta'
+ // time elapses.
+ //
+ // Returns true if all submissions are complete, false otherwise.
+ bool WaitFor(const MonoDelta& delta);
+
+ private:
+ // All possible token states. Legal state transitions:
+ // IDLE -> RUNNING: task is submitted via token
+ // IDLE -> QUIESCED: token or pool is shut down
+ // RUNNING -> IDLE: worker thread finishes executing a task and
+ // there are no more tasks queued to the token
+ // RUNNING -> QUIESCING: token or pool is shut down while worker thread
+ // is executing a task
+ // RUNNING -> QUIESCED: token or pool is shut down
+ // QUIESCING -> QUIESCED: worker thread finishes executing a task
+ // belonging to a shut down token or pool
+ enum class State {
+ // Token has no queued tasks.
+ IDLE,
+
+ // A worker thread is running one of the token's previously queued tasks.
+ RUNNING,
+
+ // No new tasks may be submitted to the token. A worker thread is still
+ // running a previously queued task.
+ QUIESCING,
+
+ // No new tasks may be submitted to the token. There are no active tasks
+ // either. At this state, the token may only be destroyed.
+ QUIESCED,
+ };
+
+ // Writes a textual representation of the token state in 's' to 'o'.
+ friend std::ostream& operator<<(std::ostream& o, ThreadPoolToken::State s);
+
+ friend class ThreadPool;
+
+ // Returns a textual representation of 's' suitable for debugging.
+ static const char* StateToString(State s);
+
+ // Constructs a new token.
+ //
+ // The token may not outlive its thread pool ('pool').
+ ThreadPoolToken(ThreadPool* pool, ThreadPool::ExecutionMode mode);
+
+ // Changes this token's state to 'new_state' taking actions as needed.
+ void Transition(State new_state);
+
+ // Returns true if this token has a task queued and ready to run, or if a
+ // task belonging to this token is already running.
+ bool IsActive() const {
+ return state_ == State::RUNNING ||
+ state_ == State::QUIESCING;
+ }
+
+ // Returns true if new tasks may be submitted to this token.
+ bool MaySubmitNewTasks() const {
+ return state_ != State::QUIESCING &&
+ state_ != State::QUIESCED;
+ }
+
+ State state() const { return state_; }
+ ThreadPool::ExecutionMode mode() const { return mode_; }
+
+ // Token's configured execution mode.
+ const ThreadPool::ExecutionMode mode_;
+
+ // Pointer to the token's thread pool.
+ ThreadPool* pool_;
+
+ // Token state machine.
+ State state_;
+
+ // Queued client tasks.
+ std::deque<ThreadPool::Task> entries_;
+
+ // Condition variable for "token is idle". Waiters wake up when the token
+ // transitions to IDLE or QUIESCED.
+ ConditionVariable not_running_cond_;
+
+ // Number of worker threads currently executing tasks belonging to this
+ // token.
+ int active_threads_;
+
+ DISALLOW_COPY_AND_ASSIGN(ThreadPoolToken);
+};
+
} // namespace kudu
#endif