hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ecl...@apache.org
Subject [20/50] [abbrv] hbase git commit: HBASE-15770 Stop using wangle's global executor
Date Fri, 13 May 2016 21:37:14 GMT
HBASE-15770 Stop using wangle's global executor

Summary: Connection pool and connection factory now get thread pools through their constructor.
This means that the client has the whole control over the threads.

Test Plan: simple-client still writes.

Differential Revision: https://reviews.facebook.net/D57801


Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/1e69edac
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/1e69edac
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/1e69edac

Branch: refs/heads/HBASE-14850
Commit: 1e69edacafb6a4db63da55765097308b1fa42e15
Parents: a0c2244
Author: Elliott Clark <eclark@apache.org>
Authored: Fri May 6 14:32:16 2016 -0700
Committer: Elliott Clark <eclark@apache.org>
Committed: Fri May 13 14:36:28 2016 -0700

----------------------------------------------------------------------
 .../connection/client-dispatcher.h              |  3 ++-
 .../connection/client-handler.cc                |  2 +-
 .../connection/connection-factory.cc            |  8 +++---
 .../connection/connection-factory.h             |  2 +-
 .../connection/connection-pool-test.cc          |  1 +
 .../connection/connection-pool.cc               |  7 ++---
 .../connection/connection-pool.h                |  2 +-
 hbase-native-client/core/client.cc              | 23 +++++++++++-----
 hbase-native-client/core/client.h               |  9 +++++--
 hbase-native-client/core/location-cache-test.cc |  5 ++--
 hbase-native-client/core/location-cache.cc      | 28 +++++++++++---------
 hbase-native-client/core/location-cache.h       |  8 ++++--
 hbase-native-client/core/simple-client.cc       |  8 +++---
 hbase-native-client/serde/server-name-test.cc   |  1 -
 14 files changed, 63 insertions(+), 44 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/1e69edac/hbase-native-client/connection/client-dispatcher.h
