absl::BlockingCounter·Barrier — 다중 thread 조율

1
#include "absl/synchronization/blocking_counter.h"
2

3
void DoWork(absl::BlockingCounter* counter, int i) {
4
  // ...
5
  counter->DecrementCount();
6
}
7

8
void Run() {
9
  constexpr int kN = 10;
10
  absl::BlockingCounter counter(kN);
11

12
  for (int i = 0; i < kN; ++i) {
13
    std::thread([&counter, i] { DoWork(&counter, i); }).detach();
14
  }
15

16
  counter.Wait();   // 모든 worker가 DecrementCount할 때까지
17
}

1
namespace absl {
2
class BlockingCounter {
3
 public:
4
  explicit BlockingCounter(int initial_count);
5
  bool DecrementCount();   // counter==0이 되면 true 반환
6
  void Wait();
7
};
8

9
class Barrier {
10
 public:
11
  explicit Barrier(int num_threads);
12
  bool Block();   // 마지막 도착 thread가 true, 나머지는 false
13
};
14
}

1
#include "absl/synchronization/barrier.h"
2

3
void Worker(absl::Barrier* barrier, int id) {
4
  Prepare(id);
5
  if (barrier->Block()) {
6
    // 마지막 도착자 — cleanup 책임
7
    LOG(INFO) << "all workers arrived";
8
  }
9
  // 여기서부터 모두 동시에 진행
10
  Process(id);
11
}
12

13
void Run() {
14
  constexpr int kN = 8;
15
  absl::Barrier b(kN);
16
  std::vector<std::thread> threads;
17
  for (int i = 0; i < kN; ++i) {
18
    threads.emplace_back([&b, i] { Worker(&b, i); });
19
  }
20
  for (auto& t : threads) t.join();
21
}

1
// absl/synchronization/blocking_counter.cc (요약)
2
class BlockingCounter {
3
  absl::Mutex lock_;
4
  int count_;
5
  int num_waiting_;
6
  bool done_;
7
};
8

9
void BlockingCounter::Wait() {
10
  absl::MutexLock l(&lock_);
11
  lock_.Await(absl::Condition(&done_));
12
}
13

14
bool BlockingCounter::DecrementCount() {
15
  absl::MutexLock l(&lock_);
16
  if (--count_ <= 0) done_ = true;
17
  return done_;
18
}

1
// 회피 — busy wait
2
std::atomic<int> remaining{N};
3
for (int i = 0; i < N; ++i) {
4
  std::thread([&] { Work(); remaining.fetch_sub(1); }).detach();
5
}
6
while (remaining.load() > 0) std::this_thread::yield();
7

8
// Good
9
absl::BlockingCounter counter(N);
10
for (int i = 0; i < N; ++i) {
11
  std::thread([&counter] { Work(); counter.DecrementCount(); }).detach();
12
}
13
counter.Wait();

1
TEST(Race, RareCondition) {
2
  constexpr int kN = 16;
3
  absl::Barrier start(kN);
4
  std::vector<std::thread> threads;
5
  for (int i = 0; i < kN; ++i) {
6
    threads.emplace_back([&] {
7
      start.Block();   // 모두 도착할 때까지 대기 — 정확히 같은 시점에 진행
8
      DoRaceyOp();
9
    });
10
  }
11
  for (auto& t : threads) t.join();
12
}

1
std::vector<Result> results(N);
2
absl::BlockingCounter c(N);
3
for (int i = 0; i < N; ++i) {
4
  pool->Schedule([&, i] {
5
    results[i] = ComputeShard(i);
6
    c.DecrementCount();
7
  });
8
}
9
c.Wait();
10
Result merged = Merge(results);

1
// 회피 — Barrier는 1회용
2
absl::Barrier b(N);
3
for (round : rounds) {
4
  // ...
5
  b.Block();   // 첫 라운드 OK, 두 번째부터 UB
6
}

1
// 회피 — worker가 예외로 die하면 counter는 N에서 멈춤 → Wait 영원
2
std::thread([&] { Work(); counter.DecrementCount(); }).detach();

1
std::thread([&] {
2
  absl::Cleanup dec = [&counter] { counter.DecrementCount(); };
3
  Work();   // throw 해도 dec가 호출됨
4
}).detach();

1
absl::BlockingCounter c(0);
2
c.Wait();   // 즉시 반환 — 의미 없음