C++ migration: add AsyncQueue, the C++ version of FSTDispatchQueue (#1176)

AsyncQueue is a queue that executes given operations asynchronously,
enforcing that only a single operation is executing at any given time,
and that in-progress operations don't spawn more operations. The actual
execution is delegated to a platform-specific executor. Executor is an
interface for a FIFO queue that executes given operations serially. Two
implementations of Executor, one using libdispatch and the other using
C++11 standard library, are provided.

AsyncQueue is not used anywhere in the code base at this point.

10 files changed, 941 insertions, 0 deletions

diff --git a/Firestore/core/test/firebase/firestore/util/CMakeLists.txt b/Firestore/core/test/firebase/firestore/util/CMakeLists.txt
index e4da8d3..2e1e2f9 100644
--- a/Firestore/core/test/firebase/firestore/util/CMakeLists.txt
+++ b/Firestore/core/test/firebase/firestore/util/CMakeLists.txt
@@ -61,6 +61,60 @@ if(HAVE_OPENSSL_RAND_H)
   )
 endif()
 
+## executors
+
+cc_test(
+  firebase_firestore_util_executor_std_test
+  SOURCES
+    executor_test.h
+    executor_test.cc
+    executor_std_test.cc
+    async_tests_util.h
+  DEPENDS
+    firebase_firestore_util_executor_std
+)
+
+if(HAVE_LIBDISPATCH)
+  cc_test(
+    firebase_firestore_util_executor_libdispatch_test
+    SOURCES
+      executor_test.h
+      executor_test.cc
+      executor_libdispatch_test.cc
+      async_tests_util.h
+    DEPENDS
+      firebase_firestore_util_executor_libdispatch
+  )
+endif()
+
+## async queue
+
+cc_test(
+  firebase_firestore_util_async_queue_std_test
+  SOURCES
+    async_queue_test.h
+    async_queue_test.cc
+    async_queue_test_std.cc
+    async_tests_util.h
+  DEPENDS
+    firebase_firestore_util_executor_std
+    firebase_firestore_util_async_queue
+)
+
+if(HAVE_LIBDISPATCH)
+  cc_test(
+    firebase_firestore_util_async_queue_libdispatch_test
+    SOURCES
+      async_queue_test.h
+      async_queue_test.cc
+      async_queue_test_libdispatch.cc
+      async_tests_util.h
+    DEPENDS
+    firebase_firestore_util_executor_libdispatch
+    firebase_firestore_util_async_queue
+  )
+endif()
+
 ## main library
 
 cc_test(
diff --git a/Firestore/core/test/firebase/firestore/util/async_queue_test.cc b/Firestore/core/test/firebase/firestore/util/async_queue_test.cc
new file mode 100644
index 0000000..bcee2e3
--- /dev/null
+++ b/Firestore/core/test/firebase/firestore/util/async_queue_test.cc
@@ -0,0 +1,184 @@
+/*
+ * Copyright 2018 Google
+ *
+ * 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.
+ */
+
+#include "Firestore/core/test/firebase/firestore/util/async_queue_test.h"
+
+#include <chrono>  // NOLINT(build/c++11)
+#include <future>  // NOLINT(build/c++11)
+#include <string>
+
+#include "Firestore/core/src/firebase/firestore/util/executor.h"
+#include "absl/memory/memory.h"
+#include "gtest/gtest.h"
+
+namespace firebase {
+namespace firestore {
+namespace util {
+
+namespace {
+
+// In these generic tests the specific timer ids don't matter.
+const TimerId kTimerId1 = TimerId::ListenStreamConnectionBackoff;
+const TimerId kTimerId2 = TimerId::ListenStreamIdle;
+const TimerId kTimerId3 = TimerId::WriteStreamConnectionBackoff;
+
+}  // namespace
+
+TEST_P(AsyncQueueTest, Enqueue) {
+  queue.Enqueue([&] { signal_finished(); });
+  EXPECT_TRUE(WaitForTestToFinish());
+}
+
+TEST_P(AsyncQueueTest, EnqueueDisallowsNesting) {
+  queue.Enqueue([&] {  // clang-format off
+    // clang-format on
+    EXPECT_ANY_THROW(queue.Enqueue([] {}));
+    signal_finished();
+  });
+
+  EXPECT_TRUE(WaitForTestToFinish());
+}
+
+TEST_P(AsyncQueueTest, EnqueueRelaxedWorksFromWithinEnqueue) {
+  queue.Enqueue([&] {  // clang-format off
+    queue.EnqueueRelaxed([&] { signal_finished(); });
+    // clang-format on
+  });
+
+  EXPECT_TRUE(WaitForTestToFinish());
+}
+
+TEST_P(AsyncQueueTest, EnqueueBlocking) {
+  bool finished = false;
+  queue.EnqueueBlocking([&] { finished = true; });
+  EXPECT_TRUE(finished);
+}
+
+TEST_P(AsyncQueueTest, EnqueueBlockingDisallowsNesting) {
+  queue.EnqueueBlocking([&] {  // clang-format off
+    EXPECT_ANY_THROW(queue.EnqueueBlocking([] {}););
+    // clang-format on
+  });
+}
+
+TEST_P(AsyncQueueTest, ExecuteBlockingDisallowsNesting) {
+  queue.EnqueueBlocking(
+      [&] { EXPECT_ANY_THROW(queue.ExecuteBlocking([] {});); });
+}
+
+TEST_P(AsyncQueueTest, VerifyIsCurrentQueueWorksWithOperationInProgress) {
+  queue.EnqueueBlocking([&] { EXPECT_NO_THROW(queue.VerifyIsCurrentQueue()); });
+}
+
+TEST_P(AsyncQueueTest, CanScheduleOperationsInTheFuture) {
+  std::string steps;
+
+  queue.Enqueue([&steps] { steps += '1'; });
+  queue.Enqueue([&] {
+    queue.EnqueueAfterDelay(AsyncQueue::Milliseconds(5), kTimerId1, [&] {
+      steps += '4';
+      signal_finished();
+    });
+    queue.EnqueueAfterDelay(AsyncQueue::Milliseconds(1), kTimerId2,
+                            [&steps] { steps += '3'; });
+    queue.EnqueueRelaxed([&steps] { steps += '2'; });
+  });
+
+  EXPECT_TRUE(WaitForTestToFinish());
+  EXPECT_EQ(steps, "1234");
+}
+
+TEST_P(AsyncQueueTest, CanCancelDelayedOperations) {
+  std::string steps;
+
+  queue.Enqueue([&] {
+    // Queue everything from the queue to ensure nothing completes before we
+    // cancel.
+
+    queue.EnqueueRelaxed([&steps] { steps += '1'; });
+
+    DelayedOperation delayed_operation = queue.EnqueueAfterDelay(
+        AsyncQueue::Milliseconds(1), kTimerId1, [&steps] { steps += '2'; });
+
+    queue.EnqueueAfterDelay(AsyncQueue::Milliseconds(5), kTimerId2, [&] {
+      steps += '3';
+      signal_finished();
+    });
+
+    EXPECT_TRUE(queue.IsScheduled(kTimerId1));
+    delayed_operation.Cancel();
+    EXPECT_FALSE(queue.IsScheduled(kTimerId1));
+  });
+
+  EXPECT_TRUE(WaitForTestToFinish());
+  EXPECT_EQ(steps, "13");
+  EXPECT_FALSE(queue.IsScheduled(kTimerId1));
+}
+
+TEST_P(AsyncQueueTest, CanCallCancelOnDelayedOperationAfterTheOperationHasRun) {
+  DelayedOperation delayed_operation;
+  queue.Enqueue([&] {
+    delayed_operation = queue.EnqueueAfterDelay(
+        AsyncQueue::Milliseconds(10), kTimerId1, [&] { signal_finished(); });
+    EXPECT_TRUE(queue.IsScheduled(kTimerId1));
+  });
+
+  EXPECT_TRUE(WaitForTestToFinish());
+  EXPECT_FALSE(queue.IsScheduled(kTimerId1));
+  EXPECT_NO_THROW(delayed_operation.Cancel());
+}
+
+TEST_P(AsyncQueueTest, CanManuallyDrainAllDelayedOperationsForTesting) {
+  std::string steps;
+
+  queue.Enqueue([&] {
+    queue.EnqueueRelaxed([&steps] { steps += '1'; });
+    queue.EnqueueAfterDelay(AsyncQueue::Milliseconds(20000), kTimerId1,
+                            [&] { steps += '4'; });
+    queue.EnqueueAfterDelay(AsyncQueue::Milliseconds(10000), kTimerId2,
+                            [&steps] { steps += '3'; });
+    queue.EnqueueRelaxed([&steps] { steps += '2'; });
+    signal_finished();
+  });
+
+  EXPECT_TRUE(WaitForTestToFinish());
+  queue.RunScheduledOperationsUntil(TimerId::All);
+  EXPECT_EQ(steps, "1234");
+}
+
+TEST_P(AsyncQueueTest, CanManuallyDrainSpecificDelayedOperationsForTesting) {
+  std::string steps;
+
+  queue.Enqueue([&] {
+    queue.EnqueueRelaxed([&] { steps += '1'; });
+    queue.EnqueueAfterDelay(AsyncQueue::Milliseconds(20000), kTimerId1,
+                            [&steps] { steps += '5'; });
+    queue.EnqueueAfterDelay(AsyncQueue::Milliseconds(10000), kTimerId2,
+                            [&steps] { steps += '3'; });
+    queue.EnqueueAfterDelay(AsyncQueue::Milliseconds(15000), kTimerId3,
+                            [&steps] { steps += '4'; });
+    queue.EnqueueRelaxed([&] { steps += '2'; });
+    signal_finished();
+  });
+
+  EXPECT_TRUE(WaitForTestToFinish());
+  queue.RunScheduledOperationsUntil(kTimerId3);
+  EXPECT_EQ(steps, "1234");
+}
+
+}  // namespace util
+}  // namespace firestore
+}  // namespace firebase
diff --git a/Firestore/core/test/firebase/firestore/util/async_queue_test.h b/Firestore/core/test/firebase/firestore/util/async_queue_test.h
new file mode 100644
index 0000000..61c7ab6
--- /dev/null
+++ b/Firestore/core/test/firebase/firestore/util/async_queue_test.h
@@ -0,0 +1,47 @@
+/*
+ * Copyright 2018 Google
+ *
+ * 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_TEST_FIREBASE_FIRESTORE_UTIL_ASYNC_QUEUE_TEST_H_
+#define FIRESTORE_CORE_TEST_FIREBASE_FIRESTORE_UTIL_ASYNC_QUEUE_TEST_H_
+
+#include <memory>
+
+#include "gtest/gtest.h"
+
+#include "Firestore/core/src/firebase/firestore/util/async_queue.h"
+#include "Firestore/core/test/firebase/firestore/util/async_tests_util.h"
+
+namespace firebase {
+namespace firestore {
+namespace util {
+
+using FactoryFunc = std::unique_ptr<internal::Executor> (*)();
+
+class AsyncQueueTest : public TestWithTimeoutMixin,
+                       public ::testing::TestWithParam<FactoryFunc> {
+ public:
+  // `GetParam()` must return a factory function.
+  AsyncQueueTest() : queue{GetParam()()} {
+  }
+
+  AsyncQueue queue;
+};
+
+}  // namespace util
+}  // namespace firestore
+}  // namespace firebase
+
+#endif  // FIRESTORE_CORE_TEST_FIREBASE_FIRESTORE_UTIL_ASYNC_QUEUE_TEST_H_
diff --git a/Firestore/core/test/firebase/firestore/util/async_queue_test_libdispatch.cc b/Firestore/core/test/firebase/firestore/util/async_queue_test_libdispatch.cc
new file mode 100644
index 0000000..b4b9c63
--- /dev/null
+++ b/Firestore/core/test/firebase/firestore/util/async_queue_test_libdispatch.cc
@@ -0,0 +1,86 @@
+/*
+ * Copyright 2018 Google
+ *
+ * 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.
+ */
+
+#include "Firestore/core/test/firebase/firestore/util/async_queue_test.h"
+
+#include "Firestore/core/src/firebase/firestore/util/executor_libdispatch.h"
+
+#include "absl/memory/memory.h"
+#include "gtest/gtest.h"
+
+namespace firebase {
+namespace firestore {
+namespace util {
+
+namespace {
+
+dispatch_queue_t CreateDispatchQueue() {
+  return dispatch_queue_create("AsyncQueueTests", DISPATCH_QUEUE_SERIAL);
+}
+
+std::unique_ptr<internal::Executor> CreateExecutorFromQueue(
+    const dispatch_queue_t queue) {
+  return absl::make_unique<internal::ExecutorLibdispatch>(queue);
+}
+
+std::unique_ptr<internal::Executor> CreateExecutorLibdispatch() {
+  return CreateExecutorFromQueue(CreateDispatchQueue());
+}
+
+}  // namespace
+
+INSTANTIATE_TEST_CASE_P(AsyncQueueLibdispatch,
+                        AsyncQueueTest,
+                        ::testing::Values(CreateExecutorLibdispatch));
+
+class AsyncQueueTestLibdispatchOnly : public TestWithTimeoutMixin,
+                                      public ::testing::Test {
+ public:
+  AsyncQueueTestLibdispatchOnly()
+      : underlying_queue{CreateDispatchQueue()},
+        queue{CreateExecutorFromQueue(underlying_queue)} {
+  }
+
+  dispatch_queue_t underlying_queue;
+  AsyncQueue queue;
+};
+
+// Additional tests to see how libdispatch-based version of `AsyncQueue`
+// interacts with raw usage of libdispatch.
+
+TEST_F(AsyncQueueTestLibdispatchOnly, SameQueueIsAllowedForUnownedActions) {
+  internal::DispatchAsync(underlying_queue, [this] {
+    queue.Enqueue([this] { signal_finished(); });
+  });
+  EXPECT_TRUE(WaitForTestToFinish());
+}
+
+TEST_F(AsyncQueueTestLibdispatchOnly,
+       VerifyIsCurrentQueueRequiresOperationInProgress) {
+  internal::DispatchSync(underlying_queue, [this] {
+    EXPECT_ANY_THROW(queue.VerifyIsCurrentQueue());
+  });
+}
+
+TEST_F(AsyncQueueTestLibdispatchOnly,
+       VerifyIsCurrentQueueRequiresBeingCalledAsync) {
+  ASSERT_NE(underlying_queue, dispatch_get_main_queue());
+  EXPECT_ANY_THROW(queue.VerifyIsCurrentQueue());
+}
+
+}  // namespace util
+}  // namespace firestore
+}  // namespace firebase
diff --git a/Firestore/core/test/firebase/firestore/util/async_queue_test_std.cc b/Firestore/core/test/firebase/firestore/util/async_queue_test_std.cc
new file mode 100644
index 0000000..9e69ad0
--- /dev/null
+++ b/Firestore/core/test/firebase/firestore/util/async_queue_test_std.cc
@@ -0,0 +1,41 @@
+/*
+ * Copyright 2018 Google
+ *
+ * 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.
+ */
+
+#include "Firestore/core/test/firebase/firestore/util/async_queue_test.h"
+
+#include "Firestore/core/src/firebase/firestore/util/executor_std.h"
+
+#include "absl/memory/memory.h"
+#include "gtest/gtest.h"
+
+namespace firebase {
+namespace firestore {
+namespace util {
+
+namespace {
+
+std::unique_ptr<internal::Executor> ExecutorFactory() {
+  return absl::make_unique<internal::ExecutorStd>();
+}
+
+}  // namespace
+
+INSTANTIATE_TEST_CASE_P(AsyncQueueStd,
+                        AsyncQueueTest,
+                        ::testing::Values(ExecutorFactory));
+}  // namespace util
+}  // namespace firestore
+}  // namespace firebase
diff --git a/Firestore/core/test/firebase/firestore/util/async_tests_util.h b/Firestore/core/test/firebase/firestore/util/async_tests_util.h
new file mode 100644
index 0000000..422745b
--- /dev/null
+++ b/Firestore/core/test/firebase/firestore/util/async_tests_util.h
@@ -0,0 +1,90 @@
+/*
+ * Copyright 2018 Google
+ *
+ * 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_TEST_FIREBASE_FIRESTORE_UTIL_ASYNC_TESTS_UTIL_H_
+#define FIRESTORE_CORE_TEST_FIREBASE_FIRESTORE_UTIL_ASYNC_TESTS_UTIL_H_
+
+#include <chrono>  // NOLINT(build/c++11)
+#include <cstdlib>
+#include <future>  // NOLINT(build/c++11)
+
+#include "gtest/gtest.h"
+
+namespace firebase {
+namespace firestore {
+namespace util {
+
+inline std::chrono::time_point<std::chrono::system_clock,
+                               std::chrono::milliseconds>
+now() {
+  return std::chrono::time_point_cast<std::chrono::milliseconds>(
+      std::chrono::system_clock::now());
+}
+
+constexpr auto kTimeout = std::chrono::seconds(5);
+
+// Waits for the future to become ready and returns whether it timed out.
+inline bool Await(const std::future<void>& future,
+                  const std::chrono::milliseconds timeout = kTimeout) {
+  return future.wait_for(timeout) == std::future_status::ready;
+}
+
+// Unfortunately, the future returned from std::async blocks in its destructor
+// until the async call is finished. If the function called from std::async is
+// buggy and hangs forever, the future's destructor will also hang forever. To
+// avoid all tests freezing, the only thing to do is to abort (which skips
+// destructors).
+inline void Abort() {
+  ADD_FAILURE();
+  std::abort();
+}
+
+// Calls std::abort if the future times out.
+inline void AbortOnTimeout(const std::future<void>& future) {
+  if (!Await(future, kTimeout)) {
+    Abort();
+  }
+}
+
+// The macro calls AbortOnTimeout, but preserves stack trace.
+#define ABORT_ON_TIMEOUT(future)                            \
+  do {                                                      \
+    SCOPED_TRACE("Async operation timed out, aborting..."); \
+    AbortOnTimeout(future);                                 \
+  } while (0)
+
+class TestWithTimeoutMixin {
+ public:
+  TestWithTimeoutMixin() : signal_finished{[] {}} {
+  }
+
+  // Googletest doesn't contain built-in functionality to block until an async
+  // operation completes, and there is no timeout by default. Work around both
+  // by resolving a packaged_task in the async operation and blocking on the
+  // associated future (with timeout).
+  bool WaitForTestToFinish(const std::chrono::seconds timeout = kTimeout) {
+    return signal_finished.get_future().wait_for(timeout) ==
+           std::future_status::ready;
+  }
+
+  std::packaged_task<void()> signal_finished;
+};
+
+}  // namespace util
+}  // namespace firestore
+}  // namespace firebase
+
+#endif  // FIRESTORE_CORE_TEST_FIREBASE_FIRESTORE_UTIL_ASYNC_TESTS_UTIL_H_
diff --git a/Firestore/core/test/firebase/firestore/util/executor_libdispatch_test.cc b/Firestore/core/test/firebase/firestore/util/executor_libdispatch_test.cc
new file mode 100644
index 0000000..0167c83
--- /dev/null
+++ b/Firestore/core/test/firebase/firestore/util/executor_libdispatch_test.cc
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2018 Google
+ *
+ * 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.
+ */
+
+#include "Firestore/core/test/firebase/firestore/util/executor_test.h"
+
+#include <memory>
+
+#include "Firestore/core/src/firebase/firestore/util/executor_libdispatch.h"
+#include "absl/memory/memory.h"
+#include "gtest/gtest.h"
+
+namespace firebase {
+namespace firestore {
+namespace util {
+
+namespace {
+
+std::unique_ptr<internal::Executor> ExecutorFactory() {
+  return absl::make_unique<internal::ExecutorLibdispatch>();
+}
+
+}  // namespace
+
+INSTANTIATE_TEST_CASE_P(ExecutorTestLibdispatch,
+                        ExecutorTest,
+                        ::testing::Values(ExecutorFactory));
+
+}  // namespace util
+}  // namespace firestore
+}  // namespace firebase
diff --git a/Firestore/core/test/firebase/firestore/util/executor_std_test.cc b/Firestore/core/test/firebase/firestore/util/executor_std_test.cc
new file mode 100644
index 0000000..43cad60
--- /dev/null
+++ b/Firestore/core/test/firebase/firestore/util/executor_std_test.cc
@@ -0,0 +1,240 @@
+/*
+ * Copyright 2018 Google
+ *
+ * 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.
+ */
+
+#include "Firestore/core/test/firebase/firestore/util/executor_test.h"
+
+#include <chrono>  // NOLINT(build/c++11)
+#include <cstdlib>
+#include <future>  // NOLINT(build/c++11)
+#include <string>
+#include <thread>  // NOLINT(build/c++11)
+
+#include "Firestore/core/src/firebase/firestore/util/executor_std.h"
+#include "Firestore/core/test/firebase/firestore/util/async_tests_util.h"
+#include "absl/memory/memory.h"
+#include "gtest/gtest.h"
+
+namespace firebase {
+namespace firestore {
+namespace util {
+
+namespace chr = std::chrono;
+using async::Schedule;
+
+class ScheduleTest : public ::testing::Test {
+ public:
+  ScheduleTest() : start_time{now()} {
+  }
+
+  using ScheduleT = Schedule<int>;
+
+  ScheduleT schedule;
+  ScheduleT::TimePoint start_time;
+};
+
+// Schedule tests
+
+TEST_F(ScheduleTest, PopIfDue_Immediate) {
+  EXPECT_FALSE(schedule.PopIfDue().has_value());
+
+  // Push values in a deliberately non-sorted order.
+  schedule.Push(3, start_time);
+  schedule.Push(1, start_time);
+  schedule.Push(2, start_time);
+  EXPECT_FALSE(schedule.empty());
+  EXPECT_EQ(schedule.size(), 3u);
+
+  EXPECT_EQ(schedule.PopIfDue().value(), 3);
+  EXPECT_EQ(schedule.PopIfDue().value(), 1);
+  EXPECT_EQ(schedule.PopIfDue().value(), 2);
+  EXPECT_FALSE(schedule.PopIfDue().has_value());
+  EXPECT_TRUE(schedule.empty());
+  EXPECT_EQ(schedule.size(), 0u);
+}
+
+TEST_F(ScheduleTest, PopIfDue_Delayed) {
+  schedule.Push(1, start_time + chr::milliseconds(5));
+  schedule.Push(2, start_time + chr::milliseconds(3));
+  schedule.Push(3, start_time + chr::milliseconds(1));
+
+  EXPECT_FALSE(schedule.PopIfDue().has_value());
+  std::this_thread::sleep_for(chr::milliseconds(5));
+
+  EXPECT_EQ(schedule.PopIfDue().value(), 3);
+  EXPECT_EQ(schedule.PopIfDue().value(), 2);
+  EXPECT_EQ(schedule.PopIfDue().value(), 1);
+  EXPECT_TRUE(schedule.empty());
+}
+
+TEST_F(ScheduleTest, PopBlocking) {
+  schedule.Push(1, start_time + chr::milliseconds(3));
+  EXPECT_FALSE(schedule.PopIfDue().has_value());
+
+  EXPECT_EQ(schedule.PopBlocking(), 1);
+  EXPECT_GE(now(), start_time + chr::milliseconds(3));
+  EXPECT_TRUE(schedule.empty());
+}
+
+TEST_F(ScheduleTest, RemoveIf) {
+  schedule.Push(1, start_time);
+  schedule.Push(2, now() + chr::minutes(1));
+
+  auto maybe_removed = schedule.RemoveIf([](const int v) { return v == 1; });
+  EXPECT_TRUE(maybe_removed.has_value());
+  EXPECT_EQ(maybe_removed.value(), 1);
+
+  // Non-existent value.
+  maybe_removed = schedule.RemoveIf([](const int v) { return v == 1; });
+  EXPECT_FALSE(maybe_removed.has_value());
+
+  maybe_removed = schedule.RemoveIf([](const int v) { return v == 2; });
+  EXPECT_TRUE(maybe_removed.has_value());
+  EXPECT_EQ(maybe_removed.value(), 2);
+  EXPECT_TRUE(schedule.empty());
+}
+
+TEST_F(ScheduleTest, Ordering) {
+  schedule.Push(11, start_time + chr::milliseconds(5));
+  schedule.Push(1, start_time);
+  schedule.Push(2, start_time);
+  schedule.Push(9, start_time + chr::milliseconds(2));
+  schedule.Push(3, start_time);
+  schedule.Push(10, start_time + chr::milliseconds(3));
+  schedule.Push(12, start_time + chr::milliseconds(5));
+  schedule.Push(4, start_time);
+  schedule.Push(5, start_time);
+  schedule.Push(6, start_time);
+  schedule.Push(8, start_time + chr::milliseconds(1));
+  schedule.Push(7, start_time);
+
+  std::vector<int> values;
+  while (!schedule.empty()) {
+    values.push_back(schedule.PopBlocking());
+  }
+  const std::vector<int> expected = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12};
+  EXPECT_EQ(values, expected);
+}
+
+TEST_F(ScheduleTest, AddingEntryUnblocksEmptyQueue) {
+  const auto future = std::async(std::launch::async, [&] {
+    ASSERT_FALSE(schedule.PopIfDue().has_value());
+    EXPECT_EQ(schedule.PopBlocking(), 1);
+  });
+
+  std::this_thread::sleep_for(chr::milliseconds(5));
+  schedule.Push(1, start_time);
+  ABORT_ON_TIMEOUT(future);
+}
+
+TEST_F(ScheduleTest, PopBlockingUnblocksOnNewPastDueEntries) {
+  const auto far_away = start_time + chr::seconds(10);
+  schedule.Push(5, far_away);
+
+  const auto future = std::async(std::launch::async, [&] {
+    ASSERT_FALSE(schedule.PopIfDue().has_value());
+    EXPECT_EQ(schedule.PopBlocking(), 3);
+  });
+
+  std::this_thread::sleep_for(chr::milliseconds(5));
+  schedule.Push(3, start_time);
+  ABORT_ON_TIMEOUT(future);
+}
+
+TEST_F(ScheduleTest, PopBlockingAdjustsWaitTimeOnNewSoonerEntries) {
+  const auto far_away = start_time + chr::seconds(10);
+  schedule.Push(5, far_away);
+
+  const auto future = std::async(std::launch::async, [&] {
+    ASSERT_FALSE(schedule.PopIfDue().has_value());
+    EXPECT_EQ(schedule.PopBlocking(), 3);
+    // Make sure schedule hasn't been waiting longer than necessary.
+    EXPECT_LT(now(), far_away);
+  });
+
+  std::this_thread::sleep_for(chr::milliseconds(5));
+  schedule.Push(3, start_time + chr::milliseconds(100));
+  ABORT_ON_TIMEOUT(future);
+}
+
+TEST_F(ScheduleTest, PopBlockingCanReadjustTimeIfSeveralElementsAreAdded) {
+  const auto far_away = start_time + chr::seconds(5);
+  const auto very_far_away = start_time + chr::seconds(10);
+  schedule.Push(3, very_far_away);
+
+  const auto future = std::async(std::launch::async, [&] {
+    ASSERT_FALSE(schedule.PopIfDue().has_value());
+    EXPECT_EQ(schedule.PopBlocking(), 1);
+    EXPECT_LT(now(), far_away);
+  });
+
+  std::this_thread::sleep_for(chr::milliseconds(5));
+  schedule.Push(2, far_away);
+  std::this_thread::sleep_for(chr::milliseconds(1));
+  schedule.Push(1, start_time + chr::milliseconds(100));
+  ABORT_ON_TIMEOUT(future);
+}
+
+TEST_F(ScheduleTest, PopBlockingNoticesRemovals) {
+  const auto future = std::async(std::launch::async, [&] {
+    schedule.Push(1, start_time + chr::milliseconds(50));
+    schedule.Push(2, start_time + chr::milliseconds(100));
+    ASSERT_FALSE(schedule.PopIfDue().has_value());
+    EXPECT_EQ(schedule.PopBlocking(), 2);
+  });
+
+  while (schedule.empty()) {
+    std::this_thread::sleep_for(chr::milliseconds(1));
+  }
+  const auto maybe_removed =
+      schedule.RemoveIf([](const int v) { return v == 1; });
+  EXPECT_EQ(maybe_removed.value(), 1);
+  ABORT_ON_TIMEOUT(future);
+}
+
+TEST_F(ScheduleTest, PopBlockingIsNotAffectedByIrrelevantRemovals) {
+  const auto future = std::async(std::launch::async, [&] {
+    schedule.Push(1, start_time + chr::milliseconds(50));
+    schedule.Push(2, start_time + chr::seconds(10));
+    ASSERT_FALSE(schedule.PopIfDue().has_value());
+    EXPECT_EQ(schedule.PopBlocking(), 1);
+  });
+
+  while (schedule.empty()) {
+    std::this_thread::sleep_for(chr::milliseconds(1));
+  }
+  const auto maybe_removed =
+      schedule.RemoveIf([](const int v) { return v == 2; });
+  EXPECT_EQ(maybe_removed.value(), 2);
+  ABORT_ON_TIMEOUT(future);
+}
+
+// ExecutorStd tests
+
+namespace {
+
+inline std::unique_ptr<internal::Executor> ExecutorFactory() {
+  return absl::make_unique<internal::ExecutorStd>();
+}
+
+}  // namespace
+
+INSTANTIATE_TEST_CASE_P(ExecutorTestStd,
+                        ExecutorTest,
+                        ::testing::Values(ExecutorFactory));
+
+}  // namespace util
+}  // namespace firestore
+}  // namespace firebase
diff --git a/Firestore/core/test/firebase/firestore/util/executor_test.cc b/Firestore/core/test/firebase/firestore/util/executor_test.cc
new file mode 100644
index 0000000..5cf389b
--- /dev/null
+++ b/Firestore/core/test/firebase/firestore/util/executor_test.cc
@@ -0,0 +1,110 @@
+/*
+ * Copyright 2018 Google
+ *
+ * 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.
+ */
+
+#include "Firestore/core/test/firebase/firestore/util/executor_test.h"
+
+#include <chrono>  // NOLINT(build/c++11)
+#include <cstdlib>
+#include <future>  // NOLINT(build/c++11)
+#include <string>
+#include <thread>  // NOLINT(build/c++11)
+
+#include "Firestore/core/src/firebase/firestore/util/executor.h"
+#include "gtest/gtest.h"
+
+namespace firebase {
+namespace firestore {
+namespace util {
+
+namespace chr = std::chrono;
+using internal::Executor;
+
+namespace {
+
+DelayedOperation Schedule(Executor* const executor,
+                          const Executor::Milliseconds delay,
+                          Executor::Operation&& operation) {
+  const Executor::Tag no_tag = -1;
+  return executor->Schedule(
+      delay, Executor::TaggedOperation{no_tag, std::move(operation)});
+}
+
+}  // namespace
+
+TEST_P(ExecutorTest, Execute) {
+  executor->Execute([&] { signal_finished(); });
+  EXPECT_TRUE(WaitForTestToFinish());
+}
+
+TEST_P(ExecutorTest, DestructorDoesNotBlockIfThereArePendingTasks) {
+  const auto future = std::async(std::launch::async, [&] {
+    auto another_executor = GetParam()();
+    Schedule(another_executor.get(), chr::minutes(5), [] {});
+    Schedule(another_executor.get(), chr::minutes(10), [] {});
+    // Destructor shouldn't block waiting for the 5/10-minute-away operations.
+  });
+
+  ABORT_ON_TIMEOUT(future);
+}
+
+TEST_P(ExecutorTest, CanScheduleOperationsInTheFuture) {
+  std::string steps;
+
+  executor->Execute([&steps] { steps += '1'; });
+  Schedule(executor.get(), Executor::Milliseconds(5), [&] {
+    steps += '4';
+    signal_finished();
+  });
+  Schedule(executor.get(), Executor::Milliseconds(1),
+           [&steps] { steps += '3'; });
+  executor->Execute([&steps] { steps += '2'; });
+
+  EXPECT_TRUE(WaitForTestToFinish());
+  EXPECT_EQ(steps, "1234");
+}
+
+TEST_P(ExecutorTest, CanCancelDelayedOperations) {
+  std::string steps;
+
+  executor->Execute([&] {
+    executor->Execute([&steps] { steps += '1'; });
+
+    DelayedOperation delayed_operation = Schedule(
+        executor.get(), Executor::Milliseconds(1), [&steps] { steps += '2'; });
+
+    Schedule(executor.get(), Executor::Milliseconds(5), [&] {
+      steps += '3';
+      signal_finished();
+    });
+
+    delayed_operation.Cancel();
+  });
+
+  EXPECT_TRUE(WaitForTestToFinish());
+  EXPECT_EQ(steps, "13");
+}
+
+TEST_P(ExecutorTest, DelayedOperationIsValidAfterTheOperationHasRun) {
+  DelayedOperation delayed_operation = Schedule(
+      executor.get(), Executor::Milliseconds(1), [&] { signal_finished(); });
+
+  EXPECT_TRUE(WaitForTestToFinish());
+  EXPECT_NO_THROW(delayed_operation.Cancel());
+}
+
+}  // namespace util
+}  // namespace firestore
+}  // namespace firebase
diff --git a/Firestore/core/test/firebase/firestore/util/executor_test.h b/Firestore/core/test/firebase/firestore/util/executor_test.h
new file mode 100644
index 0000000..8b78d50
--- /dev/null
+++ b/Firestore/core/test/firebase/firestore/util/executor_test.h
@@ -0,0 +1,46 @@
+/*
+ * Copyright 2018 Google
+ *
+ * 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_TEST_FIREBASE_FIRESTORE_UTIL_EXECUTOR_TEST_H_
+#define FIRESTORE_CORE_TEST_FIREBASE_FIRESTORE_UTIL_EXECUTOR_TEST_H_
+
+#include <memory>
+
+#include "gtest/gtest.h"
+
+#include "Firestore/core/src/firebase/firestore/util/executor.h"
+#include "Firestore/core/test/firebase/firestore/util/async_tests_util.h"
+
+namespace firebase {
+namespace firestore {
+namespace util {
+
+using FactoryFunc = std::unique_ptr<internal::Executor> (*)();
+
+class ExecutorTest : public TestWithTimeoutMixin,
+                     public ::testing::TestWithParam<FactoryFunc> {
+ public:
+  // `GetParam()` must return a factory function.
+  ExecutorTest() : executor{GetParam()()} {
+  }
+
+  std::unique_ptr<internal::Executor> executor;
+};
+}  // namespace util
+}  // namespace firestore
+}  // namespace firebase
+
+#endif  // FIRESTORE_CORE_TEST_FIREBASE_FIRESTORE_UTIL_EXECUTOR_TEST_H_