From e1fa945f4bb7418d99b3afe4831046aec6efaf4a Mon Sep 17 00:00:00 2001
From: Kefu Chai <kefu.chai@scylladb.com>
Date: Fri, 3 May 2024 22:33:39 +0800
Subject: [PATCH 1/3] coroutine/async_generator: reimplement async generator

this generator implementation is inspired by https://wg21.link/P2502R2.

Refs #2190
Refs #1913
Refs #1677
Signed-off-by: Kefu Chai <kefu.chai@scylladb.com>
---
 include/seastar/coroutine/async_generator.hh | 491 +++++++++++++++++++
 tests/unit/CMakeLists.txt                    |   3 +
 tests/unit/generator_test.cc                 | 236 +++++++++
 3 files changed, 730 insertions(+)
 create mode 100644 include/seastar/coroutine/async_generator.hh
 create mode 100644 tests/unit/generator_test.cc

diff --git a/include/seastar/coroutine/async_generator.hh b/include/seastar/coroutine/async_generator.hh
new file mode 100644
index 00000000000..170e651e07b
--- /dev/null
+++ b/include/seastar/coroutine/async_generator.hh
@@ -0,0 +1,491 @@
+/*
+ * This file is open source software, licensed to you under the terms
+ * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
+ * distributed with this work for additional information regarding copyright
+ * ownership.  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.
+ */
+
+#pragma once
+
+#include <cassert>
+#include <coroutine>
+#include <exception>
+#include <iterator>
+#include <memory>
+#include <type_traits>
+#include <utility>
+#include <seastar/core/future.hh>
+
+// seastar::coroutine::experimental::async_generator is inspired by the C++23 proposal
+// P2502R2 (https://wg21.link/P2502R2), which introduced std::generator for
+// synchronous coroutine-based range generation.
+//
+// Similar to P2502R2's generator, seastar::coroutine::experimental::async_generator
+// * prioritizes storing references to yielded objects instead of copying them
+// * generates a range with iterators that yield values
+//
+// However, there are key differences in seastar::coroutine::experimental::generator:
+//
+// * Alocator support:
+//   Seastar's generator does not support the Allocator template parameter.
+//   Unlike Seastar uses its built-in allocator eliminating the need for
+//   additional flexibility.
+// * Asynchronous Operations:
+//   - generator::begin() is a coroutine, unlike P2502R2's synchronous approach
+//   - generator::iterator::operator++() is a coroutine
+//   - generator::iterator::operator++(int) is a coroutine
+//   Note: Due to its asynchronous nature, this generator cannot be used in
+//   range-based for loops.
+// * Ranges Integration:
+//   seastar's generator is not a std::ranges::view_interface. So it lacks
+//   integration with the C++20 Ranges library due to its asynchronous operations.
+// * Nesting:
+//   Nesting generators is not supported. You cannot yield another generator
+//   from within a generator. This prevents implementation asynchronous,
+//   recursive algorithms like depth-first search on trees.
+namespace seastar::coroutine::experimental {
+
+template<typename Ref, typename Value = void>
+class async_generator;
+
+namespace internal {
+
+template <typename Yielded> class next_awaiter;
+
+template <typename Yielded>
+class async_generator_promise_base : public seastar::task {
+protected:
+    // a glvalue yield expression is passed to co_yield as its operand. and
+    // the object denoted by this expression is guaranteed to live until the
+    // coroutine resumes. we take advantage of this fact by storing only a
+    // pointer to the denoted object in the promise as long as the result of
+    // dereferencing this pointer is convertible to the Ref type.
+    std::add_pointer_t<Yielded> _value = nullptr;
+
+protected:
+    std::exception_ptr _exception;
+    std::coroutine_handle<> _consumer;
+    task* _waiting_task = nullptr;
+
+    /// awaiter returned by the generator when it produces a new element
+    ///
+    /// There are different combinations of expression types passed to
+    /// \c co_yield and \c Ref. In most cases, zero copies are made. Copies
+    /// are only necessary when \c co_yield requires type conversion.
+    ///
+    /// The following table summarizes the number of copies made for different
+    /// scenarios:
+    ///
+    /// | Ref       | co_yield const T& | co_yield T& | co_yield T&& | co_yield U&& |
+    /// | --------- | ----------------- | ----------- | ------------ | ------------ |
+    /// | T         | 0                 | 0           | 0            | 1            |
+    /// | const T&  | 0                 | 0           | 0            | 1            |
+    /// | T&        | ill-formed        | 0           | ill-formed   | ill-formed   |
+    /// | T&&       | ill-formed        | ill-formed  | 0            | 1            |
+    /// | const T&& | ill-formed        | ill-formed  | 0            | 1            |
+    ///
+    /// When no copies are required, \c yield_awaiter is used. Otherwise,
+    /// \c copy_awaiter is used. The latter converts \c U to \c T, and keeps the converted
+    /// value in it.
+    class yield_awaiter final {
+        async_generator_promise_base* _promise;
+        std::coroutine_handle<> _consumer;
+    public:
+        yield_awaiter(async_generator_promise_base* promise,
+                      std::coroutine_handle<> consumer) noexcept
+            : _promise{promise}
+            , _consumer{consumer}
+        {}
+        bool await_ready() const noexcept {
+            return false;
+        }
+        template <typename Promise>
+        std::coroutine_handle<> await_suspend(std::coroutine_handle<Promise> producer) noexcept {
+            _promise->_waiting_task = &producer.promise();
+            if (seastar::need_preempt()) {
+                auto consumer = std::coroutine_handle<seastar::task>::from_address(
+                    _consumer.address());
+                seastar::schedule(&consumer.promise());
+                return std::noop_coroutine();
+            }
+            return _consumer;
+        }
+        void await_resume() noexcept {}
+    };
+
+    class copy_awaiter {
+        using value_type = std::remove_cvref_t<Yielded>;
+        async_generator_promise_base* _promise;
+        std::coroutine_handle<> _consumer;
+        value_type _value;
+    public:
+        copy_awaiter(async_generator_promise_base* promise,
+                     std::coroutine_handle<> consumer,
+                     const value_type& value)
+            : _promise{promise}
+            , _consumer{consumer}
+            , _value{value}
+        {}
+        constexpr bool await_ready() const noexcept {
+            return false;
+        }
+        template <typename Promise>
+        std::coroutine_handle<> await_suspend(std::coroutine_handle<Promise> producer) noexcept {
+            auto& current = producer.promise();
+            current._value = std::addressof(_value);
+            _promise->_waiting_task = &current;
+            if (seastar::need_preempt()) {
+                auto consumer = std::coroutine_handle<seastar::task>::from_address(
+                    _consumer.address());
+                seastar::schedule(&consumer.promise());
+                return std::noop_coroutine();
+            }
+            return _consumer;
+        }
+        constexpr void await_resume() const noexcept {}
+    };
+
+public:
+    async_generator_promise_base() noexcept = default;
+    async_generator_promise_base(const async_generator_promise_base &) = delete;
+    async_generator_promise_base& operator=(const async_generator_promise_base &) = delete;
+    async_generator_promise_base(async_generator_promise_base &&) noexcept = default;
+    async_generator_promise_base& operator=(async_generator_promise_base &&) noexcept = default;
+
+    // lazily-started coroutine, do not execute the coroutine until
+    // the coroutine is awaited.
+    std::suspend_always initial_suspend() const noexcept {
+        return {};
+    }
+
+    yield_awaiter final_suspend() noexcept {
+        _value = nullptr;
+        return yield_awaiter{this, this->_consumer};
+    }
+
+    void unhandled_exception() noexcept {
+        _exception = std::current_exception();
+    }
+
+    yield_awaiter yield_value(Yielded value) noexcept {
+        this->_value = std::addressof(value);
+        return yield_awaiter{this, this->_consumer};
+    }
+
+    copy_awaiter yield_value(const std::remove_reference_t<Yielded>& value)
+    requires (std::is_rvalue_reference_v<Yielded> &&
+              std::constructible_from<
+                  std::remove_cvref_t<Yielded>,
+                  const std::remove_reference_t<Yielded>&>) {
+        return copy_awaiter{this, this->_consumer, value};
+    }
+
+    void return_void() noexcept {}
+
+    // @return if the async_generator has reached the end of the sequence
+    bool finished() const noexcept {
+        return _value == nullptr;
+    }
+
+    void rethrow_if_unhandled_exception() {
+        if (_exception) {
+            std::rethrow_exception(std::move(_exception));
+        }
+    }
+
+    void run_and_dispose() noexcept final {
+        using handle_type = std::coroutine_handle<async_generator_promise_base>;
+        handle_type::from_promise(*this).resume();
+    }
+
+    seastar::task* waiting_task() noexcept final {
+        return _waiting_task;
+    }
+
+private:
+    friend class next_awaiter<Yielded>;
+};
+
+/// awaiter returned when the consumer gets the \c begin iterator or
+/// when it advances the iterator.
+template <typename Yielded>
+class next_awaiter {
+protected:
+    async_generator_promise_base<Yielded>* _promise = nullptr;
+    std::coroutine_handle<> _producer = nullptr;
+
+    explicit next_awaiter(std::nullptr_t) noexcept {}
+    next_awaiter(async_generator_promise_base<Yielded>& promise,
+                 std::coroutine_handle<> producer) noexcept
+        : _promise{std::addressof(promise)}
+        , _producer{producer} {}
+
+public:
+    bool await_ready() const noexcept {
+        return false;
+    }
+
+    template <typename Promise>
+    std::coroutine_handle<> await_suspend(std::coroutine_handle<Promise> consumer) noexcept {
+        _promise->_consumer = consumer;
+        return _producer;
+    }
+};
+
+} // namespace internal
+
+/// async_generator represents a view modeling std::ranges::input_range,
+/// and has move-only iterators.
+///
+/// async_generator has 2 template parameters:
+///
+/// - Ref
+/// - Value
+///
+/// From Ref and Value, we derive types:
+/// - value_type: a cv-unqualified object type that specifies the value type of
+///   the async_generator's range and iterators
+/// - reference_Type: the reference type of the async_generator's range and iterators
+/// - yielded_type: the type of the parameter to the primary overload of \c
+///   yield_value in the async_generator's associated promise type
+///
+/// Under the most circumstances, only the first parameter is specified: like
+/// \c async_generator<meow>. The resulting async_generator:
+/// - has a value type of \c remove_cvref_t<meow>
+/// - has a reference type of \c meow, if it is a reference type, or \c meow&&
+///   otherwise
+/// - the operand of \c co_yield in the body of the async_generator should be
+///   convertible to \c meow, if it is a reference type, otherwise the operand
+///   type should be <tt>const meow&</tt>
+///
+/// Consider following code snippet:
+/// \code
+/// async_generator<const std::string&> send_query(std::string query) {
+///   auto result_set = db.execute(query);
+///   for (auto row : result_set) {
+///       co_yield std::format("{}", row);
+///   }
+/// }
+/// \endcode
+///
+/// In this case, \c Ref is a reference type of \c <tt>const std::string&</tt>,
+/// and \c Value is the default value of \c void. So the \c value_type is
+/// \c std::string. As the async_generator always returns a \c std::string, its
+/// iterator has the luxury of returning a reference to it.
+///
+/// But if some rare users want to use a proxy reference type, or to generate a
+/// range whose iterators yield prvalue for whatever reason, they should use
+/// the two-argument \c async_generator, like <tt>async_generator<meow, woof></tt>.
+/// The resulting async_generator:
+/// - has a value type of \c woof
+/// - has a reference type of \c meow
+///
+/// For instance, consider following code snippet:
+/// \code
+/// async_generator<std::string_view, std::string> generate_strings() {
+///   co_yield "[";
+///   std::string s;
+///   for (auto sv : {"1"sv, "2"sv}) {
+///     s = sv;
+///     s.push_back(',');
+///     co_yield s;
+///   }
+///   co_yield "]";
+/// }
+/// \endcode
+///
+/// In this case, \c Ref is \c std::string_view, and \Value is \c std::string.
+/// So we can ensure that the caller cannot invalidate the yielded values by
+/// mutating the defererenced value of iterator. As the \c std::string_view
+/// instance is immutable. But in the meanwhile, the async_generator can return
+/// a \c std::string by \c co_yield a \c std::string_view or a \c std::string.
+/// And the caller can still access the element of the range via the same type:
+/// \c std::string_view.
+///
+/// Current Limitation and Future Plans:
+///
+/// This generator implementation does not address the "Pingpong problem":
+/// where the producer generates elements one at a time, forcing frequent
+/// context switches between producer and consumer. This can lead to suboptimal
+/// performance, especially when bulk generation and consumption would be more
+/// efficient.
+///
+/// We intend to extend the existing implementation to allow the producer
+/// to yield a range of elements. This will enable batch processing,
+/// potentially improving performance by reducing context switches.
+///
+/// TODO: Implement range-based yielding to mitigate the Ping-pong problem.
+template<typename Ref, typename Value>
+class [[nodiscard]] async_generator {
+    using value_type = std::conditional_t<std::is_void_v<Value>,
+                                          std::remove_cvref_t<Ref>,
+                                          Value>;
+    using reference_type = std::conditional_t<std::is_void_v<Value>,
+                                              Ref&&,
+                                              Ref>;
+    using yielded_type = std::conditional_t<std::is_reference_v<reference_type>,
+                                            reference_type,
+                                            const reference_type&>;
+
+public:
+    class promise_type;
+
+private:
+    using handle_type = std::coroutine_handle<promise_type>;
+    handle_type _coro = {};
+
+public:
+    class iterator;
+
+    async_generator() noexcept = default;
+    explicit async_generator(promise_type& promise) noexcept
+        : _coro(std::coroutine_handle<promise_type>::from_promise(promise))
+    {}
+    async_generator(async_generator&& other) noexcept
+        : _coro{std::exchange(other._coro, {})}
+    {}
+    async_generator(const async_generator&) = delete;
+    async_generator& operator=(const async_generator&) = delete;
+
+    ~async_generator() {
+        if (_coro) {
+            _coro.destroy();
+        }
+    }
+
+    friend void swap(async_generator& lhs, async_generator& rhs) noexcept {
+        std::swap(lhs._coro, rhs._coro);
+    }
+
+    async_generator& operator=(async_generator&& other) noexcept {
+        if (this == &other) {
+            return *this;
+        }
+        if (_coro) {
+            _coro.destroy();
+        }
+        _coro = std::exchange(other._coro, nullptr);
+        return *this;
+    }
+
+    [[nodiscard]] auto begin() noexcept {
+        using base_awaiter = internal::next_awaiter<yielded_type>;
+        class begin_awaiter final : public base_awaiter {
+            using base_awaiter::_promise;
+
+        public:
+            explicit begin_awaiter(std::nullptr_t) noexcept
+                : base_awaiter{nullptr}
+            {}
+            explicit begin_awaiter(handle_type producer_coro) noexcept
+                : base_awaiter{producer_coro.promise(), producer_coro}
+            {}
+            bool await_ready() const noexcept {
+                return _promise == nullptr || base_awaiter::await_ready();
+            }
+
+            iterator await_resume() {
+                if (_promise == nullptr) {
+                    return iterator{nullptr};
+                }
+                if (_promise->finished()) {
+                    _promise->rethrow_if_unhandled_exception();
+                    return iterator{nullptr};
+                }
+                return iterator{
+                    handle_type::from_promise(*static_cast<promise_type *>(_promise))
+                };
+            }
+        };
+
+        if (_coro) {
+            return begin_awaiter{_coro};
+        } else {
+            return begin_awaiter{nullptr};
+        }
+    }
+
+    [[nodiscard]] std::default_sentinel_t end() const noexcept {
+        return {};
+    }
+};
+
+template <typename Ref, typename Value>
+class async_generator<Ref, Value>::promise_type final : public internal::async_generator_promise_base<yielded_type> {
+public:
+    async_generator get_return_object() noexcept {
+        return async_generator{*this};
+    }
+
+    yielded_type value() const noexcept {
+        return static_cast<yielded_type>(*this->_value);
+    }
+};
+
+template <typename Ref, typename Value>
+class async_generator<Ref, Value>::iterator final {
+private:
+    using handle_type = async_generator::handle_type;
+    handle_type _coro = nullptr;
+
+public:
+    using iterator_category = std::input_iterator_tag;
+    using difference_type = std::ptrdiff_t;
+    using value_type = async_generator::value_type;
+    using reference = async_generator::reference_type;
+    using pointer = std::add_pointer_t<value_type>;
+
+    explicit iterator(handle_type coroutine) noexcept
+        : _coro{coroutine}
+    {}
+
+    explicit operator bool() const noexcept {
+        return _coro && !_coro.done();
+    }
+
+    [[nodiscard]] auto operator++() noexcept {
+        using base_awaiter = internal::next_awaiter<yielded_type>;
+        class increment_awaiter final : public base_awaiter {
+            iterator& _iterator;
+            using base_awaiter::_promise;
+
+        public:
+            explicit increment_awaiter(iterator& iterator) noexcept
+                : base_awaiter{iterator._coro.promise(), iterator._coro}
+                , _iterator{iterator}
+            {}
+            iterator& await_resume() {
+                if (_promise->finished()) {
+                    // update iterator to end()
+                    _iterator = iterator{nullptr};
+                    _promise->rethrow_if_unhandled_exception();
+                }
+                return _iterator;
+            }
+        };
+
+        assert(bool(*this) && "cannot increment end iterator");
+        return increment_awaiter{*this};
+    }
+
+    reference operator*() const noexcept {
+        return _coro.promise().value();
+    }
+
+    bool operator==(std::default_sentinel_t) const noexcept {
+        return !bool(*this);
+    }
+};
+
+} // namespace seastar::coroutine::experimental
diff --git a/tests/unit/CMakeLists.txt b/tests/unit/CMakeLists.txt
index 65a274c7acf..8deff2e8365 100644
--- a/tests/unit/CMakeLists.txt
+++ b/tests/unit/CMakeLists.txt
@@ -282,6 +282,9 @@ seastar_add_test (content_source
 seastar_add_test (coroutines
   SOURCES coroutines_test.cc)
 
+seastar_add_test (generator
+  SOURCES generator_test.cc)
+
 seastar_add_test (defer
   KIND BOOST
   SOURCES defer_test.cc)
diff --git a/tests/unit/generator_test.cc b/tests/unit/generator_test.cc
new file mode 100644
index 00000000000..a240109e917
--- /dev/null
+++ b/tests/unit/generator_test.cc
@@ -0,0 +1,236 @@
+/*
+ * This file is open source software, licensed to you under the terms
+ * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
+ * distributed with this work for additional information regarding copyright
+ * ownership.  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.
+ */
+/*
+ * Copyright (C) 2024 ScyllaDB Ltd.
+ */
+
+#include <seastar/core/coroutine.hh>
+#include <seastar/coroutine/as_future.hh>
+#include <seastar/coroutine/async_generator.hh>
+#include <seastar/coroutine/exception.hh>
+#include <seastar/testing/test_case.hh>
+#include <seastar/util/bool_class.hh>
+#include <seastar/util/later.hh>
+#include <string_view>
+#if __cplusplus >= 202302L && defined(__cpp_lib_generator)
+#include <generator>
+template <typename T>
+using sync_generator = std::generator<const T&>;
+#else
+#include "tl-generator.hh"
+template <typename T>
+using sync_generator = tl::generator<T>;
+#endif
+
+using namespace seastar;
+
+using do_suspend = bool_class<struct do_suspend_tag>;
+
+sync_generator<int>
+sync_fibonacci_sequence(unsigned count) {
+    auto a = 0, b = 1;
+    for (unsigned i = 0; i < count; ++i) {
+        if (std::numeric_limits<decltype(a)>::max() - a < b) {
+            throw std::out_of_range(
+                fmt::format("fibonacci[{}] is greater than the largest value of int", i));
+        }
+        int next = std::exchange(a, std::exchange(b, a + b));
+        // tl::generator::yield_value() only accepts arguments of reference type,
+        // instead of a cv-qualified value.
+        co_yield next;
+    }
+}
+
+coroutine::experimental::async_generator<int>
+async_fibonacci_sequence(unsigned count, do_suspend suspend) {
+    auto a = 0, b = 1;
+    for (unsigned i = 0; i < count; ++i) {
+        if (std::numeric_limits<decltype(a)>::max() - a < b) {
+            throw std::out_of_range(
+                fmt::format("fibonacci[{}] is greater than the largest value of int", i));
+        }
+        if (suspend) {
+            co_await yield();
+        }
+        co_yield std::exchange(a, std::exchange(b, a + b));
+    }
+}
+
+seastar::future<>
+verify_fib_drained(coroutine::experimental::async_generator<int> actual_fibs, unsigned count) {
+    auto expected_fibs = sync_fibonacci_sequence(count);
+    auto expected_fib = std::begin(expected_fibs);
+
+    auto actual_fib = co_await actual_fibs.begin();
+
+    for (; actual_fib != actual_fibs.end(); co_await ++actual_fib) {
+        BOOST_REQUIRE(expected_fib != std::end(expected_fibs));
+        BOOST_REQUIRE_EQUAL(*actual_fib, *expected_fib);
+        ++expected_fib;
+    }
+    BOOST_REQUIRE(actual_fib == actual_fibs.end());
+}
+
+SEASTAR_TEST_CASE(test_async_generator_drained_with_suspend) {
+    constexpr int count = 4;
+    return verify_fib_drained(async_fibonacci_sequence(count, do_suspend::yes),
+                              count);
+}
+
+SEASTAR_TEST_CASE(test_async_generator_drained_without_suspend) {
+    constexpr int count = 4;
+    return verify_fib_drained(async_fibonacci_sequence(count, do_suspend::no),
+                              count);
+}
+
+seastar::future<> test_async_generator_not_drained(do_suspend suspend) {
+    auto fib = async_fibonacci_sequence(42, suspend);
+    auto actual_fib = co_await fib.begin();
+    BOOST_REQUIRE_EQUAL(*actual_fib, 0);
+}
+
+SEASTAR_TEST_CASE(test_async_generator_not_drained_with_suspend) {
+    return test_async_generator_not_drained(do_suspend::yes);
+}
+
+SEASTAR_TEST_CASE(test_async_generator_not_drained_without_suspend) {
+    return test_async_generator_not_drained(do_suspend::no);
+}
+
+coroutine::experimental::async_generator<std::string_view, std::string>
+generate_value_and_ref(std::vector<std::string_view> strings) {
+    co_yield "[";
+    std::string s;
+    for (auto sv : strings) {
+        s = sv;
+        s.push_back(',');
+        co_yield s;
+    }
+    co_yield "]";
+}
+
+SEASTAR_TEST_CASE(test_async_generator_value_reference) {
+    using namespace std::literals;
+    std::vector<std::string_view> expected_quoted = {"["sv, "foo,"sv, "bar,"sv, "]"sv};
+    auto actual_quoted = generate_value_and_ref({"foo"sv, "bar"sv});
+    auto actual = co_await actual_quoted.begin();
+    for (auto expected : expected_quoted) {
+        BOOST_REQUIRE_EQUAL(*actual, expected);
+        co_await ++actual;
+    }
+}
+
+coroutine::experimental::async_generator<std::string>
+generate_yield_rvalue_reference(const std::vector<std::string> strings) {
+    for (auto& s: strings) {
+        co_yield s;
+    }
+}
+
+SEASTAR_TEST_CASE(test_async_generator_rvalue_reference) {
+    std::vector<std::string> expected_strings = {"hello", "world"};
+    auto actual_strings = generate_yield_rvalue_reference(expected_strings);
+    auto actual = co_await actual_strings.begin();
+    for (auto expected : expected_strings) {
+        BOOST_REQUIRE_EQUAL(*actual, expected);
+        co_await ++actual;
+    }
+}
+
+SEASTAR_TEST_CASE(test_async_generator_move_ctor) {
+    constexpr int count = 4;
+    auto actual_fibs = async_fibonacci_sequence(count, do_suspend::no);
+    return verify_fib_drained(std::move(actual_fibs), count);
+}
+
+SEASTAR_TEST_CASE(test_async_generator_swap) {
+    int count_a = 4;
+    int count_b = 42;
+    auto fibs_a = async_fibonacci_sequence(count_a, do_suspend::no);
+    auto fibs_b = async_fibonacci_sequence(count_b, do_suspend::no);
+    std::swap(fibs_a, fibs_b);
+    std::swap(count_a, count_b);
+    co_await verify_fib_drained(std::move(fibs_a), count_a);
+    co_await verify_fib_drained(std::move(fibs_b), count_b);
+}
+
+struct counter_t {
+    int n;
+    int* count;
+    counter_t(counter_t&& other) noexcept
+        : n{std::exchange(other.n, -1)},
+          count{std::exchange(other.count, nullptr)}
+    {}
+    counter_t(int n, int* count) noexcept
+        : n{n}, count{count} {
+        ++(*count);
+    }
+    ~counter_t() noexcept {
+        if (count) {
+            --(*count);
+        }
+    }
+};
+
+std::ostream& operator<<(std::ostream& os, const counter_t& c) {
+    return os << c.n;
+}
+
+coroutine::experimental::async_generator<counter_t>
+fiddle(int n, int* total) {
+    int i = 0;
+    while (true) {
+        if (i++ == n) {
+            throw std::invalid_argument("Eureka from generator!");
+        }
+        co_yield counter_t{i, total};
+    }
+}
+
+SEASTAR_TEST_CASE(test_async_generator_throws_from_generator) {
+    int total = 0;
+    auto count_to = [total=&total](unsigned n) -> seastar::future<> {
+        auto count = fiddle(n, total);
+        auto it = co_await count.begin();
+        for (unsigned i = 0; i < 2 * n; i++) {
+            co_await ++it;
+        }
+    };
+    auto f = co_await coroutine::as_future(count_to(42));
+    BOOST_REQUIRE(f.failed());
+    BOOST_REQUIRE_THROW(std::rethrow_exception(f.get_exception()), std::invalid_argument);
+    BOOST_REQUIRE_EQUAL(total, 0);
+}
+
+SEASTAR_TEST_CASE(test_async_generator_throws_from_consumer) {
+    int total = 0;
+    auto count_to = [total=&total](unsigned n) -> seastar::future<> {
+        auto count = fiddle(n, total);
+        auto it = co_await count.begin();
+        for (unsigned i = 0; i < n; i++) {
+            if (i == n / 2) {
+                throw std::invalid_argument("Eureka from consumer!");
+            }
+            co_await ++it;
+        }
+    };
+    auto f = co_await coroutine::as_future(count_to(42));
+    BOOST_REQUIRE(f.failed());
+    BOOST_REQUIRE_THROW(std::rethrow_exception(f.get_exception()), std::invalid_argument);
+    BOOST_REQUIRE_EQUAL(total, 0);
+}

From b0024b324f83589b544ac89f57d0c532e06aefa3 Mon Sep 17 00:00:00 2001
From: Kefu Chai <kefu.chai@scylladb.com>
Date: Mon, 22 Jul 2024 12:37:43 +0800
Subject: [PATCH 2/3] treewide: replace generator with async_generator

* replace coroutine::experimental::generator with
  coroutine::experimental::async_generator
* remove the generator related tests in coroutines_test.cc. since
  we already have tests/unit/generator_test.cc, there are no needs
  to keep two copies of these tests.
* update `experimental_list_directory()` to return `generator<const
  directory_entry&>`, for better performance, so we can point the
  promise's value pointer to the yielded value, without copying it
  using the copy_awaiter.

Fixes #2190
Fixes #1913
Fixes #1677

Signed-off-by: Kefu Chai <kefu.chai@scylladb.com>
---
 include/seastar/core/file.hh                 |   4 +-
 include/seastar/coroutine/async_generator.hh | 491 ------------
 include/seastar/coroutine/generator.hh       | 799 +++++++++----------
 src/core/file-impl.hh                        |   2 +-
 src/core/file.cc                             |  10 +-
 tests/unit/coroutines_test.cc                | 142 ----
 tests/unit/directory_test.cc                 |  12 +-
 tests/unit/generator_test.cc                 |  38 +-
 8 files changed, 407 insertions(+), 1091 deletions(-)
 delete mode 100644 include/seastar/coroutine/async_generator.hh

diff --git a/include/seastar/core/file.hh b/include/seastar/core/file.hh
index 1c9b0b4651f..ffe089e8247 100644
--- a/include/seastar/core/file.hh
+++ b/include/seastar/core/file.hh
@@ -182,7 +182,7 @@ public:
     virtual subscription<directory_entry> list_directory(std::function<future<> (directory_entry de)> next) = 0;
     // due to https://github.com/scylladb/seastar/issues/1913, we cannot use
     // buffered generator yet.
-    virtual coroutine::experimental::generator<directory_entry> experimental_list_directory();
+    virtual coroutine::experimental::generator<const directory_entry&> experimental_list_directory();
 };
 
 future<shared_ptr<file_impl>> make_file_impl(int fd, file_open_options options, int oflags, struct stat st) noexcept;
