Remove iterator interface for folly::coro::AsyncGenerator

Summary:
Simplified the interface for consuming an AsyncGenerator to now just have a single async next() method instead of an iterator-based API.

The iterator-based API was added to integrate with the 'for co_await' syntax present in the Coroutines TS, however this syntax has been removed from C++20 to allow for future exploration of the AsyncRanges design-space.

The `co_await gen.next()` expression produces an optional-like object that lets you query whether the result is a sentinel or contains a value.

Old:
```
Task<void> consume(AsyncGenerator<T> gen) {
  for (auto it = co_await gen.begin();
       it != gen.end();
       co_await ++it) {
    use(*it);
  }
}
```

New:
```
Task<void> consume(AsyncGenerator<T> gen) {
  while (auto item = co_await gen.next()) {
    use(*item);
  }
}
```

Reviewed By: andriigrynenko, kirkshoop

Differential Revision: D16586151

fbshipit-source-id: 4b0bf31ba9291d894a18e9553513eddee0cde33a

folly/experimental/coro/AsyncGenerator.h

@@ -131,21 +131,17 @@ class AsyncGeneratorPromise {
     }
   }
 
-  Reference value() noexcept {
+  decltype(auto) getRvalue() noexcept {
     DCHECK(hasValue_);
-    return value_.get();
-  }
-
-  std::add_pointer_t<Reference> valuePointer() noexcept {
-    DCHECK(hasValue_);
-    return std::addressof(value_.get());
+    return std::move(value_).get();
   }
 
   void clearValue() noexcept {
-    DCHECK(hasValue_);
+    if (hasValue_) {
       hasValue_ = false;
       value_.destruct();
     }
+  }
 
   bool hasValue() const noexcept {
     return hasValue_;
@@ -238,10 +234,8 @@ class AsyncGeneratorPromise {
 //
 //  folly::coro::Task<void> consumer() {
 //    auto records = getRecordsAsync();
-//    for (auto it = co_await records.begin();
-//         it != records.end();
-//         co_await ++it) {
-//      auto&& record = *it;
+//    while (auto item = co_await records.next()) {
+//      auto&& record = *item;
 //      process(record);
 //    }
 //  }
@@ -263,198 +257,182 @@ class FOLLY_NODISCARD AsyncGenerator {
   using reference = Reference;
   using pointer = std::add_pointer_t<Reference>;
 
-  struct sentinel {};
-
-  class async_iterator {
-    class FOLLY_NODISCARD AdvanceAwaiter {
  public:
-      explicit AdvanceAwaiter(async_iterator& iter) noexcept : iter_(iter) {}
+  AsyncGenerator() noexcept : coro_() {}
 
-      bool await_ready() noexcept {
-        return false;
-      }
+  AsyncGenerator(AsyncGenerator&& other) noexcept
+      : coro_(std::exchange(other.coro_, {})) {}
 
-      handle_t await_suspend(
-          std::experimental::coroutine_handle<> continuation) noexcept {
-        auto& promise = iter_.coro_.promise();
-        promise.setContinuation(continuation);
-        promise.clearValue();
-        return iter_.coro_;
+  ~AsyncGenerator() {
+    if (coro_) {
+      coro_.destroy();
+    }
   }
 
-      async_iterator& await_resume() {
-        if (iter_.coro_.done()) {
-          iter_.coro_.promise().throwIfException();
+  AsyncGenerator& operator=(AsyncGenerator&& other) noexcept {
+    auto oldCoro = std::exchange(coro_, std::exchange(other.coro_, {}));
+    if (oldCoro) {
+      oldCoro.destroy();
     }
-        return iter_;
+    return *this;
   }
 
-     private:
-      async_iterator& iter_;
-    };
+  void swap(AsyncGenerator& other) noexcept {
+    std::swap(coro_, other.coro_);
+  }
+
+  class NextAwaitable;
+  class NextSemiAwaitable;
 
-    class FOLLY_NODISCARD AdvanceSemiAwaitable {
+  class NextResult {
    public:
-      explicit AdvanceSemiAwaitable(async_iterator& iter) noexcept
-          : iter_(iter) {}
+    NextResult() noexcept : hasValue_(false) {}
 
-      friend AdvanceAwaiter co_viaIfAsync(
-          folly::Executor::KeepAlive<> executor,
-          AdvanceSemiAwaitable awaitable) noexcept {
-        awaitable.iter_.coro_.promise().setExecutor(std::move(executor));
-        return AdvanceAwaiter{awaitable.iter_};
+    NextResult(NextResult&& other) noexcept : hasValue_(other.hasValue_) {
+      if (hasValue_) {
+        value_.construct(std::move(other.value_).get());
+      }
     }
 
-     private:
-      async_iterator& iter_;
-    };
+    ~NextResult() {
+      if (hasValue_) {
+        value_.destruct();
+      }
+    }
 
-    friend class AdvanceAwaiter;
-    friend class AdvanceSemiAwaitable;
+    NextResult& operator=(NextResult&& other) {
+      if (&other != this) {
+        if (has_value()) {
+          hasValue_ = false;
+          value_.destruct();
+        }
 
-   public:
-    using async_iterator_category = std::input_iterator_tag;
-    using value_type = typename AsyncGenerator::value_type;
-    using reference = typename AsyncGenerator::reference;
-    using pointer = typename AsyncGenerator::pointer;
+        if (other.has_value()) {
+          value_.construct(std::move(other.value_).get());
+          hasValue_ = true;
+        }
+      }
+      return *this;
+    }
 
-    async_iterator() noexcept = default;
+    bool has_value() const noexcept {
+      return hasValue_;
+    }
 
-    explicit async_iterator(handle_t coro) noexcept : coro_(coro) {}
+    explicit operator bool() const noexcept {
+      return has_value();
+    }
 
-    async_iterator(async_iterator&& other) noexcept
-        : coro_(std::exchange(other.coro_, {})) {}
+    decltype(auto) value() & {
+      DCHECK(has_value());
+      return value_.get();
+    }
 
-    async_iterator& operator=(async_iterator&& other) noexcept {
-      coro_ = std::exchange(other.coro_, {});
-      return *this;
+    decltype(auto) value() && {
+      DCHECK(has_value());
+      return std::move(value_).get();
     }
 
-    AdvanceSemiAwaitable operator++() noexcept {
-      return AdvanceSemiAwaitable(*this);
+    decltype(auto) value() const& {
+      DCHECK(has_value());
+      return value_.get();
     }
 
-    typename AsyncGenerator::reference operator*() const
-        noexcept(std::is_nothrow_copy_constructible<Reference>::value) {
-      return coro_.promise().value();
+    decltype(auto) value() const&& {
+      DCHECK(has_value());
+      return std::move(value_).get();
     }
 
-    typename AsyncGenerator::pointer operator->() const noexcept {
-      return coro_.promise().valuePointer();
+    decltype(auto) operator*() & {
+      return value();
     }
 
-    friend bool operator==(const async_iterator& it, sentinel) noexcept {
-      return !it.coro_ || it.coro_.done();
+    decltype(auto) operator*() && {
+      return std::move(*this).value();
     }
 
-    friend bool operator!=(const async_iterator& it, sentinel s) noexcept {
-      return !(it == s);
+    decltype(auto) operator*() const& {
+      return value();
     }
 
-    friend bool operator==(sentinel s, const async_iterator& it) noexcept {
-      return it == s;
+    decltype(auto) operator*() const&& {
+      return std::move(*this).value();
     }
 
-    friend bool operator!=(sentinel s, const async_iterator& it) noexcept {
-      return it != s;
+    decltype(auto) operator-> () {
+      DCHECK(has_value());
+      auto&& x = value_.get();
+      return std::addressof(x);
+    }
+
+    decltype(auto) operator-> () const {
+      DCHECK(has_value());
+      auto&& x = value_.get();
+      return std::addressof(x);
     }
 
    private:
-    handle_t coro_;
+    friend NextAwaitable;
+    explicit NextResult(handle_t coro) noexcept : hasValue_(true) {
+      value_.construct(coro.promise().getRvalue());
+    }
+
+    detail::ManualLifetime<Reference> value_;
+    bool hasValue_ = false;
   };
 
- private:
-  class FOLLY_NODISCARD BeginAwaiter {
+  class NextAwaitable {
    public:
-    BeginAwaiter(handle_t coro) noexcept : coro_(coro) {}
-
-    bool await_ready() noexcept {
+    bool await_ready() {
       return !coro_;
     }
 
     handle_t await_suspend(
         std::experimental::coroutine_handle<> continuation) noexcept {
-      coro_.promise().setContinuation(continuation);
+      auto& promise = coro_.promise();
+      promise.setContinuation(continuation);
+      promise.clearValue();
       return coro_;
     }
 
-    FOLLY_NODISCARD async_iterator await_resume() {
-      if (coro_ && coro_.done()) {
+    NextResult await_resume() {
+      if (!coro_) {
+        return NextResult{};
+      } else if (coro_.done()) {
         coro_.promise().throwIfException();
+        return NextResult{};
+      } else {
+        return NextResult{coro_};
       }
-      return async_iterator{coro_};
     }
 
    private:
-    handle_t coro_;
-  };
-
-  class FOLLY_NODISCARD BeginSemiAwaitable {
-   public:
-    explicit BeginSemiAwaitable(handle_t coro) noexcept : coro_(coro) {}
-
-    // A BeginSemiAwaitable requires an executor to be injected by calling
-    // the folly::coro::co_viaIfAsync() function. This is done implicitly
-    // by coroutine-types such as Task<T> and AsyncGenerator<T> which call
-    // co_viaIfAsync() from their promise_type::await_transform() method
-    // to inject the awaiting coroutine's current executor.
-    friend BeginAwaiter co_viaIfAsync(
-        folly::Executor::KeepAlive<> executor,
-        BeginSemiAwaitable&& awaitable) noexcept {
-      if (awaitable.coro_) {
-        awaitable.coro_.promise().setExecutor(std::move(executor));
-      }
-      return BeginAwaiter{awaitable.coro_};
-    }
+    friend NextSemiAwaitable;
+    explicit NextAwaitable(handle_t coro) noexcept : coro_(coro) {}
 
-   private:
     handle_t coro_;
   };
 
+  class NextSemiAwaitable {
    public:
-  AsyncGenerator() noexcept : coro_() {}
-
-  AsyncGenerator(AsyncGenerator&& other) noexcept
-      : coro_(std::exchange(other.coro_, {})) {}
-
-  ~AsyncGenerator() {
+    NextAwaitable viaIfAsync(Executor::KeepAlive<> executor) noexcept {
       if (coro_) {
-      coro_.destroy();
+        coro_.promise().setExecutor(std::move(executor));
       }
+      return NextAwaitable{coro_};
     }
 
-  AsyncGenerator& operator=(AsyncGenerator&& other) noexcept {
-    auto oldCoro = std::exchange(coro_, std::exchange(other.coro_, {}));
-    if (oldCoro) {
-      oldCoro.destroy();
-    }
-    return *this;
-  }
+   private:
+    friend AsyncGenerator; //<Reference, Value>;
 
-  void swap(AsyncGenerator& other) noexcept {
-    std::swap(coro_, other.coro_);
-  }
+    explicit NextSemiAwaitable(handle_t coro) noexcept : coro_(coro) {}
 
-  // begin() returns a SemiAwaitable type that must either be awaited within
-  // the context of a coroutine that has an associated folly::Executor (eg.
-  // a folly::coro::Task<T> or a folly::coro::AsyncGenerator<T>) or otherwise
-  // must have an executor explicitly injected by calling
-  // folly::co_viaIfAsync(executor, gen.begin()).
-  //
-  // The result of `co_await this->begin()` is an 'async_iterator' that can
-  // be used to access the current elements of the sequence and advance to
-  // the next element.
-  //
-  // Note that the AsyncGenerator is an input-range and the elements can only
-  // be consumed once. It is undefined behaviour to call the .begin() method
-  // multiple times for the same generator object.
-  FOLLY_NODISCARD BeginSemiAwaitable begin() noexcept {
-    DCHECK(!hasStarted());
-    return BeginSemiAwaitable{coro_};
-  }
-
-  FOLLY_NODISCARD sentinel end() noexcept {
-    return {};
+    handle_t coro_;
+  };
+
+  NextSemiAwaitable next() noexcept {
+    DCHECK(!coro_ || !coro_.done());
+    return NextSemiAwaitable{coro_};
   }
 
  private:
@@ -464,10 +442,6 @@ class FOLLY_NODISCARD AsyncGenerator {
       std::experimental::coroutine_handle<promise_type> coro) noexcept
       : coro_(coro) {}
 
-  bool hasStarted() const noexcept {
-    return coro_ && (coro_.done() || coro_.promise().hasValue());
-  }
-
   std::experimental::coroutine_handle<promise_type> coro_;
 };
 
@@ -504,11 +478,8 @@ auto co_invoke(Func func, Args... args) -> std::enable_if_t<
     invoke_result_t<Func, Args...>> {
   auto asyncRange =
       folly::invoke(static_cast<Func&&>(func), static_cast<Args&&>(args)...);
-  const auto itEnd = asyncRange.end();
-  auto it = co_await asyncRange.begin();
-  while (it != itEnd) {
-    co_yield* it;
-    co_await++ it;
+  while (auto result = co_await asyncRange.next()) {
+    co_yield* result;
   }
 }
 

folly/experimental/coro/test/AsyncGeneratorTest.cpp

@@ -32,17 +32,11 @@
 #include <string>
 #include <tuple>
 
-// AsyncGenerator's iterator type is move-only.
-static_assert(!std::is_copy_constructible_v<
-              folly::coro::AsyncGenerator<int>::async_iterator>);
-static_assert(std::is_move_constructible_v<
-              folly::coro::AsyncGenerator<int>::async_iterator>);
-
 TEST(AsyncGenerator, DefaultConstructedGeneratorIsEmpty) {
   folly::coro::blockingWait([]() -> folly::coro::Task<void> {
     folly::coro::AsyncGenerator<int> g;
-    auto it = co_await g.begin();
-    CHECK(it == g.end());
+    auto result = co_await g.next();
+    CHECK(!result);
   }());
 }
 
@@ -79,11 +73,11 @@ TEST(AsyncGenerator, PartiallyConsumingSequenceDestroysObjectsInScope) {
       auto gen = makeGenerator();
       CHECK(!started);
       CHECK(!destroyed);
-      auto it = co_await gen.begin();
+      auto result = co_await gen.next();
       CHECK(started);
       CHECK(!destroyed);
-      CHECK(it != gen.end());
-      CHECK_EQ(1, *it);
+      CHECK(result);
+      CHECK_EQ(1, *result);
     }
     CHECK(destroyed);
   }());
@@ -98,20 +92,20 @@ TEST(AsyncGenerator, FullyConsumeSequence) {
 
   folly::coro::blockingWait([&]() -> folly::coro::Task<void> {
     auto gen = makeGenerator();
-    auto it = co_await gen.begin();
-    CHECK(it != gen.end());
-    CHECK_EQ(0, *it);
-    co_await(++it);
-    CHECK(it != gen.end());
-    CHECK_EQ(1, *it);
-    co_await(++it);
-    CHECK(it != gen.end());
-    CHECK_EQ(2, *it);
-    co_await(++it);
-    CHECK(it != gen.end());
-    CHECK_EQ(3, *it);
-    co_await(++it);
-    CHECK(it == gen.end());
+    auto result = co_await gen.next();
+    CHECK(result);
+    CHECK_EQ(0, *result);
+    result = co_await gen.next();
+    CHECK(result);
+    CHECK_EQ(1, *result);
+    result = co_await gen.next();
+    CHECK(result);
+    CHECK_EQ(2, *result);
+    result = co_await gen.next();
+    CHECK(result);
+    CHECK_EQ(3, *result);
+    result = co_await gen.next();
+    CHECK(!result);
   }());
 }
 
@@ -131,7 +125,7 @@ TEST(AsyncGenerator, ThrowExceptionBeforeFirstYield) {
     auto gen = makeGenerator();
     bool caughtException = false;
     try {
-      (void)co_await gen.begin();
+      (void)co_await gen.next();
       CHECK(false);
     } catch (const SomeError&) {
       caughtException = true;
@@ -148,12 +142,12 @@ TEST(AsyncGenerator, ThrowExceptionAfterFirstYield) {
 
   folly::coro::blockingWait([&]() -> folly::coro::Task<void> {
     auto gen = makeGenerator();
-    auto it = co_await gen.begin();
-    CHECK(it != gen.end());
-    CHECK_EQ(42, *it);
+    auto result = co_await gen.next();
+    CHECK(result);
+    CHECK_EQ(42, *result);
     bool caughtException = false;
     try {
-      (void)co_await++ it;
+      (void)co_await gen.next();
       CHECK(false);
     } catch (const SomeError&) {
       caughtException = true;
@@ -172,8 +166,8 @@ TEST(AsyncGenerator, ConsumingManySynchronousElementsDoesNotOverflowStack) {
   folly::coro::blockingWait([&]() -> folly::coro::Task<void> {
     auto gen = makeGenerator();
     std::uint64_t sum = 0;
-    for (auto it = co_await gen.begin(); it != gen.end(); co_await++ it) {
-      sum += *it;
+    while (auto result = co_await gen.next()) {
+      sum += *result;
     }
     CHECK_EQ(499999500000u, sum);
   }());
@@ -198,12 +192,12 @@ TEST(AsyncGenerator, ProduceResultsAsynchronously) {
     };
 
     auto gen = makeGenerator();
-    auto it = co_await gen.begin();
-    CHECK_EQ(1, *it);
-    co_await++ it;
-    CHECK_EQ(2, *it);
-    co_await++ it;
-    CHECK(it == gen.end());
+    auto result = co_await gen.next();
+    CHECK_EQ(1, *result);
+    result = co_await gen.next();
+    CHECK_EQ(2, *result);
+    result = co_await gen.next();
+    CHECK(!result);
   }());
 }
 
@@ -229,13 +223,13 @@ TEST(AsyncGenerator, GeneratorOfLValueReference) {
 
   folly::coro::blockingWait([&]() -> folly::coro::Task<void> {
     auto gen = makeGenerator();
-    auto it = co_await gen.begin();
-    CHECK_EQ(10, *it);
-    *it = 20;
-    co_await++ it;
-    CHECK_EQ(30, *it);
-    co_await++ it;
-    CHECK(it == gen.end());
+    auto result = co_await gen.next();
+    CHECK_EQ(10, result.value());
+    *result = 20;
+    result = co_await gen.next();
+    CHECK_EQ(30, result.value());
+    result = co_await gen.next();
+    CHECK(!result.has_value());
   }());
 }
 
@@ -259,14 +253,14 @@ TEST(AsyncGenerator, GeneratorOfConstLValueReference) {
 
   folly::coro::blockingWait([&]() -> folly::coro::Task<void> {
     auto gen = makeGenerator();
-    auto it = co_await gen.begin();
-    CHECK_EQ(10, *it);
-    co_await++ it;
-    CHECK_EQ(30, *it);
-    co_await++ it;
-    CHECK_EQ(99, *it);
-    co_await++ it;
-    CHECK(it == gen.end());
+    auto result = co_await gen.next();
+    CHECK_EQ(10, *result);
+    result = co_await gen.next();
+    CHECK_EQ(30, *result);
+    result = co_await gen.next();
+    CHECK_EQ(99, *result);
+    result = co_await gen.next();
+    CHECK(!result);
   }());
 }
 
@@ -283,16 +277,16 @@ TEST(AsyncGenerator, GeneratorOfRValueReference) {
   folly::coro::blockingWait([&]() -> folly::coro::Task<void> {
     auto gen = makeGenerator();
 
-    auto it = co_await gen.begin();
-    CHECK_EQ(10, **it);
+    auto result = co_await gen.next();
+    CHECK_EQ(10, **result);
     // Don't move it to a local var.
 
-    co_await++ it;
-    CHECK_EQ(20, **it);
-    auto ptr = *it; // Move it to a local var.
+    result = co_await gen.next();
+    CHECK_EQ(20, **result);
+    auto ptr = *result; // Move it to a local var.
 
-    co_await++ it;
-    CHECK(it == gen.end());
+    result = co_await gen.next();
+    CHECK(!result);
   }());
 }
 
@@ -321,17 +315,17 @@ TEST(AsyncGenerator, GeneratorOfMoveOnlyType) {
 
   folly::coro::blockingWait([&]() -> folly::coro::Task<void> {
     auto gen = makeGenerator();
-    auto it = co_await gen.begin();
+    auto result = co_await gen.next();
 
     // NOTE: It's an error to dereference using '*it' as this returns a copy
     // of the iterator's reference type, which in this case is 'MoveOnly'.
-    CHECK_EQ(1, it->value());
+    CHECK_EQ(1, result->value());
 
-    co_await++ it;
-    CHECK_EQ(2, it->value());
+    result = co_await gen.next();
+    CHECK_EQ(2, result->value());
 
-    co_await++ it;
-    CHECK(it == gen.end());
+    result = co_await gen.next();
+    CHECK(!result);
   }());
 }
 
@@ -349,15 +343,15 @@ TEST(AsyncGenerator, GeneratorOfConstValue) {
 
   folly::coro::blockingWait([&]() -> folly::coro::Task<void> {
     auto gen = makeGenerator();
-    auto it = co_await gen.begin();
-    CHECK_EQ(42, *it);
-    static_assert(std::is_same_v<decltype(*it), int>);
-    co_await++ it;
-    CHECK_EQ(123, *it);
-    co_await++ it;
-    CHECK_EQ(99, *it);
-    co_await++ it;
-    CHECK(it == gen.end());
+    auto result = co_await gen.next();
+    CHECK_EQ(42, *result);
+    static_assert(std::is_same_v<decltype(*result), const int&>);
+    result = co_await gen.next();
+    CHECK_EQ(123, *result);
+    result = co_await gen.next();
+    CHECK_EQ(99, *result);
+    result = co_await gen.next();
+    CHECK(!result);
   }());
 }
 
@@ -376,15 +370,15 @@ TEST(AsyncGenerator, ExplicitValueType) {
 
   folly::coro::blockingWait([&]() -> folly::coro::Task<void> {
     auto gen = makeGenerator();
-    auto it = co_await gen.begin();
+    auto result = co_await gen.next();
     {
-      auto [kRef, vRef] = *it;
+      auto [kRef, vRef] = *result;
       CHECK_EQ("bar", kRef);
       CHECK_EQ("goodbye", vRef);
-      decltype(gen)::value_type copy = *it;
+      decltype(gen)::value_type copy = *result;
       vRef = "au revoir";
       CHECK_EQ("goodbye", std::get<1>(copy));
-      CHECK_EQ("au revoir", std::get<1>(*it));
+      CHECK_EQ("au revoir", std::get<1>(*result));
     }
   }());
 
@@ -400,9 +394,9 @@ TEST(AsyncGenerator, InvokeLambda) {
           co_yield std::move(p);
         });
 
-    auto it = co_await gen.begin();
-    CHECK(it != gen.end());
-    ptr = *it;
+    auto result = co_await gen.next();
+    CHECK(result);
+    ptr = *result;
     CHECK(ptr);
     CHECK(*ptr == 123);
   }());

folly/python/async_generator.h

@@ -30,20 +30,16 @@ class AsyncGeneratorWrapper {
       : gen_(std::move(gen)) {}
 
   coro::Task<Optional<T>> getNext() {
-    if (!iter_) {
-      iter_ = co_await gen_.begin();
+    auto item = co_await gen_.next();
+    if (item) {
+      co_return std::move(item).value();
     } else {
-      co_await(++*iter_);
-    }
-    if (iter_ == gen_.end()) {
       co_return none;
     }
-    co_return(**iter_);
   }
 
  private:
   coro::AsyncGenerator<T> gen_;
-  Optional<typename coro::AsyncGenerator<T>::async_iterator> iter_;
 };
 
 } // namespace python