Decode documents in a background thread (#4857)

* Add Index-free benchmarks

* Add BackgroundQueue

Port of firebase/firebase-android-sdk#559

* Decode documents in background threads to improve performance

Run on (12 X 2900 MHz CPU s)
CPU Caches:
  L1 Data 32K (x6)
  L1 Instruction 32K (x6)
  L2 Unified 262K (x6)
  L3 Unified 12582K (x1)
Load Average: 2.24, 2.18, 1.96
----------------------------------------------------------------------------------
Benchmark                           Time             CPU   Iterations  Improvement
----------------------------------------------------------------------------------
BM_QueryMatching/0/1              132 us         12.7 us        53524        21.4%
BM_QueryMatching/1/1              150 us         18.1 us        37305        16.2%
BM_QueryMatching/1/10             371 us         27.6 us        10000        48.5%
BM_QueryMatching/10/10            429 us         64.3 us        11349        41.0%
BM_QueryMatching/1/100           2573 us         29.0 us         1000        51.7% (max)
BM_QueryMatching/100/100         3123 us          416 us         1000        45.5%
BM_QueryMatching/1/1000         26785 us         33.5 us          100        50.5%
BM_QueryMatching/10/1000        28389 us         74.5 us          100        48.0%
BM_QueryMatching/100/1000       27883 us          470 us          100        49.4%
BM_QueryMatching/1000/1000      32189 us         34.3 us          100        44.8%
BM_QueryAll/1                     151 us         15.2 us        45385       -49.5% (min)
BM_QueryAll/10                    444 us         60.7 us        12829        32.5%
BM_QueryAll/100                  3175 us          414 us         1000        45.1%
BM_QueryAll/1000                30222 us         18.8 us          100        49.7%
Mean                                                                         35.3%

* Insert into a vector to reduce contention

Run on (12 X 2900 MHz CPU s)
CPU Caches:
  L1 Data 32K (x6)
  L1 Instruction 32K (x6)
  L2 Unified 262K (x6)
  L3 Unified 12582K (x1)
Load Average: 1.43, 2.10, 1.98
----------------------------------------------------------------------------------
Benchmark                           Time             CPU   Iterations  Improvement
----------------------------------------------------------------------------------
BM_QueryMatching/0/1              127 us         12.2 us        56131         3.8%
BM_QueryMatching/1/1              142 us         16.6 us        39200         5.3%
BM_QueryMatching/1/10             369 us         26.6 us        10000         0.5%
BM_QueryMatching/10/10            425 us         59.2 us        12724         0.9%
BM_QueryMatching/1/100           2194 us         28.9 us        10000        14.7%
BM_QueryMatching/100/100         2765 us          405 us         1000        11.5%
BM_QueryMatching/1/1000         21095 us         32.8 us         1000        21.2%
BM_QueryMatching/10/1000        20797 us         66.4 us         1000        26.7% (max)
BM_QueryMatching/100/1000       21915 us          408 us         1000        21.4%
BM_QueryMatching/1000/1000      23700 us         30.9 us         1000        26.4%
BM_QueryAll/1                     136 us         14.3 us        44799         9.9%
BM_QueryAll/10                    443 us         57.3 us        10000         0.2% (min)
BM_QueryAll/100                  2607 us          389 us         1000        17.9%
BM_QueryAll/1000                23512 us         19.4 us         1000        22.2%
Mean                                                                         13.1%

Author: wilhuff

Firestore/CHANGELOG.md

@@ -1,4 +1,5 @@
 # Unreleased
+- [changed] Improved performance of queries with large result sets.
 
 # v1.10.2
 - [changed] Internal improvements.

Firestore/Example/Benchmarks/remote_document_cache_benchmark.mm

@@ -61,7 +61,7 @@
   return doc;
 }
 
