stub: optimize ThreadlessExecutor used for blocking calls

The `ThreadlessExecutor` currently used for blocking calls uses `LinkedBlockingQueue` which is relatively heavy both in terms of allocations and synchronization overhead (e.g. when compared to `ConcurrentLinkedQueue`). It accounts for ~10% of allocations and ~5% of allocated bytes per-call in the `TransportBenchmark` when using in-process transport with [stats and tracing disabled](#5510).

Changing to use a `ConcurrentLinkedQueue` results in a ~5% speedup of that benchmark.

Before:
```
Benchmark                         (direct)  (transport)  Mode  Cnt      Score     Error  Units
TransportBenchmark.unaryCall1024      true    INPROCESS  avgt   60   1877.339 ±  46.309  ns/op
TransportBenchmark.unaryCall1024     false    INPROCESS  avgt   60  12680.525 ± 208.684  ns/op
```

After:
```
Benchmark                         (direct)  (transport)  Mode  Cnt      Score     Error  Units
TransportBenchmark.unaryCall1024      true    INPROCESS  avgt   60   1779.188 ±  36.769  ns/op
TransportBenchmark.unaryCall1024     false    INPROCESS  avgt   60  12532.470 ± 238.271  ns/op
```

Author: njhill

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

@@ -34,10 +34,11 @@
 import java.util.NoSuchElementException;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.locks.LockSupport;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import javax.annotation.Nullable;
@@ -627,32 +628,54 @@ public void onClose(Status status, Metadata trailers) {
     }
   }
 
-  private static final class ThreadlessExecutor implements Executor {
+  @SuppressWarnings("serial")
+  private static final class ThreadlessExecutor extends ConcurrentLinkedQueue<Runnable>
+      implements Executor {
     private static final Logger log = Logger.getLogger(ThreadlessExecutor.class.getName());
 
-    private final BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
+    private volatile Thread waiter;
 
     // Non private to avoid synthetic class
     ThreadlessExecutor() {}
 
     /**
      * Waits until there is a Runnable, then executes it and all queued Runnables after it.
+     * Must only be called by one thread at a time.
      */
     public void waitAndDrain() throws InterruptedException {
-      Runnable runnable = queue.take();
-      while (runnable != null) {
+      final Thread currentThread = Thread.currentThread();
+      throwIfInterrupted(currentThread);
+      Runnable runnable = poll();
+      if (runnable == null) {
+        waiter = currentThread;
+        try {
+          while ((runnable = poll()) == null) {
+            LockSupport.park(this);
+            throwIfInterrupted(currentThread);
+          }
+        } finally {
+          waiter = null;
+        }
+      }
+      do {
         try {
           runnable.run();
         } catch (Throwable t) {
           log.log(Level.WARNING, "Runnable threw exception", t);
         }
-        runnable = queue.poll();
+      } while ((runnable = poll()) != null);
+    }
+
+    private static void throwIfInterrupted(Thread currentThread) throws InterruptedException {
+      if (currentThread.isInterrupted()) {
+        throw new InterruptedException();
       }
     }
 
     @Override
     public void execute(Runnable runnable) {
-      queue.add(runnable);
+      add(runnable);
+      LockSupport.unpark(waiter); // no-op if null
     }
   }
 }