coroutine/generator: reimplement generator

[tchaikov]

this generator implementation is inspired by https://wg21.link/P2502R2. instead of keeping an internal queue, it reuses the yielded value of the producer. in this change:

• replace the existing seastar::experimental::coroutine::generator with this implementation
• drop the generator tests in coroutines_test.cc. and keep generator_test.cc which is dedicated to generator.

before this change, we were using a bounded buffer for holding the yielded elements. this proved to be error-prone, that's why we had "_wait_for_free_space assertion failure" in the directory_test. after this change, instead of using an internal buffer, we just hold a pointer referencing the yielded range or element, this saves us the pain of tracking the free space in the bounded buffer. hence address all of the three issues.

Fixes #2190
Fixes #1913
Fixes #1677

[bhalevy]

nit && sticks to type

[bhalevy]

Also, better do the below inside if (this != &other) to protect against self-assignment;

[tchaikov]

to check self-assignment in move ctor was a controversial topic. but i think it has been clarified by https://cplusplus.github.io/LWG/issue2468 and formalized in the standard, see https://eel.is/c++draft/utility.arg.requirements#tab:cpp17.moveassignable .

If t and rv do not refer to the same object, t is equivalent to the value of rv before the assignment

so, if they refer to the same object, there are no requirements on the value of the lhs of the assignment operator after the assignment expression is evaluated. hence there is no need to perform the self-assignment here.

[bhalevy]

@tchaikov we have such protections against self move all over the place.
The conservative approach is to make such a move be a no-op.
This way the lvalue contains the value of itself before the move, and it remains in a valid state (though unspecified).
Clearing self contents does not satisfy the first requirement of holding the state of the rvalue prior to move.

[bhalevy]

@avikivity do we have a documented convention regarding self-move assignment semantics?
I see that future<> just guarantees validity but not the state after self-move is undetermined.
Maybe for control classes this is good enough, and that may be applicable in this case.
But for containers, self-move should be a no-op to prevent unpleasant surprises IMO (even though the stl isn't consistent about it).

[tchaikov]

@tchaikov we have such protections against self move all over the place. The conservative approach is to make such a move be a no-op. This way the lvalue contains the value of itself before the move, and it remains in a valid state (though unspecified).

yeah, i understand the concern.

Clearing self contents does not satisfy the first requirement of holding the state of the rvalue prior to move.

the postcondition has a requirement of "If t and rv do not refer to the same object, t is equivalent to the value of rv before the assignment", hence i am not sure we need this check in the case of move assignment operator .

[tchaikov]

i changed it anyway. but personally, i think since the standard allows the undefined behavior in this case, we don't need to be conservative here.

[bhalevy]

Let's get @avikivity's opinion on this too.
I can see cases where the generator needs to be moved around

[tchaikov]

I can see cases where the generator needs to be moved around

i agree. but IMHO, moving a variable into itself is not quite a typical use case.

[tchaikov]

v2:

• document the template parameter usage
• let & and && stick to the type
• check for self-assignment in the move assignment operator
• always yield in the test where suspend is true

[tchaikov]

v3:

• replaced .then_wrapped() with coroutine::as_future()
• refactor generator tests to use std::generator when available
• added test cases to verify the move constructor and std::swap()
• do not include coroutine/maybe_yield.hh. should have removed the #include in the previous revision.

[tchaikov]

v4:

• add test to exercise the case where both template parameters are used
• add test to exercise the case where the yielded value is a rvalue reference
• rename element_awaiter to copy_awaiter to highlight that it performs a copy of the yielded value. the idea is stolen from libstdc++'s generator header.
• remove the generator tests in coroutine_test.cc
• replace generator with async_generator
• update experimental_list_directory() to use const directory_entry& as the Ref template parameter.

[tchaikov]

@scylladb/seastar-maint hello maintainers, could you help review this change?

[tchaikov]

@scylladb/seastar-maint hello maintainers, could you help review this change?

[avikivity]

replace the existing seastar::coroutine::experimental with this implementation

What's the motivation?

Shouldn't we fix the ping-pong problem? IIRC we had some internal queuing to avoid ping-pongs.

[avikivity]

Should have a constraint (type suitable to be carried in a future).

[tchaikov]

in this implementation, the return value is passed to the consumer via a pointer -- the producer assigns the value to the pointer, so seastar::future<> is not involved. and there are two cases:

• producer produces a lvalue: so we can just reuse it, and memorize its address, and return it to the consumer
• producer produces a rvalue: so we have to create a copy of it, and return the copy to the consumer

[avikivity]

Here we have symmetric transfer, which is nice, the consumer is awakened without going through the scheduler. However, we must make at least one of the two directions has a preemption check.

I think it's okay now because await_ready always returns false, but I also think await_ready shouldn't return false.

If we are able to do symmetric transfer in both directions, then the ping-pong problem is much reduced: the two coroutines exchange control until one of them detects that preemption is needed or has to block itself.

[tchaikov]

However, we must make at least one of the two directions has a preemption check.

i will take a closer look to see how to season the awaiter gently.

[tchaikov]

should be fixed in the latest revision, which yields on the producer's end.

[avikivity]

Aha, so this is how we can be sure the consumer is a coroutine.

I think we should relax this, though my brain will probably explode.

[tchaikov]

to accommodate both asynchronous and synchronous producers, we could specialize the generator based on the Promise type passed to std::coroutine_handle. However, this approach would introduce additional complexity to the implementation. i recommend utilizing the standard library's generator implementation,
which offers a more comprehensive set of features.

[avikivity]

You mean synchronous and asynchronous consumers, no? The producer is always a coroutine.

[tchaikov]

You mean synchronous and asynchronous consumers, no?

ahh, sorry. i was thinking about the async generator versus sync generator. but seastar::coroutine::experimental::generator implementation always assumes the promise is a seastar task. To obtain elements from an instance of this generator, one must use co_await. consequently, it's challenging to envision a consumer of this generator that isn't itself a coroutine.

The producer is always a coroutine.

indeed. i just realized that it's more about async versus sync not about regular function versus coroutine.

[avikivity]

Maybe I should pause the internals review and look at the interface. Why an iterator range interface? It can't be be used in a range for loop. How do you determine the generator is exhausted? Here you know it generates 2*n, but in general, you don't.

I think we should go back to std::optional<T>

[avikivity]

With such an interface, you could implement both future<> operator(), and operator co_await directly on the generator object itself, using symmetric tranfser.

[tchaikov]

Maybe I should pause the internals review and look at the interface. Why an iterator range interface?

to maintain consistency with the interface proposed in https://wg21.link/P2502R2, i opted for a design that closely aligns with the standard. the proposal introduces std::generator, a standard library type implementing a coroutine generator that forms to the std::ranges::input_range model. by adhering to this proposed interface, we ensure that the developers familiar with the standard will feel at home when using the asynchronous generator. hopefully, this approach makes the asynchronous generator more intuitive and easier to use.

It can't be be used in a range for loop.

it can't due to the inherent difference between async generator and synchronous generator. and unfortunately, Remove for co_await statement was accepted. so we cannot use a ranged-for loop with async generator. but as explained above. probably we can preserve this interface? even though the async variant cannot allow the caller to use a nice "for" loop for enumerating the elements produced by the generator.

How do you determine the generator is exhausted? Here you know it generates 2*n, but in general, you don't.

we do have tests exercising the "you don't" cases, for example, see test_generator_drained_with_suspend. i am quoting part of it as a reference

seastar::future<>
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);

    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());
}

