pitrou commented on a change in pull request #9533:
URL: https://github.com/apache/arrow/pull/9533#discussion_r592424432
##########
File path: cpp/src/arrow/util/async_generator.h
##########
@@ -177,6 +184,99 @@ class TransformingGenerator {
std::shared_ptr<TransformingGeneratorState> state_;
};
+template <typename T>
+class SerialReadaheadGenerator {
+ public:
+ SerialReadaheadGenerator(AsyncGenerator<T> source_generator, int
max_readahead)
+ : state_(std::make_shared<State>(std::move(source_generator),
max_readahead)) {}
+
+ Future<T> operator()() {
+ if (state_->first_) {
+ // Lazy generator, need to wait for the first ask to prime the pump
+ state_->first_ = false;
+ auto next = state_->source_();
+ return next.Then(Callback{state_});
+ }
+
+ // This generator is not async-reentrant. We won't be called until the
last
+ // future finished so we know there is something in the queue
+ auto finished = state_->finished_.load();
+ if (finished && state_->readahead_queue_.IsEmpty()) {
+ return AsyncGeneratorEnd<T>();
+ }
+
+ auto next_ptr = state_->readahead_queue_.FrontPtr();
+ auto next = std::move(**next_ptr);
+ state_->readahead_queue_.PopFront();
Review comment:
Wouldn't it better to call `readahead_queue_.Read()` here? It would do
those three things at once AFAICT.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
