Merge pull request #8843 from apolcyn/fix_ruby_shutdown

Fix ruby shutdown

src/ruby/ext/grpc/rb_server.c

@@ -37,6 +37,7 @@
 #include "rb_server.h"
 
 #include <grpc/grpc.h>
+#include <grpc/support/atm.h>
 #include <grpc/grpc_security.h>
 #include <grpc/support/log.h>
 #include "rb_call.h"
@@ -59,10 +60,13 @@ typedef struct grpc_rb_server {
   /* The actual server */
   grpc_server *wrapped;
   grpc_completion_queue *queue;
+  gpr_atm shutdown_started;
 } grpc_rb_server;
 
 static void destroy_server(grpc_rb_server *server, gpr_timespec deadline) {
   grpc_event ev;
+  // This can be started by app or implicitly by GC. Avoid a race between these.
+  if (gpr_atm_full_fetch_add(&server->shutdown_started, (gpr_atm)1) == 0) {
     if (server->wrapped != NULL) {
       grpc_server_shutdown_and_notify(server->wrapped, server->queue, NULL);
       ev = rb_completion_queue_pluck(server->queue, NULL, deadline, NULL);
@@ -77,6 +81,7 @@ static void destroy_server(grpc_rb_server *server, gpr_timespec deadline) {
       server->queue = NULL;
     }
   }
+}
 
 /* Destroys server instances. */
 static void grpc_rb_server_free(void *p) {
@@ -115,6 +120,7 @@ static const rb_data_type_t grpc_rb_server_data_type = {
 static VALUE grpc_rb_server_alloc(VALUE cls) {
   grpc_rb_server *wrapper = ALLOC(grpc_rb_server);
   wrapper->wrapped = NULL;
+  wrapper->shutdown_started = (gpr_atm)0;
   return TypedData_Wrap_Struct(cls, &grpc_rb_server_data_type, wrapper);
 }
 

src/ruby/qps/client.rb

@@ -134,6 +134,7 @@ class BenchmarkClient
     resp = stub.streaming_call(q.each_item)
     start = Time.now
     q.push(req)
+    pushed_sentinal = false
     resp.each do |r|
       @histogram.add((Time.now-start)*1e9)
       if !@done
@@ -141,8 +142,9 @@ class BenchmarkClient
         start = Time.now
         q.push(req)
       else
-        q.push(self)
-        break
+        q.push(self) unless pushed_sentinal
+	# Continue polling on the responses to consume and release resources
+        pushed_sentinal = true
       end
     end
   end