folly::coro::AsyncGenerator — 비동기 스트림

1
folly::coro::AsyncGenerator<int> Range(int n) {
2
  for (int i = 0; i < n; ++i) {
3
    co_await folly::coro::sleep(std::chrono::milliseconds(10));
4
    co_yield i;
5
  }
6
}
7

8
folly::coro::Task<int> Sum(int n) {
9
  int total = 0;
10
  auto gen = Range(n);
11
  while (auto v = co_await gen.next()) {
12
    total += *v;
13
  }
14
  co_return total;
15
}

1
#include <folly/coro/AsyncGenerator.h>
2

3
folly::coro::AsyncGenerator<int> Source();
4

5
folly::coro::Task<void> Consume() {
6
  auto gen = Source();
7

8
  // 방법 1: next() — Optional 반환
9
  while (auto opt = co_await gen.next()) {
10
    process(*opt);
11
  }
12

13
  // 방법 2: for co_await (C++23 range-based)
14
  // CO_FOREACH macro, 또는 직접 풀어쓴 loop
15
}

1
// Value generator — co_yield가 복사/이동
2
folly::coro::AsyncGenerator<std::string> Lines(std::istream& in) {
3
  std::string line;
4
  while (std::getline(in, line)) {
5
    co_yield line;
6
  }
7
}
8

9
// Reference generator — co_yield가 참조 전달, 큰 객체 효율
10
folly::coro::AsyncGenerator<const LargeMessage&> Messages(Stream& s) {
11
  for (;;) {
12
    auto msg = co_await s.next();
13
    if (!msg) break;
14
    co_yield *msg;
15
  }
16
}

1
folly::coro::Task<void> Print(folly::coro::AsyncGenerator<int> gen) {
2
  while (auto v = co_await gen.next()) {
3
    std::cout << *v << "\n";
4
  }
5
}

1
folly::coro::Task<void> Print(folly::coro::AsyncGenerator<int> gen) {
2
  CO_FOREACH (int v, gen) {        // folly macro
3
    std::cout << v << "\n";
4
  }
5
}

1
// folly/coro/AsyncGenerator.h 약식
2
template <class Ref, class Value = std::remove_cvref_t<Ref>>
3
class AsyncGenerator {
4
 public:
5
  class promise_type {
6
   public:
7
    AsyncGenerator get_return_object() noexcept;
8
    std::suspend_always initial_suspend() noexcept;       // lazy
9
    auto                final_suspend() noexcept;         // notify consumer
10
    auto                yield_value(Ref v) noexcept;      // co_yield
11
    void                return_void() noexcept;
12
    void                unhandled_exception() noexcept;
13

14
   private:
15
    Ref* current_ = nullptr;
16
    std::coroutine_handle<> consumer_;
17
    folly::exception_wrapper exception_;
18
  };
19

20
  class NextAwaitable {
21
    auto await_suspend(std::coroutine_handle<> h) noexcept;
22
    auto await_resume();
23
  };
24

25
  NextAwaitable next() noexcept;
26
};

1
template <class T, class F>
2
folly::coro::AsyncGenerator<std::invoke_result_t<F, T>>
3
transform(folly::coro::AsyncGenerator<T> src, F f) {
4
  while (auto v = co_await src.next()) {
5
    co_yield f(std::move(*v));
6
  }
7
}
8

9
folly::coro::Task<void> Pipeline() {
10
  auto evens = transform(Range(10), [](int x) { return x * 2; });
11
  while (auto v = co_await evens.next()) {
12
    std::cout << *v << "\n";
13
  }
14
}

1
folly::coro::Task<void> ConsumeWithTimeout(
2
    folly::coro::AsyncGenerator<int> gen) {
3
  auto src = folly::CancellationSource{};
4
  auto token = src.getToken();
5

6
  auto consumeTask = [&]() -> folly::coro::Task<void> {
7
    while (auto v = co_await gen.next()) {
8
      process(*v);
9
    }
10
  }();
11

12
  // 5초 뒤 cancel
13
  std::thread([s = src]() mutable {
14
    std::this_thread::sleep_for(std::chrono::seconds(5));
15
    s.requestCancellation();
16
  }).detach();
17

18
  co_await folly::coro::co_withCancellation(token, std::move(consumeTask));
19
}

1
// 1. reference yield 결과를 collection에 보관
2
std::vector<const LargeMessage*> all;
3
while (auto v = co_await gen.next()) {
4
  all.push_back(&*v);   // 다음 iter에서 dangling
5
}
6

7
// 2. generator를 다중 consumer에게 공유
8
auto gen = Source();
9
std::thread t1([&] { folly::coro::blockingWait(consume(gen)); });
10
std::thread t2([&] { folly::coro::blockingWait(consume(gen)); });  // race
11

12
// 3. eager 평가 시도
13
auto all = collectAll(gen);   // AsyncGenerator는 collect 직접 안 됨