Implement thrift brpc transport.

example/thrift_extension_c++/Makefile

@@ -3,7 +3,6 @@ include $(BRPC_PATH)/config.mk
 # Notes on the flags:
 # 1. Added -fno-omit-frame-pointer: perf/tcmalloc-profiler use frame pointers by default
 # 2. Added -D__const__= : Avoid over-optimizations of TLS variables by GCC>=4.8
-#CXXFLAGS = -std=c++0x -g -DDEBUG -D__const__= -DBRPC_ENABLE_CPU_PROFILER -pipe -W -Wall -Werror -Wno-unused-parameter -fPIC -fno-omit-frame-pointer
 CXXFLAGS = -std=c++0x -g -DDEBUG -D__const__= -pipe -W -Wall -Werror -Wno-unused-parameter -fPIC -fno-omit-frame-pointer
 HDRS+=$(BRPC_PATH)/output/include
 LIBS+=$(BRPC_PATH)/output/lib
@@ -14,8 +13,8 @@ SOPATHS=$(addprefix -Wl$(COMMA)-rpath=, $(LIBS))
 
 STATIC_LINKINGS += -lbrpc -lthrift -lgflags
 
-CLIENT_SOURCES = client.cpp
-SERVER_SOURCES = server.cpp
+CLIENT_SOURCES = client.cpp thrift_brpc_helper_transport.cpp
+SERVER_SOURCES = server.cpp thrift_brpc_helper_transport.cpp
 PROTOS = $(wildcard *.proto)
 
 PROTO_OBJS = $(PROTOS:.proto=.pb.o)
@@ -24,7 +23,7 @@ CLIENT_OBJS = $(addsuffix .o, $(basename $(CLIENT_SOURCES)))
 SERVER_OBJS = $(addsuffix .o, $(basename $(SERVER_SOURCES))) 
 
 .PHONY:all
-all: echo_client echo_server thrift_server thrift_client libechothrift.a
+all: echo_client echo_server thrift_server thrift_client libechothrift.a client.o server.o
 
 .PHONY:clean
 clean:

example/thrift_extension_c++/client.cpp