I think we should go back to std::optional

it is a step back. the gist of the rewrite is to reuse the reference design proposed by https://wg21.link/P2502R2. so that:

• we don't need to track the emptiness or the free slots in the internal buffer. this is probably most important.
• we can save the overhead of calling move constructor or conversion.
• we can expose a similar interface to the one proposed by P2502R2

returning std::optinoal<T> defeats the purpose of the efforts of rewriting. especially it does not work with the first item in the list above.

[avikivity]

But what does co_await ++it mean? Shouldn't it be co_await ++*it? or *co_await ++it?

Using iterators doesn't help, because there's no async for loop, and all the range algorithms don't work on such iterators.

How can you compare the iterator to the end iterator, before you co_await? How is the empty sequence represented?

[tchaikov]

But what does co_await ++it mean?

move to (wait for) the next element.

Shouldn't it be co_await ++*it? or *co_await ++it?

i think *co_await ++it works.

Using iterators doesn't help, because there's no async for loop, and all the range algorithms don't work on such iterators.

hmm, iterator helps with for loop. but i think it's also a way to represent a range. imaging we are walking from home to a bus stop, the start point is begin(journey), and the bus stop is represented by end(journey).

How can you compare the iterator to the end iterator, before you co_await? How is the empty sequence represented?

please see the code above, i am replicating it here:

for (; actual_fib != actual_fibs.end(); co_await ++actual_fib) {
   // ..
}

if we hit the sentinel in the first iteration, then it is an empty sequence.

[avikivity]

I see it's inspired by https://github.com/lewissbaker/cppcoro#async_generatort. Does it have users? We can learn from its experience.

[tchaikov]

It seems to work. But it's just odd. The begin/end methods don't return iterators. You don't compare begin() to end(), you compare co_await begin() to end(). It looks like a range but it isn't a range.

The interface is also odd. You await begin() and ++i, but not anything else. You can't do regular iterator stuff like *++i.

