iface/common/task.hh

@@ -233,6 +233,12 @@ class SimpleTaskQueue : public TaskQueue<T> {
     cv_.notify_all();
   }
 
+  // THREAD-SAFE
+  bool WaitForEmpty(auto dur) noexcept {
+    std::unique_lock<std::mutex> k {mtx_};
+    return cv_.wait_for(k, dur, [this]() { return tasks_.empty(); });
+  }
+
   template <TaskDriverLike<Item> Driver>
   void Drive(Driver& driver) {
     while (!driver.nextIdleInterruption()) {
@@ -257,6 +263,7 @@ class SimpleTaskQueue : public TaskQueue<T> {
 
       try {
         std::unique_lock<std::mutex> k{mtx_};
+        cv_.notify_all();
 
         const auto until = nextAwake();
         const auto pred  = [&]() {

iface/common/task_test.cc

@@ -281,3 +281,36 @@ TEST(SimpleTaskQueue, ChaoticPushAndDrive) {
     EXPECT_EQ(execCount, kPushPerThread);
   }
 }
+
+TEST(SimpleTaskQueue, WaitForEmpty) {
+  nf7::test::SimpleTaskQueueMock<nf7::Task<int32_t&>> sut;
+  EXPECT_CALL(sut, onErrorWhilePush).Times(0);
+
+  // use NiceMock to suppress annoying warnings that slowed unittests
+  ::testing::NiceMock<
+      nf7::test::SimpleTaskQueueDriverMock<nf7::Task<int32_t&>>> driver;
+
+  for (uint32_t i = 0; i < 1000; ++i) {
+    sut.Exec([](auto&){});
+  }
+
+  auto ctx = int32_t {0};
+  ON_CALL(driver, Drive)
+      .WillByDefault([&](auto&& task) { task(ctx); });
+
+  std::atomic<bool> exit = false;
+  ON_CALL(driver, nextIdleInterruption)
+      .WillByDefault([&]() -> bool { return exit; });
+
+  std::thread th {[&]() { sut.Drive(driver); }};
+  EXPECT_TRUE(sut.WaitForEmpty(1s));
+
+  exit = true;
+  sut.Wake();
+  th.join();
+}
+
+TEST(SimpleTaskQueue, WaitForEmptyWhenEmpty) {
+  nf7::test::SimpleTaskQueueMock<nf7::Task<int32_t&>> sut;
+  EXPECT_TRUE(sut.WaitForEmpty(1s));
+}