Limited concurrency/pausable executors (#4459)

* Prototype: limited pausable executors

* Update

* Add copyrights

* fix race

* Apply suggestions from code review

Co-authored-by: Rodrigo Lazo <[email protected]>

* Address review comments.

* fix

---------

Co-authored-by: Rodrigo Lazo <[email protected]>

Author: vkryachko

firebase-common/src/main/java/com/google/firebase/concurrent/ExecutorsRegistrar.java

@@ -64,7 +64,7 @@ public class ExecutorsRegistrar implements ComponentRegistrar {
                           Process.THREAD_PRIORITY_BACKGROUND
                               + Process.THREAD_PRIORITY_LESS_FAVORABLE))));
 
-  private static final Lazy<ScheduledExecutorService> SCHEDULER =
+  static final Lazy<ScheduledExecutorService> SCHEDULER =
       new Lazy<>(
           () ->
               Executors.newSingleThreadScheduledExecutor(

firebase-common/src/main/java/com/google/firebase/concurrent/FirebaseExecutors.java

@@ -15,6 +15,8 @@
 package com.google.firebase.concurrent;
 
 import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
 
 /** Provides commonly useful executors. */
 public class FirebaseExecutors {
@@ -33,6 +35,94 @@ public static Executor newSequentialExecutor(Executor delegate) {
     return new SequentialExecutor(delegate);
   }
 
+  /**
+   * Returns an {@link Executor} that limits the number of running tasks at a given time.
+   *
+   * <p>The executor uses the {@code delegate} in order to {@link Executor#execute(Runnable)
+   * execute} each task, and does not create any threads of its own.
+   *
+   * @param delegate {@link Executor} used to execute tasks
+   * @param concurrency max number of tasks executing concurrently
+   */
+  public static Executor newLimitedConcurrencyExecutor(Executor delegate, int concurrency) {
+    return new LimitedConcurrencyExecutor(delegate, concurrency);
+  }
+
+  /**
+   * Returns a {@link ExecutorService} that limits the number of running tasks at a given time.
+   *
+   * <p>The executor uses delegate in order to {@link Executor#execute(Runnable) execute} each task,
+   * and does not create any threads of its own.
+   *
+   * @param delegate {@link ExecutorService} used to execute tasks
+   * @param concurrency max number of tasks executing concurrently
+   */
+  public static ExecutorService newLimitedConcurrencyExecutorService(
+      ExecutorService delegate, int concurrency) {
+    return new LimitedConcurrencyExecutorService(delegate, concurrency);
+  }
+
+  /**
+   * Returns a {@link ScheduledExecutorService} that limits the number of running tasks at a given
+   * time.
+   *
+   * <p>The executor uses delegate in order to {@link Executor#execute(Runnable) execute} each task,
+   * and does not create any threads of its own.
+   *
+   * @param delegate {@link ExecutorService} used to execute tasks
+   * @param concurrency max number of tasks executing concurrently
+   */
+  public static ScheduledExecutorService newLimitedConcurrencyScheduledExecutorService(
+      ExecutorService delegate, int concurrency) {
+    return new DelegatingScheduledExecutorService(
+        newLimitedConcurrencyExecutorService(delegate, concurrency),
+        ExecutorsRegistrar.SCHEDULER.get());
+  }
+
+  /**
+   * Returns a {@link PausableExecutor }.
+   *
+   * <p>The executor does not create any threads of its own and instead delegates to the {@code
+   * delegate} executor.
+   *
+   * <p>While {@link PausableExecutor#pause() paused}, the executor queues tasks which are executed
+   * when the executor is {@link PausableExecutor#resume() resumed}, tasks that are already being
+   * executed will not be paused and will run to completion.
+   */
+  public static PausableExecutor newPausableExecutor(Executor delegate) {
+    return new PausableExecutorImpl(false, delegate);
+  }
+
+  /**
+   * Returns a {@link PausableExecutorService }.
+   *
+   * <p>The executor does not create any threads of its own and instead delegates to the {@code
+   * delegate} executor.
+   *
+   * <p>While {@link PausableExecutor#pause() paused}, the executor queues tasks which are executed
+   * when the executor is {@link PausableExecutor#resume() resumed}, tasks that are already being
+   * executed will not be paused and will run to completion.
+   */
+  public static PausableExecutorService newPausableExecutorService(ExecutorService delegate) {
+    return new PausableExecutorServiceImpl(false, delegate);
+  }
+
+  /**
+   * Returns a {@link PausableScheduledExecutorService }.
+   *
+   * <p>The executor does not create any threads of its own and instead delegates to the {@code
+   * delegate} executor.
+   *
+   * <p>While {@link PausableExecutor#pause() paused}, the executor queues tasks which are executed
+   * when the executor is {@link PausableExecutor#resume() resumed}, tasks that are already being
+   * executed will not be paused and will run to completion.
+   */
+  public static PausableScheduledExecutorService newPausableScheduledExecutorService(
+      ScheduledExecutorService delegate) {
+    return new PausableScheduledExecutorServiceImpl(
+        newPausableExecutorService(delegate), ExecutorsRegistrar.SCHEDULER.get());
+  }
+
   /** Returns a direct executor. */
   public static Executor directExecutor() {
     return DirectExecutor.INSTANCE;

firebase-common/src/main/java/com/google/firebase/concurrent/LimitedConcurrencyExecutor.java

@@ -0,0 +1,72 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.firebase.concurrent;
+
+import com.google.firebase.components.Preconditions;
+import java.util.concurrent.Executor;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.Semaphore;
+
+/**
+ * An {@link Executor}that limits the number of concurrent tasks executing at a given time.
+ *
+ * <p>Delegates actual execution to the {@code delegate} executor and does not create threads of it
+ * own.
+ *
+ * <p>The executor is fair: has FIFO semantics for submitted tasks.
+ */
+class LimitedConcurrencyExecutor implements Executor {
+
+  private final Executor delegate;
+
+  private final Semaphore semaphore;
+
+  private final LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
+
+  LimitedConcurrencyExecutor(Executor delegate, int concurrency) {
+    Preconditions.checkArgument(concurrency > 0, "concurrency must be positive.");
+    this.delegate = delegate;
+    semaphore = new Semaphore(concurrency, true);
+  }
+
+  @Override
+  public void execute(Runnable command) {
+    queue.offer(command);
+    maybeEnqueueNext();
+  }
+
+  private void maybeEnqueueNext() {
+    while (semaphore.tryAcquire()) {
+      Runnable next = queue.poll();
+      if (next != null) {
+        delegate.execute(decorate(next));
+      } else {
+        semaphore.release();
+        break;
+      }
+    }
+  }
+
+  private Runnable decorate(Runnable command) {
+    return () -> {
+      try {
+        command.run();
+      } finally {
+        semaphore.release();
+        maybeEnqueueNext();
+      }
+    };
+  }
+}

firebase-common/src/main/java/com/google/firebase/concurrent/LimitedConcurrencyExecutorService.java

@@ -0,0 +1,110 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.firebase.concurrent;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+final class LimitedConcurrencyExecutorService extends LimitedConcurrencyExecutor
+    implements ExecutorService {
+  private final ExecutorService delegate;
+
+  LimitedConcurrencyExecutorService(ExecutorService delegate, int concurrency) {
+    super(delegate, concurrency);
+    this.delegate = delegate;
+  }
+
+  @Override
+  public void shutdown() {
+    throw new UnsupportedOperationException("Shutting down is not allowed.");
+  }
+
+  @Override
+  public List<Runnable> shutdownNow() {
+    throw new UnsupportedOperationException("Shutting down is not allowed.");
+  }
+
+  @Override
+  public boolean isShutdown() {
+    return delegate.isShutdown();
+  }
+
+  @Override
+  public boolean isTerminated() {
+    return delegate.isTerminated();
+  }
+
+  @Override
+  public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
+    return delegate.awaitTermination(timeout, unit);
+  }
+
+  @Override
+  public <T> Future<T> submit(Callable<T> task) {
+    FutureTask<T> ft = new FutureTask<>(task);
+    execute(ft);
+    return ft;
+  }
+
+  @Override
+  public <T> Future<T> submit(Runnable task, T result) {
+    return submit(
+        () -> {
+          task.run();
+          return result;
+        });
+  }
+
+  @Override
+  public Future<?> submit(Runnable task) {
+    return submit(
+        () -> {
+          task.run();
+          return null;
+        });
+  }
+
+  @Override
+  public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
+      throws InterruptedException {
+    return delegate.invokeAll(tasks);
+  }
+
+  @Override
+  public <T> List<Future<T>> invokeAll(
+      Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
+      throws InterruptedException {
+    return delegate.invokeAll(tasks, timeout, unit);
+  }
+
+  @Override
+  public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
+      throws ExecutionException, InterruptedException {
+    return delegate.invokeAny(tasks);
+  }
+
+  @Override
+  public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
+      throws ExecutionException, InterruptedException, TimeoutException {
+    return delegate.invokeAny(tasks, timeout, unit);
+  }
+}

firebase-common/src/main/java/com/google/firebase/concurrent/PausableExecutor.java

@@ -0,0 +1,32 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.firebase.concurrent;
+
+import java.util.concurrent.Executor;
+
+/**
+ * An {@link Executor} that can be paused/resumed.
+ *
+ * <p>When the executor is {@link #pause() paused}, tasks get queued and execute when {@link
+ * #resume()} is called.
+ */
+public interface PausableExecutor extends Executor {
+
+  void pause();
+
+  void resume();
+
+  boolean isPaused();
+}

firebase-common/src/main/java/com/google/firebase/concurrent/PausableExecutorImpl.java

@@ -0,0 +1,68 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.firebase.concurrent;
+
+import androidx.annotation.VisibleForTesting;
+import java.util.concurrent.Executor;
+import java.util.concurrent.LinkedBlockingQueue;
+
+final class PausableExecutorImpl implements PausableExecutor {
+  private volatile boolean paused;
+  private final Executor delegate;
+
+  @VisibleForTesting final LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
+
+  PausableExecutorImpl(boolean paused, Executor delegate) {
+    this.paused = paused;
+    this.delegate = delegate;
+  }
+
+  @Override
+  public void execute(Runnable command) {
+    queue.offer(command);
+    maybeEnqueueNext();
+  }
+
+  private void maybeEnqueueNext() {
+    if (paused) {
+      return;
+    }
+    Runnable next = queue.poll();
+    while (next != null) {
+      delegate.execute(next);
+      if (!paused) {
+        next = queue.poll();
+      } else {
+        next = null;
+      }
+    }
+  }
+
+  @Override
+  public void pause() {
+    paused = true;
+  }
+
+  @Override
+  public void resume() {
+    paused = false;
+    maybeEnqueueNext();
+  }
+
+  @Override
+  public boolean isPaused() {
+    return paused;
+  }
+}