hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From e...@apache.org
Subject [2/2] hbase git commit: HBASE-18214 Replace the folly::AtomicHashMap usage in the RPC layer
Date Mon, 26 Jun 2017 21:33:56 GMT
HBASE-18214 Replace the folly::AtomicHashMap usage in the RPC layer


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

Branch: refs/heads/HBAES-14850
Commit: a499d6a819b2464c5e6500ef0d93808ef6f46d73
Parents: bb10756
Author: Enis Soztutar <enis@apache.org>
Authored: Mon Jun 26 14:30:49 2017 -0700
Committer: Enis Soztutar <enis@apache.org>
Committed: Mon Jun 26 14:30:49 2017 -0700

----------------------------------------------------------------------
 .../connection/client-dispatcher.cc             |  16 ++-
 .../connection/client-dispatcher.h              |   6 +-
 .../connection/client-handler.cc                |  17 +--
 hbase-native-client/connection/client-handler.h |   5 +-
 .../connection/connection-factory.cc            |   8 +-
 hbase-native-client/utils/BUCK                  |  17 ++-
 .../utils/concurrent-map-test.cc                |  36 ++++++
 hbase-native-client/utils/concurrent-map.h      | 125 +++++++++++++++++++
 8 files changed, 196 insertions(+), 34 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/a499d6a8/hbase-native-client/connection/client-dispatcher.cc
