KUDU-3633 shutdown DnsResolver in ServerBase::ShutdownImpl()

The thread pool of the DNS resolver should be shut down along with the
messenger in ServerBase to prevent retrying of RPCs that failed as a
collateral of the shutdown process in progress.  Those RPCs might be
retried by invoking rpc::Proxy::RefreshDnsAndEnqueueRequest(), etc.

On the related note, I also added a guard to protect ThreadPool::tokens_
in the destructor of the ThreadPool class, as elsewhere.  I also snuck
in an update to call DCHECK() in a loop only when DCHECK_IS_ON()
macro evaluates to 'true'.

This addresses flakiness reported at least in one of the RemoteKsckTest
scenarios (e.g., TestFilterOnNotabletTable in [1]).  One of the related
TSAN reports looked like below:

RemoteKsckTest.TestFilterOnNotabletTable: WARNING: ThreadSanitizer: data race
  Read of size 8 at 0x7b54001e5118 by main thread:
    #0 std::__1::__hash_table<kudu::ThreadPoolToken*, ...>::size() const
    #1 std::__1::unordered_set<kudu::ThreadPoolToken*, ...>::size() const
    #2 kudu::ThreadPool::~ThreadPool()
    ...
    #6 kudu::kserver::KuduServer::~KuduServer()
    #7 kudu::tserver::TabletServer::~TabletServer()
    ...

  Previous write of size 8 at 0x7b54001e5118 by thread T262 ...:
    #0 std::__1::__hash_table<kudu::ThreadPoolToken*, ...>::remove(...)
    ...
    #4 kudu::ThreadPool::ReleaseToken(...)
    #5 kudu::ThreadPoolToken::~ThreadPoolToken()
    ...
    #24 kudu::consensus::LeaderElection::~LeaderElection()
    ...
    #35 kudu::rpc::Proxy::RefreshDnsAndEnqueueRequest(...)
    ...
    #41 kudu::DnsResolver::RefreshAddressesAsync()
    ...

  Thread T262 'dns-resolver [w' (tid=29102, running) created by thread T182 at:
    #0 pthread_create
    #1 kudu::Thread::StartThread(...)
    #2 kudu::Thread::Create(...)
    #3 kudu::ThreadPool::CreateThread()
    #4 kudu::ThreadPool::DoSubmit(..., kudu::ThreadPoolToken*)
    #5 kudu::ThreadPool::Submit(...)
    #6 kudu::DnsResolver::RefreshAddressesAsync(..)
    #7 kudu::rpc::Proxy::RefreshDnsAndEnqueueRequest(...)
    #8 kudu::rpc::Proxy::AsyncRequest(...)
    ...
    #15 kudu::rpc::OutboundCall::CallCallback()
    #16 kudu::rpc::OutboundCall::SetFailed()
    #17 kudu::rpc::Connection::Shutdown()
    #18 kudu::rpc::ReactorThread::ShutdownInternal()
    ...
    #25 kudu::rpc::ReactorThread::RunThread()
    ...

[1] http://dist-test.cloudera.org:8080/test_drilldown?test_name=ksck_remote-test

Change-Id: I525f1078a349dbd2926938bb4fcc3e80888dfbb4
Reviewed-on: http://gerrit.cloudera.org:8080/22434
Tested-by: Alexey Serbin <alexey@apache.org>
Reviewed-by: Abhishek Chennaka <achennaka@cloudera.com>
(cherry picked from commit fc40fcd)
Reviewed-on: http://gerrit.cloudera.org:8080/22449

Author: alexeyserbin

src/kudu/server/server_base.cc

@@ -1222,6 +1222,7 @@ void ServerBase::ShutdownImpl() {
     web_server_->Stop();
   }
   rpc_server_->Shutdown();
+  dns_resolver_->Shutdown();
   if (messenger_) {
     messenger_->Shutdown();
   }

src/kudu/util/net/dns_resolver.cc

@@ -67,6 +67,10 @@ DnsResolver::DnsResolver(int max_threads_num,
 }
 
 DnsResolver::~DnsResolver() {
+  Shutdown();
+}
+
+void DnsResolver::Shutdown() {
   pool_->Shutdown();
 }
 
@@ -87,7 +91,7 @@ void DnsResolver::ResolveAddressesAsync(const HostPort& hostport,
   const auto s = pool_->Submit([=]() {
     this->DoResolutionCb(hostport, addresses, cb);
   });
-  if (!s.ok()) {
+  if (PREDICT_FALSE(!s.ok())) {
     cb(s);
   }
 }
@@ -107,7 +111,7 @@ void DnsResolver::RefreshAddressesAsync(const HostPort& hostport,
     }
     this->DoResolutionCb(hostport, addresses, cb);
   });
-  if (!s.ok()) {
+  if (PREDICT_FALSE(!s.ok())) {
     cb(s);
   }
 }

src/kudu/util/net/dns_resolver.h

@@ -52,6 +52,14 @@ class DnsResolver {
                        MonoDelta cache_ttl = MonoDelta::FromSeconds(60));
   ~DnsResolver();
 
+  // Shutdown the resolver's thread pool. After shutting down the thread pool,
+  // RefreshAddressesAsync() returns Status::ServiceUnavailable(),
+  // while ResolveAddressesAsync() returns cached not-yet-expired entry
+  // from the cache (if present) or Status::ServiceUnavailable() otherwise.
+  // The behavior of ResolveAddresses() isn't affected by the status of the
+  // resolver's thread pool.
+  void Shutdown();
+
   // Synchronously resolve addresses corresponding to the specified host:port
   // pair in 'hostport'. Note that a host may resolve to more than one IP
   // address.

src/kudu/util/threadpool.cc

@@ -387,10 +387,15 @@ 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());
+#if DCHECK_IS_ON()
+  {
+    // There should only be one live token: the one used in tokenless submission.
+    std::lock_guard guard(lock_);
+    DCHECK_EQ(1, tokens_.size()) << Substitute(
+        "Threadpool $0 destroyed with $1 allocated tokens",
+        name_, tokens_.size());
+  }
+#endif
   Shutdown();
 }
 
@@ -472,11 +477,13 @@ void ThreadPool::Shutdown() {
     no_threads_cond_.Wait();
   }
 
+#if DCHECK_IS_ON()
   // 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);
   }
+#endif
 
   // Finally release the queued tasks, outside the lock.
   lock.unlock();