Decode documents in background thread

Author: schmidt-sebastian

firebase-firestore/CHANGELOG.md

@@ -1,4 +1,6 @@
 # Unreleased
+- [changed] Reduced execution time of queries with large result sets by up to
+  60%.
 - [changed] Instead of failing silently, Firestore now crashes the client app
   if it fails to load SSL Ciphers. To avoid these crashes, you must bundle 
   Conscrypt to support non-GMSCore devices on Android KitKat or JellyBean (see

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

@@ -35,7 +35,11 @@
 import java.util.List;
 import java.util.Map;
 
-/** Serializer for values stored in the LocalStore. */
+/**
+ * Serializer for values stored in the LocalStore.
+ *
+ * <p>This class is thread-safe.
+ */
 public final class LocalSerializer {
 
   private final RemoteSerializer rpcSerializer;

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

@@ -23,12 +23,16 @@
 import com.google.firebase.firestore.model.DocumentKey;
 import com.google.firebase.firestore.model.MaybeDocument;
 import com.google.firebase.firestore.model.ResourcePath;
+import com.google.firebase.firestore.util.Executors;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.MessageLite;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Semaphore;
 import javax.annotation.Nullable;
 
 final class SQLiteRemoteDocumentCache implements RemoteDocumentCache {
@@ -118,7 +122,10 @@ public ImmutableSortedMap<DocumentKey, Document> getAllDocumentsMatchingQuery(Qu
     String prefixPath = EncodedPath.encode(prefix);
     String prefixSuccessorPath = EncodedPath.prefixSuccessor(prefixPath);
 
-    Map<DocumentKey, Document> results = new HashMap<>();
+    Map<DocumentKey, Document> allDocuments = new ConcurrentHashMap<>();
+
+    int[] pendingTaskCount = new int[] {0};
+    Semaphore completedTasks = new Semaphore(0);
 
     db.query("SELECT path, contents FROM remote_documents WHERE path >= ? AND path < ?")
         .binding(prefixPath, prefixSuccessorPath)
@@ -136,20 +143,38 @@ public ImmutableSortedMap<DocumentKey, Document> getAllDocumentsMatchingQuery(Qu
                 return;
               }
 
-              MaybeDocument maybeDoc = decodeMaybeDocument(row.getBlob(1));
-              if (!(maybeDoc instanceof Document)) {
-                return;
-              }
-
-              Document doc = (Document) maybeDoc;
-              if (!query.matches(doc)) {
-                return;
-              }
-
-              results.put(doc.getKey(), doc);
+              byte[] rawContents = row.getBlob(1);
+
+              ++pendingTaskCount[0];
+
+              // Since scheduling background tasks incurs overhead, we only dispatch to a background
+              // thread if there are still some documents remaining.
+              Executor deserializationExecutor =
+                  row.isLast() ? Executors.DIRECT_EXECUTOR : Executors.BACKGROUND_EXECUTOR;
+              deserializationExecutor.execute(
+                  () -> {
+                    MaybeDocument maybeDoc = decodeMaybeDocument(rawContents);
+                    if (maybeDoc instanceof Document) {
+                      allDocuments.put(maybeDoc.getKey(), (Document) maybeDoc);
+                    }
+                    completedTasks.release();
+                  });
             });
 
-    return ImmutableSortedMap.Builder.fromMap(results, DocumentKey.comparator());
+    try {
+      completedTasks.acquire(pendingTaskCount[0]);
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+    }
+
+    ImmutableSortedMap<DocumentKey, Document> matchingDocuments =
+        ImmutableSortedMap.Builder.emptyMap(DocumentKey.comparator());
+    for (Map.Entry<DocumentKey, Document> entry : allDocuments.entrySet()) {
+      if (query.matches(entry.getValue())) {
+        matchingDocuments = matchingDocuments.insert(entry.getKey(), entry.getValue());
+      }
+    }
+    return matchingDocuments;
   }
 
   private String pathForKey(DocumentKey key) {

firebase-firestore/src/main/java/com/google/firebase/firestore/model/DatabaseId.java

@@ -16,7 +16,11 @@
 
 import androidx.annotation.NonNull;
 
-/** Represents a particular database in Firestore */
+/**
+ * Represents a particular database in Firestore.
+ *
+ * <p>This class is thread-safe.
+ */
 public final class DatabaseId implements Comparable<DatabaseId> {
   public static final String DEFAULT_DATABASE_ID = "(default)";
 

firebase-firestore/src/main/java/com/google/firebase/firestore/remote/GrpcCallProvider.java

@@ -15,7 +15,6 @@
 package com.google.firebase.firestore.remote;
 
 import android.content.Context;
-import android.os.AsyncTask;
 import androidx.annotation.VisibleForTesting;
 import com.google.android.gms.common.GooglePlayServicesNotAvailableException;
 import com.google.android.gms.common.GooglePlayServicesRepairableException;
@@ -24,6 +23,7 @@
 import com.google.android.gms.tasks.Tasks;
 import com.google.firebase.firestore.core.DatabaseInfo;
 import com.google.firebase.firestore.util.AsyncQueue;
+import com.google.firebase.firestore.util.Executors;
 import com.google.firebase.firestore.util.Logger;
 import com.google.firebase.firestore.util.Supplier;
 import com.google.firestore.v1.FirestoreGrpc;
@@ -69,11 +69,11 @@ public static void overrideChannelBuilder(
       CallCredentials firestoreHeaders) {
     this.asyncQueue = asyncQueue;
 
-    // We execute network initialization on a separate thred to not block operations that depend on
+    // We execute network initialization on a separate thread to not block operations that depend on
     // the AsyncQueue.
     this.channelTask =
         Tasks.call(
-            AsyncTask.THREAD_POOL_EXECUTOR,
+            Executors.BACKGROUND_EXECUTOR,
             () -> {
               ManagedChannel channel = initChannel(context, databaseInfo);
               FirestoreGrpc.FirestoreStub firestoreStub =

firebase-firestore/src/main/java/com/google/firebase/firestore/remote/RemoteSerializer.java

@@ -105,7 +105,11 @@
 import java.util.Map;
 import java.util.Set;
 
-/** Serializer that converts to and from Firestore API protos. */
+/**
+ * Serializer that converts to and from Firestore API protos.
+ *
+ * <p>This class is thread-safe.
+ */
 public final class RemoteSerializer {
 
   private final DatabaseId databaseId;

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

@@ -14,11 +14,22 @@
 
 package com.google.firebase.firestore.util;
 
+import android.os.AsyncTask;
 import com.google.android.gms.tasks.TaskExecutors;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /** Helper class for executors. */
 public final class Executors {
+  /**
+   * The maximum number of tasks we submit to AsyncTask.THREAD_POOL_EXECUTOR.
+   *
+   * <p>The limit is based on the number of core threads spun by THREAD_POOL_EXECUTOR and addresses
+   * its queue size limit of 120 pending tasks.
+   */
+  private static final int MAXIMUM_CONCURRENT_BACKGROUND_TASKS = 4;
 
   /**
    * The default executor for user visible callbacks. It is an executor scheduling callbacks on
@@ -29,6 +40,39 @@ public final class Executors {
   /** An executor that executes the provided runnable immediately on the current thread. */
   public static final Executor DIRECT_EXECUTOR = Runnable::run;
 
+  /**
+   * An executor that runs tasks in parallel on Android's AsyncTask.THREAD_POOL_EXECUTOR.
+   *
+   * <p>Unlike the main THREAD_POOL_EXECUTOR, this executor manages its own queue of tasks and can
+   * handle an unbounded number of pending tasks.
+   */
+  public static final Executor BACKGROUND_EXECUTOR =
+      new Executor() {
+        AtomicInteger activeRunnerCount = new AtomicInteger(0);
+        Queue<Runnable> pendingTasks = new ConcurrentLinkedQueue<>();
+
+        @Override
+        public void execute(Runnable command) {
+          pendingTasks.add(command);
+
+          if (activeRunnerCount.get() < MAXIMUM_CONCURRENT_BACKGROUND_TASKS) {
+            // Note that the runner count could temporarily exceed
+            // MAXIMUM_CONCURRENT_BACKGROUND_TASKS if this is code path was executed in parallel.
+            // While undesired, this would merely queue another task on THREAD_POOL_EXECUTOR,
+            // and we are unlikely to hit the 120 pending task limit.
+            activeRunnerCount.incrementAndGet();
+            AsyncTask.THREAD_POOL_EXECUTOR.execute(
+                () -> {
+                  Runnable r;
+                  while ((r = pendingTasks.poll()) != null) {
+                    r.run();
+                  }
+                  activeRunnerCount.decrementAndGet();
+                });
+          }
+        }
+      };
+
   private Executors() {
     // Private constructor to prevent initialization
   }