[impala] 02/06: IMPALA-7565: Set TAcceptQueueServer connection_setup_pool to be multi-threaded by default

[ta...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

tarmstrong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git

commit 26b8b427188052735ce52f319aebe2ff1ac17e2a
Author: Bikramjeet Vig <bi...@cloudera.com>
AuthorDate: Wed Dec 19 15:41:25 2018 -0800

    IMPALA-7565: Set TAcceptQueueServer connection_setup_pool to be
    multi-threaded by default
    
    Bumping up the thread count for the threads that process the
    post-accept, pre-setup connection queue to 2 in order to minimize the
    chances of a single client holding up others if the thread gets stuck
    in that step. This patch also un-hides the advanced startup flag
    'accepted_cnxn_setup_thread_pool_size'.
    
    Testing:
    - Ran exhaustive tests with a thread pool of 10.
    - Scanned manually through the code and parts of thrift lib to make
      sure the APIs are used in a thread safe manner.
    - Rapidly executed openssl connect-disconnect on the impalad's
      hs2 server port on a thread sanitizer build. No data races were
      flagged by the thread sanitizer.
    - Ran a stress test on an 8 node secure cluster with 100 concurrent
      streams executing the tpch workload with a scale of 5g and a
      connection_setup_pool of 100.
    
    Change-Id: I053120d4c3153ddbe5261acd28388be6cd191908
    Reviewed-on: http://gerrit.cloudera.org:8080/12249
    Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
    Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
 be/src/rpc/TAcceptQueueServer.cpp         | 13 ++++---------
 be/src/transport/TSaslServerTransport.cpp |  1 +
 2 files changed, 5 insertions(+), 9 deletions(-)

diff --git a/be/src/rpc/TAcceptQueueServer.cpp b/be/src/rpc/TAcceptQueueServer.cpp
index 730901b..8f4a01a 100644
--- a/be/src/rpc/TAcceptQueueServer.cpp
+++ b/be/src/rpc/TAcceptQueueServer.cpp
@@ -29,11 +29,10 @@ DEFINE_int32(accepted_cnxn_queue_depth, 10000,
     "(Advanced) The size of the post-accept, pre-setup connection queue in each thrift "
     "server set up to service Impala internal and external connections.");
 
-DEFINE_int32_hidden(accepted_cnxn_setup_thread_pool_size, 1,
+DEFINE_int32(accepted_cnxn_setup_thread_pool_size, 2,
     "(Advanced) The size of the thread pool that is used to process the "
     "post-accept, pre-setup connection queue in each thrift server set up to service "
-    "Impala internal and external connections. Warning: This is untested for values "
-    "greater than 1 which might exhibit unpredictable behavior and/or cause crashes.");
+    "Impala internal and external connections.");
 
 namespace apache {
 namespace thrift {
@@ -216,14 +215,10 @@ void TAcceptQueueServer::serve() {
   }
 
   if (FLAGS_accepted_cnxn_setup_thread_pool_size > 1) {
-    LOG(WARNING) << "connection_setup_thread_pool_size is set to "
-                 << FLAGS_accepted_cnxn_setup_thread_pool_size
-                 << ". Values greater than 1 are untested and might exhibit "
-                    "unpredictable behavior and/or cause crashes.";
+    LOG(INFO) << "connection_setup_thread_pool_size is set to "
+              << FLAGS_accepted_cnxn_setup_thread_pool_size;
   }
   // New - this is the thread pool used to process the internal accept queue.
-  // TODO: IMPALA-7565: Make sure the related thrift code is thread safe and subsequently
-  // enable multi-threading by default.
   ThreadPool<shared_ptr<TTransport>> connection_setup_pool("setup-server", "setup-worker",
       FLAGS_accepted_cnxn_setup_thread_pool_size, FLAGS_accepted_cnxn_queue_depth,
       [this](int tid, const shared_ptr<TTransport>& item) {
diff --git a/be/src/transport/TSaslServerTransport.cpp b/be/src/transport/TSaslServerTransport.cpp
index 15d548e..9bd7795 100644
--- a/be/src/transport/TSaslServerTransport.cpp
+++ b/be/src/transport/TSaslServerTransport.cpp
@@ -177,6 +177,7 @@ boost::shared_ptr<TTransport> TSaslServerTransport::Factory::getTransport(
   socket->setSendTimeout(0);
   {
     lock_guard<mutex> l(transportMap_mutex_);
+    DCHECK(transportMap_.find(trans) == transportMap_.end());
     transportMap_[trans] = ret_transport;
   }
   return ret_transport;