-FIRQuerySnapshot* GetDocuments(FIRQuery* query) {
+FIRQuerySnapshot* GetDocumentsFromCache(FIRQuery* query) {
   __block FIRQuerySnapshot* result;
   dispatch_semaphore_t done = dispatch_semaphore_create(0);
   [query getDocumentsWithSource:FIRFirestoreSourceCache
@@ -74,6 +74,19 @@
   return result;
 }
 
+FIRQuerySnapshot* GetDocumentsFromServer(FIRQuery* query) {
+  __block FIRQuerySnapshot* result;
+  dispatch_semaphore_t done = dispatch_semaphore_create(0);
+  [query getDocumentsWithSource:FIRFirestoreSourceServer
+                     completion:^(FIRQuerySnapshot* snap, NSError* error) {
+                       HARD_ASSERT(error == nil, "Failed: %s", MakeString([error description]));
+                       result = snap;
+                       dispatch_semaphore_signal(done);
+                     }];
+  dispatch_semaphore_wait(done, DISPATCH_TIME_FOREVER);
+  return result;
+}
+
 void WaitForPendingWrites(FIRFirestore* db) {
   dispatch_semaphore_t done = dispatch_semaphore_create(0);
   [db waitForPendingWritesWithCompletion:^(NSError* error) {
@@ -100,17 +113,51 @@ void Shutdown(FIRFirestore* db) {
   dispatch_semaphore_wait(done, DISPATCH_TIME_FOREVER);
 }
 
+void BM_QueryIndexFree(benchmark::State& state) {
+  int64_t matching_docs = state.range(0);
+  int64_t total_docs = state.range(1);
+
+  FIRFirestore* db = OpenFirestore();
+  auto collection = [db collectionWithPath:MakeNSString("docs-" + CreateAutoId())];
+  WriteDocs(collection, matching_docs, /*match=*/true);
+  WriteDocs(collection, total_docs - matching_docs, /*match=*/false);
+
+  FIRQuery* query = [collection queryWhereField:@"match" isEqualTo:@YES];
+
+  // Query the server to force the target tables to be updated.
+  GetDocumentsFromServer(query);
+
+  for (auto _ : state) {
+    auto docs = GetDocumentsFromCache(query);
+    (void)docs;
+  }
+
+  Shutdown(db);
+}
+BENCHMARK(BM_QueryIndexFree)
+    ->Unit(benchmark::kMicrosecond)
+    ->Args({0, 1})
+    ->Args({1, 1})
+    ->Args({1, 10})
+    ->Args({10, 10})
+    ->Args({1, 100})
+    ->Args({100, 100})
+    ->Args({1, 1000})
+    ->Args({10, 1000})
+    ->Args({100, 1000})
+    ->Args({1000, 1000});
+
 void BM_QueryMatching(benchmark::State& state) {
   int64_t matching_docs = state.range(0);
   int64_t total_docs = state.range(1);
 
   FIRFirestore* db = OpenFirestore();
   auto collection = [db collectionWithPath:MakeNSString("docs-" + CreateAutoId())];
-  WriteDocs(collection, matching_docs, /* match= */ true);
-  WriteDocs(collection, total_docs - matching_docs, /* match= */ false);
+  WriteDocs(collection, matching_docs, /*match=*/true);
+  WriteDocs(collection, total_docs - matching_docs, /*match=*/false);
 
   for (auto _ : state) {
-    auto docs = GetDocuments([collection queryWhereField:@"match" isEqualTo:@YES]);
+    auto docs = GetDocumentsFromCache([collection queryWhereField:@"match" isEqualTo:@YES]);
     (void)docs;
   }
 
@@ -134,10 +181,10 @@ void BM_QueryAll(benchmark::State& state) {
 
   FIRFirestore* db = OpenFirestore();
   auto collection = [db collectionWithPath:MakeNSString("docs-" + CreateAutoId())];
-  WriteDocs(collection, total_docs, /* match= */ true);
+  WriteDocs(collection, total_docs, /*match=*/true);
 
   for (auto _ : state) {
-    auto docs = GetDocuments(collection);
+    auto docs = GetDocumentsFromCache(collection);
     (void)docs;
   }
 

Firestore/core/src/firebase/firestore/core/target_id_generator.cc

@@ -23,14 +23,9 @@ namespace firebase {
 namespace firestore {
 namespace core {
 
-TargetIdGenerator::TargetIdGenerator(const TargetIdGenerator& value)
-    : generator_id_(value.generator_id_), next_id_(value.next_id_) {
-}
-
 TargetIdGenerator::TargetIdGenerator(TargetIdGeneratorId generator_id,
                                      TargetId seed)
     : generator_id_(generator_id) {
-  generator_id_ = generator_id;
   seek(seed);
 }
 

Firestore/core/src/firebase/firestore/core/target_id_generator.h

@@ -49,7 +49,8 @@ class TargetIdGenerator {
   // Makes Objective-C++ code happy to provide a default ctor.
   TargetIdGenerator() = default;
 
-  TargetIdGenerator(const TargetIdGenerator& value);
+  TargetIdGenerator(const TargetIdGenerator& value) = default;
+  TargetIdGenerator& operator=(const TargetIdGenerator& value) = default;
 
   /**
    * Creates and returns the TargetIdGenerator for the local store.
@@ -84,8 +85,9 @@ class TargetIdGenerator {
  private:
   TargetIdGenerator(TargetIdGeneratorId generator_id, model::TargetId seed);
   void seek(model::TargetId target_id);
-  TargetIdGeneratorId generator_id_;
-  model::TargetId next_id_;
+
+  TargetIdGeneratorId generator_id_ = TargetIdGeneratorId::TargetCache;
+  model::TargetId next_id_ = 0;
 
   static const int kReservedBits = 1;
 };

Firestore/core/src/firebase/firestore/local/leveldb_persistence.cc

@@ -98,7 +98,7 @@ util::StatusOr<std::unique_ptr<LevelDbPersistence>> LevelDbPersistence::Create(
   std::unique_ptr<LevelDbPersistence> result(
       new LevelDbPersistence(std::move(db), std::move(dir), std::move(users),
                              std::move(serializer), lru_params));
-  return std::move(result);
+  return {std::move(result)};
 }
 
 LevelDbPersistence::LevelDbPersistence(std::unique_ptr<leveldb::DB> db,

Firestore/core/src/firebase/firestore/local/leveldb_remote_document_cache.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 Google
+ * Copyright 2018 Google LLC
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@
 #include "Firestore/core/src/firebase/firestore/local/leveldb_remote_document_cache.h"
 
 #include <string>
+#include <thread>  // NOLINT(build/c++11)
 #include <utility>
 
 #include "Firestore/Protos/nanopb/firestore/local/maybe_document.nanopb.h"
@@ -26,13 +27,16 @@
 #include "Firestore/core/src/firebase/firestore/local/local_serializer.h"
 #include "Firestore/core/src/firebase/firestore/nanopb/message.h"
 #include "Firestore/core/src/firebase/firestore/nanopb/reader.h"
+#include "Firestore/core/src/firebase/firestore/util/background_queue.h"
+#include "Firestore/core/src/firebase/firestore/util/executor.h"
 #include "Firestore/core/src/firebase/firestore/util/status.h"
 #include "Firestore/core/src/firebase/firestore/util/string_util.h"
 #include "leveldb/db.h"
 
 namespace firebase {
 namespace firestore {
 namespace local {
+namespace {
 
 using core::Query;
 using leveldb::Status;
@@ -48,12 +52,58 @@ using model::SnapshotVersion;
 using nanopb::ByteString;
 using nanopb::Message;
 using nanopb::StringReader;
+using util::BackgroundQueue;
+using util::Executor;
+
+/**
+ * An accumulator for results produced asynchronously. This accumulates
+ * values in a vector to avoid contention caused by accumulating into more
+ * complex structures like immutable::SortedMap.
+ */
+template <typename T>
+class AsyncResults {
+ public:
+  void Insert(T&& value) {
+    std::lock_guard<std::mutex> lock(mutex_);
+    values_.push_back(std::move(value));
+  }
+
+  void Insert(const T& value) {
+    std::lock_guard<std::mutex> lock(mutex_);
+    values_.push_back(value);
+  }
+
+  /**
+   * Returns the accumulated result, moving it out of AsyncResults. The
+   * AsyncResults object should not be reused.
+   */
+  std::vector<T> Result() {
+    std::lock_guard<std::mutex> lock(mutex_);
+    return std::move(values_);
+  }
+
+ private:
+  std::vector<T> values_;
+  std::mutex mutex_;
+};
+
+}  // namespace
 
 LevelDbRemoteDocumentCache::LevelDbRemoteDocumentCache(
     LevelDbPersistence* db, LocalSerializer* serializer)
     : db_(db), serializer_(NOT_NULL(serializer)) {
+  auto hw_concurrency = std::thread::hardware_concurrency();
+  if (hw_concurrency == 0) {
+    // If the standard library doesn't know, guess something reasonable.
+    hw_concurrency = 4;
+  }
+  executor_ = Executor::CreateConcurrent("com.google.firebase.firestore.query",
+                                         static_cast<int>(hw_concurrency));
 }
 
+// Out of line because of unique_ptrs to incomplete types.
+LevelDbRemoteDocumentCache::~LevelDbRemoteDocumentCache() = default;
+
 void LevelDbRemoteDocumentCache::Add(const MaybeDocument& document,
                                      const SnapshotVersion& read_time) {
   const DocumentKey& key = document.key();
@@ -93,7 +143,8 @@ absl::optional<MaybeDocument> LevelDbRemoteDocumentCache::Get(
 
 OptionalMaybeDocumentMap LevelDbRemoteDocumentCache::GetAll(
     const DocumentKeySet& keys) {
-  OptionalMaybeDocumentMap results;
+  BackgroundQueue tasks(executor_.get());
+  AsyncResults<std::pair<DocumentKey, absl::optional<MaybeDocument>>> results;
 
   LevelDbRemoteDocumentKey current_key;
   auto it = db_->current_transaction()->NewIterator();
@@ -102,13 +153,22 @@ OptionalMaybeDocumentMap LevelDbRemoteDocumentCache::GetAll(
     it->Seek(LevelDbRemoteDocumentKey::Key(key));
     if (!it->Valid() || !current_key.Decode(it->key()) ||
         current_key.document_key() != key) {
-      results = results.insert(key, absl::nullopt);
+      results.Insert(std::make_pair(key, absl::nullopt));
     } else {
-      results = results.insert(key, DecodeMaybeDocument(it->value(), key));
+      const std::string& contents = it->value();
+      tasks.Execute([this, &results, &key, contents] {
+        results.Insert(std::make_pair(key, DecodeMaybeDocument(contents, key)));
+      });
     }
   }
 
-  return results;
+  tasks.AwaitAll();
+
+  OptionalMaybeDocumentMap map;
+  for (const auto& entry : results.Result()) {
+    map = map.insert(entry.first, entry.second);
+  }
+  return map;
 }
 
 DocumentMap LevelDbRemoteDocumentCache::GetAllExisting(
@@ -165,7 +225,8 @@ DocumentMap LevelDbRemoteDocumentCache::GetMatching(
 
     return LevelDbRemoteDocumentCache::GetAllExisting(remote_keys);
   } else {
-    DocumentMap results;
+    BackgroundQueue tasks(executor_.get());
+    AsyncResults<Document> results;
 
     // Documents are ordered by key, so we can use a prefix scan to narrow down
     // the documents we need to match the query against.
@@ -185,15 +246,26 @@ DocumentMap LevelDbRemoteDocumentCache::GetMatching(
         continue;
       }
 
-      MaybeDocument maybe_doc = DecodeMaybeDocument(it->value(), document_key);
-      if (!query_path.IsPrefixOf(maybe_doc.key().path())) {
+      if (!query_path.IsPrefixOf(document_key.path())) {
         break;
-      } else if (maybe_doc.is_document()) {
-        results = results.insert(maybe_doc.key(), Document(maybe_doc));
       }
+
+      const std::string& contents = it->value();
+      tasks.Execute([this, &results, document_key, contents] {
+        MaybeDocument maybe_doc = DecodeMaybeDocument(contents, document_key);
+        if (maybe_doc.is_document()) {
+          results.Insert(Document(maybe_doc));
+        }
+      });
     }
 
-    return results;
+    tasks.AwaitAll();
+
+    DocumentMap map;
+    for (const Document& doc : results.Result()) {
+      map = map.insert(doc.key(), doc);
+    }
+    return map;
   }
 }
 

Firestore/core/src/firebase/firestore/local/leveldb_remote_document_cache.h

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 Google
+ * Copyright 2018 Google LLC
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,6 +17,8 @@
 #ifndef FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_LOCAL_LEVELDB_REMOTE_DOCUMENT_CACHE_H_
 #define FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_LOCAL_LEVELDB_REMOTE_DOCUMENT_CACHE_H_
 
+#include <memory>
+#include <thread>  // NOLINT(build/c++11)
 #include <vector>
 
 #include "Firestore/core/src/firebase/firestore/local/remote_document_cache.h"
@@ -29,6 +31,11 @@
 
 namespace firebase {
 namespace firestore {
+
+namespace util {
+class Executor;
+}  // namespace util
+
 namespace local {
 
 class LevelDbPersistence;
@@ -39,6 +46,7 @@ class LevelDbRemoteDocumentCache : public RemoteDocumentCache {
  public:
   LevelDbRemoteDocumentCache(LevelDbPersistence* db,
                              LocalSerializer* serializer);
+  ~LevelDbRemoteDocumentCache();
 
   void Add(const model::MaybeDocument& document,
            const model::SnapshotVersion& read_time) override;
@@ -66,6 +74,8 @@ class LevelDbRemoteDocumentCache : public RemoteDocumentCache {
   LevelDbPersistence* db_;
   // Owned by LevelDbPersistence.
   LocalSerializer* serializer_ = nullptr;
+
+  std::unique_ptr<util::Executor> executor_;
 };
 
 }  // namespace local

Firestore/core/src/firebase/firestore/local/leveldb_target_cache.cc

@@ -66,7 +66,7 @@ LevelDbTargetCache::TryReadMetadata(leveldb::DB* db) {
     }
   }
 
-  return std::move(result);
+  return {std::move(result)};
 }
 
 Message<firestore_client_TargetGlobal> LevelDbTargetCache::ReadMetadata(

Firestore/core/src/firebase/firestore/model/document_map.h

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 Google
+ * Copyright 2018 Google LLC
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -57,6 +57,9 @@ using OptionalMaybeDocumentMap =
  */
 class DocumentMap {
  public:
+  using key_type = DocumentKey;
+  using mapped_type = Document;
+
   DocumentMap() = default;
 
   ABSL_MUST_USE_RESULT DocumentMap insert(const DocumentKey& key,