stub: ensure BlockingClientCall tasks run after cancellation

After canceling a `BlockingClientCall`, the caller may not interact with it further. That means the call executor, a `ThreadSafeThreadlessExecutor`, will not execute any more tasks. There may still be tasks submitted to the call executor, though, until the underlying call completes. Some of these tasks (e.g., server messages available) may leak native resources unless executed. So, we convert the executor to a "direct" executor during cancellation in order to ensure all call tasks run.

Fixes grpc#12355.

Author: benjaminp

stub/src/main/java/io/grpc/stub/BlockingClientCall.java

@@ -249,6 +249,13 @@ void sendSingleRequest(ReqT request) {
   public void cancel(String message, Throwable cause) {
     writeClosed = true;
     call.cancel(message, cause);
+    // After canceling a BlockingClientCall, the caller may not interact with it further. That means
+    // the call executor, a ThreadSafeThreadlessExecutor, will not execute any more tasks. There may
+    // still be tasks submitted to the call executor, though, until the underlying call completes.
+    // Some of these tasks (e.g., server messages available) may leak native resources unless
+    // executed. So, we convert the executor to a "direct" executor in order to ensure all call
+    // tasks run.
+    executor.becomeDirect();
   }
 
   /**

stub/src/main/java/io/grpc/stub/ClientCalls.java

@@ -871,6 +871,7 @@ static final class ThreadSafeThreadlessExecutor extends ConcurrentLinkedQueue<Ru
 
     private final Lock waiterLock = new ReentrantLock();
     private final Condition waiterCondition = waiterLock.newCondition();
+    private boolean direct;
 
     // Non private to avoid synthetic class
     ThreadSafeThreadlessExecutor() {}
@@ -950,6 +951,20 @@ void drain() {
       }
     }
 
+    /**
+     * Turn the executor into a "direct" executor. Further calls to {@link #execute(Runnable)} will
+     * executable the runnable immediately in the calling thread.
+     */
+    void becomeDirect() {
+      waiterLock.lock();
+      try {
+        direct = true;
+      } finally {
+        waiterLock.unlock();
+      }
+      drain();
+    }
+
     private void signalAll() {
       waiterLock.lock();
       try {
@@ -975,13 +990,22 @@ private static void throwIfInterrupted() throws InterruptedException {
 
     @Override
     public void execute(Runnable runnable) {
+      boolean added = false;
       waiterLock.lock();
       try {
-        add(runnable);
-        waiterCondition.signalAll(); // If anything is waiting let it wake up and process this task
+        if (!direct) {
+          added = true;
+          add(runnable);
+          // If anything is waiting, let it wake up and process this task.
+          waiterCondition.signalAll();
+        }
       } finally {
         waiterLock.unlock();
       }
+      if (!added) {
+        runQuietly(runnable);
+        signalAll();
+      }
     }
   }