@@ -59,26 +59,21 @@ int main(int argc, char* argv[]) {
     // Send a request and wait for the response every 1 second.
     int log_id = 0;
     
-    boost::shared_ptr<BrpcThriftClient<example::EchoServiceClient>> client = 
-        boost::make_shared<BrpcThriftClient<example::EchoServiceClient>>();
+    apache::thrift::transport::TThriftBrpcHelperTransport* transport;
+    auto client = InitThriftClient<example::EchoServiceClient>(&channel, &transport);
 
     while (!brpc::IsAskedToQuit()) {
         brpc::Controller cntl;
         cntl.set_log_id(log_id ++);  // set by user
 
+        transport->set_controller(&cntl);
+
         // Thrift Req
         example::EchoRequest thrift_request;
         example::EchoResponse thrift_response;
         thrift_request.data = "hello";
 
-        // util the Thrift client's send_XXX method, actuall do serilize work inside
-        client->get_thrift_client()->send_Echo(thrift_request);
-
-        // do rpc call actually
-        client->call_method(&channel, &cntl);
-
-        // util the Thrift client's recv_XXX method, actuall do deserilize work inside
-        client->get_thrift_client()->recv_Echo(thrift_response);
+        client->Echo(thrift_response, thrift_request);
 
         if (cntl.Failed()) {
             LOG(ERROR) << "Fail to send thrift request, " << cntl.ErrorText();

example/thrift_extension_c++/thrift_brpc_helper_transport.cpp

+/*
+ * 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 <cassert>
+#include <algorithm>
+
+#include "thrift_brpc_helper_transport.h"
+#include <brpc/thrift_binary_message.h>
+
+using std::string;
+
+namespace apache {
+namespace thrift {
+namespace transport {
+
+void TThriftBrpcHelperTransport::computeRead(uint32_t len, uint8_t** out_start, uint32_t* out_give) {
+  // Correct rBound_ so we can use the fast path in the future.
+  rBound_ = wBase_;
+
+  // Decide how much to give.
+  uint32_t give = (std::min)(len, available_read());
+
+  *out_start = rBase_;
+  *out_give = give;
+
+  // Preincrement rBase_ so the caller doesn't have to.
+  rBase_ += give;
+}
+
+uint32_t TThriftBrpcHelperTransport::readSlow(uint8_t* buf, uint32_t len) {
+  uint8_t* start;
+  uint32_t give;
+  computeRead(len, &start, &give);
+
+  // Copy into the provided buffer.
+  memcpy(buf, start, give);
+
+  return give;
+}
+
+uint32_t TThriftBrpcHelperTransport::readAppendToString(std::string& str, uint32_t len) {
+  // Don't get some stupid assertion failure.
+  if (buffer_ == NULL) {
+    return 0;
+  }
+
+  uint8_t* start;
+  uint32_t give;
+  computeRead(len, &start, &give);
+
+  // Append to the provided string.
+  str.append((char*)start, give);
+
+  return give;
+}
+
+void TThriftBrpcHelperTransport::ensureCanWrite(uint32_t len) {
+  // Check available space
+  uint32_t avail = available_write();
+  if (len <= avail) {
+    return;
+  }
+
+  if (!owner_) {
+    throw TTransportException("Insufficient space in external TThriftBrpcHelperTransport memory");
+  }
+
+  // Grow the buffer as necessary.
+  uint32_t new_size = bufferSize_;
+  while (len > avail) {
+    new_size = new_size > 0 ? new_size * 2 : 1;
+    avail = available_write() + (new_size - bufferSize_);
+  }
+
+  // Allocate into a new pointer so we don't bork ours if it fails.
+  uint8_t* new_buffer = static_cast<uint8_t*>(std::realloc(buffer_, new_size));
+  if (new_buffer == NULL) {
+    throw std::bad_alloc();
+  }
+
+  rBase_ = new_buffer + (rBase_ - buffer_);
+  rBound_ = new_buffer + (rBound_ - buffer_);
+  wBase_ = new_buffer + (wBase_ - buffer_);
+  wBound_ = new_buffer + new_size;
+  buffer_ = new_buffer;
+  bufferSize_ = new_size;
+}
+
+void TThriftBrpcHelperTransport::writeSlow(const uint8_t* buf, uint32_t len) {
+  ensureCanWrite(len);
+
+  // Copy into the buffer and increment wBase_.
+  memcpy(wBase_, buf, len);
+  wBase_ += len;
+}
+
+void TThriftBrpcHelperTransport::wroteBytes(uint32_t len) {
+  uint32_t avail = available_write();
+  if (len > avail) {
+    throw TTransportException("Client wrote more bytes than size of buffer.");
+  }
+  wBase_ += len;
+}
+
+const uint8_t* TThriftBrpcHelperTransport::borrowSlow(uint8_t* buf, uint32_t* len) {
+  (void)buf;
+  rBound_ = wBase_;
+  if (available_read() >= *len) {
+    *len = available_read();
+    return rBase_;
+  }
+  return NULL;
+}
+
+// return number of bytes read
+uint32_t TThriftBrpcHelperTransport::readEnd() {
+  // This cast should be safe, because buffer_'s size is a uint32_t
+  uint32_t bytes = static_cast<uint32_t>(rBase_ - buffer_);
+  if (rBase_ == wBase_) {
+    resetBuffer();
+  }
+  return bytes;
+}
+
+// Return number of bytes written
+uint32_t TThriftBrpcHelperTransport::writeEnd() {
+  // This cast should be safe, because buffer_'s size is a uint32_t
+
+  brpc::ThriftBinaryMessage request;
+  brpc::ThriftBinaryMessage response;
+  
+  butil::IOBuf buf;
+  buf.append(this->getBufferAsString());
+  request.body = buf;    
+  
+  // send the request the server
+  // Because `done'(last parameter) is NULL, this function waits until
+  // the response comes back or error occurs(including timedout).
+  channel_->CallMethod(NULL, cntl_, &request, &response, NULL);
+  if (!cntl_->Failed()) {
+      size_t body_len  = response.head.body_len;  
+      uint8_t* thrift_buffer = (uint8_t*)malloc(body_len);
+
+      const size_t k = response.body.copy_to(thrift_buffer, body_len);
+      if ( k != body_len) {
+          cntl_->SetFailed("copy response body to thrift buffer failed!");
+          free(thrift_buffer);
+          return 0;
+      }
+      this->resetBuffer(thrift_buffer, body_len, TAKE_OWNERSHIP);          
+  }
+  
+  return static_cast<uint32_t>(wBase_ - buffer_);
+}
+
+
+}
+}
+} // apache::thrift::transport
+

example/thrift_extension_c++/thrift_client.cpp

@@ -51,7 +51,7 @@ int main(int argc, char **argv) {
 
         LOG(INFO) << "Req: " << req.data
                   << "Res: " << res.data;
-        sleep(1);
+        //sleep(1);
     }
     transport->close();
 

example/thrift_extension_c++/thrift_utils.h

@@ -18,67 +18,26 @@
 
 #include <boost/make_shared.hpp>
 
+#include "thrift_brpc_helper_transport.h"
+
 #include <thrift/transport/TBufferTransports.h>
 #include <thrift/protocol/TBinaryProtocol.h>
 
-template <class T>
-class BrpcThriftClient {
-
-public:
-  BrpcThriftClient() {
-
-    out_buffer_ = boost::make_shared<apache::thrift::transport::TMemoryBuffer>();
-    out_ = boost::make_shared<apache::thrift::protocol::TBinaryProtocol>(out_buffer_);
-
-    in_buffer_ = boost::make_shared<apache::thrift::transport::TMemoryBuffer>();
-    in_ = boost::make_shared<apache::thrift::protocol::TBinaryProtocol>(in_buffer_);
-
-    client_ = boost::make_shared<T>(in_, out_);
-  }
-
-  boost::shared_ptr<T> get_thrift_client() {
-    return client_;
-  }
-
- void call_method(brpc::Channel* channel, brpc::Controller* cntl) {
-
-      brpc::ThriftBinaryMessage request;
-      brpc::ThriftBinaryMessage response;
-
-      in_buffer_->resetBuffer();
-
-      butil::IOBuf buf;
-      buf.append(out_buffer_->getBufferAsString());
-      request.body = buf;
 
-      // send the request the server
-      // Because `done'(last parameter) is NULL, this function waits until
-      // the response comes back or error occurs(including timedout).
-      channel->CallMethod(NULL, cntl, &request, &response, NULL);
-      if (!cntl->Failed()) {
-          size_t body_len  = response.head.body_len;
-          uint8_t* thrift_buffer = (uint8_t*)malloc(body_len);
-          const size_t k = response.body.copy_to(thrift_buffer, body_len);
-          if ( k != body_len) {
-              free(thrift_buffer);
-              cntl->SetFailed("copy response buf failed!");
-              return;
-          }
-          in_buffer_->resetBuffer(thrift_buffer, body_len);
-      }
-      free(thrift_buffer);
-      return;
-  }
+template <typename T>
+boost::shared_ptr<T> InitThriftClient(brpc::Channel* channel,
+    apache::thrift::transport::TThriftBrpcHelperTransport** transport) {
+    auto thrift_brpc_transport = 
+        boost::make_shared<apache::thrift::transport::TThriftBrpcHelperTransport>();
 
-private:
-    boost::shared_ptr<T> client_;
-    boost::shared_ptr<apache::thrift::transport::TMemoryBuffer> out_buffer_;
-    boost::shared_ptr<apache::thrift::transport::TMemoryBuffer> in_buffer_;
-    boost::shared_ptr<apache::thrift::protocol::TBinaryProtocol> in_;
-    boost::shared_ptr<apache::thrift::protocol::TBinaryProtocol> out_;
+    thrift_brpc_transport->set_channel(channel);
+    *transport = thrift_brpc_transport.get();
 
-};
+    auto out = boost::make_shared<apache::thrift::protocol::TBinaryProtocol>(thrift_brpc_transport);
+    auto in = boost::make_shared<apache::thrift::protocol::TBinaryProtocol>(thrift_brpc_transport);
 
+    return boost::make_shared<T>(in, out);
+}
 
 bool brpc_thrift_server_helper(const brpc::ThriftBinaryMessage& request,
                               brpc::ThriftBinaryMessage* response,
@@ -95,8 +54,8 @@ bool brpc_thrift_server_helper(const brpc::ThriftBinaryMessage& request,
 
     // Cut the thrift buffer and parse thrift message
     size_t body_len  = request.head.body_len;
+    //std::shared_ptr<uint8_t> thrift_buffer(new uint8_t[10],std::default_delete<uint8_t[]>()); 
     uint8_t* thrift_buffer = (uint8_t*)malloc(body_len);
-
     const size_t k = request.body.copy_to(thrift_buffer, body_len);
     if ( k != body_len) {
         free(thrift_buffer);
@@ -106,16 +65,14 @@ bool brpc_thrift_server_helper(const brpc::ThriftBinaryMessage& request,
     in_buffer->resetBuffer(thrift_buffer, body_len);
 
     if (processor->process(in, out, NULL)) {
-        butil::IOBuf buf;
-        std::string s = out_buffer->getBufferAsString();
-        buf.append(s);
-        response->body = buf;
+        response->body.append(out_buffer->getBufferAsString());
     } else {
+    
         free(thrift_buffer);
         return false;
     }
-
     free(thrift_buffer);
+
     return true;
 }