yes, it's a known limit. i put this in the comment in the very beginning of the header file. see 5ef1a6e#diff-b7dc64e5af0d4dc90acdd0c4f909a940dbb2c74e88d65e503508ec1eb54253c5R48 . you are practically repeating it =( . i am quoting the comment here

// * 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.

On the other hand, there's an advantage to being able to pass a reference, which isn't possible with std::optional. (there will be optional<T&> in C++26).

you mean returning a reference from the producer or returning a reference to the consumer?

[tchaikov]

I see it's inspired by https://github.com/lewissbaker/cppcoro#async_generatort. Does it have users? We can learn from its experience.

yes, this implementation is also inspired by the cppcoro's async_generator<T> and folly's generator implementation. i searched around with the keyword of <cppcoro/async_generator.hpp> . but couldn't find any users of it. i think, C++20 coroutines are a relatively new feature in the C++ world. generators, which are based on coroutines, are still gaining traction among developers. as a result, the number of programmers actively using generators remains limited. asynchronous generators, being an even more advanced concept, are even less commonly utilized in applications with source code indexed by search engines.

[bhalevy]

@avikivity so are you okay with using an async iterator like: for (auto it = co_await gen.begin(); it != gen.end(); co_await ++it) { auto val = *it; }?

[niekbouman]

Qt seems to use the same interface as proposed here for its AsyncGenerator<T> type (which is a good thing IMHO):

https://qcoro.dev/reference/coro/asyncgenerator/

[tchaikov]

v9

• removed the unused do_yield() helper (it was removed in a succeeding commit, but it existed in the first commit)

[tchaikov]

v10:

• instead of transferring to the consumer immediately, schedule the consumer continuation if seastar::need_preempt(). this should avoid the reactor stall if we are handling a large batch of elements.

[tchaikov]

@avikivity could you take another look at your convenience?

[tchaikov]

@avikivity zealous ping.

[niekbouman]

@tchaikov Regarding the naming of your async-generator class: at various places in this PR, I see:

replace generator with async_generator

treewide: replace generator with async_generator

but when I look at the current implementation (under the "Files changed" tab), then I see that the class is named generator. What's the rationale for calling it generator and not async_generator ?

(btw thanks for your hard work on this!!)

[tchaikov]

@tchaikov Regarding the naming of your async-generator class: at various places in this PR, I see:

replace generator with async_generator

treewide: replace generator with async_generator

but when I look at the current implementation (under the "Files changed" tab), then I see that the class is named generator. What's the rationale for calling it generator and not async_generator ?

please see #2218 (comment) , the idea is to reimplement the existing generator, as the title of this PR suggests. also, because Seastar is inherently async, adding an "async" prefix is unnecessary and would be redundant. In the context of Seastar, it's understood that operations are generally asynchronous.

[niekbouman]

@tchaikov Regarding the naming of your async-generator class: at various places in this PR, I see:

replace generator with async_generator

treewide: replace generator with async_generator

but when I look at the current implementation (under the "Files changed" tab), then I see that the class is named generator. What's the rationale for calling it generator and not async_generator ?

please see #2218 (comment) , the idea is to reimplement the existing generator, as the title of this PR suggests. also, because the Seastar is inherently async, adding an "async" prefix is unnecessary and would be redundant. In the context of Seastar, it's understood that operations are generally asynchronous.

I see, thanks for pointing me to that comment, I had missed it. (And the rationale makes sense indeed.)

[tchaikov]

@avikivity gentle ping.

[tchaikov]

@avikivity yet another gentle ping.

[tchaikov]

@avikivity yet another ping.

[tchaikov]

@avikivity yet another gentle ping.

[niekbouman]

@tchaikov Dear Kefu,

I have another question. With your current generator , I can imagine how I could make an async file reader, that, assuming the (binary) file contains ints, yields those one-by-one, but still efficiently reads from disk using seastar's file api.
If I understand correctly, using a generator will be much more efficient compared to the following code (a wrapper type around input_stream<char>):

class int_input_stream {
    seastar::input_stream<char> is_;

  public:
    future<int> read()
    {
        int x{};
        auto buf = co_await is_.read_exactly(sizeof(int));
        memcpy(&x, buf.get(), sizeof(int));
        co_return x;
    }
};

which passes through seastar's reactor for every time I call co_await read() to read a single integer.

Now imagine I want to make a file writer which enjoys the same kind of "ping-pong" efficiency improvement: I can pass single integers into the generator, which, inside the generator fills some buffer which will then flush once in a while (essentially a more efficient version of a wrapper like the above around an output_stream<char>, featuring a future<> write(int) function). Would this be possible? I.e., is it possible according to the c++ standard, like in Python, to send values into a generator?
https://docs.python.org/3/reference/expressions.html#agen.asend

