Fixes for issues discovered in the Unity desktop build (#4795)

* Make Filesystem's destructor virtual

* Avoid capturing a strong reference to FirestoreClient in GC tasks.

This prevents this task from keeping the FirestoreClient alive even
after the api::Firestore instance has been destroyed.

* Ensure that FirestoreClient is terminated

In the public C++ API, it's possible to delete the owning FirebaseApp
which triggers the destruction of the api::Firestore object even though
we haven't called Terminate.

This makes it such that api::Firestore calls a blocking Terminate call
on FirestoreClient in its destructor.

Additionally, change AsyncQueue to always run the operation enqueued on
shutdown, even if already shutdown. Make FirestoreClient capable of
short circuiting the termination work for itself. This makes terminate
idempotent in a more staightforward way.

* Clean up excess logging

* Remove defintion of weak_from_this

* Review feedback

* Make AsyncQueue extend std::enable_shared_from_this

Also hide the AsyncQueue's public constructor and force all users to
manage their reference to it via std::shared_ptr.

* Handle Executor shutdown from within its own pool

Author: wilhuff

Firestore/Source/API/FSTFirestoreComponent.mm

@@ -99,7 +99,7 @@ - (FIRFirestore *)firestoreForDatabase:(NSString *)database {
       }
 
       auto executor = Executor::CreateSerial(queue_name.c_str());
-      auto workerQueue = absl::make_unique<AsyncQueue>(std::move(executor));
+      auto workerQueue = AsyncQueue::Create(std::move(executor));
 
       id<FIRAuthInterop> auth = FIR_COMPONENT(FIRAuthInterop, self.app.container);
       auto credentialsProvider = std::make_shared<FirebaseCredentialsProvider>(self.app, auth);

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

@@ -62,6 +62,16 @@ Firestore::Firestore(model::DatabaseId database_id,
       extension_{extension} {
 }
 
+Firestore::~Firestore() {
+  std::lock_guard<std::mutex> lock{mutex_};
+
+  // If the client hasn't been configured yet we don't need to create it just
+  // to tear it down.
+  if (!client_) return;
+
+  client_->Terminate();
+}
+
 const std::shared_ptr<FirestoreClient>& Firestore::client() {
   HARD_ASSERT(client_, "Client is not yet configured.");
   return client_;
@@ -133,7 +143,7 @@ void Firestore::Terminate(util::StatusCallback callback) {
   // The client must be initialized to ensure that all subsequent API usage
   // throws an exception.
   EnsureClientConfigured();
-  client_->Terminate(std::move(callback));
+  client_->TerminateAsync(std::move(callback));
 }
 
 void Firestore::WaitForPendingWrites(util::StatusCallback callback) {

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

@@ -60,6 +60,8 @@ class Firestore : public std::enable_shared_from_this<Firestore> {
             std::shared_ptr<util::AsyncQueue> worker_queue,
             void* extension);
 
+  ~Firestore();
+
   const model::DatabaseId& database_id() const {
     return database_id_;
   }

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

@@ -213,9 +213,12 @@ void FirestoreClient::Initialize(const User& user, const Settings& settings) {
 void FirestoreClient::ScheduleLruGarbageCollection() {
   std::chrono::milliseconds delay =
       gc_has_run_ ? regular_gc_delay_ : initial_gc_delay_;
-  auto shared_this = shared_from_this();
+  std::weak_ptr<FirestoreClient> weak_this = shared_from_this();
   lru_callback_ = worker_queue()->EnqueueAfterDelay(
-      delay, TimerId::GarbageCollectionDelay, [shared_this] {
+      delay, TimerId::GarbageCollectionDelay, [weak_this] {
+        auto shared_this = weak_this.lock();
+        if (!shared_this) return;
+
         shared_this->local_store_->CollectGarbage(
             shared_this->lru_delegate_->garbage_collector());
         shared_this->gc_has_run_ = true;
@@ -245,30 +248,42 @@ void FirestoreClient::EnableNetwork(StatusCallback callback) {
   });
 }
 
-void FirestoreClient::Terminate(StatusCallback callback) {
+void FirestoreClient::TerminateAsync(StatusCallback callback) {
   auto shared_this = shared_from_this();
   worker_queue()->EnqueueAndInitiateShutdown([shared_this, callback] {
-    shared_this->credentials_provider_->SetCredentialChangeListener(nullptr);
+    shared_this->TerminateInternal();
 
-    // If we've scheduled LRU garbage collection, cancel it.
-    if (shared_this->lru_callback_) {
-      shared_this->lru_callback_.Cancel();
-    }
-    shared_this->remote_store_->Shutdown();
-    shared_this->persistence_->Shutdown();
-  });
-
-  // This separate enqueue ensures if `terminate` is called multiple times
-  // every time the callback is triggered. If it is in the above
-  // enqueue, it might not get executed because after first `terminate`
-  // all operations are not executed.
-  worker_queue()->EnqueueEvenAfterShutdown([shared_this, callback] {
     if (callback) {
       shared_this->user_executor()->Execute([=] { callback(Status::OK()); });
     }
   });
 }
 
+void FirestoreClient::Terminate() {
+  std::promise<void> signal_terminated;
+  worker_queue()->EnqueueAndInitiateShutdown([&, this] {
+    TerminateInternal();
+    signal_terminated.set_value();
+  });
+  signal_terminated.get_future().wait();
+}
+
+void FirestoreClient::TerminateInternal() {
+  if (!remote_store_) return;
+
+  credentials_provider_->SetCredentialChangeListener(nullptr);
+
+  // If we've scheduled LRU garbage collection, cancel it.
+  if (lru_callback_) {
+    lru_callback_.Cancel();
+  }
+  remote_store_->Shutdown();
+  persistence_->Shutdown();
+
+  // Clear the remote store to indicate terminate is complete.
+  remote_store_.reset();
+}
+
 void FirestoreClient::WaitForPendingWrites(StatusCallback callback) {
   VerifyNotTerminated();
 

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

@@ -88,7 +88,13 @@ class FirestoreClient : public std::enable_shared_from_this<FirestoreClient> {
    * Terminates this client, cancels all writes / listeners, and releases all
    * resources.
    */
-  void Terminate(util::StatusCallback callback);
+  void TerminateAsync(util::StatusCallback callback);
+
+  /**
+   * Synchronously terminates this client, cancels all writes / listeners, and
+   * releases all resources.
+   */
+  void Terminate();
 
   /**
    * Passes a callback that is triggered when all the pending writes at the
@@ -183,6 +189,8 @@ class FirestoreClient : public std::enable_shared_from_this<FirestoreClient> {
 
   void VerifyNotTerminated();
 
+  void TerminateInternal();
+
   void ScheduleLruGarbageCollection();
 
   DatabaseInfo database_info_;

Firestore/core/src/firebase/firestore/util/async_queue.cc

@@ -26,6 +26,13 @@ namespace firebase {
 namespace firestore {
 namespace util {
 
+std::shared_ptr<AsyncQueue> AsyncQueue::Create(
+    std::unique_ptr<Executor> executor) {
+  // Use new because make_shared cannot access a private constructor.
+  auto queue = new AsyncQueue(std::move(executor));
+  return std::shared_ptr<AsyncQueue>(queue);
+}
+
 AsyncQueue::AsyncQueue(std::unique_ptr<Executor> executor)
     : executor_{std::move(executor)} {
   is_operation_in_progress_ = false;
@@ -72,9 +79,7 @@ void AsyncQueue::Enqueue(const Operation& operation) {
 void AsyncQueue::EnqueueAndInitiateShutdown(const Operation& operation) {
   std::lock_guard<std::mutex> lock{shut_down_mutex_};
   VerifySequentialOrder();
-  if (is_shutting_down_) {
-    return;
-  }
+
   is_shutting_down_ = true;
   executor_->Execute(Wrap(operation));
 }
@@ -109,7 +114,7 @@ DelayedOperation AsyncQueue::EnqueueAfterDelay(Milliseconds delay,
     return DelayedOperation();
   }
 
-  // Skip delays for timer_ids that have been overriden
+  // Skip delays for timer_ids that have been overridden
   if (absl::c_linear_search(timer_ids_to_skip_, timer_id)) {
     delay = Milliseconds(0);
   }
@@ -123,7 +128,8 @@ AsyncQueue::Operation AsyncQueue::Wrap(const Operation& operation) {
   // ensure that it doesn't spawn any nested operations.
 
   // Note: can't move `operation` into lambda until C++14.
-  return [this, operation] { ExecuteBlocking(operation); };
+  auto shared_this = shared_from_this();
+  return [shared_this, operation] { shared_this->ExecuteBlocking(operation); };
 }
 
 void AsyncQueue::VerifySequentialOrder() const {

Firestore/core/src/firebase/firestore/util/async_queue.h

@@ -84,12 +84,12 @@ enum class TimerId {
 //
 // A significant portion of `AsyncQueue` interface only exists for test purposes
 // and must *not* be used in regular code.
-class AsyncQueue {
+class AsyncQueue : public std::enable_shared_from_this<AsyncQueue> {
  public:
   using Operation = Executor::Operation;
   using Milliseconds = Executor::Milliseconds;
 
-  explicit AsyncQueue(std::unique_ptr<Executor> executor);
+  static std::shared_ptr<AsyncQueue> Create(std::unique_ptr<Executor> executor);
 
   // Asserts for the caller that it is being invoked as part of an operation on
   // the `AsyncQueue`.
@@ -186,6 +186,8 @@ class AsyncQueue {
   void SkipDelaysForTimerId(TimerId timer_id);
 
  private:
+  explicit AsyncQueue(std::unique_ptr<Executor> executor);
+
   Operation Wrap(const Operation& operation);
 
   // Asserts that the current invocation happens asynchronously on the queue.

Firestore/core/src/firebase/firestore/util/executor_std.cc

@@ -40,7 +40,8 @@ std::string ThreadIdToString(const std::thread::id thread_id) {
 
 // MARK: - ExecutorStd
 
-ExecutorStd::ExecutorStd(int threads) {
+ExecutorStd::ExecutorStd(int threads)
+    : shutting_down_(std::make_shared<std::atomic<bool>>()) {
   HARD_ASSERT(threads > 0);
 
   // Somewhat counter-intuitively, constructor of `std::atomic` assigns the
@@ -49,14 +50,14 @@ ExecutorStd::ExecutorStd(int threads) {
   // See [this thread](https://stackoverflow.com/questions/25609858) for context
   // on the constructor.
   current_id_ = 0;
-  shutting_down_ = false;
+  *shutting_down_ = false;
   for (int i = 0; i < threads; ++i) {
     worker_thread_pool_.emplace_back(&ExecutorStd::PollingThread, this);
   }
 }
 
 ExecutorStd::~ExecutorStd() {
-  shutting_down_ = true;
+  *shutting_down_ = true;
 
   // Make sure the worker threads are not blocked, so that the call to `join`
   // doesn't hang. It's not deterministic which thread will pick up an entry,
@@ -66,7 +67,14 @@ ExecutorStd::~ExecutorStd() {
   }
 
   for (std::thread& thread : worker_thread_pool_) {
-    thread.join();
+    // If the current thread is running this destructor, we can't join the
+    // thread. Instead detach it and rely on PollingThread to notice that
+    // *shutting_down_ is now true.
+    if (std::this_thread::get_id() == thread.get_id()) {
+      thread.detach();
+    } else {
+      thread.join();
+    }
   }
 }
 
@@ -105,7 +113,10 @@ ExecutorStd::Id ExecutorStd::PushOnSchedule(Operation&& operation,
 }
 
 void ExecutorStd::PollingThread() {
-  while (!shutting_down_) {
+  // Keep a local shared_ptr here to ensure that the atomic pointed to by
+  // shutting_down_ remains valid even after the destruction of the executor.
+  std::shared_ptr<std::atomic<bool>> local_shutting_down = shutting_down_;
+  while (!*local_shutting_down) {
     Entry entry = schedule_.PopBlocking();
     if (entry.tagged.operation) {
       entry.tagged.operation();

Firestore/core/src/firebase/firestore/util/executor_std.h

@@ -21,6 +21,7 @@
 #include <atomic>
 #include <condition_variable>  // NOLINT(build/c++11)
 #include <deque>
+#include <memory>
 #include <mutex>  // NOLINT(build/c++11)
 #include <string>
 #include <thread>  // NOLINT(build/c++11)
@@ -273,7 +274,7 @@ class ExecutorStd : public Executor {
 
   std::vector<std::thread> worker_thread_pool_;
   // Used to stop the worker thread.
-  std::atomic<bool> shutting_down_{false};
+  std::shared_ptr<std::atomic<bool>> shutting_down_;
 
   std::atomic<Id> current_id_{0};
 };

Firestore/core/src/firebase/firestore/util/filesystem.h

@@ -37,6 +37,8 @@ class StatusOr;
  */
 class Filesystem {
  public:
+  virtual ~Filesystem() = default;
+
   Filesystem(const Filesystem&) = delete;
   Filesystem& operator=(const Filesystem&) = delete;