Make Executors have deterministic shutdown

Firestore/core/src/util/async_queue.cc

@@ -19,6 +19,7 @@
 #include <utility>
 
 #include "Firestore/core/src/util/hard_assert.h"
+#include "Firestore/core/src/util/task.h"
 #include "absl/algorithm/container.h"
 #include "absl/memory/memory.h"
 
@@ -119,8 +120,8 @@ DelayedOperation AsyncQueue::EnqueueAfterDelay(Milliseconds delay,
     delay = Milliseconds(0);
   }
 
-  Executor::TaggedOperation tagged{static_cast<int>(timer_id), Wrap(operation)};
-  return executor_->Schedule(delay, std::move(tagged));
+  auto tag = static_cast<Executor::Tag>(timer_id);
+  return executor_->Schedule(delay, tag, Wrap(operation));
 }
 
 AsyncQueue::Operation AsyncQueue::Wrap(const Operation& operation) {
@@ -162,10 +163,12 @@ void AsyncQueue::RunScheduledOperationsUntil(const TimerId last_timer_id) {
         "Attempted to run scheduled operations until missing timer id: %s",
         last_timer_id);
 
-    for (auto next = executor_->PopFromSchedule(); next.has_value();
+    for (auto next = executor_->PopFromSchedule(); next != nullptr;
          next = executor_->PopFromSchedule()) {
-      next->operation();
-      if (next->tag == static_cast<int>(last_timer_id)) {
+      bool should_break = next->tag() == static_cast<int>(last_timer_id);
+
+      next->Execute();
+      if (should_break) {
         break;
       }
     }

Firestore/core/src/util/async_queue.h

@@ -112,7 +112,7 @@ class AsyncQueue : public std::enable_shared_from_this<AsyncQueue> {
   // Like `Enqueue`, but also starts the shutdown process. Once the shutdown
   // process has started, calling any Enqueue* methods becomes a no-op
   //
-  // The exception is `EnqueueEvenAfterShutdown`, operations requsted via
+  // The exception is `EnqueueEvenAfterShutdown`, operations requested via
   // this will still be scheduled.
   void EnqueueAndInitiateShutdown(const Operation& operation);
 

Firestore/core/src/util/defer.h

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2020 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.
+ */
+
+#ifndef FIRESTORE_CORE_SRC_UTIL_DEFER_H_
+#define FIRESTORE_CORE_SRC_UTIL_DEFER_H_
+
+#include <utility>
+
+namespace firebase {
+namespace firestore {
+namespace util {
+
+template <typename Action>
+class Deferred;
+
+/**
+ * Creates a deferred action that will be destroyed at the close of the lexical
+ * scope containing the result of this function call. This is useful for
+ * performing ad-hoc RAII-style actions, without having to create the wrapper
+ * class. For example:
+ *
+ *     FILE* file = fopen(filename, "rb");
+ *     auto cleanup = defer([&] {
+ *       if (file) {
+ *         fclose(file);
+ *       }
+ *     });
+ *
+ * @param action a callable object; usually a lambda. Even if exceptions are
+ *     enabled, when `action` is invoked it must not throw. This is similar to
+ *     the restriction that exists on destructors generally.
+ */
+template <typename Action>
+Deferred<Action> defer(Action&& action) {
+  return Deferred<Action>(std::forward<Action>(action));
+}
+
+/**
+ * Storage for a deferred action. The `action` is invoked during the destructor
+ * of the `Deferred`.
+ */
+template <typename Action>
+class Deferred {
+ public:
+  explicit Deferred(Action&& action) : action_(std::move(action)) {
+  }
+
+  ~Deferred() {
+    action_();
+  }
+
+ private:
+  Action action_;
+};
+
+}  // namespace util
+}  // namespace firestore
+}  // namespace firebase
+
+#endif  // FIRESTORE_CORE_SRC_UTIL_DEFER_H_

Firestore/core/src/util/executor.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.
@@ -21,46 +21,13 @@
 #include <functional>
 #include <memory>
 #include <string>
-#include <utility>
-
-#include "absl/types/optional.h"
 
 namespace firebase {
 namespace firestore {
 namespace util {
 
-// A handle to an operation scheduled for future execution. The handle may
-// outlive the operation, but it *cannot* outlive the executor that created it.
-class DelayedOperation {
- public:
-  // Creates an empty `DelayedOperation` not associated with any actual
-  // operation. Calling `Cancel` on it is a no-op.
-  DelayedOperation() {
-  }
-
-  // Returns whether this `DelayedOperation` is associated with an actual
-  // operation.
-  explicit operator bool() const {
-    return static_cast<bool>(cancel_func_);
-  }
-
-  // If the operation has not been run yet, cancels the operation. Otherwise,
-  // this function is a no-op.
-  void Cancel() {
-    if (cancel_func_) {
-      cancel_func_();
-      cancel_func_ = {};
-    }
-  }
-
-  // Internal use only.
-  explicit DelayedOperation(std::function<void()>&& cancel_func)
-      : cancel_func_{std::move(cancel_func)} {
-  }
-
- private:
-  std::function<void()> cancel_func_;
-};
+class DelayedOperation;
+class Task;
 
 // An interface to a platform-specific executor of asynchronous operations
 // (called tasks on other platforms).
@@ -74,22 +41,20 @@ class DelayedOperation {
 // Delayed operations may be canceled if they have not already been run.
 class Executor {
  public:
+  // An opaque name for a kind of operation. All operations of the same type
+  // should share a tag.
   using Tag = int;
+  static constexpr Tag kNoTag = -1;
+
+  // An opaque, monotonically increasing identifier for each operation that does
+  // not depend on their address. Where the `Tag` identifies the kind of
+  // operation, the `Id` identifies the specific instance.
+  using Id = uint32_t;
   using Operation = std::function<void()>;
-  using Milliseconds = std::chrono::milliseconds;
 
-  // Operations scheduled for future execution have an opaque tag. The value of
-  // the tag is ignored by the executor but can be used to find operations with
-  // a given tag after they are scheduled.
-  struct TaggedOperation {
-    TaggedOperation() {
-    }
-    TaggedOperation(const Tag tag, Operation&& operation)
-        : tag{tag}, operation{std::move(operation)} {
-    }
-    Tag tag = 0;
-    Operation operation;
-  };
+  using Milliseconds = std::chrono::milliseconds;
+  using Clock = std::chrono::steady_clock;
+  using TimePoint = std::chrono::time_point<Clock, Milliseconds>;
 
   // Creates a new serial Executor of the platform-appropriate type, and gives
   // it the given label, if the implementation supports it.
@@ -113,15 +78,20 @@ class Executor {
   // Like `Execute`, but blocks until the `operation` finishes, consequently
   // draining immediate operations from the executor.
   virtual void ExecuteBlocking(Operation&& operation) = 0;
+
   // Scheduled the given `operation` to be executed after `delay` milliseconds
   // from now, and returns a handle that allows to cancel the operation
-  // (provided it hasn't been run already). The operation is tagged to allow
-  // retrieving it later.
+  // (provided it hasn't been run already).
+  //
+  // Operations scheduled for future execution have an opaque tag. The value of
+  // the tag is ignored by the executor but can be used to find operations with
+  // a given tag after they are scheduled.
   //
   // `delay` must be non-negative; use `Execute` to schedule operations for
   // immediate execution.
   virtual DelayedOperation Schedule(Milliseconds delay,
-                                    TaggedOperation&& operation) = 0;
+                                    Tag tag,
+                                    Operation&& operation) = 0;
 
   // Checks for the caller whether it is being invoked by this executor.
   virtual bool IsCurrentExecutor() const = 0;
@@ -136,12 +106,68 @@ class Executor {
   // Checks whether an operation tagged with the given `tag` is currently
   // scheduled for future execution.
   virtual bool IsScheduled(Tag tag) const = 0;
+  virtual bool IsTaskScheduled(Id id) const = 0;
+
   // Removes the nearest due scheduled operation from the schedule and returns
-  // it to the caller. This function may be used to reschedule operations.
-  // Immediate operations don't count; only operations scheduled for delayed
-  // execution may be removed. If no such operations are currently scheduled, an
-  // empty `optional` is returned.
-  virtual absl::optional<TaggedOperation> PopFromSchedule() = 0;
+  // it to the caller.
+  //
+  // Only operations scheduled for delayed execution can be removed with this
+  // method; immediate operations don't count. If no such operations are
+  // currently scheduled, `nullptr` is returned.
+  //
+  // The caller is responsible for either Executing or Canceling (and Releasing)
+  // the returned Task.
+  virtual Task* PopFromSchedule() = 0;
+
+ private:
+  // Mark a task completed, removing it from any internal schedule or tracking.
+  //
+  // Called by Task once it has completed execution. Implementations of
+  // `Complete` should not call back to the Task: it is responsible for marking
+  // itself completed.
+  virtual void Complete(Task* task) = 0;
+  friend class Task;
+
+  // If the operation hasn't yet been run, it will be removed from the queue.
+  // Otherwise, this function is a no-op.
+  //
+  // Called by `DelayedOperation` when its user calls `Cancel`. Implementations
+  // of `Cancel` should also `Dispose` the underlying `Task` to actually prevent
+  // execution.
+  virtual void Cancel(Id operation_id) = 0;
+  friend class DelayedOperation;
+};
+
+// A handle to an operation scheduled for future execution. The handle may
+// outlive the operation, but it *cannot* outlive the executor that created it.
+class DelayedOperation {
+ public:
+  // Creates an empty `DelayedOperation` not associated with any actual
+  // operation. Calling `Cancel` on it is a no-op.
+  DelayedOperation() = default;
+
+  // Returns whether this `DelayedOperation` is associated with an actual
+  // operation.
+  explicit operator bool() const {
+    return executor_ && executor_->IsTaskScheduled(id_);
+  }
+
+  // If the operation has not been run yet, cancels the operation. Otherwise,
+  // this function is a no-op.
+  void Cancel() {
+    if (executor_) {
+      executor_->Cancel(id_);
+    }
+  }
+
+  // Internal use only.
+  explicit DelayedOperation(Executor* executor, Executor::Id id)
+      : executor_(executor), id_(id) {
+  }
+
+ private:
+  Executor* executor_ = nullptr;
+  Executor::Id id_ = 0;
 };
 
 }  // namespace util