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:42 UTC
[2/5] kudu git commit: KUDU-1863: improve overall safety of graceful
server shutdown
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.";
}
}