[tchaikov]

@tchaikov Dear Kefu,

I have another question. With your current generator , I can imagine how I could make an async file reader, that, assuming the (binary) file contains ints, yields those one-by-one, but still efficiently reads from disk using seastar's file api. If I understand correctly, using a generator will be much more efficient compared to the following code (a wrapper type around input_stream<char>):

class int_input_stream {
    seastar::input_stream<char> is_;

  public:
    future<int> read()
    {
        int x{};
        auto buf = co_await is_.read_exactly(sizeof(int));
        memcpy(&x, buf.get(), sizeof(int));
        co_return x;
    }
};

which passes through seastar's reactor for every time I call co_await read() to read a single integer.

yes. it suffers in multiple ways, every time we produces and consumes a number we do this dance:

1. spawn a task for consuming it, and this task is scheduled in the reactor, waiting for the producer
2. spawn a task for producing it, and this task is scheduled in the reactor, waiting for the underlying disk i/o
3. resume the producer
4. resume the consumer

Now imagine I want to make a file writer which enjoys the same kind of "ping-pong" efficiency improvement: I can pass single integers into the generator, which, inside the generator fills some buffer which will then flush once in a while (essentially a more efficient version of a wrapper like the above around an output_stream<char>, featuring a future<> write(int) function). Would this be possible? I.e., is it possible according to the c++ standard, like in Python, to send values into a generator? https://docs.python.org/3/reference/expressions.html#agen.asend

i think it is possible in theory. as generator is a customized type, and we are able to resume the consumer's coroutine once one sends something to the generator. IIUC, asend() feeds a generator, and it yields the "sent" value.

but my question is why/how we would use it for implementing a reader in a more natural way. what i can think of is that it can be used to implement a pipeline, so that the producer can "send" values on one end of the pipeline explicitly instead of using yield , and the consumer can consume it as a regular iterator using next() at the other end.

[niekbouman]

Now imagine I want to make a file writer [...]

but my question is why/how we would use it for implementing a reader in a more natural way. what i can think of is that it can be used to implement a pipeline, so that the producer can "send" values on one end of the pipeline explicitly instead of using yield , and the consumer can consume it as a regular iterator using next() at the other end.

My question was indeed about implementing a writer, in other words something that sits at the "sink" part of a pipeline.

(A file reader can already be implemented using your current implementation)

[tchaikov]

sorry, i misread your question. so instead of consuming the element generated from the generator, you want to feed the generator with well, something else (for instance another generator =)). i don't really think this is what a generator is designed for. i think it allows us to generate a series of elements not to consume elements. what do you think?

[niekbouman]

sorry, i misread your question. so instead of consuming the element generated from the generator, you want to feed the generator with well, something else (for instance another generator =)). i don't really think this is what a generator is designed for. i think it allows us to generate a series of elements not to consume elements. what do you think?

Well, for sure the term "generator" becomes misleading when we want to build a data sink.

But let's regard generator as a "coroutine that can produce several values", with which we could for example build a more efficient input_stream.
Then it would be logical to say that there should exist a "coroutine that can consume several values", with which we could build a more efficient output_stream.

[tchaikov]

I see. this is indeed an interesting idea. and as you put, the points is the symmetric transferring between the producer and consumer, and producing the data in batch, so we can alleviate the pain of the ping-pong problem. i think the beauty of generator is that one is able to use yield to "return" values multiple times in a single execution of coroutine.

if we have to mirror the paradigm of passing a value to a certain function we only a single option: input parameter. but we cannot re-evaluate a function without executing its body multiple times. however, we can "yield" / "return" values multiple times.

i have no idea how to make an output_stream in coroutine consuming values multiple times in a way similar to generator.

[niekbouman]

i have no idea how to make an output_stream in coroutine consuming values multiple times in a way similar to generator.

Wouldn't it be simply possible to make it legal (only in cases where applicable) to assign to the iterator?

for (auto it = co_await gen.begin(); it != gen.end(); co_await ++it) { *it = my_value_to_be_consumed; }

[niekbouman]

i have no idea how to make an output_stream in coroutine consuming values multiple times in a way similar to generator.

hi Kefu @tchaikov
I found a Stackoverflow post where they suggest two ways to send values into a C++20-style generator:
https://stackoverflow.com/questions/64083104/making-python-generator-via-c20-coroutines

I've tried those approaches on your PR; the await_transform route was not successful, as it interferes with other uses of the co_await operator (in unrelated parts in the seastar codebase). But the technique in the "Be more yielding" section does work with your generator type! See the raw proof-of-principle (last commit) in my branch, which is based on your async-generator branch:

https://github.com/niekbouman/seastar/tree/niek-async-generator-send