Merge pull request #110 from fengli79/master

Fix memory leaks in proto/streambody decoder.

Makefile

@@ -2,6 +2,7 @@ OS := $(shell uname)
 CC := g++
 ROOT_DIR := $(shell pwd)
 GRPC_GATEWAY_PROTOS := $(ROOT_DIR)/net/grpc/gateway/protos
+PROTO_INC := $(ROOT_DIR)/third_party/grpc/third_party/protobuf/include
 PROTO_SRC := $(ROOT_DIR)/third_party/grpc/third_party/protobuf/src
 PROTO_LIB := $(PROTO_SRC)/.libs
 PROTOC := $(PROTO_SRC)/protoc
@@ -41,7 +42,7 @@ nginx_config:
 	auto/configure \
 	--with-http_ssl_module \
 	--with-http_v2_module \
-	--with-cc-opt="-I /usr/local/include -I $(ROOT_DIR) -I $(PROTO_SRC) \
+	--with-cc-opt="-I /usr/local/include -I $(ROOT_DIR) -I $(PROTO_INC) -I $(PROTO_SRC) \
 -I $(GRPC_INC) -I $(GRPC_SRC)" \
 	--with-ld-opt="$(NGINX_LD_OPT)" \
 	--with-openssl="$(ROOT_DIR)/third_party/openssl" \
@@ -52,7 +53,7 @@ nginx_config_static:
 	auto/configure \
 	--with-http_ssl_module \
 	--with-http_v2_module \
-	--with-cc-opt="-I /usr/local/include -I $(ROOT_DIR) -I $(PROTO_SRC) \
+	--with-cc-opt="-I /usr/local/include -I $(ROOT_DIR) -I $(PROTO_INC) -I $(PROTO_SRC) \
 -I $(GRPC_INC) -I $(GRPC_SRC)" \
 	--with-ld-opt="$(NGINX_STATIC_LD_OPT)" \
 	--with-openssl="$(ROOT_DIR)/third_party/openssl" \

net/grpc/gateway/backend/grpc_backend.cc

@@ -48,6 +48,10 @@ GrpcBackend::GrpcBackend()
 
 GrpcBackend::~GrpcBackend() {
   BACKEND_DEBUG("Deleting GRPC backend proxy.");
+  for (auto& m : request_initial_metadata_) {
+    grpc_slice_unref(m.key);
+    grpc_slice_unref(m.value);
+  }
   grpc_metadata_array_destroy(&response_initial_metadata_);
   grpc_metadata_array_destroy(&response_trailing_metadata_);
   if (request_buffer_ != nullptr) {
@@ -171,9 +175,9 @@ void GrpcBackend::OnResponseMessage(bool result) {
   grpc_slice slice;
   while (grpc_byte_buffer_reader_next(&reader, &slice)) {
     message->push_back(Slice(slice, Slice::STEAL_REF));
-    grpc_slice_unref(slice);
   }
   grpc_byte_buffer_reader_destroy(&reader);
+  grpc_byte_buffer_destroy(response_buffer_);
   response->set_message(std::move(message));
   frontend()->Send(std::move(response));
 
@@ -234,8 +238,8 @@ void GrpcBackend::Send(std::unique_ptr<Request> request, Tag* on_done) {
         continue;
       }
       grpc_metadata initial_metadata;
-      initial_metadata.key = grpc_slice_intern(
+      initial_metadata.key =
-          grpc_slice_from_copied_string(header.first.c_str()));
+          grpc_slice_from_copied_string(header.first.c_str());
       initial_metadata.value = grpc_slice_from_copied_buffer(
           header.second.data(), header.second.size());
       initial_metadata.flags = 0;

net/grpc/gateway/codec/proto_decoder.cc

@@ -12,8 +12,8 @@ ProtoDecoder::~ProtoDecoder() {}
 
 Status ProtoDecoder::Decode() {
   if (inputs()->empty()) {
-    Slice* slice = new Slice(grpc_empty_slice(), Slice::STEAL_REF);
+    Slice slice(grpc_empty_slice(), Slice::STEAL_REF);
-    ByteBuffer* buffer = new ByteBuffer(slice, 1);
+    ByteBuffer* buffer = new ByteBuffer(&slice, 1);
     results()->push_back(std::unique_ptr<ByteBuffer>(buffer));
   } else {
     ByteBuffer* buffer = new ByteBuffer(inputs()->data(), inputs()->size());

net/grpc/gateway/codec/stream_body_decoder.cc

@@ -49,7 +49,7 @@ Status StreamBodyDecoder::Decode() {
             if (varint_value_ == 0) {
               buffer_.reset(new Slice(grpc_empty_slice(), Slice::STEAL_REF));
               results()->push_back(std::unique_ptr<ByteBuffer>(
-                  new ByteBuffer(buffer_.release(), 1)));
+                  new ByteBuffer(buffer_.get(), 1)));
               state_ = EXPECTING_MESSAGE_KEY_TYPE;
             } else {
               grpc_slice slice = grpc_slice_malloc(varint_value_);

net/grpc/gateway/frontend/nginx_http_frontend.cc

@@ -43,8 +43,10 @@ grpc::gateway::Frontend *get_frontend(ngx_http_request_t *r) {
 void grpc_gateway_request_cleanup_handler(void *data) {
   grpc_gateway_request_context *context =
       static_cast<grpc_gateway_request_context *>(data);
-  static_cast<grpc::gateway::NginxHttpFrontend *>(context->frontend.get())
+  auto *frontend =
-      ->set_http_request(nullptr);
+      static_cast<grpc::gateway::NginxHttpFrontend *>(context->frontend.get());
+  frontend->StopClientLivenessDetection();
+  frontend->set_http_request(nullptr);
   context->frontend.reset();
 }
 

net/grpc/gateway/frontend/nginx_http_frontend.h

@@ -81,6 +81,8 @@ class NginxHttpFrontend : public Frontend {
 
   void OnClientLivenessDetectionEvent(ngx_event_t *event);
 
+  void StopClientLivenessDetection();
+
  private:
   friend void ::continue_read_request_body(ngx_http_request_t *r);
 
@@ -116,8 +118,6 @@ class NginxHttpFrontend : public Frontend {
 
   void WriteToNginxResponse(uint8_t *data, size_t size);
 
-  void StopClientLivenessDetection();
-
   ngx_http_request_t *http_request_;
   std::unique_ptr<Decoder> decoder_;
   std::unique_ptr<Encoder> encoder_;

net/grpc/gateway/runtime/runtime.cc

@@ -4,6 +4,7 @@
 #include <cstdlib>
 #include <cstring>
 
+#include "google/protobuf/stubs/common.h"
 #include "net/grpc/gateway/backend/grpc_backend.h"
 #include "net/grpc/gateway/codec/b64_proto_decoder.h"
 #include "net/grpc/gateway/codec/b64_proto_encoder.h"
@@ -78,6 +79,8 @@ void Runtime::Shutdown() {
     grpc_channel_destroy(entry.second);
   }
   grpc_backend_channels_.clear();
+  grpc_shutdown();
+  google::protobuf::ShutdownProtobufLibrary();
 }
 
 std::shared_ptr<Frontend> Runtime::CreateNginxFrontend(