@@ -700,7 +700,7 @@ public:
     /// Returns a directory listing, given that this file object is a directory.
     // due to https://github.com/scylladb/seastar/issues/1913, we cannot use
     // buffered generator yet.
-    coroutine::experimental::generator<directory_entry> experimental_list_directory();
+    coroutine::experimental::generator<const directory_entry&> experimental_list_directory();
 
 #if SEASTAR_API_LEVEL < 7
     /**
diff --git a/include/seastar/coroutine/async_generator.hh b/include/seastar/coroutine/async_generator.hh
deleted file mode 100644
index 170e651e07b..00000000000
--- a/include/seastar/coroutine/async_generator.hh
+++ /dev/null
@@ -1,491 +0,0 @@
-/*
- * This file is open source software, licensed to you under the terms
- * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
- * distributed with this work for additional information regarding copyright
- * ownership.  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.
- */
-
-#pragma once
-
-#include <cassert>
-#include <coroutine>
-#include <exception>
-#include <iterator>
-#include <memory>
-#include <type_traits>
-#include <utility>
-#include <seastar/core/future.hh>
-
-// seastar::coroutine::experimental::async_generator is inspired by the C++23 proposal
-// P2502R2 (https://wg21.link/P2502R2), which introduced std::generator for
-// synchronous coroutine-based range generation.
-//
-// Similar to P2502R2's generator, seastar::coroutine::experimental::async_generator
-// * prioritizes storing references to yielded objects instead of copying them
-// * generates a range with iterators that yield values
-//
-// However, there are key differences in seastar::coroutine::experimental::generator:
-//
-// * Alocator support:
-//   Seastar's generator does not support the Allocator template parameter.
-//   Unlike Seastar uses its built-in allocator eliminating the need for
-//   additional flexibility.
-// * Asynchronous Operations:
-//   - generator::begin() is a coroutine, unlike P2502R2's synchronous approach
-//   - generator::iterator::operator++() is a coroutine
-//   - generator::iterator::operator++(int) is a coroutine
-//   Note: Due to its asynchronous nature, this generator cannot be used in
-//   range-based for loops.
-// * Ranges Integration:
-//   seastar's generator is not a std::ranges::view_interface. So it lacks
-//   integration with the C++20 Ranges library due to its asynchronous operations.
-// * Nesting:
-//   Nesting generators is not supported. You cannot yield another generator
-//   from within a generator. This prevents implementation asynchronous,
-//   recursive algorithms like depth-first search on trees.
-namespace seastar::coroutine::experimental {
-
-template<typename Ref, typename Value = void>
-class async_generator;
-
-namespace internal {
-
-template <typename Yielded> class next_awaiter;
-
-template <typename Yielded>
-class async_generator_promise_base : public seastar::task {
-protected:
-    // a glvalue yield expression is passed to co_yield as its operand. and
-    // the object denoted by this expression is guaranteed to live until the
-    // coroutine resumes. we take advantage of this fact by storing only a
-    // pointer to the denoted object in the promise as long as the result of
-    // dereferencing this pointer is convertible to the Ref type.
-    std::add_pointer_t<Yielded> _value = nullptr;
-
-protected:
-    std::exception_ptr _exception;
-    std::coroutine_handle<> _consumer;
-    task* _waiting_task = nullptr;
-
-    /// awaiter returned by the generator when it produces a new element
-    ///
-    /// There are different combinations of expression types passed to
-    /// \c co_yield and \c Ref. In most cases, zero copies are made. Copies
-    /// are only necessary when \c co_yield requires type conversion.
-    ///
-    /// The following table summarizes the number of copies made for different
-    /// scenarios:
-    ///
-    /// | Ref       | co_yield const T& | co_yield T& | co_yield T&& | co_yield U&& |
-    /// | --------- | ----------------- | ----------- | ------------ | ------------ |
-    /// | T         | 0                 | 0           | 0            | 1            |
-    /// | const T&  | 0                 | 0           | 0            | 1            |
-    /// | T&        | ill-formed        | 0           | ill-formed   | ill-formed   |
-    /// | T&&       | ill-formed        | ill-formed  | 0            | 1            |
-    /// | const T&& | ill-formed        | ill-formed  | 0            | 1            |
-    ///
-    /// When no copies are required, \c yield_awaiter is used. Otherwise,
-    /// \c copy_awaiter is used. The latter converts \c U to \c T, and keeps the converted
-    /// value in it.
-    class yield_awaiter final {
-        async_generator_promise_base* _promise;
-        std::coroutine_handle<> _consumer;
-    public:
-        yield_awaiter(async_generator_promise_base* promise,
-                      std::coroutine_handle<> consumer) noexcept
-            : _promise{promise}
-            , _consumer{consumer}
-        {}
-        bool await_ready() const noexcept {
-            return false;
-        }
-        template <typename Promise>
-        std::coroutine_handle<> await_suspend(std::coroutine_handle<Promise> producer) noexcept {
-            _promise->_waiting_task = &producer.promise();
-            if (seastar::need_preempt()) {
-                auto consumer = std::coroutine_handle<seastar::task>::from_address(
-                    _consumer.address());
-                seastar::schedule(&consumer.promise());
-                return std::noop_coroutine();
-            }
-            return _consumer;
-        }
-        void await_resume() noexcept {}
-    };
-
-    class copy_awaiter {
-        using value_type = std::remove_cvref_t<Yielded>;
-        async_generator_promise_base* _promise;
-        std::coroutine_handle<> _consumer;
-        value_type _value;
-    public:
-        copy_awaiter(async_generator_promise_base* promise,
-                     std::coroutine_handle<> consumer,
-                     const value_type& value)
-            : _promise{promise}
-            , _consumer{consumer}
-            , _value{value}
-        {}
-        constexpr bool await_ready() const noexcept {
-            return false;
-        }
-        template <typename Promise>
-        std::coroutine_handle<> await_suspend(std::coroutine_handle<Promise> producer) noexcept {
-            auto& current = producer.promise();
-            current._value = std::addressof(_value);
-            _promise->_waiting_task = &current;
-            if (seastar::need_preempt()) {
-                auto consumer = std::coroutine_handle<seastar::task>::from_address(
-                    _consumer.address());
-                seastar::schedule(&consumer.promise());
-                return std::noop_coroutine();
-            }
-            return _consumer;
-        }
-        constexpr void await_resume() const noexcept {}
-    };
-
-public:
-    async_generator_promise_base() noexcept = default;
-    async_generator_promise_base(const async_generator_promise_base &) = delete;
-    async_generator_promise_base& operator=(const async_generator_promise_base &) = delete;
-    async_generator_promise_base(async_generator_promise_base &&) noexcept = default;
-    async_generator_promise_base& operator=(async_generator_promise_base &&) noexcept = default;
-
-    // lazily-started coroutine, do not execute the coroutine until
-    // the coroutine is awaited.
-    std::suspend_always initial_suspend() const noexcept {
-        return {};
-    }
-
-    yield_awaiter final_suspend() noexcept {
-        _value = nullptr;
-        return yield_awaiter{this, this->_consumer};
-    }
-
-    void unhandled_exception() noexcept {
-        _exception = std::current_exception();
-    }
-
-    yield_awaiter yield_value(Yielded value) noexcept {
-        this->_value = std::addressof(value);
-        return yield_awaiter{this, this->_consumer};
-    }
-
-    copy_awaiter yield_value(const std::remove_reference_t<Yielded>& value)
-    requires (std::is_rvalue_reference_v<Yielded> &&
-              std::constructible_from<
-                  std::remove_cvref_t<Yielded>,
-                  const std::remove_reference_t<Yielded>&>) {
-        return copy_awaiter{this, this->_consumer, value};
-    }
-
-    void return_void() noexcept {}
-
-    // @return if the async_generator has reached the end of the sequence
-    bool finished() const noexcept {
-        return _value == nullptr;
-    }
-
-    void rethrow_if_unhandled_exception() {
-        if (_exception) {
-            std::rethrow_exception(std::move(_exception));
-        }
-    }
-
-    void run_and_dispose() noexcept final {
-        using handle_type = std::coroutine_handle<async_generator_promise_base>;
-        handle_type::from_promise(*this).resume();
-    }
-
-    seastar::task* waiting_task() noexcept final {
-        return _waiting_task;
-    }
-
-private:
-    friend class next_awaiter<Yielded>;
-};
-
-/// awaiter returned when the consumer gets the \c begin iterator or
-/// when it advances the iterator.
-template <typename Yielded>
-class next_awaiter {
-protected:
-    async_generator_promise_base<Yielded>* _promise = nullptr;
-    std::coroutine_handle<> _producer = nullptr;
-
-    explicit next_awaiter(std::nullptr_t) noexcept {}
-    next_awaiter(async_generator_promise_base<Yielded>& promise,
-                 std::coroutine_handle<> producer) noexcept
-        : _promise{std::addressof(promise)}
-        , _producer{producer} {}
-
-public:
-    bool await_ready() const noexcept {
-        return false;
-    }
-
-    template <typename Promise>
-    std::coroutine_handle<> await_suspend(std::coroutine_handle<Promise> consumer) noexcept {
-        _promise->_consumer = consumer;
-        return _producer;
-    }
-};
-
-} // namespace internal
-
-/// async_generator represents a view modeling std::ranges::input_range,
-/// and has move-only iterators.
-///
-/// async_generator has 2 template parameters:
-///
-/// - Ref
-/// - Value
-///
-/// From Ref and Value, we derive types:
-/// - value_type: a cv-unqualified object type that specifies the value type of
-///   the async_generator's range and iterators
-/// - reference_Type: the reference type of the async_generator's range and iterators
-/// - yielded_type: the type of the parameter to the primary overload of \c
-///   yield_value in the async_generator's associated promise type
-///
-/// Under the most circumstances, only the first parameter is specified: like
-/// \c async_generator<meow>. The resulting async_generator:
-/// - has a value type of \c remove_cvref_t<meow>
-/// - has a reference type of \c meow, if it is a reference type, or \c meow&&
-///   otherwise
-/// - the operand of \c co_yield in the body of the async_generator should be
-///   convertible to \c meow, if it is a reference type, otherwise the operand
-///   type should be <tt>const meow&</tt>
-///
-/// Consider following code snippet:
-/// \code
-/// async_generator<const std::string&> send_query(std::string query) {
-///   auto result_set = db.execute(query);
-///   for (auto row : result_set) {
-///       co_yield std::format("{}", row);
-///   }
-/// }
-/// \endcode
-///
-/// In this case, \c Ref is a reference type of \c <tt>const std::string&</tt>,
-/// and \c Value is the default value of \c void. So the \c value_type is
-/// \c std::string. As the async_generator always returns a \c std::string, its
-/// iterator has the luxury of returning a reference to it.
-///
-/// But if some rare users want to use a proxy reference type, or to generate a
-/// range whose iterators yield prvalue for whatever reason, they should use
-/// the two-argument \c async_generator, like <tt>async_generator<meow, woof></tt>.
-/// The resulting async_generator:
-/// - has a value type of \c woof
-/// - has a reference type of \c meow
-///
-/// For instance, consider following code snippet:
-/// \code
-/// async_generator<std::string_view, std::string> generate_strings() {
-///   co_yield "[";
-///   std::string s;
-///   for (auto sv : {"1"sv, "2"sv}) {
-///     s = sv;
-///     s.push_back(',');
-///     co_yield s;
-///   }
-///   co_yield "]";
-/// }
-/// \endcode
-///
-/// In this case, \c Ref is \c std::string_view, and \Value is \c std::string.
-/// So we can ensure that the caller cannot invalidate the yielded values by
-/// mutating the defererenced value of iterator. As the \c std::string_view
-/// instance is immutable. But in the meanwhile, the async_generator can return
-/// a \c std::string by \c co_yield a \c std::string_view or a \c std::string.
-/// And the caller can still access the element of the range via the same type:
-/// \c std::string_view.
-///
-/// Current Limitation and Future Plans:
-///
-/// This generator implementation does not address the "Pingpong problem":
-/// where the producer generates elements one at a time, forcing frequent
-/// context switches between producer and consumer. This can lead to suboptimal
-/// performance, especially when bulk generation and consumption would be more
-/// efficient.
-///
-/// We intend to extend the existing implementation to allow the producer
-/// to yield a range of elements. This will enable batch processing,
-/// potentially improving performance by reducing context switches.
-///
-/// TODO: Implement range-based yielding to mitigate the Ping-pong problem.
-template<typename Ref, typename Value>
-class [[nodiscard]] async_generator {
-    using value_type = std::conditional_t<std::is_void_v<Value>,
-                                          std::remove_cvref_t<Ref>,
-                                          Value>;
-    using reference_type = std::conditional_t<std::is_void_v<Value>,
-                                              Ref&&,
-                                              Ref>;
-    using yielded_type = std::conditional_t<std::is_reference_v<reference_type>,
-                                            reference_type,
-                                            const reference_type&>;
-
-public:
-    class promise_type;
-
-private:
-    using handle_type = std::coroutine_handle<promise_type>;
-    handle_type _coro = {};
-
-public:
-    class iterator;
-
-    async_generator() noexcept = default;
-    explicit async_generator(promise_type& promise) noexcept
-        : _coro(std::coroutine_handle<promise_type>::from_promise(promise))
-    {}
-    async_generator(async_generator&& other) noexcept
-        : _coro{std::exchange(other._coro, {})}
-    {}
-    async_generator(const async_generator&) = delete;
-    async_generator& operator=(const async_generator&) = delete;
-
-    ~async_generator() {
-        if (_coro) {
-            _coro.destroy();
-        }
-    }
-
-    friend void swap(async_generator& lhs, async_generator& rhs) noexcept {
-        std::swap(lhs._coro, rhs._coro);
-    }
-
-    async_generator& operator=(async_generator&& other) noexcept {
-        if (this == &other) {
-            return *this;
-        }
-        if (_coro) {
-            _coro.destroy();
-        }
-        _coro = std::exchange(other._coro, nullptr);
-        return *this;
-    }
-
-    [[nodiscard]] auto begin() noexcept {
-        using base_awaiter = internal::next_awaiter<yielded_type>;
-        class begin_awaiter final : public base_awaiter {
-            using base_awaiter::_promise;
-
-        public:
-            explicit begin_awaiter(std::nullptr_t) noexcept
-                : base_awaiter{nullptr}
-            {}
-            explicit begin_awaiter(handle_type producer_coro) noexcept
-                : base_awaiter{producer_coro.promise(), producer_coro}
-            {}
-            bool await_ready() const noexcept {
-                return _promise == nullptr || base_awaiter::await_ready();
-            }
-
-            iterator await_resume() {
-                if (_promise == nullptr) {
-                    return iterator{nullptr};
-                }
-                if (_promise->finished()) {
-                    _promise->rethrow_if_unhandled_exception();
-                    return iterator{nullptr};
-                }
-                return iterator{
-                    handle_type::from_promise(*static_cast<promise_type *>(_promise))
-                };
-            }
-        };
-
-        if (_coro) {
-            return begin_awaiter{_coro};
-        } else {
-            return begin_awaiter{nullptr};
-        }
-    }
-
-    [[nodiscard]] std::default_sentinel_t end() const noexcept {
-        return {};
-    }
-};
-
-template <typename Ref, typename Value>
-class async_generator<Ref, Value>::promise_type final : public internal::async_generator_promise_base<yielded_type> {
-public:
-    async_generator get_return_object() noexcept {
-        return async_generator{*this};
-    }
-
-    yielded_type value() const noexcept {
-        return static_cast<yielded_type>(*this->_value);
-    }
-};
-
-template <typename Ref, typename Value>
-class async_generator<Ref, Value>::iterator final {
-private:
-    using handle_type = async_generator::handle_type;
-    handle_type _coro = nullptr;
-
-public:
-    using iterator_category = std::input_iterator_tag;
-    using difference_type = std::ptrdiff_t;
-    using value_type = async_generator::value_type;
-    using reference = async_generator::reference_type;
-    using pointer = std::add_pointer_t<value_type>;
-
-    explicit iterator(handle_type coroutine) noexcept
-        : _coro{coroutine}
-    {}
-
-    explicit operator bool() const noexcept {
-        return _coro && !_coro.done();
-    }
-
-    [[nodiscard]] auto operator++() noexcept {
-        using base_awaiter = internal::next_awaiter<yielded_type>;
-        class increment_awaiter final : public base_awaiter {
-            iterator& _iterator;
-            using base_awaiter::_promise;
-
-        public:
-            explicit increment_awaiter(iterator& iterator) noexcept
-                : base_awaiter{iterator._coro.promise(), iterator._coro}
-                , _iterator{iterator}
-            {}
-            iterator& await_resume() {
-                if (_promise->finished()) {
-                    // update iterator to end()
-                    _iterator = iterator{nullptr};
-                    _promise->rethrow_if_unhandled_exception();
-                }
-                return _iterator;
-            }
-        };
-
-        assert(bool(*this) && "cannot increment end iterator");
-        return increment_awaiter{*this};
-    }
-
-    reference operator*() const noexcept {
-        return _coro.promise().value();
-    }
-
-    bool operator==(std::default_sentinel_t) const noexcept {
-        return !bool(*this);
-    }
-};
-
-} // namespace seastar::coroutine::experimental
diff --git a/include/seastar/coroutine/generator.hh b/include/seastar/coroutine/generator.hh
index 24598584617..499695fd08f 100644
--- a/include/seastar/coroutine/generator.hh
+++ b/include/seastar/coroutine/generator.hh
@@ -23,525 +23,472 @@
 
 #include <cassert>
 #include <coroutine>
