hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ecl...@apache.org
Subject [38/50] [abbrv] hbase git commit: HBASE-15687 Allow decoding more than GetResponse from the server
Date Fri, 13 May 2016 21:37:32 GMT
HBASE-15687 Allow decoding more than GetResponse from the server

Summary:
We'll need more than get's for the client to be usable. So now the
Request class contains the protobufs needed to encode and decode
rpc's.

I also added some helper methods to create initial requests.

Test Plan: It compiles and still gets data from HBase meta

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


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

Branch: refs/heads/HBASE-14850
Commit: 5a222062859d416f2dc5d55f56bce47fd8a058d9
Parents: 14e2166
Author: Elliott Clark <eclark@apache.org>
Authored: Wed Apr 27 15:27:09 2016 -0700
Committer: Elliott Clark <eclark@apache.org>
Committed: Fri May 13 14:36:28 2016 -0700

----------------------------------------------------------------------
 hbase-native-client/connection/BUCK             |  1 +
 .../connection/client-dispatcher.cc             |  6 +--
 .../connection/client-dispatcher.h              |  6 +--
 .../connection/client-handler.cc                | 30 ++++++++++---
 hbase-native-client/connection/client-handler.h | 18 ++++++--
 .../connection/connection-factory.cc            |  6 +--
 .../connection/connection-factory.h             |  2 +-
 hbase-native-client/connection/pipeline.h       |  3 +-
 hbase-native-client/connection/request.cc       | 45 ++++++++++++++++++++
 hbase-native-client/connection/request.h        | 25 +++++++++--
 hbase-native-client/core/simple-client.cc       | 13 +++---
 11 files changed, 123 insertions(+), 32 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/5a222062/hbase-native-client/connection/BUCK