----------------------------------------------------------------------
diff --git a/hbase-native-client/connection/client-dispatcher.cc b/hbase-native-client/connection/client-dispatcher.cc
index 35a1f7d..b9b2c34 100644
--- a/hbase-native-client/connection/client-dispatcher.cc
+++ b/hbase-native-client/connection/client-dispatcher.cc
@@ -25,16 +25,11 @@ using std::unique_ptr;
 
 namespace hbase {
 
-ClientDispatcher::ClientDispatcher() : requests_(5000), current_call_id_(9) {}
+ClientDispatcher::ClientDispatcher() : current_call_id_(9), requests_(5000) {}
 
 void ClientDispatcher::read(Context *ctx, unique_ptr<Response> in) {
   auto call_id = in->call_id();
-
-  auto search = requests_.find(call_id);
-  CHECK(search != requests_.end());
-  auto p = std::move(search->second);
-
-  requests_.erase(call_id);
+  auto p = requests_.find_and_erase(call_id);
 
   if (in->exception()) {
     p.setException(in->exception());
@@ -46,8 +41,11 @@ void ClientDispatcher::read(Context *ctx, unique_ptr<Response> in)
{
 folly::Future<unique_ptr<Response>> ClientDispatcher::operator()(unique_ptr<Request>
arg) {
   auto call_id = current_call_id_++;
   arg->set_call_id(call_id);
-  requests_.insert(call_id, folly::Promise<unique_ptr<Response>>{});
-  auto &p = requests_.find(call_id)->second;
+
+  // TODO: if the map is full (or we have more than hbase.client.perserver.requests.threshold)
+  // then throw ServerTooBusyException so that upper layers will retry.
+  auto &p = requests_[call_id];
+
   auto f = p.getFuture();
   p.setInterruptHandler([call_id, this](const folly::exception_wrapper &e) {
     LOG(ERROR) << "e = " << call_id;

http://git-wip-us.apache.org/repos/asf/hbase/blob/a499d6a8/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 857042c..1f8e6b3 100644
--- a/hbase-native-client/connection/client-dispatcher.h
+++ b/hbase-native-client/connection/client-dispatcher.h
@@ -19,16 +19,18 @@
 
 #pragma once
 
-#include <folly/AtomicHashMap.h>
 #include <folly/Logging.h>
 #include <wangle/service/ClientDispatcher.h>
 
 #include <atomic>
+#include <map>
 #include <memory>
+#include <mutex>
 
 #include "connection/pipeline.h"
 #include "connection/request.h"
 #include "connection/response.h"
+#include "utils/concurrent-map.h"
 
 namespace hbase {
 /**
@@ -51,7 +53,7 @@ class ClientDispatcher
   folly::Future<folly::Unit> close() override;
 
  private:
-  folly::AtomicHashMap<uint32_t, folly::Promise<std::unique_ptr<Response>>>
requests_;
+  concurrent_map<uint32_t, folly::Promise<std::unique_ptr<Response>>> requests_;
   // Start at some number way above what could
   // be there for un-initialized call id counters.
   //

http://git-wip-us.apache.org/repos/asf/hbase/blob/a499d6a8/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 775df68..052c171 100644
--- a/hbase-native-client/connection/client-handler.cc
+++ b/hbase-native-client/connection/client-handler.cc
@@ -40,8 +40,9 @@ ClientHandler::ClientHandler(std::string user_name, std::shared_ptr<Codec>
codec
       serde_(codec),
       server_(server),
       once_flag_(std::make_unique<std::once_flag>()),
-      resp_msgs_(std::make_unique<folly::AtomicHashMap<uint32_t, std::shared_ptr<Message>>>(5000))
{
-}
+      resp_msgs_(
+          std::make_unique<concurrent_map<uint32_t, std::shared_ptr<google::protobuf::Message>>>(
+              5000)) {}
 
 void ClientHandler::read(Context *ctx, std::unique_ptr<folly::IOBuf> buf) {
   if (LIKELY(buf != nullptr)) {
@@ -53,15 +54,7 @@ void ClientHandler::read(Context *ctx, std::unique_ptr<folly::IOBuf>
buf) {
     VLOG(3) << "Read RPC ResponseHeader size=" << used_bytes << " call_id="
<< header.call_id()
             << " has_exception=" << header.has_exception() << ", server:
" << server_;
 
-    // Get the response protobuf from the map
-    auto search = resp_msgs_->find(header.call_id());
-    // It's an error if it's not there.
-    CHECK(search != resp_msgs_->end());
-    auto resp_msg = search->second;
-    CHECK(resp_msg != nullptr);
-
-    // Make sure we don't leak the protobuf
-    resp_msgs_->erase(header.call_id());
+    auto resp_msg = resp_msgs_->find_and_erase(header.call_id());
 
     // set the call_id.
     // This will be used to by the dispatcher to match up
@@ -133,7 +126,7 @@ folly::Future<folly::Unit> ClientHandler::write(Context *ctx, std::unique_ptr<Re
   VLOG(3) << "Writing RPC Request:" << r->DebugString() << ", server:
" << server_;
 
   // Now store the call id to response.
-  resp_msgs_->insert(r->call_id(), r->resp_msg());
+  resp_msgs_->insert(std::make_pair(r->call_id(), r->resp_msg()));
 
   // Send the data down the pipeline.
   return ctx->fireWrite(serde_.Request(r->call_id(), r->method(), r->req_msg().get()));

http://git-wip-us.apache.org/repos/asf/hbase/blob/a499d6a8/hbase-native-client/connection/client-handler.h
----------------------------------------------------------------------
diff --git a/hbase-native-client/connection/client-handler.h b/hbase-native-client/connection/client-handler.h
index 4c106e0..8de3a8b 100644
--- a/hbase-native-client/connection/client-handler.h
+++ b/hbase-native-client/connection/client-handler.h
@@ -18,7 +18,6 @@
  */
 #pragma once
 
-#include <folly/AtomicHashMap.h>
 #include <wangle/channel/Handler.h>
 
 #include <atomic>
@@ -30,6 +29,7 @@
 #include "exceptions/exception.h"
 #include "serde/codec.h"
 #include "serde/rpc.h"
+#include "utils/concurrent-map.h"
 
 // Forward decs.
 namespace hbase {
@@ -81,7 +81,6 @@ class ClientHandler
   std::string server_;  // for logging
 
   // in flight requests
-  std::unique_ptr<folly::AtomicHashMap<uint32_t, std::shared_ptr<google::protobuf::Message>>>
-      resp_msgs_;
+  std::unique_ptr<concurrent_map<uint32_t, std::shared_ptr<google::protobuf::Message>>>
resp_msgs_;
 };
 }  // namespace hbase

http://git-wip-us.apache.org/repos/asf/hbase/blob/a499d6a8/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 d1bfbce..a0c7f96 100644
--- a/hbase-native-client/connection/connection-factory.cc
+++ b/hbase-native-client/connection/connection-factory.cc
@@ -17,15 +17,15 @@
  *
  */
 
-#include "connection/connection-factory.h"
-#include "connection/sasl-handler.h"
+#include <glog/logging.h>
+#include <wangle/channel/Handler.h>
 
 #include <chrono>
 
-#include <glog/logging.h>
-#include <wangle/channel/Handler.h>
 #include "connection/client-dispatcher.h"
+#include "connection/connection-factory.h"
 #include "connection/pipeline.h"
+#include "connection/sasl-handler.h"
 #include "connection/service.h"
 
 using std::chrono::milliseconds;

http://git-wip-us.apache.org/repos/asf/hbase/blob/a499d6a8/hbase-native-client/utils/BUCK
----------------------------------------------------------------------
diff --git a/hbase-native-client/utils/BUCK b/hbase-native-client/utils/BUCK
index 788056b..96f02b8 100644
--- a/hbase-native-client/utils/BUCK
+++ b/hbase-native-client/utils/BUCK
@@ -20,6 +20,7 @@ cxx_library(
     exported_headers=[
         "bytes-util.h",
         "connection-util.h",
+        "concurrent-map.h",
         "optional.h",
         "sys-util.h",
         "time-util.h",
@@ -38,18 +39,26 @@ cxx_library(
     ],
     compiler_flags=['-Weffc++'],)
 cxx_test(
-    name="user-util-test",
+    name="bytes-util-test",
     srcs=[
-        "user-util-test.cc",
+        "bytes-util-test.cc",
     ],
     deps=[
         ":utils",
     ],)
 cxx_test(
-    name="bytes-util-test",
+    name="concurrent-map-test",
     srcs=[
-        "bytes-util-test.cc",
+        "concurrent-map-test.cc",
     ],
     deps=[
         ":utils",
     ],)
+cxx_test(
+    name="user-util-test",
+    srcs=[
+        "user-util-test.cc",
+    ],
+    deps=[
+        ":utils",
+    ],)
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hbase/blob/a499d6a8/hbase-native-client/utils/concurrent-map-test.cc
----------------------------------------------------------------------
diff --git a/hbase-native-client/utils/concurrent-map-test.cc b/hbase-native-client/utils/concurrent-map-test.cc
new file mode 100644
index 0000000..588bd08
--- /dev/null
+++ b/hbase-native-client/utils/concurrent-map-test.cc
@@ -0,0 +1,36 @@
+/*
+ * 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 <folly/Logging.h>
+#include <gtest/gtest.h>
+#include <string>
+
+#include "utils/concurrent-map.h"
+
+using hbase::concurrent_map;
+
+TEST(TestConcurrentMap, TestFindAndErase) {
+  concurrent_map<std::string, std::string> map{500};
+
+  map.insert(std::make_pair("foo", "bar"));
+  auto prev = map.find_and_erase("foo");
+  ASSERT_EQ("bar", prev);
+
+  ASSERT_EQ(map.end(), map.find("foo"));
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/a499d6a8/hbase-native-client/utils/concurrent-map.h
----------------------------------------------------------------------
diff --git a/hbase-native-client/utils/concurrent-map.h b/hbase-native-client/utils/concurrent-map.h
new file mode 100644
index 0000000..d9703e1
--- /dev/null
+++ b/hbase-native-client/utils/concurrent-map.h
@@ -0,0 +1,125 @@
+/*
+ * 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 <map>
+#include <memory>
+#include <shared_mutex>
+#include <unordered_map>
+#include <utility>
+
+namespace hbase {
+
+/**
+ * A concurrent version of std::unordered_map where we acquire a shared or exclusive
+ * lock for operations. This is NOT a highly-concurrent and scalable implementation
+ * since there is only one lock object.
+ * Replace this with tbb::concurrent_unordered_map or similar.
+ *
+ * Concurrency here is different than in Java. For example, the iterators returned from
+ * find() will not copy the key, value pairs.
+ */
+template <typename K, typename V>
+class concurrent_map {
+ public:
+  typedef K key_type;
+  typedef V mapped_type;
+  typedef std::pair<const key_type, mapped_type> value_type;
+  typedef typename std::unordered_map<K, V>::iterator iterator;
+  typedef typename std::unordered_map<K, V>::const_iterator const_iterator;
+
+  concurrent_map() : map_(), mutex_() {}
+  explicit concurrent_map(int32_t n) : map_(n), mutex_() {}
+
+  void insert(const value_type& value) {
+    std::unique_lock<std::shared_timed_mutex> lock(mutex_);
+    map_.insert(value);
+  }
+
+  /**
+   * Return the mapped object for this key. Be careful to not use the return reference
+   * to do assignment. I think it won't be thread safe
+   */
+  mapped_type& at(const key_type& key) {
+    std::shared_lock<std::shared_timed_mutex> lock(mutex_);
+    iterator where = map_.find(key);
+    if (where == end()) {
+      std::runtime_error("Key not found");
+    }
+    return where->second;
+  }
+
+  mapped_type& operator[](const key_type& key) {
+    std::unique_lock<std::shared_timed_mutex> lock(mutex_);
+    iterator where = map_.find(key);
+    if (where == end()) {
+      return map_[key];
+    }
+    return where->second;
+  }
+
+  /**
+   * Atomically finds the entry and removes it from the map, returning
+   * the previously associated value.
+   */
+  mapped_type find_and_erase(const K& key) {
+    std::unique_lock<std::shared_timed_mutex> lock(mutex_);
+    auto search = map_.find(key);
+    // It's an error if it's not there.
+    CHECK(search != end());
+    auto val = std::move(search->second);
+    map_.erase(key);
+    return val;
+  }
+
+  void erase(const K& key) {
+    std::unique_lock<std::shared_timed_mutex> lock(mutex_);
+    map_.erase(key);
+  }
+
+  iterator begin() { return map_.begin(); }
+
+  const_iterator begin() const { return map_.begin(); }
+
+  const_iterator cbegin() const { return map_.begin(); }
+
+  iterator end() { return map_.end(); }
+
+  const_iterator end() const { return map_.end(); }
+
+  const_iterator cend() const { return map_.end(); }
+
+  iterator find(const K& key) {
+    std::shared_lock<std::shared_timed_mutex> lock(mutex_);
+    return map_.find(key);
+  }
+
+  // TODO: find(), at() returning const_iterator
+
+  bool empty() const {
+    std::unique_lock<std::shared_timed_mutex> lock(mutex_);
+    return map_.empty();
+  }
+
+ private:
+  std::shared_timed_mutex mutex_;
+  std::unordered_map<K, V> map_;
+};
+} /* namespace hbase */


Mime
View raw message