-#include <optional>
+#include <exception>
+#include <iterator>
+#include <memory>
+#include <type_traits>
 #include <utility>
 #include <seastar/core/future.hh>
 
+// seastar::coroutine::generator is inspired by the C++23 proposal
+// P2502R2 (https://wg21.link/P2502R2), which introduced std::generator for
+// synchronous coroutine-based range generation.
+//
+// Similar to P2502R2's generator, seastar::coroutine::experimental::generator
+// * prioritizes storing references to yielded objects instead of copying them
+// * generates a range with iterators that yield values
+//
+// However, there are key differences in seastar::coroutine::experimental::generator:
+//
+// * Alocator support:
+//   Seastar's generator does not support the Allocator template parameter.
+//   Unlike Seastar uses its built-in allocator eliminating the need for
+//   additional flexibility.
+// * Asynchronous Operations:
+//   - generator::begin() is a coroutine, unlike P2502R2's synchronous approach
+//   - generator::iterator::operator++() is a coroutine
+//   - generator::iterator::operator++(int) is a coroutine
+//   Note: Due to its asynchronous nature, this generator cannot be used in
+//   range-based for loops.
+// * Ranges Integration:
+//   seastar's generator is not a std::ranges::view_interface. So it lacks
+//   integration with the C++20 Ranges library due to its asynchronous operations.
+// * Nesting:
+//   Nesting generators is not supported. You cannot yield another generator
+//   from within a generator. This prevents implementation asynchronous,
+//   recursive algorithms like depth-first search on trees.
 namespace seastar::coroutine::experimental {
 
-template<typename T, template <typename> class Container = std::optional>
+template<typename Ref, typename Value = void>
 class generator;
 
-/// `seastar::coroutine::experimental` is used as the type of the first
-/// parameter of a buffered generator coroutine.
-///
-/// the value of a `buffer_size_t` specifies the size of the buffer holding the
-/// values produced by the generator coroutine. Unlike its unbuffered variant,
-/// the bufferred generator does not wait for its caller to consume every single
-/// produced values. Instead, it puts the produced values into an internal
-/// buffer, before the buffer is full or the generator is suspended. This helps
-/// to alleviate the problem of pingpong between the generator coroutine and
-/// its caller.
-enum class buffer_size_t : size_t;
-
 namespace internal {
 
-using std::coroutine_handle;
-using std::suspend_never;
-using std::suspend_always;
-using std::suspend_never;
-using std::noop_coroutine;
-
-template<typename T>
-using next_value_t = std::optional<T>;
-
-template <template <typename> class Container, typename T>
-concept Fifo = requires(Container<T>&& c, T&& value) {
-    // better off returning a reference though, so we can move away from it
-    { c.front() } -> std::same_as<T&>;
-    c.pop_front();
-    c.push_back(value);
-    bool(c.empty());
-    { c.size() } -> std::convertible_to<size_t>;
-};
+template <typename Yielded> class next_awaiter;
 
-template<typename T>
-concept NothrowMoveConstructible = std::is_nothrow_move_constructible_v<T>;
+template <typename Yielded>
+class generator_promise_base : public seastar::task {
+protected:
+    // a glvalue yield expression is passed to co_yield as its operand. and
+    // the object denoted by this expression is guaranteed to live until the
+    // coroutine resumes. we take advantage of this fact by storing only a
+    // pointer to the denoted object in the promise as long as the result of
+    // dereferencing this pointer is convertible to the Ref type.
+    std::add_pointer_t<Yielded> _value = nullptr;
 
-template<NothrowMoveConstructible T>
-class generator_unbuffered_promise final : public seastar::task {
-    using generator_type = seastar::coroutine::experimental::generator<T, std::optional>;
-    std::optional<seastar::promise<>> _wait_for_next_value;
-    generator_type* _generator = nullptr;
+protected:
+    std::exception_ptr _exception;
+    std::coroutine_handle<> _consumer;
+    task* _waiting_task = nullptr;
+
+    /// awaiter returned by the generator when it produces a new element
+    ///
+    /// There are different combinations of expression types passed to
+    /// \c co_yield and \c Ref. In most cases, zero copies are made. Copies
+    /// are only necessary when \c co_yield requires type conversion.
+    ///
+    /// The following table summarizes the number of copies made for different
+    /// scenarios:
+    ///
+    /// | Ref       | co_yield const T& | co_yield T& | co_yield T&& | co_yield U&& |
+    /// | --------- | ----------------- | ----------- | ------------ | ------------ |
+    /// | T         | 0                 | 0           | 0            | 1            |
+    /// | const T&  | 0                 | 0           | 0            | 1            |
+    /// | T&        | ill-formed        | 0           | ill-formed   | ill-formed   |
+    /// | T&&       | ill-formed        | ill-formed  | 0            | 1            |
+    /// | const T&& | ill-formed        | ill-formed  | 0            | 1            |
+    ///
+    /// When no copies are required, \c yield_awaiter is used. Otherwise,
+    /// \c copy_awaiter is used. The latter converts \c U to \c T, and keeps the converted
+    /// value in it.
+    class yield_awaiter final {
+        generator_promise_base* _promise;
+        std::coroutine_handle<> _consumer;
+    public:
+        yield_awaiter(generator_promise_base* promise,
+                      std::coroutine_handle<> consumer) noexcept
+            : _promise{promise}
+            , _consumer{consumer}
+        {}
+        bool await_ready() const noexcept {
+            return false;
+        }
+        template <typename Promise>
+        std::coroutine_handle<> await_suspend(std::coroutine_handle<Promise> producer) noexcept {
+            _promise->_waiting_task = &producer.promise();
+            if (seastar::need_preempt()) {
+                auto consumer = std::coroutine_handle<seastar::task>::from_address(
+                    _consumer.address());
+                seastar::schedule(&consumer.promise());
+                return std::noop_coroutine();
+            }
+            return _consumer;
+        }
+        void await_resume() noexcept {}
+    };
+
+    class copy_awaiter {
+        using value_type = std::remove_cvref_t<Yielded>;
+        generator_promise_base* _promise;
+        std::coroutine_handle<> _consumer;
+        value_type _value;
+    public:
+        copy_awaiter(generator_promise_base* promise,
+                     std::coroutine_handle<> consumer,
+                     const value_type& value)
+            : _promise{promise}
+            , _consumer{consumer}
+            , _value{value}
+        {}
+        constexpr bool await_ready() const noexcept {
+            return false;
+        }
+        template <typename Promise>
+        std::coroutine_handle<> await_suspend(std::coroutine_handle<Promise> producer) noexcept {
+            auto& current = producer.promise();
+            current._value = std::addressof(_value);
+            _promise->_waiting_task = &current;
+            if (seastar::need_preempt()) {
+                auto consumer = std::coroutine_handle<seastar::task>::from_address(
+                    _consumer.address());
+                seastar::schedule(&consumer.promise());
+                return std::noop_coroutine();
+            }
+            return _consumer;
+        }
+        constexpr void await_resume() const noexcept {}
+    };
 
 public:
-    generator_unbuffered_promise() = default;
-    generator_unbuffered_promise(generator_unbuffered_promise&&) = delete;
-    generator_unbuffered_promise(const generator_unbuffered_promise&) = delete;
-
-    void return_void() noexcept;
-    void unhandled_exception() noexcept;
-
-    template<std::convertible_to<T> U>
-    suspend_always yield_value(U&& value) noexcept {
-        assert(_generator);
-        _generator->put_next_value(std::forward<U>(value));
-        assert(_wait_for_next_value);
-        _wait_for_next_value->set_value();
-        _wait_for_next_value = {};
+    generator_promise_base() noexcept = default;
+    generator_promise_base(const generator_promise_base &) = delete;
+    generator_promise_base& operator=(const generator_promise_base &) = delete;
+    generator_promise_base(generator_promise_base &&) noexcept = default;
+    generator_promise_base& operator=(generator_promise_base &&) noexcept = default;
+
+    // lazily-started coroutine, do not execute the coroutine until
+    // the coroutine is awaited.
+    std::suspend_always initial_suspend() const noexcept {
         return {};
     }
 
-    generator_type get_return_object() noexcept;
-    void set_generator(generator_type* g) noexcept {
-        assert(!_generator);
-        _generator = g;
-    }
-
-    suspend_always initial_suspend() const noexcept { return {}; }
-    suspend_never final_suspend() const noexcept {
-        assert(_generator);
-        _generator->on_finished();
-        return {};
+    yield_awaiter final_suspend() noexcept {
+        _value = nullptr;
+        return yield_awaiter{this, this->_consumer};
     }
 
-    seastar::future<> wait_for_next_value() noexcept {
-        assert(!_wait_for_next_value);
-        return _wait_for_next_value.emplace().get_future();
+    void unhandled_exception() noexcept {
+        _exception = std::current_exception();
     }
 
-    void run_and_dispose() noexcept final {
-        using handle_type = coroutine_handle<generator_unbuffered_promise>;
-        handle_type::from_promise(*this).resume();
+    yield_awaiter yield_value(Yielded value) noexcept {
+        this->_value = std::addressof(value);
+        return yield_awaiter{this, this->_consumer};
     }
 
-    seastar::task* waiting_task() noexcept final {
-        if (_wait_for_next_value) {
-            return _wait_for_next_value->waiting_task();
-        } else {
-            return nullptr;
-        }
+    copy_awaiter yield_value(const std::remove_reference_t<Yielded>& value)
+    requires (std::is_rvalue_reference_v<Yielded> &&
+              std::constructible_from<
+                  std::remove_cvref_t<Yielded>,
+                  const std::remove_reference_t<Yielded>&>) {
+        return copy_awaiter{this, this->_consumer, value};
     }
-};
-
-template <NothrowMoveConstructible T, template <typename> class Container>
-requires Fifo<Container, T>
-class generator_buffered_promise;
 
-template<typename T, template <typename> class Container>
-struct yield_awaiter final {
-    using promise_type = generator_buffered_promise<T, Container>;
-    seastar::future<> _future;
+    void return_void() noexcept {}
 
-public:
-    yield_awaiter(seastar::future<>&& f) noexcept
-        : _future{std::move(f)} {}
-
-    bool await_ready() noexcept {
-        return _future.available();
+    // @return if the generator has reached the end of the sequence
+    bool finished() const noexcept {
+        return _value == nullptr;
     }
 
-    coroutine_handle<> await_suspend(coroutine_handle<promise_type> coro) noexcept;
-    void await_resume() noexcept { }
-};
-
-
-template<NothrowMoveConstructible T, template <typename> class Container>
-requires Fifo<Container, T>
-class generator_buffered_promise final : public seastar::task {
-    using generator_type = seastar::coroutine::experimental::generator<T, Container>;
-
-    std::optional<seastar::promise<>> _wait_for_next_value;
-    std::optional<seastar::promise<>> _wait_for_free_space;
-    generator_type* _generator = nullptr;
-    const size_t _buffer_capacity;
-
-public:
-    template<typename... Args>
-    generator_buffered_promise(buffer_size_t buffer_capacity, Args&&... args)
-        : _buffer_capacity{static_cast<size_t>(buffer_capacity)} {}
-    generator_buffered_promise(generator_buffered_promise&&) = delete;
-    generator_buffered_promise(const generator_buffered_promise&) = delete;
-    ~generator_buffered_promise() = default;
-    void return_void() noexcept {
-        if (_wait_for_next_value) {
-            _wait_for_next_value->set_value();
-            _wait_for_next_value = {};
+    void rethrow_if_unhandled_exception() {
+        if (_exception) {
+            std::rethrow_exception(std::move(_exception));
         }
     }
-    void unhandled_exception() noexcept;
 
-    template<std::convertible_to<T> U>
-    yield_awaiter<T, Container> yield_value(U&& value) noexcept {
-        bool ready = _generator->put_next_value(std::forward<U>(value));
-
-        if (_wait_for_next_value) {
-            _wait_for_next_value->set_value();
-            _wait_for_next_value = {};
-            return {make_ready_future()};
-        }
-        if (ready) {
-            return {make_ready_future()};
-        } else {
-            assert(!_wait_for_free_space);
-            return {_wait_for_free_space.emplace().get_future()};
-        }
-    }
-
-    auto get_return_object() noexcept -> generator_type;
-    void set_generator(generator_type* g) noexcept {
-        assert(!_generator);
-        _generator = g;
-    }
-
-    suspend_always initial_suspend() const noexcept { return {}; }
-    suspend_never final_suspend() const noexcept {
-        assert(_generator);
-        _generator->on_finished();
-        return {};
-    }
-
-    bool is_awaiting() const noexcept {
-        return _wait_for_next_value.has_value();
-    }
-
-    coroutine_handle<> coroutine() const noexcept {
-        return coroutine_handle<>::from_address(_wait_for_next_value->waiting_task());
-    }
-
-    seastar::future<> wait_for_next_value() noexcept {
-        assert(!_wait_for_next_value);
-        return _wait_for_next_value.emplace().get_future();
-    }
-
-    void on_reclaim_free_space() noexcept {
-        assert(_wait_for_free_space);
-        _wait_for_free_space->set_value();
-        _wait_for_free_space = {};
-    }
-
-private:
     void run_and_dispose() noexcept final {
-        using handle_type = coroutine_handle<generator_buffered_promise>;
+        using handle_type = std::coroutine_handle<generator_promise_base>;
         handle_type::from_promise(*this).resume();
     }
 
     seastar::task* waiting_task() noexcept final {
-        if (_wait_for_next_value) {
-            return _wait_for_next_value->waiting_task();
-        } else if (_wait_for_free_space) {
-            return _wait_for_free_space->waiting_task();
-        } else {
-            return nullptr;
-        }
+        return _waiting_task;
     }
+
+private:
+    friend class next_awaiter<Yielded>;
 };
 
-template<typename T, typename Generator>
-struct next_awaiter final {
-    using next_value_type = next_value_t<T>;
-    Generator* const _generator;
-    seastar::task* const _task;
-    seastar::future<> _next_value_future;
+/// awaiter returned when the consumer gets the \c begin iterator or
+/// when it advances the iterator.
+template <typename Yielded>
+class next_awaiter {
+protected:
+    generator_promise_base<Yielded>* _promise = nullptr;
+    std::coroutine_handle<> _producer = nullptr;
 
-public:
-    next_awaiter(Generator* generator,
-                 seastar::task* task,
-                 seastar::future<>&& f) noexcept
-        : _generator{generator}
-        , _task(task)
-        , _next_value_future(std::move(f)) {}
-
-    next_awaiter(const next_awaiter&) = delete;
-    next_awaiter(next_awaiter&&) = delete;
-
-    constexpr bool await_ready() const noexcept {
-        return _next_value_future.available() && !seastar::need_preempt();
-    }
+    explicit next_awaiter(std::nullptr_t) noexcept {}
+    next_awaiter(generator_promise_base<Yielded>& promise,
+                 std::coroutine_handle<> producer) noexcept
+        : _promise{std::addressof(promise)}
+        , _producer{producer} {}
 
-    template<typename Promise>
-    void await_suspend(coroutine_handle<Promise> coro) noexcept {
-        auto& current_task = coro.promise();
-        if (_next_value_future.available()) {
-            seastar::schedule(&current_task);
-        } else {
-            _next_value_future.set_coroutine(current_task);
-            seastar::schedule(_task);
-        }
+public:
+    bool await_ready() const noexcept {
+        return false;
     }
 
-    next_value_type await_resume() {
-        assert(_next_value_future.available());
-        assert(_generator);
-        return _generator->take_next_value();
+    template <typename Promise>
+    std::coroutine_handle<> await_suspend(std::coroutine_handle<Promise> consumer) noexcept {
+        _promise->_consumer = consumer;
+        return _producer;
     }
 };
 
 } // namespace internal
 
-/// `seastar::coroutine::experimental::generator<T>` can be used to model a
-/// generator which yields a sequence of values asynchronously.
+/// generator represents a view modeling std::ranges::input_range,
+/// and has move-only iterators.
 ///
-/// Typically, it is used as the return type of a coroutine which is only
-/// allowed to either `co_yield` the next element in the sequence, or
-/// `co_return` nothing to indicate the end of this sequence. Please note,
-/// despite that `tl::generator` defined in `tests/unit/tl-generator.hh` also
-/// represents a generator, it can only produce the elements in a synchronous
-/// coroutine which does not suspend itself because of any asynchronous
-/// operation.
+/// generator has 2 template parameters:
 ///
-/// Example
+/// - Ref
+/// - Value
 ///
-/// ```
-/// auto generate_request = [&input_stream](coroutine::experimental::buffer_size_t)
-///     -> seastar::coroutine::generator<Request> {
-///     while (!input_stream.eof()) {
-///         co_yield co_await input_stream.read_exactly(42);
-///     }
+/// From Ref and Value, we derive types:
+/// - value_type: a cv-unqualified object type that specifies the value type of
+///   the generator's range and iterators
+/// - reference_Type: the reference type of the generator's range and iterators
+/// - yielded_type: the type of the parameter to the primary overload of \c
+///   yield_value in the generator's associated promise type
+///
+/// Under the most circumstances, only the first parameter is specified: like
+/// \c generator<meow>. The resulting generator:
+/// - has a value type of \c remove_cvref_t<meow>
+/// - has a reference type of \c meow, if it is a reference type, or \c meow&&
+///   otherwise
+/// - the operand of \c co_yield in the body of the generator should be
+///   convertible to \c meow, if it is a reference type, otherwise the operand
+///   type should be <tt>const meow&</tt>
+///
+/// Consider following code snippet:
+/// \code
+/// generator<const std::string&> send_query(std::string query) {
+///   auto result_set = db.execute(query);
+///   for (auto row : result_set) {
+///       co_yield std::format("{}", row);
+///   }
 /// }
+/// \endcode
+///
+/// In this case, \c Ref is a reference type of \c <tt>const std::string&</tt>,
+/// and \c Value is the default value of \c void. So the \c value_type is
+/// \c std::string. As the generator always returns a \c std::string, its
+/// iterator has the luxury of returning a reference to it.
+///
+/// But if some rare users want to use a proxy reference type, or to generate a
+/// range whose iterators yield prvalue for whatever reason, they should use
+/// the two-argument \c generator, like <tt>generator<meow, woof></tt>.
+/// The resulting generator:
+/// - has a value type of \c woof
+/// - has a reference type of \c meow
 ///
-/// const coroutine::experimental::buffer_size_t backlog = 42;
-/// while (true) {
-///     auto request = co_await generate_request(backlog);
-///     if (!request) {
-///         break;
-///     }
-///     co_await process(*std::move(request));
+/// For instance, consider following code snippet:
+/// \code
+/// generator<std::string_view, std::string> generate_strings() {
+///   co_yield "[";
+///   std::string s;
+///   for (auto sv : {"1"sv, "2"sv}) {
+///     s = sv;
+///     s.push_back(',');
+///     co_yield s;
+///   }
+///   co_yield "]";
 /// }
-/// ````
-template <typename T, template <typename> class Container>
-class generator {
+/// \endcode
+///
+/// In this case, \c Ref is \c std::string_view, and \Value is \c std::string.
+/// So we can ensure that the caller cannot invalidate the yielded values by
+/// mutating the defererenced value of iterator. As the \c std::string_view
+/// instance is immutable. But in the meanwhile, the generator can return
+/// a \c std::string by \c co_yield a \c std::string_view or a \c std::string.
+/// And the caller can still access the element of the range via the same type:
+/// \c std::string_view.
+///
+/// Current Limitation and Future Plans:
+///
+/// This generator implementation does not address the "Pingpong problem":
+/// where the producer generates elements one at a time, forcing frequent
+/// context switches between producer and consumer. This can lead to suboptimal
+/// performance, especially when bulk generation and consumption would be more
+/// efficient.
+///
+/// We intend to extend the existing implementation to allow the producer
+/// to yield a range of elements. This will enable batch processing,
+/// potentially improving performance by reducing context switches.
+///
+/// TODO: Implement range-based yielding to mitigate the Ping-pong problem.
+template<typename Ref, typename Value>
+class [[nodiscard]] generator {
+    using value_type = std::conditional_t<std::is_void_v<Value>,
+                                          std::remove_cvref_t<Ref>,
+                                          Value>;
+    using reference_type = std::conditional_t<std::is_void_v<Value>,
+                                              Ref&&,
+                                              Ref>;
+    using yielded_type = std::conditional_t<std::is_reference_v<reference_type>,
+                                            reference_type,
+                                            const reference_type&>;
+
 public:
-    using promise_type = internal::generator_buffered_promise<T, Container>;
+    class promise_type;
 
 private:
-    using handle_type = internal::coroutine_handle<promise_type>;
-    handle_type _coro;
-    promise_type* _promise;
-    Container<T> _values;
-    const size_t _buffer_capacity;
-    std::exception_ptr _exception;
+    using handle_type = std::coroutine_handle<promise_type>;
+    handle_type _coro = {};
 
 public:
-    generator(size_t buffer_capacity,
-              handle_type coro,
-              promise_type* promise) noexcept
-        : _coro{coro}
-        , _promise{promise}
-        , _buffer_capacity{buffer_capacity} {
-        assert(_promise);
-        _promise->set_generator(this);
-    }
-    generator(const generator&) = delete;
+    class iterator;
+
+    generator() noexcept = default;
+    explicit generator(promise_type& promise) noexcept
+        : _coro(std::coroutine_handle<promise_type>::from_promise(promise))
+    {}
     generator(generator&& other) noexcept
         : _coro{std::exchange(other._coro, {})}
-        , _buffer_capacity{other._buffer_capacity} {}
-    generator& operator=(generator&& other) noexcept {
-        if (std::addressof(other) != this) {
-            auto old_coro = std::exchange(_coro, std::exchange(other._coro, {}));
-            if (old_coro) {
-                old_coro.destroy();
-            }
-        }
-        return *this;
-    }
+    {}
+    generator(const generator&) = delete;
+    generator& operator=(const generator&) = delete;
+
     ~generator() {
         if (_coro) {
             _coro.destroy();
         }
     }
 
-    void swap(generator& other) noexcept {
-        std::swap(_coro, other._coro);
+    friend void swap(generator& lhs, generator& rhs) noexcept {
+        std::swap(lhs._coro, rhs._coro);
     }
 
-    internal::next_awaiter<T, generator> operator()() noexcept {
-        if (!_values.empty()) {
-            return {this, nullptr, make_ready_future<>()};
-        } else if (_exception) [[unlikely]] {
-            return {this, nullptr, make_ready_future<>()};
-        } else if (_promise) {
-            return {this, _promise, _promise->wait_for_next_value()};
-        } else {
-            return {this, nullptr, make_ready_future<>()};
+    generator& operator=(generator&& other) noexcept {
+        if (this == &other) {
+            return *this;
         }
+        if (_coro) {
+            _coro.destroy();
+        }
+        _coro = std::exchange(other._coro, nullptr);
+        return *this;
     }
 
-    template<typename U>
-    bool put_next_value(U&& value) {
-        _values.push_back(std::forward<U>(value));
-        return _values.size() < _buffer_capacity;
-    }
+    [[nodiscard]] auto begin() noexcept {
+        using base_awaiter = internal::next_awaiter<yielded_type>;
+        class begin_awaiter final : public base_awaiter {
+            using base_awaiter::_promise;
+
+        public:
+            explicit begin_awaiter(std::nullptr_t) noexcept
+                : base_awaiter{nullptr}
+            {}
+            explicit begin_awaiter(handle_type producer_coro) noexcept
+                : base_awaiter{producer_coro.promise(), producer_coro}
+            {}
+            bool await_ready() const noexcept {
+                return _promise == nullptr || base_awaiter::await_ready();
+            }
 
-    internal::next_value_t<T> take_next_value() {
-        if (!_values.empty()) [[likely]] {
-            auto value = std::move(_values.front());
-            bool maybe_reclaim = _values.size() == _buffer_capacity;
-            _values.pop_front();
-            if (maybe_reclaim) {
-                if (_promise) [[likely]] {
-                    _promise->on_reclaim_free_space();
+            iterator await_resume() {
+                if (_promise == nullptr) {
+                    return iterator{nullptr};
                 }
+                if (_promise->finished()) {
+                    _promise->rethrow_if_unhandled_exception();
+                    return iterator{nullptr};
+                }
+                return iterator{
+                    handle_type::from_promise(*static_cast<promise_type *>(_promise))
+                };
             }
-            return internal::next_value_t<T>(std::move(value));
-        } else if (_exception) [[unlikely]] {
-            std::rethrow_exception(std::exchange(_exception, nullptr));
+        };
+
+        if (_coro) {
+            return begin_awaiter{_coro};
         } else {
-            return std::nullopt;
+            return begin_awaiter{nullptr};
         }
     }
 
-    void on_finished() {
-        _promise = nullptr;
-        _coro = nullptr;
-    }
-
-    void unhandled_exception() noexcept {
-        // called by promise's unhandled_exception()
-        assert(!_exception);
-        _exception = std::current_exception();
+    [[nodiscard]] std::default_sentinel_t end() const noexcept {
+        return {};
     }
 };
 
-template <typename T>
-class generator<T, std::optional> {
-public:
-    using promise_type = internal::generator_unbuffered_promise<T>;
-
-private:
-    using handle_type = internal::coroutine_handle<promise_type>;
-    handle_type _coro;
-    promise_type* _promise;
-    std::optional<T> _maybe_value;
-    std::exception_ptr _exception;
-
+template <typename Ref, typename Value>
+class generator<Ref, Value>::promise_type final : public internal::generator_promise_base<yielded_type> {
 public:
-    generator(handle_type coro,
-              promise_type* promise) noexcept
-        : _coro{coro}
-        , _promise{promise} {
-        assert(_promise);
-        _promise->set_generator(this);
-    }
-    generator(const generator&) = delete;
-    generator(generator&& other) noexcept
-        : _coro{std::exchange(other._coro, {})} {}
-    generator& operator=(generator&& other) noexcept {
-        if (std::addressof(other) != this) {
-            auto old_coro = std::exchange(_coro, std::exchange(other._coro, {}));
-            if (old_coro) {
-                old_coro.destroy();
-            }
-        }
-        return *this;
-    }
-    ~generator() {
-        if (_coro) {
-            _coro.destroy();
-        }
+    generator get_return_object() noexcept {
+        return generator{*this};
     }
 
-    void swap(generator& other) noexcept {
-        std::swap(_coro, other._coro);
+    yielded_type value() const noexcept {
+        return static_cast<yielded_type>(*this->_value);
     }
+};
 
-    internal::next_awaiter<T, generator> operator()() noexcept {
-        if (_promise) [[likely]] {
-            return {this, _promise, _promise->wait_for_next_value()};
-        } else {
-            return {this, nullptr, make_ready_future<>()};
-        }
-    }
+template <typename Ref, typename Value>
+class generator<Ref, Value>::iterator final {
+private:
+    using handle_type = generator::handle_type;
+    handle_type _coro = nullptr;
 
-    template<typename U>
-    void put_next_value(U&& value) noexcept {
-        _maybe_value.emplace(std::forward<U>(value));
-    }
+public:
+    using iterator_category = std::input_iterator_tag;
+    using difference_type = std::ptrdiff_t;
+    using value_type = generator::value_type;
+    using reference = generator::reference_type;
+    using pointer = std::add_pointer_t<value_type>;
+
+    explicit iterator(handle_type coroutine) noexcept
+        : _coro{coroutine}
+    {}
+
+    explicit operator bool() const noexcept {
+        return _coro && !_coro.done();
+    }
+
+    [[nodiscard]] auto operator++() noexcept {
+        using base_awaiter = internal::next_awaiter<yielded_type>;
+        class increment_awaiter final : public base_awaiter {
+            iterator& _iterator;
+            using base_awaiter::_promise;
+
+        public:
+            explicit increment_awaiter(iterator& iterator) noexcept
+                : base_awaiter{iterator._coro.promise(), iterator._coro}
+                , _iterator{iterator}
+            {}
+            iterator& await_resume() {
+                if (_promise->finished()) {
+                    // update iterator to end()
+                    _iterator = iterator{nullptr};
+                    _promise->rethrow_if_unhandled_exception();
+                }
+                return _iterator;
+            }
+        };
 
-    internal::next_value_t<T> take_next_value() {
-        if (_maybe_value.has_value()) [[likely]] {
-            return std::exchange(_maybe_value, std::nullopt);
-        } else if (_exception) [[unlikely]] {
-            std::rethrow_exception(std::exchange(_exception, nullptr));
-        } else {
-            return std::nullopt;
-        }
+        assert(bool(*this) && "cannot increment end iterator");
+        return increment_awaiter{*this};
     }
 
-    void on_finished() {
-        _promise = nullptr;
-        _coro = nullptr;
+    reference operator*() const noexcept {
+        return _coro.promise().value();
     }
 
-    void unhandled_exception() noexcept {
-        assert(!_exception);
-        _exception = std::current_exception();
+    bool operator==(std::default_sentinel_t) const noexcept {
+        return !bool(*this);
     }
 };
 
-namespace internal {
-
-template<NothrowMoveConstructible T>
-void generator_unbuffered_promise<T>::return_void() noexcept {
-    assert(_wait_for_next_value);
-    _wait_for_next_value->set_value();
-    _wait_for_next_value = {};
-}
-
-template<NothrowMoveConstructible T>
-void generator_unbuffered_promise<T>::unhandled_exception() noexcept {
-    // instead of storing the current exception into promise, in order to be
-    // more consistent, we let generator preserve all the output of produced
-    // value, including the values and the exception if any. so we just signal
-    // _wait_for_next_value, and delegate generator's unhandled_exception() to
-    // store the exception.
-    _generator->unhandled_exception();
-    if (_wait_for_next_value.has_value()) {
-        _wait_for_next_value->set_value();
-        _wait_for_next_value = {};
-    }
-}
-
-template<NothrowMoveConstructible T>
-auto generator_unbuffered_promise<T>::get_return_object() noexcept -> generator_type {
-    using handle_type = coroutine_handle<generator_unbuffered_promise<T>>;
-    return generator_type{handle_type::from_promise(*this), this};
-}
-
-template<NothrowMoveConstructible T, template <typename> class Container>
-requires Fifo<Container, T>
-void generator_buffered_promise<T, Container>::unhandled_exception() noexcept {
-    _generator->unhandled_exception();
-    if (_wait_for_next_value.has_value()) {
-        _wait_for_next_value->set_value();
-        _wait_for_next_value = {};
-    }
-}
-
-template<NothrowMoveConstructible T, template <typename> class Container>
-requires Fifo<Container, T>
-auto generator_buffered_promise<T, Container>::get_return_object() noexcept -> generator_type {
-    using handle_type = coroutine_handle<generator_buffered_promise<T, Container>>;
-    return generator_type{_buffer_capacity, handle_type::from_promise(*this), this};
-}
-
-template<typename T, template <typename> class Container>
-coroutine_handle<> yield_awaiter<T, Container>::await_suspend(
-    coroutine_handle<generator_buffered_promise<T, Container>> coro) noexcept {
-    if (_future.available()) {
-        auto& current_task = coro.promise();
-        seastar::schedule(&current_task);
-        return coro;
-    } else {
-        // we cannot do something like `task.set_coroutine(consumer_task)`.
-        // because, instead of waiting for a subcoroutine, we are pending on
-        // the caller of current coroutine to consume the produced values to
-        // free up at least a free slot in the buffer, if we set the `_task`
-        // of the of the awaiting task, we would have an infinite loop of
-        // "promise->_task".
-        return noop_coroutine();
-    }
-}
-
-} // namespace internal
 } // namespace seastar::coroutine::experimental
diff --git a/src/core/file-impl.hh b/src/core/file-impl.hh
index 3f29cff559f..10790281f03 100644
--- a/src/core/file-impl.hh
+++ b/src/core/file-impl.hh
@@ -103,7 +103,7 @@ public:
     virtual future<> close() noexcept override;
     virtual std::unique_ptr<seastar::file_handle_impl> dup() override;
     virtual subscription<directory_entry> list_directory(std::function<future<> (directory_entry de)> next) override;
-    virtual coroutine::experimental::generator<directory_entry> experimental_list_directory() override;
+    virtual coroutine::experimental::generator<const directory_entry&> experimental_list_directory() override;
 
 #if SEASTAR_API_LEVEL >= 7
     virtual future<size_t> read_dma(uint64_t pos, void* buffer, size_t len, io_intent* intent) noexcept override = 0;
diff --git a/src/core/file.cc b/src/core/file.cc
index 8c914534863..b2495446f78 100644
--- a/src/core/file.cc
+++ b/src/core/file.cc
@@ -393,7 +393,7 @@ static std::optional<directory_entry_type> dirent_type(const linux_dirent64& de)
     return type;
 }
 
-static coroutine::experimental::generator<directory_entry> make_list_directory_generator(int fd) {
+static coroutine::experimental::generator<const directory_entry&> make_list_directory_generator(int fd) {
     temporary_buffer<char> buf(8192);
 
     while (true) {
@@ -417,7 +417,7 @@ static coroutine::experimental::generator<directory_entry> make_list_directory_g
     }
 }
 
-coroutine::experimental::generator<directory_entry> posix_file_impl::experimental_list_directory() {
+coroutine::experimental::generator<const directory_entry&> posix_file_impl::experimental_list_directory() {
     // due to https://github.com/scylladb/seastar/issues/1913, we cannot use
     // buffered generator yet.
     // TODO:
@@ -1161,7 +1161,7 @@ file::list_directory(std::function<future<>(directory_entry de)> next) {
     return _file_impl->list_directory(std::move(next));
 }
 
-coroutine::experimental::generator<directory_entry> file::experimental_list_directory() {
+coroutine::experimental::generator<const directory_entry&> file::experimental_list_directory() {
     return _file_impl->experimental_list_directory();
 }
 
@@ -1368,7 +1368,7 @@ file_impl::dup() {
     throw std::runtime_error("this file type cannot be duplicated");
 }
 
-static coroutine::experimental::generator<directory_entry> make_list_directory_fallback_generator(file_impl& me) {
+static coroutine::experimental::generator<const directory_entry&> make_list_directory_fallback_generator(file_impl& me) {
     queue<std::optional<directory_entry>> ents(16);
     auto lister = me.list_directory([&ents] (directory_entry de) {
         return ents.push_eventually(std::move(de));
@@ -1388,7 +1388,7 @@ static coroutine::experimental::generator<directory_entry> make_list_directory_f
     co_await std::move(done);
 }
 
-coroutine::experimental::generator<directory_entry> file_impl::experimental_list_directory() {
+coroutine::experimental::generator<const directory_entry&> file_impl::experimental_list_directory() {
     return make_list_directory_fallback_generator(*this);
 }
 
diff --git a/tests/unit/coroutines_test.cc b/tests/unit/coroutines_test.cc
index 8e2b8a470c6..eba22bb54d0 100644
--- a/tests/unit/coroutines_test.cc
+++ b/tests/unit/coroutines_test.cc
@@ -34,7 +34,6 @@
 #include <seastar/coroutine/parallel_for_each.hh>
 #include <seastar/coroutine/as_future.hh>
 #include <seastar/coroutine/exception.hh>
-#include <seastar/coroutine/generator.hh>
 #include <seastar/testing/random.hh>
 #include <seastar/testing/test_case.hh>
 #include <seastar/util/later.hh>
@@ -741,147 +740,6 @@ SEASTAR_TEST_CASE(test_as_future_preemption) {
     BOOST_REQUIRE_THROW(f0.get(), std::runtime_error);
 }
 
-template<template<typename> class Container>
-coroutine::experimental::generator<int, Container>
-fibonacci_sequence(coroutine::experimental::buffer_size_t size, unsigned count) {
-    auto a = 0, b = 1;
-    for (unsigned i = 0; i < count; ++i) {
-        if (std::numeric_limits<decltype(a)>::max() - a < b) {
-            throw std::out_of_range(
-                fmt::format("fibonacci[{}] is greater than the largest value of int", i));
-        }
-        co_yield std::exchange(a, std::exchange(b, a + b));
-    }
-}
-
-template<template<typename> class Container>
-seastar::future<> test_async_generator_drained() {
-    auto expected_fibs = {0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55};
-    auto fib = fibonacci_sequence<Container>(coroutine::experimental::buffer_size_t{2},
-                                             std::size(expected_fibs));
-    for (auto expected_fib : expected_fibs) {
-        auto actual_fib = co_await fib();
-        BOOST_REQUIRE(actual_fib.has_value());
-        BOOST_REQUIRE_EQUAL(actual_fib.value(), expected_fib);
-    }
-    auto sentinel = co_await fib();
-    BOOST_REQUIRE(!sentinel.has_value());
-}
-
-template<typename T>
-using buffered_container = circular_buffer<T>;
-
-SEASTAR_TEST_CASE(test_async_generator_drained_buffered) {
-    return test_async_generator_drained<buffered_container>();
-}
-
-SEASTAR_TEST_CASE(test_async_generator_drained_unbuffered) {
-    return test_async_generator_drained<std::optional>();
-}
-
-template<template<typename> class Container>
-seastar::future<> test_async_generator_not_drained() {
-    auto fib = fibonacci_sequence<Container>(coroutine::experimental::buffer_size_t{2},
-                                             42);
-    auto actual_fib = co_await fib();
-    BOOST_REQUIRE(actual_fib.has_value());
-    BOOST_REQUIRE_EQUAL(actual_fib.value(), 0);
-}
-
-SEASTAR_TEST_CASE(test_async_generator_not_drained_buffered) {
-    return test_async_generator_not_drained<buffered_container>();
-}
-
-SEASTAR_TEST_CASE(test_async_generator_not_drained_unbuffered) {
-    return test_async_generator_not_drained<std::optional>();
-}
-
-struct counter_t {
-    int n;
-    int* count;
-    counter_t(counter_t&& other) noexcept
-        : n{std::exchange(other.n, -1)},
-          count{std::exchange(other.count, nullptr)}
-    {}
-    counter_t(int n, int* count) noexcept
-        : n{n}, count{count} {
-        ++(*count);
-    }
-    ~counter_t() noexcept {
-        if (count) {
-            --(*count);
-        }
-    }
-};
-std::ostream& operator<<(std::ostream& os, const counter_t& c) {
-    return os << c.n;
-}
-
-template<template<typename> class Container>
-coroutine::experimental::generator<counter_t, Container>
-fiddle(coroutine::experimental::buffer_size_t size, int n, int* total) {
-    int i = 0;
-    while (true) {
-        if (i++ == n) {
-            throw std::invalid_argument("Eureka from generator!");
-        }
-        co_yield counter_t{i, total};
-    }
-}
-
-template<template<typename> class Container>
-seastar::future<> test_async_generator_throws_from_generator() {
-    int total = 0;
-    auto count_to = [total=&total](unsigned n) -> seastar::future<> {
-        auto count = fiddle<Container>(coroutine::experimental::buffer_size_t{2},
-                                       n, total);
-        for (unsigned i = 0; i < 2 * n; i++) {
-            co_await count();
-        }
-    };
-    co_await count_to(42).then_wrapped([&total] (auto f) {
-        BOOST_REQUIRE(f.failed());
-        BOOST_REQUIRE_THROW(std::rethrow_exception(f.get_exception()), std::invalid_argument);
-        BOOST_REQUIRE_EQUAL(total, 0);
-    });
-}
-
-SEASTAR_TEST_CASE(test_async_generator_throws_from_generator_buffered) {
-    return test_async_generator_throws_from_generator<buffered_container>();
-}
-
-SEASTAR_TEST_CASE(test_async_generator_throws_from_generator_unbuffered) {
-    return test_async_generator_throws_from_generator<std::optional>();
-}
-
-template<template<typename> class Container>
-seastar::future<> test_async_generator_throws_from_consumer() {
-    int total = 0;
-    auto count_to = [total=&total](unsigned n) -> seastar::future<> {
-        auto count = fiddle<Container>(coroutine::experimental::buffer_size_t{2},
-                                       n, total);
-        for (unsigned i = 0; i < n; i++) {
-            if (i == n / 2) {
-                throw std::invalid_argument("Eureka from consumer!");
-            }
-            co_await count();
-        }
-    };
-    co_await count_to(42).then_wrapped([&total] (auto f) {
-        BOOST_REQUIRE(f.failed());
-        BOOST_REQUIRE_THROW(std::rethrow_exception(f.get_exception()), std::invalid_argument);
-        BOOST_REQUIRE_EQUAL(total, 0);
-    });
-}
-
-SEASTAR_TEST_CASE(test_async_generator_throws_from_consumer_buffered) {
-    return test_async_generator_throws_from_consumer<buffered_container>();
-}
-
-SEASTAR_TEST_CASE(test_async_generator_throws_from_consumer_unbuffered) {
-    return test_async_generator_throws_from_consumer<std::optional>();
-}
-
 SEASTAR_TEST_CASE(test_lambda_coroutine_in_continuation) {
     auto dist = std::uniform_real_distribution<>(0.0, 1.0);
     auto rand_eng = std::default_random_engine(std::random_device()());
diff --git a/tests/unit/directory_test.cc b/tests/unit/directory_test.cc
index ff37014974d..d46a3c898e1 100644
--- a/tests/unit/directory_test.cc
+++ b/tests/unit/directory_test.cc
@@ -88,14 +88,16 @@ future<> lister_test() {
 
 future<> lister_generator_test(file f) {
     auto lister = f.experimental_list_directory();
-    while (auto de = co_await lister()) {
-        auto sd = co_await file_stat(de->name, follow_symlink::no);
-        if (de->type) {
-            assert(*de->type == sd.type);
+
+    for (auto it = co_await lister.begin(); it != lister.end(); co_await ++it) {
+        const auto& de = *it;
+        auto sd = co_await file_stat(de.name, follow_symlink::no);
+        if (de.type) {
+            assert(de.type == sd.type);
         } else {
             assert(sd.type == directory_entry_type::unknown);
         }
-        fmt::print("{} (type={})\n", de->name, de_type_desc(sd.type));
+        fmt::print("{} (type={})\n", de.name, de_type_desc(sd.type));
     }
     co_await f.close();
 }
diff --git a/tests/unit/generator_test.cc b/tests/unit/generator_test.cc
index a240109e917..9b438b06f07 100644
--- a/tests/unit/generator_test.cc
+++ b/tests/unit/generator_test.cc
@@ -21,7 +21,7 @@
 
 #include <seastar/core/coroutine.hh>
 #include <seastar/coroutine/as_future.hh>
-#include <seastar/coroutine/async_generator.hh>
+#include <seastar/coroutine/generator.hh>
 #include <seastar/coroutine/exception.hh>
 #include <seastar/testing/test_case.hh>
 #include <seastar/util/bool_class.hh>
@@ -56,7 +56,7 @@ sync_fibonacci_sequence(unsigned count) {
     }
 }
 
-coroutine::experimental::async_generator<int>
+coroutine::experimental::generator<int>
 async_fibonacci_sequence(unsigned count, do_suspend suspend) {
     auto a = 0, b = 1;
     for (unsigned i = 0; i < count; ++i) {
@@ -72,7 +72,7 @@ async_fibonacci_sequence(unsigned count, do_suspend suspend) {
 }
 
 seastar::future<>
-verify_fib_drained(coroutine::experimental::async_generator<int> actual_fibs, unsigned count) {
+verify_fib_drained(coroutine::experimental::generator<int> actual_fibs, unsigned count) {
     auto expected_fibs = sync_fibonacci_sequence(count);
     auto expected_fib = std::begin(expected_fibs);
 
@@ -86,33 +86,33 @@ verify_fib_drained(coroutine::experimental::async_generator<int> actual_fibs, un
     BOOST_REQUIRE(actual_fib == actual_fibs.end());
 }
 
-SEASTAR_TEST_CASE(test_async_generator_drained_with_suspend) {
+SEASTAR_TEST_CASE(test_generator_drained_with_suspend) {
     constexpr int count = 4;
     return verify_fib_drained(async_fibonacci_sequence(count, do_suspend::yes),
                               count);
 }
 
-SEASTAR_TEST_CASE(test_async_generator_drained_without_suspend) {
+SEASTAR_TEST_CASE(test_generator_drained_without_suspend) {
     constexpr int count = 4;
     return verify_fib_drained(async_fibonacci_sequence(count, do_suspend::no),
                               count);
 }
 
-seastar::future<> test_async_generator_not_drained(do_suspend suspend) {
+seastar::future<> test_generator_not_drained(do_suspend suspend) {
     auto fib = async_fibonacci_sequence(42, suspend);
     auto actual_fib = co_await fib.begin();
     BOOST_REQUIRE_EQUAL(*actual_fib, 0);
 }
 
-SEASTAR_TEST_CASE(test_async_generator_not_drained_with_suspend) {
-    return test_async_generator_not_drained(do_suspend::yes);
+SEASTAR_TEST_CASE(test_generator_not_drained_with_suspend) {
+    return test_generator_not_drained(do_suspend::yes);
 }
 
-SEASTAR_TEST_CASE(test_async_generator_not_drained_without_suspend) {
-    return test_async_generator_not_drained(do_suspend::no);
+SEASTAR_TEST_CASE(test_generator_not_drained_without_suspend) {
+    return test_generator_not_drained(do_suspend::no);
 }
 
-coroutine::experimental::async_generator<std::string_view, std::string>
+coroutine::experimental::generator<std::string_view, std::string>
 generate_value_and_ref(std::vector<std::string_view> strings) {
     co_yield "[";
     std::string s;
@@ -124,7 +124,7 @@ generate_value_and_ref(std::vector<std::string_view> strings) {
     co_yield "]";
 }
 
-SEASTAR_TEST_CASE(test_async_generator_value_reference) {
+SEASTAR_TEST_CASE(test_generator_value_reference) {
     using namespace std::literals;
     std::vector<std::string_view> expected_quoted = {"["sv, "foo,"sv, "bar,"sv, "]"sv};
     auto actual_quoted = generate_value_and_ref({"foo"sv, "bar"sv});
@@ -135,14 +135,14 @@ SEASTAR_TEST_CASE(test_async_generator_value_reference) {
     }
 }
 
-coroutine::experimental::async_generator<std::string>
+coroutine::experimental::generator<std::string>
 generate_yield_rvalue_reference(const std::vector<std::string> strings) {
     for (auto& s: strings) {
         co_yield s;
     }
 }
 
-SEASTAR_TEST_CASE(test_async_generator_rvalue_reference) {
+SEASTAR_TEST_CASE(test_generator_rvalue_reference) {
     std::vector<std::string> expected_strings = {"hello", "world"};
     auto actual_strings = generate_yield_rvalue_reference(expected_strings);
     auto actual = co_await actual_strings.begin();
@@ -152,13 +152,13 @@ SEASTAR_TEST_CASE(test_async_generator_rvalue_reference) {
     }
 }
 
-SEASTAR_TEST_CASE(test_async_generator_move_ctor) {
+SEASTAR_TEST_CASE(test_generator_move_ctor) {
     constexpr int count = 4;
     auto actual_fibs = async_fibonacci_sequence(count, do_suspend::no);
     return verify_fib_drained(std::move(actual_fibs), count);
 }
 
-SEASTAR_TEST_CASE(test_async_generator_swap) {
+SEASTAR_TEST_CASE(test_generator_swap) {
     int count_a = 4;
     int count_b = 42;
     auto fibs_a = async_fibonacci_sequence(count_a, do_suspend::no);
@@ -191,7 +191,7 @@ std::ostream& operator<<(std::ostream& os, const counter_t& c) {
     return os << c.n;
 }
 
-coroutine::experimental::async_generator<counter_t>
+coroutine::experimental::generator<counter_t>
 fiddle(int n, int* total) {
     int i = 0;
     while (true) {
@@ -202,7 +202,7 @@ fiddle(int n, int* total) {
     }
 }
 
-SEASTAR_TEST_CASE(test_async_generator_throws_from_generator) {
+SEASTAR_TEST_CASE(test_generator_throws_from_generator) {
     int total = 0;
     auto count_to = [total=&total](unsigned n) -> seastar::future<> {
         auto count = fiddle(n, total);
@@ -217,7 +217,7 @@ SEASTAR_TEST_CASE(test_async_generator_throws_from_generator) {
     BOOST_REQUIRE_EQUAL(total, 0);
 }
 
-SEASTAR_TEST_CASE(test_async_generator_throws_from_consumer) {
+SEASTAR_TEST_CASE(test_generator_throws_from_consumer) {
     int total = 0;
     auto count_to = [total=&total](unsigned n) -> seastar::future<> {
         auto count = fiddle(n, total);

From e201340ce64e9b6b768c11607eea341a1b09cda5 Mon Sep 17 00:00:00 2001
From: Kefu Chai <kefu.chai@scylladb.com>
Date: Sun, 18 Aug 2024 20:01:50 +0800
Subject: [PATCH 3/3] coroutine/generator: introduce buffered generator for
 batch processing

before this change, we only have an unbuffered generator
implementation, where the producer and consumer operated in a
"ping-pong" fashion. elements were produced and consumed one at a time.

in this change, we introduce a buffered variant of the generator,
which allows producer to yield a range of elements in batch. it
also enables consumer to process multiple elements without suspension.

the benefits are:

* improved efficiency for batch operations
* reduced context switching between producer and consumer
* enhanced flexibility in element production and consumption patterns

Signed-off-by: Kefu Chai <kefu.chai@scylladb.com>
---
 include/seastar/coroutine/generator.hh | 352 +++++++++++++++++++++++--
 tests/unit/generator_test.cc           | 148 +++++++++++
 2 files changed, 484 insertions(+), 16 deletions(-)

diff --git a/include/seastar/coroutine/generator.hh b/include/seastar/coroutine/generator.hh
index 499695fd08f..8a6302d13dd 100644
--- a/include/seastar/coroutine/generator.hh
+++ b/include/seastar/coroutine/generator.hh
@@ -57,13 +57,16 @@
 //   Nesting generators is not supported. You cannot yield another generator
 //   from within a generator. This prevents implementation asynchronous,
 //   recursive algorithms like depth-first search on trees.
+// * Range Yielding:
+//   supports directly yielding a range. We could instead implement nesting
+//   using a dedicated awaiter and sophisticated frame tracking for nested
+//   generators, but this would significantly increase implementation complexity.
 namespace seastar::coroutine::experimental {
 
-template<typename Ref, typename Value = void>
-class generator;
-
 namespace internal {
 
+namespace unbuffered {
+
 template <typename Yielded> class next_awaiter;
 
 template <typename Yielded>
@@ -246,9 +249,7 @@ public:
     }
 };
 
-} // namespace internal
-
-/// generator represents a view modeling std::ranges::input_range,
+/// unbuffered generator represents a view modeling std::ranges::input_range,
 /// and has move-only iterators.
 ///
 /// generator has 2 template parameters:
@@ -316,19 +317,14 @@ public:
 /// And the caller can still access the element of the range via the same type:
 /// \c std::string_view.
 ///
-/// Current Limitation and Future Plans:
-///
 /// This generator implementation does not address the "Pingpong problem":
 /// where the producer generates elements one at a time, forcing frequent
 /// context switches between producer and consumer. This can lead to suboptimal
 /// performance, especially when bulk generation and consumption would be more
 /// efficient.
 ///
-/// We intend to extend the existing implementation to allow the producer
-/// to yield a range of elements. This will enable batch processing,
-/// potentially improving performance by reducing context switches.
-///
-/// TODO: Implement range-based yielding to mitigate the Ping-pong problem.
+/// If the producer is able to produce elements in batch, please consider using
+/// the buffered generator by specifying the \c Yielded template parameter.
 template<typename Ref, typename Value>
 class [[nodiscard]] generator {
     using value_type = std::conditional_t<std::is_void_v<Value>,
@@ -383,7 +379,7 @@ public:
     }
 
     [[nodiscard]] auto begin() noexcept {
-        using base_awaiter = internal::next_awaiter<yielded_type>;
+        using base_awaiter = next_awaiter<yielded_type>;
         class begin_awaiter final : public base_awaiter {
             using base_awaiter::_promise;
 
@@ -425,7 +421,7 @@ public:
 };
 
 template <typename Ref, typename Value>
-class generator<Ref, Value>::promise_type final : public internal::generator_promise_base<yielded_type> {
+class generator<Ref, Value>::promise_type final : public generator_promise_base<yielded_type> {
 public:
     generator get_return_object() noexcept {
         return generator{*this};
@@ -458,7 +454,7 @@ public:
     }
 
     [[nodiscard]] auto operator++() noexcept {
-        using base_awaiter = internal::next_awaiter<yielded_type>;
+        using base_awaiter = next_awaiter<yielded_type>;
         class increment_awaiter final : public base_awaiter {
             iterator& _iterator;
             using base_awaiter::_promise;
@@ -491,4 +487,328 @@ public:
     }
 };
 
+} // namespace unbuffered
+
+namespace buffered {
+
+template <typename Yielded> class next_awaiter;
+
+template <typename Yielded>
+class generator_promise_base : public seastar::task {
+protected:
+    // slice represents a sub sequence of the genenerated values
+    std::add_pointer_t<Yielded> _slice = nullptr;
+
+protected:
+    std::exception_ptr _exception;
+    std::coroutine_handle<> _consumer;
+    task* _waiting_task = nullptr;
+
+    class yield_awaiter final {
+        generator_promise_base* _promise;
+        std::coroutine_handle<> _consumer;
+    public:
+        yield_awaiter(generator_promise_base* promise,
+                      std::coroutine_handle<> consumer) noexcept
+            : _promise{promise}
+            , _consumer{consumer}
+        {}
+        bool await_ready() const noexcept {
+            return false;
+        }
+        template <typename Promise>
+        std::coroutine_handle<> await_suspend(std::coroutine_handle<Promise> producer) noexcept {
+            _promise->_waiting_task = &producer.promise();
+            if (seastar::need_preempt()) {
+                auto consumer = std::coroutine_handle<seastar::task>::from_address(
+                    _consumer.address());
+                seastar::schedule(&consumer.promise());
+                return std::noop_coroutine();
+            }
+            return _consumer;
+        }
+        void await_resume() noexcept {}
+    };
+
+public:
+    generator_promise_base() noexcept = default;
+    generator_promise_base(const generator_promise_base &) = delete;
+    generator_promise_base& operator=(const generator_promise_base &) = delete;
+    generator_promise_base(generator_promise_base &&) noexcept = default;
+    generator_promise_base& operator=(generator_promise_base &&) noexcept = default;
+
+    // lazily-started coroutine, do not execute the coroutine until
+    // the coroutine is awaited.
+    std::suspend_always initial_suspend() const noexcept {
+        return {};
+    }
+
+    yield_awaiter final_suspend() noexcept {
+        _slice = nullptr;
+        return yield_awaiter{this, this->_consumer};
+    }
+
+    void unhandled_exception() noexcept {
+        _exception = std::current_exception();
+    }
+
+    yield_awaiter yield_value(Yielded slice) noexcept {
+        // an empty slice is forbidden, otherwise the increment_awaiter would resume
+        // from suspension, but end up finding nothing to return to the consumer.
+        assert(!std::ranges::empty(slice));
+        this->_slice = std::addressof(slice);
+        return yield_awaiter{this, this->_consumer};
+    }
+
+    void return_void() noexcept {}
+
+    // @return if the generator has reached the end of the sequence
+    bool finished() const noexcept {
+        return _slice == nullptr;
+    }
+
+    // @return the current slice produced by the producer
+    Yielded& slice() noexcept {
+        assert(_slice);
+        return *_slice;
+    }
+
+
+    void rethrow_if_unhandled_exception() {
+        if (_exception) {
+            std::rethrow_exception(std::move(_exception));
+        }
+    }
+
+    void run_and_dispose() noexcept final {
+        using handle_type = std::coroutine_handle<generator_promise_base>;
+        handle_type::from_promise(*this).resume();
+    }
+
+    seastar::task* waiting_task() noexcept final {
+        return _waiting_task;
+    }
+
+private:
+    friend class next_awaiter<Yielded>;
+};
+
+template <typename Yielded>
+class next_awaiter {
+protected:
+    generator_promise_base<Yielded>* _promise = nullptr;
+    std::coroutine_handle<> _producer = nullptr;
+
+public:
+    explicit next_awaiter(std::nullptr_t) noexcept {}
+    next_awaiter(generator_promise_base<Yielded>& promise,
+                 std::coroutine_handle<> producer) noexcept
+        : _promise{std::addressof(promise)}
+        , _producer{producer} {}
+
+    template <typename Promise>
+    std::coroutine_handle<> await_suspend(std::coroutine_handle<Promise> consumer) noexcept {
+        _promise->_consumer = consumer;
+        return _producer;
+    }
+};
+
+template<typename Ref, typename Value, typename Yielded>
+class [[nodiscard]] generator {
+    using value_type = std::conditional_t<std::is_void_v<Value>,
+                                          std::remove_cvref_t<Ref>,
+                                          Value>;
+    using reference_type = std::conditional_t<std::is_void_v<Value>,
+                                              Ref&&,
+                                              Ref>;
+    using yielded_type = Yielded;
+
+public:
+    class promise_type;
+
+private:
+    using handle_type = std::coroutine_handle<promise_type>;
+    handle_type _coro = {};
+
+public:
+    class iterator;
+
+    generator() noexcept = default;
+    explicit generator(promise_type& promise) noexcept
+        : _coro(std::coroutine_handle<promise_type>::from_promise(promise))
+    {}
+    generator(generator&& other) noexcept
+        : _coro{std::exchange(other._coro, {})}
+    {}
+    generator(const generator&) = delete;
+    generator& operator=(const generator&) = delete;
+
+    ~generator() {
+        if (_coro) {
+            _coro.destroy();
+        }
+    }
+
+    friend void swap(generator& lhs, generator& rhs) noexcept {
+        std::swap(lhs._coro, rhs._coro);
+    }
+
+    generator& operator=(generator&& other) noexcept {
+        if (this == &other) {
+            return *this;
+        }
+        if (_coro) {
+            _coro.destroy();
+        }
+        _coro = std::exchange(other._coro, nullptr);
+        return *this;
+    }
+
+    [[nodiscard]] auto begin() noexcept {
+        using base_awaiter = next_awaiter<yielded_type>;
+        class begin_awaiter final : public base_awaiter {
+            using base_awaiter::_promise;
+
+        public:
+            using base_awaiter::base_awaiter;
+            bool await_ready() const noexcept {
+                return _promise == nullptr;
+            }
+
+            iterator await_resume() {
+                if (_promise == nullptr) {
+                    return iterator{nullptr};
+                }
+                if (_promise->finished()) {
+                    _promise->rethrow_if_unhandled_exception();
+                    return iterator{nullptr};
+                }
+                return iterator{
+                    handle_type::from_promise(*static_cast<promise_type *>(_promise))
+                };
+            }
+        };
+
+        if (_coro && !_coro.done()) {
+            return begin_awaiter{_coro.promise(), _coro};
+        } else {
+            return begin_awaiter{nullptr};
+        }
+    }
+
+    [[nodiscard]] std::default_sentinel_t end() const noexcept {
+        return {};
+    }
+};
+
+/// buffered generator has 3 template parameters:
+///
+/// - Ref
+/// - Value
+/// - Yielded
+///
+/// Unlike its unbuffered variant, the \c Yielded type can be customized. \c Yielded should be a
+/// range of elements, which are convertible to the \c Value type.
+///
+/// @note Please note, empty ranges are not allowed to be yielded.
+template <typename Ref, typename Value, typename Yielded>
+class generator<Ref, Value, Yielded>::promise_type final : public generator_promise_base<yielded_type> {
+public:
+    generator get_return_object() noexcept {
+        return generator{*this};
+    }
+};
+
+template <typename Ref, typename Value, typename Yielded>
+class generator<Ref, Value, Yielded>::iterator final {
+private:
+    using handle_type = generator::handle_type;
+    // nullptr on end
+    handle_type _coro = nullptr;
+
+    using yielded_iterator = std::ranges::iterator_t<yielded_type>;
+    static_assert(std::input_iterator<yielded_iterator>);
+    yielded_iterator _iterator;
+
+    void reset() {
+        _iterator = std::begin(_coro.promise().slice());
+    }
+
+public:
+    using iterator_category = std::input_iterator_tag;
+    using difference_type = std::ptrdiff_t;
+    using value_type = generator::value_type;
+    using reference = generator::reference_type;
+    using pointer = std::add_pointer_t<value_type>;
+
+    explicit iterator(handle_type coroutine) noexcept
+        : _coro{coroutine} {
+        if (_coro) {
+            reset();
+        }
+    }
+
+    explicit operator bool() const noexcept {
+        return _coro && !_coro.done();
+    }
+
+    [[nodiscard]] auto operator++() noexcept {
+        using base_awaiter = next_awaiter<yielded_type>;
+        class increment_awaiter final : public base_awaiter {
+            const bool _in_this_slice;
+            iterator& _iterator;
+            using base_awaiter::_promise;
+
+        public:
+            explicit increment_awaiter(bool is_ready, iterator& iterator) noexcept
+                : base_awaiter{iterator._coro.promise(), iterator._coro}
+                , _in_this_slice{is_ready}
+                , _iterator{iterator}
+            {}
+
+            bool await_ready() const noexcept {
+                return _in_this_slice;
+            }
+
+            iterator& await_resume() {
+                if (_promise->finished()) {
+                    // update iterator to end()
+                    _iterator = iterator{nullptr};
+                    _promise->rethrow_if_unhandled_exception();
+                    return _iterator;
+                }
+                if (!_in_this_slice) {
+                    _iterator.reset();
+                }
+                return _iterator;
+            }
+        };
+
+        assert(bool(*this) && "cannot increment end iterator");
+        bool is_ready = ++_iterator != std::ranges::end(_coro.promise().slice());
+        return increment_awaiter{is_ready, *this};
+    }
+
+    reference operator*() const noexcept {
+        return static_cast<reference>(*_iterator);
+    }
+
+    bool operator==(std::default_sentinel_t) const noexcept {
+        return !bool(*this);
+    }
+};
+
+} // namespace buffered
+
+template<typename T, typename U>
+concept element_of =
+  !std::convertible_to<T, U> &&
+  std::convertible_to<std::ranges::range_value_t<U>, T>;
+} // namespace internal
+
+template<typename Ref, typename Value = void, typename Yielded = void>
+using generator = std::conditional_t<internal::element_of<Value, Yielded>,
+    internal::buffered::generator<Ref, Value, Yielded>,
+    internal::unbuffered::generator<Ref, Value>>;
+
 } // namespace seastar::coroutine::experimental
diff --git a/tests/unit/generator_test.cc b/tests/unit/generator_test.cc
index 9b438b06f07..e0fc60b951d 100644
--- a/tests/unit/generator_test.cc
+++ b/tests/unit/generator_test.cc
@@ -26,6 +26,7 @@
 #include <seastar/testing/test_case.hh>
 #include <seastar/util/bool_class.hh>
 #include <seastar/util/later.hh>
+#include <string>
 #include <string_view>
 #if __cplusplus >= 202302L && defined(__cpp_lib_generator)
 #include <generator>
@@ -234,3 +235,150 @@ SEASTAR_TEST_CASE(test_generator_throws_from_consumer) {
     BOOST_REQUIRE_THROW(std::rethrow_exception(f.get_exception()), std::invalid_argument);
     BOOST_REQUIRE_EQUAL(total, 0);
 }
+
+SEASTAR_TEST_CASE(test_batch_generator_empty_sequence) {
+    using int_gen = coroutine::experimental::generator<int, int, std::vector<int>>;
+    auto seq = std::invoke([]() -> int_gen {
+        co_return;
+    });
+    for (auto i = co_await seq.begin(); i != seq.end(); co_await ++i) {
+        BOOST_FAIL("found element in an empty sequence");
+    }
+}
+
+coroutine::experimental::generator<int, int, std::vector<int>>
+async_fibonacci_sequence_batch(unsigned count, unsigned batch_size, do_suspend suspend) {
+    auto a = 0, b = 1;
+    std::vector<int> batch;
+    for (unsigned i = 0; i < count; ++i) {
+        if (std::numeric_limits<decltype(a)>::max() - a < b) {
+            throw std::out_of_range(
+                fmt::format("fibonacci[{}] is greater than the largest value of int", i));
+        }
+        if (suspend) {
+            co_await yield();
+        }
+        int next = std::exchange(a, std::exchange(b, a + b));
+        batch.push_back(next);
+        if (batch.size() == batch_size) {
+            co_yield std::exchange(batch, {});
+        }
+    }
+    if (!batch.empty()) {
+        co_yield std::move(batch);
+    }
+}
+
+seastar::future<>
+verify_fib_drained(coroutine::experimental::generator<int, int, std::vector<int>> actual_fibs, unsigned count) {
+    auto expected_fibs = sync_fibonacci_sequence(count);
+    auto expected_fib = std::begin(expected_fibs);
+
+    auto actual_fib = co_await actual_fibs.begin();
+
+    for (; actual_fib != actual_fibs.end(); co_await ++actual_fib) {
+        BOOST_REQUIRE(expected_fib != std::end(expected_fibs));
+        BOOST_REQUIRE_EQUAL(*actual_fib, *expected_fib);
+        ++expected_fib;
+    }
+    BOOST_REQUIRE(actual_fib == actual_fibs.end());
+}
+
+SEASTAR_TEST_CASE(test_batch_generator_drained_with_suspend) {
+    constexpr unsigned count = 4;
+    constexpr unsigned batch_size = 2;
+    return verify_fib_drained(async_fibonacci_sequence_batch(count, batch_size, do_suspend::yes),
+                              count);
+}
+
+SEASTAR_TEST_CASE(test_batch_generator_drained_without_suspend) {
+    constexpr int count = 4;
+    constexpr int batch_size = 2;
+    return verify_fib_drained(async_fibonacci_sequence_batch(count, batch_size, do_suspend::no),
+                              count);
+}
+
+seastar::future<> test_batch_generator_not_drained(do_suspend suspend) {
+    auto fib = async_fibonacci_sequence_batch(42, 12, suspend);
+    auto actual_fib = co_await fib.begin();
+    BOOST_REQUIRE_EQUAL(*actual_fib, 0);
+}
+
+SEASTAR_TEST_CASE(test_batch_generator_not_drained_with_suspend) {
+    return test_batch_generator_not_drained(do_suspend::yes);
+}
+
+SEASTAR_TEST_CASE(test_batch_generator_not_drained_without_suspend) {
+    return test_batch_generator_not_drained(do_suspend::no);
+}
+
+SEASTAR_TEST_CASE(test_batch_generator_move_away) {
+    struct move_only {
+        int value;
+        move_only(int value)
+            : value{value}
+        {}
+        move_only(const move_only&) = delete;
+        move_only& operator=(const move_only&) = delete;
+        move_only(move_only&&) noexcept = default;
+        move_only& operator=(move_only&&) noexcept = default;
+    };
+
+    using batch_type = std::vector<move_only>;
+    using move_only_gen = coroutine::experimental::generator<move_only&&, move_only, batch_type>;
+
+    constexpr int count = 4;
+    constexpr unsigned batch_size = 2;
+    auto numbers = std::invoke([]() -> move_only_gen {
+        batch_type batch;
+        for (int i = 0; i < count; i++) {
+            batch.push_back(i);
+            if (batch.size() == batch_size) {
+                co_yield std::exchange(batch, {});
+            }
+        }
+        if (!batch.empty()) {
+            co_yield std::move(batch);
+        }
+    });
+
+    int expected_n = 0;
+    for (auto n = co_await numbers.begin(); n != numbers.end(); co_await ++n) {
+        BOOST_REQUIRE_EQUAL((*n).value, expected_n++);
+    }
+}
+
+SEASTAR_TEST_CASE(test_batch_generator_convertible) {
+    struct convertible {
+        const std::string value;
+        convertible(std::string&& value)
+            : value{std::move(value)}
+        {}
+        explicit operator int() const {
+            return std::stoi(value);
+        }
+    };
+
+    using batch_type = std::vector<convertible>;
+    using move_only_gen = coroutine::experimental::generator<int, convertible, batch_type>;
+
+    constexpr int count = 4;
+    constexpr unsigned batch_size = 2;
+    auto numbers = std::invoke([]() -> move_only_gen {
+        batch_type batch;
+        for (int i = 0; i < count; i++) {
+            batch.push_back(fmt::to_string(i));
+            if (batch.size() == batch_size) {
+                co_yield std::exchange(batch, {});
+            }
+        }
+        if (!batch.empty()) {
+            co_yield std::move(batch);
+        }
+    });
+
+    int expected_n = 0;
+    for (auto n = co_await numbers.begin(); n != numbers.end(); co_await ++n) {
+        BOOST_REQUIRE_EQUAL(*n, expected_n++);
+    }
+}