----------------------------------------------------------------------
diff --git a/hbase-native-client/connection/client-dispatcher.h b/hbase-native-client/connection/client-dispatcher.h
index 4bfb35d..2497cc7 100644
--- a/hbase-native-client/connection/client-dispatcher.h
+++ b/hbase-native-client/connection/client-dispatcher.h
@@ -31,7 +31,8 @@
 
 namespace hbase {
 /**
- * Dispatcher that assigns a call_id and then routes the response back to the future.
+ * Dispatcher that assigns a call_id and then routes the response back to the
+ * future.
  */
 class ClientDispatcher
     : public wangle::ClientDispatcherBase<SerializePipeline,

http://git-wip-us.apache.org/repos/asf/hbase/blob/1e69edac/hbase-native-client/connection/client-handler.cc
----------------------------------------------------------------------
diff --git a/hbase-native-client/connection/client-handler.cc b/hbase-native-client/connection/client-handler.cc
index cae03c7..2e3fcd3 100644
--- a/hbase-native-client/connection/client-handler.cc
+++ b/hbase-native-client/connection/client-handler.cc
@@ -37,7 +37,7 @@ using hbase::pb::GetResponse;
 using google::protobuf::Message;
 
 ClientHandler::ClientHandler(std::string user_name)
-    : user_name_(user_name), serde_(), once_flag_(),
+    : user_name_(user_name), serde_(), once_flag_(std::make_unique<std::once_flag>()),
       resp_msgs_(
           make_unique<folly::AtomicHashMap<
               uint32_t, std::shared_ptr<google::protobuf::Message>>>(5000)) {}

http://git-wip-us.apache.org/repos/asf/hbase/blob/1e69edac/hbase-native-client/connection/connection-factory.cc
----------------------------------------------------------------------
diff --git a/hbase-native-client/connection/connection-factory.cc b/hbase-native-client/connection/connection-factory.cc
index 635d12d..beec6d5 100644
--- a/hbase-native-client/connection/connection-factory.cc
+++ b/hbase-native-client/connection/connection-factory.cc
@@ -19,8 +19,6 @@
 
 #include "connection/connection-factory.h"
 
-#include <wangle/concurrent/GlobalExecutor.h>
-
 #include "connection/client-dispatcher.h"
 #include "connection/pipeline.h"
 #include "connection/service.h"
@@ -28,9 +26,9 @@
 using namespace folly;
 using namespace hbase;
 
-ConnectionFactory::ConnectionFactory()
-    : io_pool_(std::static_pointer_cast<wangle::IOThreadPoolExecutor>(
-          wangle::getIOExecutor())),
+ConnectionFactory::ConnectionFactory(
+    std::shared_ptr<wangle::IOThreadPoolExecutor> io_pool)
+    : io_pool_(io_pool),
       pipeline_factory_(std::make_shared<RpcPipelineFactory>()) {}
 
 std::shared_ptr<wangle::ClientBootstrap<SerializePipeline>>

http://git-wip-us.apache.org/repos/asf/hbase/blob/1e69edac/hbase-native-client/connection/connection-factory.h
----------------------------------------------------------------------
diff --git a/hbase-native-client/connection/connection-factory.h b/hbase-native-client/connection/connection-factory.h
index 8b6d8d8..fb5d9fe 100644
--- a/hbase-native-client/connection/connection-factory.h
+++ b/hbase-native-client/connection/connection-factory.h
@@ -39,7 +39,7 @@ public:
    * Constructor.
    * There should only be one ConnectionFactory per client.
    */
-  ConnectionFactory();
+  ConnectionFactory(std::shared_ptr<wangle::IOThreadPoolExecutor> io_pool);
   /** Default Desctructor */
   virtual ~ConnectionFactory() = default;
 

http://git-wip-us.apache.org/repos/asf/hbase/blob/1e69edac/hbase-native-client/connection/connection-pool-test.cc
----------------------------------------------------------------------
diff --git a/hbase-native-client/connection/connection-pool-test.cc b/hbase-native-client/connection/connection-pool-test.cc
index b1a0ba0..c0c346f 100644
--- a/hbase-native-client/connection/connection-pool-test.cc
+++ b/hbase-native-client/connection/connection-pool-test.cc
@@ -34,6 +34,7 @@ using ::testing::_;
 
 class MockConnectionFactory : public ConnectionFactory {
 public:
+  MockConnectionFactory() : ConnectionFactory(nullptr) {}
   MOCK_METHOD0(MakeBootstrap,
                std::shared_ptr<wangle::ClientBootstrap<SerializePipeline>>());
   MOCK_METHOD3(Connect,

http://git-wip-us.apache.org/repos/asf/hbase/blob/1e69edac/hbase-native-client/connection/connection-pool.cc
----------------------------------------------------------------------
diff --git a/hbase-native-client/connection/connection-pool.cc b/hbase-native-client/connection/connection-pool.cc
index 75f343e..90e2056 100644
--- a/hbase-native-client/connection/connection-pool.cc
+++ b/hbase-native-client/connection/connection-pool.cc
@@ -31,9 +31,10 @@ using hbase::HBaseService;
 using folly::SharedMutexWritePriority;
 using folly::SocketAddress;
 
-ConnectionPool::ConnectionPool()
-    : cf_(std::make_shared<ConnectionFactory>()), clients_(), connections_(),
-      map_mutex_() {}
+ConnectionPool::ConnectionPool(
+    std::shared_ptr<wangle::IOThreadPoolExecutor> io_executor)
+    : cf_(std::make_shared<ConnectionFactory>(io_executor)), clients_(),
+      connections_(), map_mutex_() {}
 ConnectionPool::ConnectionPool(std::shared_ptr<ConnectionFactory> cf)
     : cf_(cf), clients_(), connections_(), map_mutex_() {}
 

http://git-wip-us.apache.org/repos/asf/hbase/blob/1e69edac/hbase-native-client/connection/connection-pool.h
----------------------------------------------------------------------
diff --git a/hbase-native-client/connection/connection-pool.h b/hbase-native-client/connection/connection-pool.h
index 5edd407..60f00de 100644
--- a/hbase-native-client/connection/connection-pool.h
+++ b/hbase-native-client/connection/connection-pool.h
@@ -61,7 +61,7 @@ struct ServerNameHash {
 class ConnectionPool {
 public:
   /** Create connection pool wit default connection factory */
-  ConnectionPool();
+  ConnectionPool(std::shared_ptr<wangle::IOThreadPoolExecutor> io_executor);
 
   /**
    * Desctructor.

http://git-wip-us.apache.org/repos/asf/hbase/blob/1e69edac/hbase-native-client/core/client.cc
----------------------------------------------------------------------
diff --git a/hbase-native-client/core/client.cc b/hbase-native-client/core/client.cc
index 4b9f844..1e80998 100644
--- a/hbase-native-client/core/client.cc
+++ b/hbase-native-client/core/client.cc
@@ -19,18 +19,27 @@
 
 #include "core/client.h"
 
-#include <folly/Logging.h>
-#include <folly/Random.h>
-#include <gflags/gflags.h>
 #include <glog/logging.h>
-#include <wangle/concurrent/GlobalExecutor.h>
 
 #include <string>
 
-#include "if/ZooKeeper.pb.h"
-
 using namespace folly;
 using namespace std;
 using namespace hbase::pb;
 
-namespace hbase {}
+namespace hbase {
+
+Client::Client(std::string zk_quorum)
+    : cpu_executor_(std::make_shared<wangle::CPUThreadPoolExecutor>(4)),
+      io_executor_(std::make_shared<wangle::IOThreadPoolExecutor>(
+          sysconf(_SC_NPROCESSORS_ONLN))),
+      location_cache_(zk_quorum, cpu_executor_, io_executor_) {}
+
+// We can't have the threads continue running after everything is done
+// that leads to an error.
+Client::~Client() {
+  cpu_executor_->stop();
+  io_executor_->stop();
+}
+
+} // namespace hbase

http://git-wip-us.apache.org/repos/asf/hbase/blob/1e69edac/hbase-native-client/core/client.h
----------------------------------------------------------------------
diff --git a/hbase-native-client/core/client.h b/hbase-native-client/core/client.h
index 4a6d23b..4db82c4 100644
--- a/hbase-native-client/core/client.h
+++ b/hbase-native-client/core/client.h
@@ -21,6 +21,8 @@
 
 #include <folly/futures/Future.h>
 #include <folly/io/IOBuf.h>
+#include <wangle/concurrent/CPUThreadPoolExecutor.h>
+#include <wangle/concurrent/IOThreadPoolExecutor.h>
 
 #include <string>
 
@@ -33,19 +35,22 @@ namespace hbase {
  * Client.
  *
  * This is the class that provides access to an HBase cluster.
- * It is thread safe and does connection pooling. Current recommendations are to have only
one Client per cluster around.
+ * It is thread safe and does connection pooling. Current recommendations are to
+ * have only one Client per cluster around.
  */
 class Client {
 public:
-
   /**
    * Create a new client.
    * @param quorum_spec Where to connect to get Zookeeper bootstrap information.
    */
   explicit Client(std::string quorum_spec);
+  ~Client();
 
 private:
   LocationCache location_cache_;
+  std::shared_ptr<wangle::CPUThreadPoolExecutor> cpu_executor_;
+  std::shared_ptr<wangle::IOThreadPoolExecutor> io_executor_;
 };
 
 } // namespace hbase

http://git-wip-us.apache.org/repos/asf/hbase/blob/1e69edac/hbase-native-client/core/location-cache-test.cc
----------------------------------------------------------------------
diff --git a/hbase-native-client/core/location-cache-test.cc b/hbase-native-client/core/location-cache-test.cc
index 172799d..8bc6383 100644
--- a/hbase-native-client/core/location-cache-test.cc
+++ b/hbase-native-client/core/location-cache-test.cc
@@ -18,14 +18,15 @@
  */
 #include <folly/Memory.h>
 #include <gtest/gtest.h>
-#include <wangle/concurrent/GlobalExecutor.h>
 
 #include "location-cache.h"
 using namespace hbase;
 
 TEST(LocationCacheTest, TestGetMetaNodeContents) {
   // TODO(elliott): need to make a test utility for this.
-  LocationCache cache{"localhost:2181", wangle::getCPUExecutor()};
+  auto cpu = std::make_shared<wangle::CPUThreadPoolExecutor>(4);
+  auto io = std::make_shared<wangle::IOThreadPoolExecutor>(4);
+  LocationCache cache{"localhost:2181", cpu, io};
   auto f = cache.LocateMeta();
   auto result = f.get();
   ASSERT_FALSE(f.hasException());

http://git-wip-us.apache.org/repos/asf/hbase/blob/1e69edac/hbase-native-client/core/location-cache.cc
----------------------------------------------------------------------
diff --git a/hbase-native-client/core/location-cache.cc b/hbase-native-client/core/location-cache.cc
index 583d305..6c018f9 100644
--- a/hbase-native-client/core/location-cache.cc
+++ b/hbase-native-client/core/location-cache.cc
@@ -20,7 +20,8 @@
 
 #include <folly/Logging.h>
 #include <folly/io/IOBuf.h>
-#include <wangle/concurrent/GlobalExecutor.h>
+#include <wangle/concurrent/CPUThreadPoolExecutor.h>
+#include <wangle/concurrent/IOThreadPoolExecutor.h>
 
 #include "connection/response.h"
 #include "if/Client.pb.h"
@@ -48,10 +49,13 @@ using hbase::pb::RegionInfo;
 // TODO(eclark): make this configurable on client creation
 static const char META_ZNODE_NAME[] = "/hbase/meta-region-server";
 
-LocationCache::LocationCache(string quorum_spec,
-                             shared_ptr<folly::Executor> executor)
-    : quorum_spec_(quorum_spec), executor_(executor), meta_promise_(nullptr),
-      meta_lock_(), cp_(), meta_util_(), zk_(nullptr) {
+LocationCache::LocationCache(
+    std::string quorum_spec,
+    std::shared_ptr<wangle::CPUThreadPoolExecutor> cpu_executor,
+    std::shared_ptr<wangle::IOThreadPoolExecutor> io_executor)
+    : quorum_spec_(quorum_spec), cpu_executor_(cpu_executor),
+      meta_promise_(nullptr), meta_lock_(), cp_(io_executor), meta_util_(),
+      zk_(nullptr) {
   zk_ = zookeeper_init(quorum_spec.c_str(), nullptr, 1000, 0, 0, 0);
 }
 
@@ -79,7 +83,7 @@ void LocationCache::InvalidateMeta() {
 /// MUST hold the meta_lock_
 void LocationCache::RefreshMetaLocation() {
   meta_promise_ = make_unique<SharedPromise<ServerName>>();
-  executor_->add([&] {
+  cpu_executor_->add([&] {
     meta_promise_->setWith([&] { return this->ReadMetaLocation(); });
   });
 }
@@ -109,10 +113,9 @@ ServerName LocationCache::ReadMetaLocation() {
 
 Future<std::shared_ptr<RegionLocation>>
 LocationCache::LocateFromMeta(const TableName &tn, const string &row) {
-  auto exec = wangle::getCPUExecutor();
   return this->LocateMeta()
-      .via(exec.get())
-      .then([ exec = exec, this ](ServerName sn) { return this->cp_.get(sn); })
+      .via(cpu_executor_.get())
+      .then([this](ServerName sn) { return this->cp_.get(sn); })
       .then([&](std::shared_ptr<HBaseService> service) {
         return (*service)(std::move(meta_util_.MetaRequest(tn, row)));
       })
@@ -121,7 +124,7 @@ LocationCache::LocateFromMeta(const TableName &tn, const string &row)
{
         // a region location.
         return this->CreateLocation(std::move(resp));
       })
-      .then([ exec = exec, this ](std::shared_ptr<RegionLocation> rl) {
+      .then([this](std::shared_ptr<RegionLocation> rl) {
         // Now fill out the connection.
         rl->set_service(cp_.get(rl->server_name()));
         return rl;
@@ -129,14 +132,14 @@ LocationCache::LocateFromMeta(const TableName &tn, const string
&row) {
 }
 
 /**
- * Filter to remove a service from the location cache and the connection cache on errors
+ * Filter to remove a service from the location cache and the connection cache
+ * on errors
  * or on cloase.
  */
 class RemoveServiceFilter
     : public ServiceFilter<std::unique_ptr<Request>, Response> {
 
 public:
-
   /** Create a new filter. */
   RemoveServiceFilter(std::shared_ptr<HBaseService> service, ServerName sn,
                       ConnectionPool &cp)
@@ -157,7 +160,6 @@ public:
     }
   }
 
-
   /** Has this been closed */
   virtual bool isAvailable() override {
     return !released && service_->isAvailable();

http://git-wip-us.apache.org/repos/asf/hbase/blob/1e69edac/hbase-native-client/core/location-cache.h
----------------------------------------------------------------------
diff --git a/hbase-native-client/core/location-cache.h b/hbase-native-client/core/location-cache.h
index 88bec18..e077750 100644
--- a/hbase-native-client/core/location-cache.h
+++ b/hbase-native-client/core/location-cache.h
@@ -21,6 +21,8 @@
 #include <folly/Executor.h>
 #include <folly/futures/Future.h>
 #include <folly/futures/SharedPromise.h>
+#include <wangle/concurrent/CPUThreadPoolExecutor.h>
+#include <wangle/concurrent/IOThreadPoolExecutor.h>
 #include <zookeeper/zookeeper.h>
 
 #include <memory>
@@ -51,7 +53,8 @@ public:
    * @param executor The cpu executor to run on.
    */
   LocationCache(std::string quorum_spec,
-                std::shared_ptr<folly::Executor> executor);
+                std::shared_ptr<wangle::CPUThreadPoolExecutor> cpu_exector,
+                std::shared_ptr<wangle::IOThreadPoolExecutor> io_executor);
   /**
    * Destructor.
    * This will clean up the zookeeper connections.
@@ -81,8 +84,9 @@ private:
   hbase::pb::ServerName ReadMetaLocation();
   std::shared_ptr<RegionLocation> CreateLocation(const Response &resp);
 
+  /* data */
   std::string quorum_spec_;
-  std::shared_ptr<folly::Executor> executor_;
+  std::shared_ptr<wangle::CPUThreadPoolExecutor> cpu_executor_;
   std::unique_ptr<folly::SharedPromise<hbase::pb::ServerName>> meta_promise_;
   std::mutex meta_lock_;
   MetaUtil meta_util_;

http://git-wip-us.apache.org/repos/asf/hbase/blob/1e69edac/hbase-native-client/core/simple-client.cc
----------------------------------------------------------------------
diff --git a/hbase-native-client/core/simple-client.cc b/hbase-native-client/core/simple-client.cc
index 39c82c3..f3f6c42 100644
--- a/hbase-native-client/core/simple-client.cc
+++ b/hbase-native-client/core/simple-client.cc
@@ -22,7 +22,7 @@
 #include <folly/futures/Future.h>
 #include <gflags/gflags.h>
 #include <wangle/concurrent/CPUThreadPoolExecutor.h>
-#include <wangle/concurrent/GlobalExecutor.h>
+#include <wangle/concurrent/IOThreadPoolExecutor.h>
 
 #include <atomic>
 #include <chrono>
@@ -89,12 +89,10 @@ int main(int argc, char *argv[]) {
   // Set up thread pools.
   auto cpu_pool =
       std::make_shared<wangle::CPUThreadPoolExecutor>(FLAGS_threads);
-  wangle::setCPUExecutor(cpu_pool);
   auto io_pool = std::make_shared<wangle::IOThreadPoolExecutor>(5);
-  wangle::setIOExecutor(io_pool);
 
   // Create the cache.
-  LocationCache cache{FLAGS_zookeeper, cpu_pool};
+  LocationCache cache{FLAGS_zookeeper, cpu_pool, io_pool};
 
   auto row = FLAGS_row;
   auto tn = folly::to<TableName>(FLAGS_table);
@@ -105,7 +103,7 @@ int main(int argc, char *argv[]) {
   auto num_puts = FLAGS_columns;
 
   auto results = std::vector<Future<Response>>{};
-  uint64_t col{0};
+  auto col = uint64_t{0};
   for (; col < num_puts; col++) {
     results.push_back(folly::makeFuture(col)
                           .via(cpu_pool.get())

http://git-wip-us.apache.org/repos/asf/hbase/blob/1e69edac/hbase-native-client/serde/server-name-test.cc
----------------------------------------------------------------------
diff --git a/hbase-native-client/serde/server-name-test.cc b/hbase-native-client/serde/server-name-test.cc
index 2281fa2..73a68d6 100644
--- a/hbase-native-client/serde/server-name-test.cc
+++ b/hbase-native-client/serde/server-name-test.cc
@@ -46,5 +46,4 @@ TEST(TestServerName, TestIPV6) {
 
   ASSERT_EQ("[::::1]", sn.host_name());
   ASSERT_EQ(123, sn.port());
-
 }


Mime
View raw message