Revert "stub: Have ClientCalls.ThreadlessExecutor reject Runnables af…

…ter end of RPC (#8847)" (#8933)

This reverts commit bb33657.

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

@@ -39,7 +39,6 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Future;
-import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.locks.LockSupport;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -162,7 +161,6 @@ public static <ReqT, RespT> RespT blockingUnaryCall(
       // Something very bad happened. All bets are off; it may be dangerous to wait for onClose().
       throw cancelThrow(call, e);
     } finally {
-      executor.shutdown();
       if (interrupt) {
         Thread.currentThread().interrupt();
       }
@@ -628,9 +626,6 @@ private Object waitForNext() {
               // Now wait for onClose() to be called, so interceptors can clean up
             }
           }
-          if (next == this || next instanceof StatusRuntimeException) {
-            threadless.shutdown();
-          }
           return next;
         }
       } finally {
@@ -717,10 +712,7 @@ private static final class ThreadlessExecutor extends ConcurrentLinkedQueue<Runn
       implements Executor {
     private static final Logger log = Logger.getLogger(ThreadlessExecutor.class.getName());
 
-    private static final Object SHUTDOWN = new Object(); // sentinel
-
-    // Set to the calling thread while it's parked, SHUTDOWN on RPC completion
-    private volatile Object waiter;
+    private volatile Thread waiter;
 
     // Non private to avoid synthetic class
     ThreadlessExecutor() {}
@@ -744,29 +736,14 @@ public void waitAndDrain() throws InterruptedException {
         }
       }
       do {
-        runQuietly(runnable);
+        try {
+          runnable.run();
+        } catch (Throwable t) {
+          log.log(Level.WARNING, "Runnable threw exception", t);
+        }
       } while ((runnable = poll()) != null);
     }
 
-    /**
-     * Called after final call to {@link #waitAndDrain()}, from same thread.
-     */
-    public void shutdown() {
-      waiter = SHUTDOWN;
-      Runnable runnable;
-      while ((runnable = poll()) != null) {
-        runQuietly(runnable);
-      }
-    }
-
-    private static void runQuietly(Runnable runnable) {
-      try {
-        runnable.run();
-      } catch (Throwable t) {
-        log.log(Level.WARNING, "Runnable threw exception", t);
-      }
-    }
-
     private static void throwIfInterrupted() throws InterruptedException {
       if (Thread.interrupted()) {
         throw new InterruptedException();
@@ -776,12 +753,7 @@ private static void throwIfInterrupted() throws InterruptedException {
     @Override
     public void execute(Runnable runnable) {
       add(runnable);
-      Object waiter = this.waiter;
-      if (waiter != SHUTDOWN) {
-        LockSupport.unpark((Thread) waiter); // no-op if null
-      } else if (remove(runnable)) {
-        throw new RejectedExecutionException();
-      }
+      LockSupport.unpark(waiter); // no-op if null
     }
   }