BackgroundQueueNoMoreAsyncTaskThreadPool #7376

Author: dconeybe

firebase-firestore/CHANGELOG.md

@@ -3,6 +3,9 @@
 - [changed] Bumped internal dependencies.
 - [changed] Improve the performance of queries in collections that contain many deleted documents.
   [#7295](//github.com/firebase/firebase-android-sdk/issues/7295)
+- [changed] Improve query performance in large result sets by replacing the deprecated AsyncTask
+  thread pool with a self-managed thread pool.
+  [#7376](//github.com/firebase/firebase-android-sdk/issues/7376)
 
 # 26.0.0
 

firebase-firestore/src/main/java/com/google/firebase/firestore/local/SQLiteDocumentOverlayCache.java

@@ -26,7 +26,6 @@
 import com.google.firebase.firestore.model.mutation.Mutation;
 import com.google.firebase.firestore.model.mutation.Overlay;
 import com.google.firebase.firestore.util.BackgroundQueue;
-import com.google.firebase.firestore.util.Executors;
 import com.google.firestore.v1.Write;
 import com.google.protobuf.InvalidProtocolBufferException;
 import java.util.ArrayList;
@@ -35,7 +34,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.SortedSet;
-import java.util.concurrent.Executor;
 
 public class SQLiteDocumentOverlayCache implements DocumentOverlayCache {
   private final SQLitePersistence db;
@@ -204,16 +202,21 @@ private void processOverlaysInBackground(
     byte[] rawMutation = row.getBlob(0);
     int largestBatchId = row.getInt(1);
 
-    // Since scheduling background tasks incurs overhead, we only dispatch to a
-    // background thread if there are still some documents remaining.
-    Executor executor = row.isLast() ? Executors.DIRECT_EXECUTOR : backgroundQueue;
-    executor.execute(
+    Runnable runnable =
         () -> {
           Overlay overlay = decodeOverlay(rawMutation, largestBatchId);
           synchronized (results) {
             results.put(overlay.getKey(), overlay);
           }
-        });
+        };
+
+    // If the cursor has exactly one row then just process that row synchronously to avoid the
+    // unnecessary overhead of scheduling its processing to run asynchronously.
+    if (row.isFirst() && row.isLast()) {
+      runnable.run();
+    } else {
+      backgroundQueue.submit(runnable);
+    }
   }
 
   private Overlay decodeOverlay(byte[] rawMutation, int largestBatchId) {

firebase-firestore/src/main/java/com/google/firebase/firestore/local/SQLiteRemoteDocumentCache.java

@@ -33,7 +33,6 @@
 import com.google.firebase.firestore.model.ResourcePath;
 import com.google.firebase.firestore.model.SnapshotVersion;
 import com.google.firebase.firestore.util.BackgroundQueue;
-import com.google.firebase.firestore.util.Executors;
 import com.google.firebase.firestore.util.Function;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.MessageLite;
@@ -47,7 +46,6 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executor;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
@@ -308,10 +306,7 @@ private void processRowInBackground(
     boolean documentTypeIsNull = row.isNull(3);
     String path = row.getString(4);
 
-    // Since scheduling background tasks incurs overhead, we only dispatch to a
-    // background thread if there are still some documents remaining.
-    Executor executor = row.isLast() ? Executors.DIRECT_EXECUTOR : backgroundQueue;
-    executor.execute(
+    Runnable runnable =
         () -> {
           MutableDocument document =
               decodeMaybeDocument(rawDocument, readTimeSeconds, readTimeNanos);
@@ -323,7 +318,15 @@ private void processRowInBackground(
               results.put(document.getKey(), document);
             }
           }
-        });
+        };
+
+    // If the cursor has exactly one row then just process that row synchronously to avoid the
+    // unnecessary overhead of scheduling its processing to run asynchronously.
+    if (row.isFirst() && row.isLast()) {
+      runnable.run();
+    } else {
+      backgroundQueue.submit(runnable);
+    }
   }
 
   @Override

firebase-firestore/src/main/java/com/google/firebase/firestore/util/BackgroundQueue.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2025 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.firestore.util
+
+import java.util.concurrent.Executor
+import java.util.concurrent.Semaphore
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.asExecutor
+
+/**
+ * Manages CPU-bound work on background threads to enable parallel processing.
+ *
+ * Instances of this class are _not_ thread-safe. All methods of an instance of this class must be
+ * called from the same thread. The behavior of an instance is undefined if any of the methods are
+ * called from multiple threads.
+ */
+internal class BackgroundQueue {
+
+  private var state: State = State.Submitting()
+
+  /**
+   * Submit a task for asynchronous execution on the executor of the owning [BackgroundQueue].
+   *
+   * @throws IllegalStateException if [drain] has been called.
+   */
+  fun submit(runnable: Runnable) {
+    val submittingState = this.state
+    check(submittingState is State.Submitting) { "submit() may not be called after drain()" }
+
+    submittingState.taskCount++
+    executor.execute {
+      try {
+        runnable.run()
+      } finally {
+        submittingState.completedTasks.release()
+      }
+    }
+  }
+
+  /**
+   * Blocks until all tasks submitted via calls to [submit] have completed.
+   *
+   * @throws IllegalStateException if called more than once.
+   */
+  fun drain() {
+    val submittingState = this.state
+    check(submittingState is State.Submitting) { "drain() may not be called more than once" }
+    this.state = State.Draining
+
+    submittingState.completedTasks.acquire(submittingState.taskCount)
+  }
+
+  private sealed interface State {
+    class Submitting : State {
+      val completedTasks = Semaphore(0)
+      var taskCount: Int = 0
+    }
+    object Draining : State
+  }
+
+  companion object {
+
+    /**
+     * The maximum amount of parallelism shared by all instances of this class.
+     *
+     * This is equal to the number of processor cores available, or 2, whichever is larger.
+     */
+    val maxParallelism = Runtime.getRuntime().availableProcessors().coerceAtLeast(2)
+
+    private val executor: Executor =
+      Dispatchers.IO.limitedParallelism(maxParallelism, "firestore.BackgroundQueue").asExecutor()
+  }
+}

firebase-firestore/src/main/java/com/google/firebase/firestore/util/BackgroundQueue.kt

@@ -40,14 +40,12 @@
 import com.google.firebase.firestore.model.ResourcePath;
 import com.google.firebase.firestore.testutil.ComparatorTester;
 import com.google.firebase.firestore.util.BackgroundQueue;
-import com.google.firebase.firestore.util.Executors;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.Executor;
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -991,10 +989,7 @@ public void testSynchronousMatchesOrderBy() {
 
     while (iterator.hasNext()) {
       MutableDocument doc = iterator.next();
-      // Only put the processing in the backgroundQueue if there are more documents
-      // in the list. This behavior matches SQLiteRemoteDocumentCache.getAll(...)
-      Executor executor = iterator.hasNext() ? backgroundQueue : Executors.DIRECT_EXECUTOR;
-      executor.execute(
+      backgroundQueue.submit(
           () -> {
             // We call query.matches() to indirectly test query.matchesOrderBy()
             boolean result = query.matches(doc);