folly::collect·collectAll·collectAny — fan-in 패턴 분석

1
#include <folly/futures/Future.h>
2

3
// (1) collect — 하나라도 실패하면 전체 실패
4
folly::SemiFuture<std::tuple<int, std::string>> sf1 =
5
    folly::collect(makeSemiFuture(1), makeSemiFuture<std::string>("ok"));
6

7
folly::SemiFuture<std::vector<int>> sf2 =
8
    folly::collect(vector_of_semifutures);
9

10
// (2) collectAll — 모두 완료 (Try로 감쌈)
11
folly::SemiFuture<std::tuple<Try<int>, Try<std::string>>> sf3 =
12
    folly::collectAll(makeSemiFuture(1), makeSemiFuture<std::string>("ok"));
13

14
folly::SemiFuture<std::vector<Try<int>>> sf4 =
15
    folly::collectAll(vector_of_semifutures);
16

17
// (3) collectAny — 가장 먼저 완료 (pair: index + Try)
18
folly::SemiFuture<std::pair<size_t, Try<int>>> sf5 =
19
    folly::collectAny(vector_of_semifutures);
20

21
// (4) collectN — n개 완료
22
folly::SemiFuture<std::vector<std::pair<size_t, Try<int>>>> sf6 =
23
    folly::collectN(vector_of_semifutures, 3);

1
auto futures = std::vector{
2
  fetchUser(1),    // SemiFuture<User>
3
  fetchUser(2),
4
  fetchUser(3),
5
};
6

7
folly::collect(std::move(futures))
8
  .via(&pool)
9
  .thenValue([](std::vector<User> users) {
10
    return aggregate(users);
11
  })
12
  .thenError(folly::tag_t<std::exception>{}, [](auto& e) {
13
    LOG(ERROR) << "at least one fetch failed: " << e.what();
14
    return Aggregate{};
15
  });

1
// folly/futures/Future-inl.h (개념)
2
template <class It>
3
SemiFuture<std::vector<value_t>> collect(It first, It last) {
4
  size_t n = std::distance(first, last);
5
  auto ctx = std::make_shared<CollectContext>(n);
6
  for (size_t i = 0; first != last; ++first, ++i) {
7
    std::move(*first).setCallback_([ctx, i](Try<T>&& t) {
8
      if (t.hasException()) {
9
        ctx->setPartialResult(t.exception());
10
      } else {
11
        ctx->setPartialResult(i, *std::move(t));
12
      }
13
    });
14
  }
15
  return ctx->promise.getSemiFuture();
16
}

1
folly::collectAll(std::move(futures))
2
  .via(&pool)
3
  .thenValue([](std::vector<Try<User>> results) {
4
    std::vector<User> ok;
5
    size_t failed = 0;
6
    for (auto& r : results) {
7
      if (r.hasValue()) ok.push_back(*std::move(r));
8
      else ++failed;
9
    }
10
    LOG(INFO) << "ok=" << ok.size() << " failed=" << failed;
11
    return aggregate(ok);
12
  });

1
auto futures = std::vector{
2
  queryReplicaA(),
3
  queryReplicaB(),
4
  queryReplicaC(),
5
};
6

7
folly::collectAny(std::move(futures))
8
  .via(&pool)
9
  .thenValue([](auto pair) {
10
    auto [idx, result] = std::move(pair);
11
    LOG(INFO) << "first replica: " << idx;
12
    if (result.hasException()) {
13
      // 가장 빠른 응답이 실패 — 다음 단계는 collectAnyWithoutException 고려
14
    }
15
    return *std::move(result);
16
  });

1
folly::collectAnyWithoutException(std::move(futures))
2
  .via(&pool)
3
  .thenValue([](auto pair) {
4
    auto [idx, value] = std::move(pair);   // 성공값
5
    return value;
6
  })
7
  .thenError([](folly::exception_wrapper&&) {
8
    return Default{};   // 모두 실패 — 마지막 예외
9
  });

1
// variadic — 타입이 다른 Future
2
auto sf = folly::collect(
3
    fetchUser(1),       // SemiFuture<User>
4
    fetchOrders(1),     // SemiFuture<vector<Order>>
5
    fetchAddress(1));   // SemiFuture<Address>
6
// sf: SemiFuture<tuple<User, vector<Order>, Address>>
7

8
// range — 같은 타입의 Future
9
std::vector<SemiFuture<User>> v = ...;
10
auto sf = folly::collect(std::move(v));
11
// sf: SemiFuture<vector<User>>

1
auto sf = folly::collectN(std::move(futures), 3);
2
// sf: SemiFuture<vector<pair<size_t, Try<T>>>>
3

4
// 가장 빠른 3개를 받는다

1
// std로 동등 — sequential get
2
auto f1 = std::async(...);
3
auto f2 = std::async(...);
4
auto f3 = std::async(...);
5
auto r1 = f1.get();   // f2/f3는 끝나도 못 받음
6
auto r2 = f2.get();
7
auto r3 = f3.get();
8
// 첫 .get()이 blocking, 그 동안 다른 결과는 idle

1
// 1. collect로 부분 실패 무시
2
folly::collect(futures).get();
3
// 하나만 실패해도 전체 throw — 의도가 partial이면 collectAll
4

5
// 2. collectAll 후 첫 예외만 체크
6
auto results = folly::collectAll(futures).get();
7
for (auto& r : results) {
8
  if (r.hasException()) throw r.exception();   // 첫 예외만 본다
9
}
10
// 의도가 all-or-nothing이면 처음부터 collect 사용
11

12
// 3. collectAny가 실패해도 진행
13
folly::collectAny(futures).thenValue([](auto pair) {
14
  return process(*pair.second);   // pair.second가 예외면 throw
15
});
16

17
// 4. 무한히 큰 vector를 collect
18
folly::collect(million_futures);   // memory 부담 — window로 분할