Add TaskSet.onEmpty() to wait until all tasks have completed.

Fixes #286.

c++/src/kj/async-test.c++

@@ -620,6 +620,30 @@ TEST(Async, TaskSet) {
   EXPECT_EQ(1u, errorHandler.exceptionCount);
 }
 
+TEST(Async, TaskSetOnEmpty) {
+  EventLoop loop;
+  WaitScope waitScope(loop);
+  ErrorHandlerImpl errorHandler;
+  TaskSet tasks(errorHandler);
+
+  KJ_EXPECT(tasks.isEmpty());
+
+  auto paf = newPromiseAndFulfiller<void>();
+  tasks.add(kj::mv(paf.promise));
+  tasks.add(evalLater([]() {}));
+
+  KJ_EXPECT(!tasks.isEmpty());
+
+  auto promise = tasks.onEmpty();
+  KJ_EXPECT(!promise.poll(waitScope));
+  KJ_EXPECT(!tasks.isEmpty());
+
+  paf.fulfiller->fulfill();
+  KJ_ASSERT(promise.poll(waitScope));
+  KJ_EXPECT(tasks.isEmpty());
+  promise.wait(waitScope);
+}
+
 class DestructorDetector {
 public:
   DestructorDetector(bool& setTrue): setTrue(setTrue) {}

c++/src/kj/async.c++

@@ -127,6 +127,14 @@ protected:
     *prev = kj::mv(next);
     next = nullptr;
     prev = nullptr;
+
+    KJ_IF_MAYBE(f, taskSet.emptyFulfiller) {
+      if (taskSet.tasks == nullptr) {
+        f->get()->fulfill();
+        taskSet.emptyFulfiller = nullptr;
+      }
+    }
+
     return mv(self);
   }
 
@@ -165,6 +173,18 @@ kj::String TaskSet::trace() {
   return kj::strArray(traces, "\n============================================\n");
 }
 
+Promise<void> TaskSet::onEmpty() {
+  KJ_REQUIRE(emptyFulfiller == nullptr, "onEmpty() can only be called once at a time");
+
+  if (tasks == nullptr) {
+    return READY_NOW;
+  } else {
+    auto paf = newPromiseAndFulfiller<void>();
+    emptyFulfiller = kj::mv(paf.fulfiller);
+    return kj::mv(paf.promise);
+  }
+}
+
 namespace _ {  // private
 
 class LoggingErrorHandler: public TaskSet::ErrorHandler {

c++/src/kj/async.h

@@ -532,11 +532,19 @@ public:
   kj::String trace();
   // Return debug info about all promises currently in the TaskSet.
 
+  bool isEmpty() { return tasks == nullptr; }
+  // Check if any tasks are running.
+
+  Promise<void> onEmpty();
+  // Returns a promise that fulfills the next time the TaskSet is empty. Only one such promise can
+  // exist at a time.
+
 private:
   class Task;
 
   TaskSet::ErrorHandler& errorHandler;
   Maybe<Own<Task>> tasks;
+  Maybe<Own<PromiseFulfiller<void>>> emptyFulfiller;
 };
 
 // =======================================================================================