Merge pull request #11849 from apolcyn/fix_ruby_md_mem_leaks_master

Fix memory leak in sent ruby metadata

src/ruby/ext/grpc/rb_call.c

@@ -24,6 +24,8 @@
 #include <grpc/grpc.h>
 #include <grpc/impl/codegen/compression_types.h>
 #include <grpc/support/alloc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
 
 #include "rb_byte_buffer.h"
 #include "rb_call_credentials.h"
@@ -368,7 +370,7 @@ static VALUE grpc_rb_call_set_credentials(VALUE self, VALUE credentials) {
    to fill grpc_metadata_array.
 
    it's capacity should have been computed via a prior call to
-   grpc_rb_md_ary_fill_hash_cb
+   grpc_rb_md_ary_capacity_hash_cb
 */
 static int grpc_rb_md_ary_fill_hash_cb(VALUE key, VALUE val, VALUE md_ary_obj) {
   grpc_metadata_array *md_ary = NULL;
@@ -376,7 +378,7 @@ static int grpc_rb_md_ary_fill_hash_cb(VALUE key, VALUE val, VALUE md_ary_obj) {
   long i;
   grpc_slice key_slice;
   grpc_slice value_slice;
-  char *tmp_str;
+  char *tmp_str = NULL;
 
   if (TYPE(key) == T_SYMBOL) {
     key_slice = grpc_slice_from_static_string(rb_id2name(SYM2ID(key)));
@@ -386,6 +388,7 @@ static int grpc_rb_md_ary_fill_hash_cb(VALUE key, VALUE val, VALUE md_ary_obj) {
   } else {
     rb_raise(rb_eTypeError,
              "grpc_rb_md_ary_fill_hash_cb: bad type for key parameter");
+    return ST_STOP;
   }
 
   if (!grpc_header_key_is_legal(key_slice)) {
@@ -413,6 +416,7 @@ static int grpc_rb_md_ary_fill_hash_cb(VALUE key, VALUE val, VALUE md_ary_obj) {
                  tmp_str);
         return ST_STOP;
       }
+      GPR_ASSERT(md_ary->count < md_ary->capacity);
       md_ary->metadata[md_ary->count].key = key_slice;
       md_ary->metadata[md_ary->count].value = value_slice;
       md_ary->count += 1;
@@ -428,6 +432,7 @@ static int grpc_rb_md_ary_fill_hash_cb(VALUE key, VALUE val, VALUE md_ary_obj) {
                tmp_str);
       return ST_STOP;
     }
+    GPR_ASSERT(md_ary->count < md_ary->capacity);
     md_ary->metadata[md_ary->count].key = key_slice;
     md_ary->metadata[md_ary->count].value = value_slice;
     md_ary->count += 1;
@@ -435,7 +440,6 @@ static int grpc_rb_md_ary_fill_hash_cb(VALUE key, VALUE val, VALUE md_ary_obj) {
     rb_raise(rb_eArgError, "Header values must be of type string or array");
     return ST_STOP;
   }
-
   return ST_CONTINUE;
 }
 
@@ -458,6 +462,7 @@ static int grpc_rb_md_ary_capacity_hash_cb(VALUE key, VALUE val,
   } else {
     md_ary->capacity += 1;
   }
+
   return ST_CONTINUE;
 }
 
@@ -480,7 +485,7 @@ void grpc_rb_md_ary_convert(VALUE md_ary_hash, grpc_metadata_array *md_ary) {
   md_ary_obj =
       TypedData_Wrap_Struct(grpc_rb_cMdAry, &grpc_rb_md_ary_data_type, md_ary);
   rb_hash_foreach(md_ary_hash, grpc_rb_md_ary_capacity_hash_cb, md_ary_obj);
-  md_ary->metadata = gpr_malloc(md_ary->capacity * sizeof(grpc_metadata));
+  md_ary->metadata = gpr_zalloc(md_ary->capacity * sizeof(grpc_metadata));
   rb_hash_foreach(md_ary_hash, grpc_rb_md_ary_fill_hash_cb, md_ary_obj);
 }
 
@@ -611,13 +616,25 @@ static void grpc_run_batch_stack_init(run_batch_stack *st,
   st->write_flag = write_flag;
 }
 
+void grpc_rb_metadata_array_destroy_including_entries(
+    grpc_metadata_array *array) {
+  size_t i;
+  if (array->metadata) {
+    for (i = 0; i < array->count; i++) {
+      grpc_slice_unref(array->metadata[i].key);
+      grpc_slice_unref(array->metadata[i].value);
+    }
+  }
+  grpc_metadata_array_destroy(array);
+}
+
 /* grpc_run_batch_stack_cleanup ensures the run_batch_stack is properly
  * cleaned up */
 static void grpc_run_batch_stack_cleanup(run_batch_stack *st) {
   size_t i = 0;
 
-  grpc_metadata_array_destroy(&st->send_metadata);
+  grpc_rb_metadata_array_destroy_including_entries(&st->send_metadata);
-  grpc_metadata_array_destroy(&st->send_trailing_metadata);
+  grpc_rb_metadata_array_destroy_including_entries(&st->send_trailing_metadata);
   grpc_metadata_array_destroy(&st->recv_metadata);
   grpc_metadata_array_destroy(&st->recv_trailing_metadata);
 
@@ -658,8 +675,6 @@ static void grpc_run_batch_stack_fill_ops(run_batch_stack *st, VALUE ops_hash) {
     st->ops[st->op_num].flags = 0;
     switch (NUM2INT(this_op)) {
       case GRPC_OP_SEND_INITIAL_METADATA:
-        /* N.B. later there is no need to explicitly delete the metadata keys
-         * and values, they are references to data in ruby objects. */
         grpc_rb_md_ary_convert(this_value, &st->send_metadata);
         st->ops[st->op_num].data.send_initial_metadata.count =
             st->send_metadata.count;
@@ -675,8 +690,6 @@ static void grpc_run_batch_stack_fill_ops(run_batch_stack *st, VALUE ops_hash) {
       case GRPC_OP_SEND_CLOSE_FROM_CLIENT:
         break;
       case GRPC_OP_SEND_STATUS_FROM_SERVER:
-        /* N.B. later there is no need to explicitly delete the metadata keys
-         * and values, they are references to data in ruby objects. */
         grpc_rb_op_update_status_from_server(
             &st->ops[st->op_num], &st->send_trailing_metadata,
             &st->send_status_details, this_value);

src/ruby/ext/grpc/rb_call.h

@@ -40,6 +40,9 @@ VALUE grpc_rb_md_ary_to_h(grpc_metadata_array *md_ary);
 */
 void grpc_rb_md_ary_convert(VALUE md_ary_hash, grpc_metadata_array *md_ary);
 
+void grpc_rb_metadata_array_destroy_including_entries(
+    grpc_metadata_array *md_ary);
+
 /* grpc_rb_eCallError is the ruby class of the exception thrown during call
    operations. */
 extern VALUE grpc_rb_eCallError;

src/ruby/ext/grpc/rb_call_credentials.c

@@ -108,7 +108,7 @@ static void grpc_rb_call_credentials_callback_with_gil(void *param) {
   error_details = StringValueCStr(details);
   params->callback(params->user_data, md_ary.metadata, md_ary.count, status,
                    error_details);
-  grpc_metadata_array_destroy(&md_ary);
+  grpc_rb_metadata_array_destroy_including_entries(&md_ary);
   gpr_free(params);
 }
 

src/ruby/spec/client_server_spec.rb

@@ -448,11 +448,18 @@ describe 'the secure http client/server' do
   it_behaves_like 'GRPC metadata delivery works OK' do
   end
 
-  it 'modifies metadata with CallCredentials' do
+  def credentials_update_test(creds_update_md)
-    auth_proc = proc { { 'k1' => 'updated-v1' } }
+    auth_proc = proc { creds_update_md }
     call_creds = GRPC::Core::CallCredentials.new(auth_proc)
-    md = { 'k2' => 'v2' }
+
-    expected_md = { 'k1' => 'updated-v1', 'k2' => 'v2' }
+    initial_md_key = 'k2'
+    initial_md_val = 'v2'
+    initial_md = {}
+    initial_md[initial_md_key] = initial_md_val
+    expected_md = creds_update_md.clone
+    fail 'bad test param' unless expected_md[initial_md_key].nil?
+    expected_md[initial_md_key] = initial_md_val
+
     recvd_rpc = nil
     rcv_thread = Thread.new do
       recvd_rpc = @server.request_call
@@ -461,7 +468,7 @@ describe 'the secure http client/server' do
     call = new_client_call
     call.set_credentials! call_creds
     client_ops = {
-      CallOps::SEND_INITIAL_METADATA => md
+      CallOps::SEND_INITIAL_METADATA => initial_md
     }
     batch_result = call.run_batch(client_ops)
     expect(batch_result.send_metadata).to be true
@@ -473,4 +480,21 @@ describe 'the secure http client/server' do
     replace_symbols = Hash[expected_md.each_pair.collect { |x, y| [x.to_s, y] }]
     expect(recvd_md).to eq(recvd_md.merge(replace_symbols))
   end
+
+  it 'modifies metadata with CallCredentials' do
+    credentials_update_test('k1' => 'updated-v1')
+  end
+
+  it 'modifies large metadata with CallCredentials' do
+    val_array = %w(
+      '00000000000000000000000000000000000000000000000000000000000000',
+      '11111111111111111111111111111111111111111111111111111111111111',
+    )
+    md = {
+      k3: val_array,
+      k4: '0000000000000000000000000000000000000000000000000000000000',
+      keeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeey5: 'v1'
+    }
+    credentials_update_test(md)
+  end
 end

src/ruby/spec/generic/client_stub_spec.rb

@@ -170,16 +170,42 @@ describe 'ClientStub' do
         th.join
       end
 
-      it 'should send metadata to the server ok' do
+      def metadata_test(md)
         server_port = create_test_server
         host = "localhost:#{server_port}"
         th = run_request_response(@sent_msg, @resp, @pass,
-                                  expected_metadata: { k1: 'v1', k2: 'v2' })
+                                  expected_metadata: md)
         stub = GRPC::ClientStub.new(host, :this_channel_is_insecure)
+        @metadata = md
         expect(get_response(stub)).to eq(@resp)
         th.join
       end
 
+      it 'should send metadata to the server ok' do
+        metadata_test(k1: 'v1', k2: 'v2')
+      end
+
+      # these tests mostly try to exercise when md might be allocated
+      # instead of inlined
+      it 'should send metadata with multiple large md to the server ok' do
+        val_array = %w(
+          '00000000000000000000000000000000000000000000000000000000000000',
+          '11111111111111111111111111111111111111111111111111111111111111',
+          '22222222222222222222222222222222222222222222222222222222222222',
+        )
+        md = {
+          k1: val_array,
+          k2: 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa',
+          k3: 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb',
+          k4: 'cccccccccccccccccccccccccccccccccccccccccccccccccccccccccc',
+          keeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeey5: 'v5',
+          'k66666666666666666666666666666666666666666666666666666' => 'v6',
+          'k77777777777777777777777777777777777777777777777777777' => 'v7',
+          'k88888888888888888888888888888888888888888888888888888' => 'v8'
+        }
+        metadata_test(md)
+      end
+
       it 'should send a request when configured using an override channel' do
         server_port = create_test_server
         alt_host = "localhost:#{server_port}"