commit e339f6feecf943c282ccc5efe4500a7d3d4b660a
author murgatroid99 <mlumish@google.com> Wed Sep 23 11:38:39 2015 -0700
committer murgatroid99 <mlumish@google.com> Wed Sep 23 11:38:39 2015 -0700
tree 65fe75ac9e851be834107cad08b4d4f72063a552
parent 1965810ede1dae0aaf72d55e1359e640d08720f8

Fixed hang when using Node gRPC with other async operations

src/node/ext/completion_queue_async_worker.cc

diff --git a/src/node/ext/completion_queue_async_worker.cc b/src/node/ext/completion_queue_async_worker.cc
index bf2cd94..efc611b 100644
--- a/src/node/ext/completion_queue_async_worker.cc
+++ b/src/node/ext/completion_queue_async_worker.cc
@@ -53,6 +53,9 @@
 
 grpc_completion_queue *CompletionQueueAsyncWorker::queue;
 
+// Invariants: current_threads <= max_queue_threads
+// (current_threads == max_queue_threads) || (waiting_next_calls == 0)
+
 int CompletionQueueAsyncWorker::current_threads;
 int CompletionQueueAsyncWorker::waiting_next_calls;
 
@@ -74,11 +77,15 @@
 void CompletionQueueAsyncWorker::Next() {
   NanScope();
   if (current_threads < max_queue_threads) {
+    current_threads += 1;
     CompletionQueueAsyncWorker *worker = new CompletionQueueAsyncWorker();
     NanAsyncQueueWorker(worker);
   } else {
     waiting_next_calls += 1;
   }
+  GPR_ASSERT(current_threads <= max_queue_threads);
+  GPR_ASSERT((current_threads == max_queue_threads) ||
+             (waiting_next_calls == 0));
 }
 
 void CompletionQueueAsyncWorker::Init(Handle<Object> exports) {
@@ -92,11 +99,15 @@
   NanScope();
   if (waiting_next_calls > 0) {
     waiting_next_calls -= 1;
+    // Old worker removed, new worker added. current_threads += 0
     CompletionQueueAsyncWorker *worker = new CompletionQueueAsyncWorker();
     NanAsyncQueueWorker(worker);
   } else {
     current_threads -= 1;
   }
+  GPR_ASSERT(current_threads <= max_queue_threads);
+  GPR_ASSERT((current_threads == max_queue_threads) ||
+             (waiting_next_calls == 0));
   NanCallback *callback = GetTagCallback(result.tag);
   Handle<Value> argv[] = {NanNull(), GetTagNodeValue(result.tag)};
   callback->Call(2, argv);
@@ -106,6 +117,17 @@
 
 void CompletionQueueAsyncWorker::HandleErrorCallback() {
   NanScope();
+  if (waiting_next_calls > 0) {
+    waiting_next_calls -= 1;
+    // Old worker removed, new worker added. current_threads += 0
+    CompletionQueueAsyncWorker *worker = new CompletionQueueAsyncWorker();
+    NanAsyncQueueWorker(worker);
+  } else {
+    current_threads -= 1;
+  }
+  GPR_ASSERT(current_threads <= max_queue_threads);
+  GPR_ASSERT((current_threads == max_queue_threads) ||
+             (waiting_next_calls == 0));
   NanCallback *callback = GetTagCallback(result.tag);
   Handle<Value> argv[] = {NanError(ErrorMessage())};