Merge github.com:grpc/grpc into error

4 files changed, 59 insertions, 44 deletions

diff --git a/src/ruby/ext/grpc/rb_call.c b/src/ruby/ext/grpc/rb_call.c
index 1b06273af9..b436057c16 100644
--- a/src/ruby/ext/grpc/rb_call.c
+++ b/src/ruby/ext/grpc/rb_call.c
@@ -101,30 +101,14 @@ static VALUE sym_message;
 static VALUE sym_status;
 static VALUE sym_cancelled;
 
-/* hash_all_calls is a hash of Call address -> reference count that is used to
- * track the creation and destruction of rb_call instances.
- */
-static VALUE hash_all_calls;
-
 /* Destroys a Call. */
 static void grpc_rb_call_destroy(void *p) {
-  grpc_call *call = NULL;
-  VALUE ref_count = Qnil;
+  grpc_call* call = NULL;
   if (p == NULL) {
     return;
-  };
-  call = (grpc_call *)p;
-
-  ref_count = rb_hash_aref(hash_all_calls, OFFT2NUM((VALUE)call));
-  if (ref_count == Qnil) {
-    return; /* No longer in the hash, so already deleted */
-  } else if (NUM2UINT(ref_count) == 1) {
-    rb_hash_delete(hash_all_calls, OFFT2NUM((VALUE)call));
-    grpc_call_destroy(call);
-  } else {
-    rb_hash_aset(hash_all_calls, OFFT2NUM((VALUE)call),
-                 UINT2NUM(NUM2UINT(ref_count) - 1));
   }
+  call = (grpc_call *)p;
+  grpc_call_destroy(call);
 }
 
 static size_t md_ary_datasize(const void *p) {
@@ -151,7 +135,7 @@ static const rb_data_type_t grpc_rb_md_ary_data_type = {
      * touches a hash object.
      * TODO(yugui) Directly use st_table and call the free function earlier?
      */
-    0,
+     0,
 #endif
 };
 
@@ -163,12 +147,7 @@ static const rb_data_type_t grpc_call_data_type = {
     NULL,
     NULL,
 #ifdef RUBY_TYPED_FREE_IMMEDIATELY
-    /* it is unsafe to specify RUBY_TYPED_FREE_IMMEDIATELY because
-     * grpc_rb_call_destroy
-     * touches a hash object.
-     * TODO(yugui) Directly use st_table and call the free function earlier?
-     */
-    0,
+    RUBY_TYPED_FREE_IMMEDIATELY
 #endif
 };
 
@@ -190,6 +169,11 @@ const char *grpc_call_error_detail_of(grpc_call_error err) {
 static VALUE grpc_rb_call_cancel(VALUE self) {
   grpc_call *call = NULL;
   grpc_call_error err;
+  if (RTYPEDDATA_DATA(self) == NULL) {
+    //This call has been closed
+    return Qnil;
+  }
+
   TypedData_Get_Struct(self, grpc_call, &grpc_call_data_type, call);
   err = grpc_call_cancel(call, NULL);
   if (err != GRPC_CALL_OK) {
@@ -200,11 +184,29 @@ static VALUE grpc_rb_call_cancel(VALUE self) {
   return Qnil;
 }
 
+/* Releases the c-level resources associated with a call
+   Once a call has been closed, no further requests can be
+   processed.
+*/
+static VALUE grpc_rb_call_close(VALUE self) {
+  grpc_call *call = NULL;
+  TypedData_Get_Struct(self, grpc_call, &grpc_call_data_type, call);
+  if(call != NULL) {
+    grpc_call_destroy(call);
+    RTYPEDDATA_DATA(self) = NULL;
+  }
+  return Qnil;
+}
+
 /* Called to obtain the peer that this call is connected to. */
 static VALUE grpc_rb_call_get_peer(VALUE self) {
   VALUE res = Qnil;
   grpc_call *call = NULL;
   char *peer = NULL;
+  if (RTYPEDDATA_DATA(self) == NULL) {
+    rb_raise(grpc_rb_eCallError, "Cannot get peer value on closed call");
+    return Qnil;
+  }
   TypedData_Get_Struct(self, grpc_call, &grpc_call_data_type, call);
   peer = grpc_call_get_peer(call);
   res = rb_str_new2(peer);
@@ -218,6 +220,10 @@ static VALUE grpc_rb_call_get_peer_cert(VALUE self) {
   grpc_call *call = NULL;
   VALUE res = Qnil;
   grpc_auth_context *ctx = NULL;
+  if (RTYPEDDATA_DATA(self) == NULL) {
+    rb_raise(grpc_rb_eCallError, "Cannot get peer cert on closed call");
+    return Qnil;
+  }
   TypedData_Get_Struct(self, grpc_call, &grpc_call_data_type, call);
 
   ctx = grpc_call_auth_context(call);
@@ -323,6 +329,10 @@ static VALUE grpc_rb_call_set_credentials(VALUE self, VALUE credentials) {
   grpc_call *call = NULL;
   grpc_call_credentials *creds;
   grpc_call_error err;
+  if (RTYPEDDATA_DATA(self) == NULL) {
+    rb_raise(grpc_rb_eCallError, "Cannot set credentials of closed call");
+    return Qnil;
+  }
   TypedData_Get_Struct(self, grpc_call, &grpc_call_data_type, call);
   creds = grpc_rb_get_wrapped_call_credentials(credentials);
   err = grpc_call_set_credentials(call, creds);
@@ -731,7 +741,7 @@ static VALUE grpc_run_batch_stack_build_result(run_batch_stack *st) {
    }
    tag = Object.new
    timeout = 10
-   call.start_batch(cqueue, tag, timeout, ops)
+   call.start_batch(cq, tag, timeout, ops)
 
    Start a batch of operations defined in the array ops; when complete, post a
    completion of type 'tag' to the completion queue bound to the call.
@@ -749,6 +759,10 @@ static VALUE grpc_rb_call_run_batch(VALUE self, VALUE cqueue, VALUE tag,
   VALUE result = Qnil;
   VALUE rb_write_flag = rb_ivar_get(self, id_write_flag);
   unsigned write_flag = 0;
+  if (RTYPEDDATA_DATA(self) == NULL) {
+    rb_raise(grpc_rb_eCallError, "Cannot run batch on closed call");
+    return Qnil;
+  }
   TypedData_Get_Struct(self, grpc_call, &grpc_call_data_type, call);
 
   /* Validate the ops args, adding them to a ruby array */
@@ -888,6 +902,7 @@ void Init_grpc_call() {
   /* Add ruby analogues of the Call methods. */
   rb_define_method(grpc_rb_cCall, "run_batch", grpc_rb_call_run_batch, 4);
   rb_define_method(grpc_rb_cCall, "cancel", grpc_rb_call_cancel, 0);
+  rb_define_method(grpc_rb_cCall, "close", grpc_rb_call_close, 0);
   rb_define_method(grpc_rb_cCall, "peer", grpc_rb_call_get_peer, 0);
   rb_define_method(grpc_rb_cCall, "peer_cert", grpc_rb_call_get_peer_cert, 0);
   rb_define_method(grpc_rb_cCall, "status", grpc_rb_call_get_status, 0);
@@ -925,11 +940,6 @@ void Init_grpc_call() {
       "BatchResult", "send_message", "send_metadata", "send_close",
       "send_status", "message", "metadata", "status", "cancelled", NULL);
 
-  /* The hash for reference counting calls, to ensure they can't be destroyed
-   * more than once */
-  hash_all_calls = rb_hash_new();
-  rb_define_const(grpc_rb_cCall, "INTERNAL_ALL_CALLs", hash_all_calls);
-
   Init_grpc_error_codes();
   Init_grpc_op_codes();
   Init_grpc_write_flags();
@@ -944,16 +954,8 @@ grpc_call *grpc_rb_get_wrapped_call(VALUE v) {
 
 /* Obtains the wrapped object for a given call */
 VALUE grpc_rb_wrap_call(grpc_call *c) {
-  VALUE obj = Qnil;
   if (c == NULL) {
     return Qnil;
   }
-  obj = rb_hash_aref(hash_all_calls, OFFT2NUM((VALUE)c));
-  if (obj == Qnil) { /* Not in the hash add it */
-    rb_hash_aset(hash_all_calls, OFFT2NUM((VALUE)c), UINT2NUM(1));
-  } else {
-    rb_hash_aset(hash_all_calls, OFFT2NUM((VALUE)c),
-                 UINT2NUM(NUM2UINT(obj) + 1));
-  }
   return TypedData_Wrap_Struct(grpc_rb_cCall, &grpc_call_data_type, c);
 }
diff --git a/src/ruby/ext/grpc/rb_completion_queue.c b/src/ruby/ext/grpc/rb_completion_queue.c
index b6ddbe88dc..9466402db0 100644
--- a/src/ruby/ext/grpc/rb_completion_queue.c
+++ b/src/ruby/ext/grpc/rb_completion_queue.c
@@ -150,6 +150,14 @@ static rb_data_type_t grpc_rb_completion_queue_data_type = {
 #endif
 };
 
+/* Releases the c-level resources associated with a completion queue */
+static VALUE grpc_rb_completion_queue_close(VALUE self) {
+  grpc_completion_queue* cq = grpc_rb_get_wrapped_completion_queue(self);
+  grpc_rb_completion_queue_destroy(cq);
+  RTYPEDDATA_DATA(self) = NULL;
+  return Qnil;
+}
+
 /* Allocates a completion queue. */
 static VALUE grpc_rb_completion_queue_alloc(VALUE cls) {
   grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
@@ -212,6 +220,11 @@ void Init_grpc_completion_queue() {
      this func, so no separate initialization step is necessary. */
   rb_define_alloc_func(grpc_rb_cCompletionQueue,
                        grpc_rb_completion_queue_alloc);
+
+  /* close: Provides a way to close the underlying file descriptor without
+     waiting for ruby garbage collection. */
+  rb_define_method(grpc_rb_cCompletionQueue, "close",
+                   grpc_rb_completion_queue_close, 0);
 }
 
 /* Gets the wrapped completion queue from the ruby wrapper */
diff --git a/src/ruby/ext/grpc/rb_grpc.c b/src/ruby/ext/grpc/rb_grpc.c
index 06a07ac646..9246893f9f 100644
--- a/src/ruby/ext/grpc/rb_grpc.c
+++ b/src/ruby/ext/grpc/rb_grpc.c
@@ -318,7 +318,7 @@ void Init_grpc_c() {
   grpc_rb_mGrpcCore = rb_define_module_under(grpc_rb_mGRPC, "Core");
   grpc_rb_sNewServerRpc =
       rb_struct_define("NewServerRpc", "method", "host",
-                       "deadline", "metadata", "call", NULL);
+                       "deadline", "metadata", "call", "cq", NULL);
   grpc_rb_sStatus =
       rb_struct_define("Status", "code", "details", "metadata", NULL);
   sym_code = ID2SYM(rb_intern("code"));
diff --git a/src/ruby/ext/grpc/rb_server.c b/src/ruby/ext/grpc/rb_server.c
index 0899feb685..f108b8acfc 100644
--- a/src/ruby/ext/grpc/rb_server.c
+++ b/src/ruby/ext/grpc/rb_server.c
@@ -234,7 +234,7 @@ static VALUE grpc_rb_server_request_call(VALUE self, VALUE cqueue,
     err = grpc_server_request_call(
         s->wrapped, &call, &st.details, &st.md_ary,
         grpc_rb_get_wrapped_completion_queue(cqueue),
-        grpc_rb_get_wrapped_completion_queue(cqueue),
+        grpc_rb_get_wrapped_completion_queue(s->mark),
         ROBJECT(tag_new));
     if (err != GRPC_CALL_OK) {
       grpc_request_call_stack_cleanup(&st);
@@ -244,7 +244,7 @@ static VALUE grpc_rb_server_request_call(VALUE self, VALUE cqueue,
       return Qnil;
     }
 
-    ev = grpc_rb_completion_queue_pluck_event(cqueue, tag_new, timeout);
+    ev = grpc_rb_completion_queue_pluck_event(s->mark, tag_new, timeout);
     if (ev.type == GRPC_QUEUE_TIMEOUT) {
       grpc_request_call_stack_cleanup(&st);
       return Qnil;
@@ -262,7 +262,7 @@ static VALUE grpc_rb_server_request_call(VALUE self, VALUE cqueue,
         rb_str_new2(st.details.host),
         rb_funcall(rb_cTime, id_at, 2, INT2NUM(deadline.tv_sec),
                    INT2NUM(deadline.tv_nsec)),
-        grpc_rb_md_ary_to_h(&st.md_ary), grpc_rb_wrap_call(call), NULL);
+        grpc_rb_md_ary_to_h(&st.md_ary), grpc_rb_wrap_call(call), cqueue, NULL);
     grpc_request_call_stack_cleanup(&st);
     return result;
   }