----------------------------------------------------------------------
diff --git a/hbase-native-client/connection/BUCK b/hbase-native-client/connection/BUCK
index 5067708..d393885 100644
--- a/hbase-native-client/connection/BUCK
+++ b/hbase-native-client/connection/BUCK
@@ -32,6 +32,7 @@ cxx_library(name="connection",
                 "client-handler.cc",
                 "connection-factory.cc",
                 "pipeline.cc",
+                "request.cc",
             ],
             deps=[
                 "//if:if",

http://git-wip-us.apache.org/repos/asf/hbase/blob/5a222062/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 25cff7d..eea0a17 100644
--- a/hbase-native-client/connection/client-dispatcher.cc
+++ b/hbase-native-client/connection/client-dispatcher.cc
@@ -35,16 +35,16 @@ void ClientDispatcher::read(Context *ctx, Response in) {
   p.setValue(in);
 }
 
-Future<Response> ClientDispatcher::operator()(Request arg) {
+Future<Response> ClientDispatcher::operator()(std::unique_ptr<Request> arg) {
   auto call_id = ++current_call_id_;
 
-  arg.set_call_id(call_id);
+  arg->set_call_id(call_id);
   auto &p = requests_[call_id];
   auto f = p.getFuture();
   p.setInterruptHandler([call_id, this](const folly::exception_wrapper &e) {
     this->requests_.erase(call_id);
   });
-  this->pipeline_->write(arg);
+  this->pipeline_->write(std::move(arg));
 
   return f;
 }

http://git-wip-us.apache.org/repos/asf/hbase/blob/5a222062/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 89c7119..877e877 100644
--- a/hbase-native-client/connection/client-dispatcher.h
+++ b/hbase-native-client/connection/client-dispatcher.h
@@ -27,11 +27,11 @@
 
 namespace hbase {
 class ClientDispatcher
-    : public wangle::ClientDispatcherBase<SerializePipeline, Request,
-                                          Response> {
+    : public wangle::ClientDispatcherBase<SerializePipeline,
+                                          std::unique_ptr<Request>, Response> {
 public:
   void read(Context *ctx, Response in) override;
-  folly::Future<Response> operator()(Request arg) override;
+  folly::Future<Response> operator()(std::unique_ptr<Request> arg) override;
   folly::Future<folly::Unit> close(Context *ctx) override;
   folly::Future<folly::Unit> close() override;
 

http://git-wip-us.apache.org/repos/asf/hbase/blob/5a222062/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 205993a7..abcf5c1 100644
--- a/hbase-native-client/connection/client-handler.cc
+++ b/hbase-native-client/connection/client-handler.cc
@@ -47,14 +47,30 @@ void ClientHandler::read(Context *ctx, std::unique_ptr<IOBuf> buf)
{
     LOG(INFO) << "Read ResponseHeader size=" << used_bytes
               << " call_id=" << header.call_id()
               << " has_exception=" << header.has_exception();
+
+    // 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(search);
+
+    // set the call_id.
+    // This will be used to by the dispatcher to match up
+    // the promise with the response.
     received.set_call_id(header.call_id());
 
+    // If there was an exception then there's no
+    // data left on the wire.
     if (header.has_exception() == false) {
       buf->trimStart(used_bytes);
-      // For now assume that everything was a get.
-      // We'll need to set this up later.
-      received.set_response(std::make_shared<GetResponse>());
-      used_bytes = deser_.parse_delimited(buf.get(), received.response().get());
+      used_bytes = deser_.parse_delimited(buf.get(), resp_msg.get());
+      // Make sure that bytes were parsed.
+      CHECK(used_bytes == buf->length());
+      received.set_response(resp_msg);
     }
     ctx->fireRead(std::move(received));
   }
@@ -62,7 +78,7 @@ void ClientHandler::read(Context *ctx, std::unique_ptr<IOBuf> buf)
{
 
 // TODO(eclark): Figure out how to handle the
 // network errors that are going to come.
-Future<Unit> ClientHandler::write(Context *ctx, Request r) {
+Future<Unit> ClientHandler::write(Context *ctx, std::unique_ptr<Request> r) {
   // Keep track of if we have sent the header.
   if (UNLIKELY(need_send_header_)) {
     need_send_header_ = false;
@@ -78,5 +94,7 @@ Future<Unit> ClientHandler::write(Context *ctx, Request r) {
     ctx->fireWrite(std::move(pre));
   }
 
-  return ctx->fireWrite(ser_.request(r.call_id(), r.method(), r.msg()));
+  resp_msgs_[r->call_id()] = r->resp_msg();
+  return ctx->fireWrite(
+      ser_.request(r->call_id(), r->method(), r->req_msg().get()));
 }

http://git-wip-us.apache.org/repos/asf/hbase/blob/5a222062/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 dbaf5a0..41bb883 100644
--- a/hbase-native-client/connection/client-handler.h
+++ b/hbase-native-client/connection/client-handler.h
@@ -30,20 +30,30 @@ namespace hbase {
 class Request;
 class Response;
 }
+namespace google {
+namespace protobuf {
+class Message;
+}
+}
 
 namespace hbase {
-class ClientHandler
-    : public wangle::Handler<std::unique_ptr<folly::IOBuf>, Response, Request,
-                             std::unique_ptr<folly::IOBuf>> {
+class ClientHandler : public wangle::Handler<std::unique_ptr<folly::IOBuf>,
+                                             Response, std::unique_ptr<Request>,
+                                             std::unique_ptr<folly::IOBuf>> {
 public:
   ClientHandler(std::string user_name);
   void read(Context *ctx, std::unique_ptr<folly::IOBuf> msg) override;
-  folly::Future<folly::Unit> write(Context *ctx, Request r) override;
+  folly::Future<folly::Unit> write(Context *ctx,
+                                   std::unique_ptr<Request> r) override;
 
 private:
   bool need_send_header_ = true;
   std::string user_name_;
   ClientSerializer ser_;
   ClientDeserializer deser_;
+
+  // in flight requests
+  std::unordered_map<uint32_t, std::shared_ptr<google::protobuf::Message>>
+      resp_msgs_;
 };
 } // namespace hbase

http://git-wip-us.apache.org/repos/asf/hbase/blob/5a222062/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 5d1b0da..7073f9d 100644
--- a/hbase-native-client/connection/connection-factory.cc
+++ b/hbase-native-client/connection/connection-factory.cc
@@ -45,14 +45,14 @@ ConnectionFactory::ConnectionFactory() {
   bootstrap_.pipelineFactory(std::make_shared<RpcPipelineFactory>());
 }
 
-std::shared_ptr<Service<Request, Response>>
+std::shared_ptr<Service<std::unique_ptr<Request>, Response>>
 ConnectionFactory::make_connection(std::string host, int port) {
   // Connect to a given server
   // Then when connected create a ClientDispactcher.
   auto pipeline = bootstrap_.connect(SocketAddress(host, port, true)).get();
   auto dispatcher = std::make_shared<ClientDispatcher>();
   dispatcher->setPipeline(pipeline);
-  auto service =
-      std::make_shared<CloseOnReleaseFilter<Request, Response>>(dispatcher);
+  auto service = std::make_shared<
+      CloseOnReleaseFilter<std::unique_ptr<Request>, Response>>(dispatcher);
   return service;
 }

http://git-wip-us.apache.org/repos/asf/hbase/blob/5a222062/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 73ac032..8d1d2f0 100644
--- a/hbase-native-client/connection/connection-factory.h
+++ b/hbase-native-client/connection/connection-factory.h
@@ -31,7 +31,7 @@ namespace hbase {
 class ConnectionFactory {
 public:
   ConnectionFactory();
-  std::shared_ptr<wangle::Service<Request, Response>>
+  std::shared_ptr<wangle::Service<std::unique_ptr<Request>, Response>>
   make_connection(std::string host, int port);
 
 private:

http://git-wip-us.apache.org/repos/asf/hbase/blob/5a222062/hbase-native-client/connection/pipeline.h
----------------------------------------------------------------------
diff --git a/hbase-native-client/connection/pipeline.h b/hbase-native-client/connection/pipeline.h
index 8114fab..6c4f4ff 100644
--- a/hbase-native-client/connection/pipeline.h
+++ b/hbase-native-client/connection/pipeline.h
@@ -26,7 +26,8 @@
 #include "utils/user-util.h"
 
 namespace hbase {
-using SerializePipeline = wangle::Pipeline<folly::IOBufQueue &, Request>;
+using SerializePipeline =
+    wangle::Pipeline<folly::IOBufQueue &, std::unique_ptr<Request>>;
 
 class RpcPipelineFactory : public wangle::PipelineFactory<SerializePipeline> {
 public:

http://git-wip-us.apache.org/repos/asf/hbase/blob/5a222062/hbase-native-client/connection/request.cc
----------------------------------------------------------------------
diff --git a/hbase-native-client/connection/request.cc b/hbase-native-client/connection/request.cc
new file mode 100644
index 0000000..50ea029
--- /dev/null
+++ b/hbase-native-client/connection/request.cc
@@ -0,0 +1,45 @@
+/*
+ * 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 "connection/request.h"
+
+#include "if/Client.pb.h"
+
+using namespace hbase;
+
+Request::Request(std::shared_ptr<google::protobuf::Message> req,
+                 std::shared_ptr<google::protobuf::Message> resp,
+                 std::string method)
+    : req_msg_(req), resp_msg_(resp), method_(method), call_id_(0) {}
+
+std::unique_ptr<Request> Request::get() {
+  return std::make_unique<Request>(std::make_shared<hbase::pb::GetRequest>(),
+                                  std::make_shared<hbase::pb::GetResponse>(),
+                                  "Get");
+}
+std::unique_ptr<Request> Request::mutate() {
+  return std::make_unique<Request>(std::make_shared<hbase::pb::MutateRequest>(),
+                                  std::make_shared<hbase::pb::MutateResponse>(),
+                                  "Mutate");
+}
+std::unique_ptr<Request> Request::scan() {
+  return std::make_unique<Request>(std::make_shared<hbase::pb::ScanRequest>(),
+                                  std::make_shared<hbase::pb::ScanResponse>(),
+                                  "Scan");
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/5a222062/hbase-native-client/connection/request.h
----------------------------------------------------------------------
diff --git a/hbase-native-client/connection/request.h b/hbase-native-client/connection/request.h
index e9e3e88..743c469 100644
--- a/hbase-native-client/connection/request.h
+++ b/hbase-native-client/connection/request.h
@@ -21,22 +21,39 @@
 #include <google/protobuf/message.h>
 
 #include <cstdint>
+#include <memory>
 #include <string>
 
 namespace hbase {
 class Request {
 public:
-  Request() : call_id_(0) {}
+  static std::unique_ptr<Request> get();
+  static std::unique_ptr<Request> mutate();
+  static std::unique_ptr<Request> scan();
+
+  Request(std::shared_ptr<google::protobuf::Message> req,
+          std::shared_ptr<google::protobuf::Message> resp, std::string method);
+
   uint32_t call_id() { return call_id_; }
   void set_call_id(uint32_t call_id) { call_id_ = call_id; }
-  google::protobuf::Message *msg() { return msg_.get(); }
-  void set_msg(std::shared_ptr<google::protobuf::Message> msg) { msg_ = msg; }
+
+  std::shared_ptr<google::protobuf::Message> req_msg() { return req_msg_; }
+  std::shared_ptr<google::protobuf::Message> resp_msg() { return resp_msg_; }
+
+  void set_req_msg(std::shared_ptr<google::protobuf::Message> msg) {
+    req_msg_ = msg;
+  }
+  void set_resp_msg(std::shared_ptr<google::protobuf::Message> msg) {
+    resp_msg_ = msg;
+  }
+
   std::string method() { return method_; }
   void set_method(std::string method) { method_ = method; }
 
 private:
   uint32_t call_id_;
-  std::shared_ptr<google::protobuf::Message> msg_ = nullptr;
+  std::shared_ptr<google::protobuf::Message> req_msg_ = nullptr;
+  std::shared_ptr<google::protobuf::Message> resp_msg_ = nullptr;
   std::string method_ = "Get";
 };
 } // namespace hbase

http://git-wip-us.apache.org/repos/asf/hbase/blob/5a222062/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 8b2fae5..2cb6200 100644
--- a/hbase-native-client/core/simple-client.cc
+++ b/hbase-native-client/core/simple-client.cc
@@ -58,24 +58,23 @@ int main(int argc, char *argv[]) {
   auto conn = cf.make_connection(result.host_name(), result.port());
 
   // Send the request
-  Request r;
+  auto r = Request::get();
 
   // This is a get request so make that
-  auto msg = make_shared<hbase::pb::GetRequest>();
+  auto req_msg = static_pointer_cast<hbase::pb::GetRequest>(r->req_msg());
 
   // Set what region
-  msg->mutable_region()->set_value(FLAGS_region);
+  req_msg->mutable_region()->set_value(FLAGS_region);
   // It's always this.
-  msg->mutable_region()->set_type(
+  req_msg->mutable_region()->set_type(
       RegionSpecifier_RegionSpecifierType::
           RegionSpecifier_RegionSpecifierType_ENCODED_REGION_NAME);
 
   // What row.
-  msg->mutable_get()->set_row(FLAGS_row);
+  req_msg->mutable_get()->set_row(FLAGS_row);
 
   // Send it.
-  r.set_msg(msg);
-  auto resp = (*conn)(r).get(milliseconds(5000));
+  auto resp = (*conn)(std::move(r)).get(milliseconds(5000));
 
   auto get_resp = std::static_pointer_cast<GetResponse>(resp.response());
   cout << "GetResponse has_result = " << get_resp->has_result() << '\n';


Mime
View raw message