absl::Mutex Conditional Critical Section — Await로 cv 없애기

1
std::mutex mu;
2
std::condition_variable cv;
3
std::deque<Task> q;
4

5
// producer
6
void Push(Task t) {
7
  {
8
    std::lock_guard<std::mutex> l(mu);
9
    q.push_back(std::move(t));
10
  }
11
  cv.notify_one();   // ① 누락하면 consumer 영원히 대기
12
}
13

14
// consumer
15
Task Pop() {
16
  std::unique_lock<std::mutex> l(mu);
17
  cv.wait(l, [&] { return !q.empty(); });   // ② predicate while 의무
18
  Task t = std::move(q.front()); q.pop_front();
19
  return t;
20
}

1
// API
2
class Mutex {
3
 public:
4
  void Await(const Condition& cond);
5
  bool AwaitWithTimeout(const Condition& cond, absl::Duration timeout);
6
  bool AwaitWithDeadline(const Condition& cond, absl::Time deadline);
7

8
  bool LockWhen(const Condition& cond);
9
  bool LockWhenWithTimeout(const Condition& cond, absl::Duration timeout);
10
};

1
absl::Mutex mu;
2
std::deque<Task> q;
3

4
void Push(Task t) {
5
  absl::MutexLock l(&mu);
6
  q.push_back(std::move(t));
7
  // notify 없음
8
}
9

10
Task Pop() {
11
  absl::MutexLock l(&mu);
12
  mu.Await(absl::Condition(+[](std::deque<Task>* q) {
13
    return !q->empty();
14
  }, &q));
15
  Task t = std::move(q.front()); q.pop_front();
16
  return t;
17
}

1
// 1) 멤버 함수 + this
2
class Server {
3
  bool IsReady() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mu_);
4
  void Wait() {
5
    absl::MutexLock l(&mu_);
6
    mu_.Await(absl::Condition(this, &Server::IsReady));
7
  }
8
};
9

10
// 2) bool* 직접
11
bool flag = false;
12
mu.Await(absl::Condition(&flag));   // *flag == true가 될 때까지
13

14
// 3) 함수 포인터 + 인자
15
mu.Await(absl::Condition(+[](Queue* q) { return !q->empty(); }, &queue));

1
mu.LockWhen(absl::Condition(&ready));
2
// ready가 true가 된 시점에 lock 보유
3
DoWork();
4
mu.Unlock();

1
absl::MutexLock l(&mu, absl::Condition(&ready));
2
DoWork();
3
// 자동 unlock

1
absl::MutexLock l(&mu);
2
if (mu.AwaitWithTimeout(cond, absl::Seconds(5))) {
3
  // 조건 만족
4
} else {
5
  // timeout
6
}

1
// 의사 코드
2
void Mutex::Unlock() {
3
  while (!wait_queue.empty()) {
4
    Waiter* w = wait_queue.front();
5
    if (w->cond.Eval()) {        // cond 평가 (mu lock 보유)
6
      w->Wake();
7
      wait_queue.pop_front();
8
    }
9
    break;   // 다음 waiter는 다음 unlock 때
10
  }
11
  ReleaseLock();
12
}

1
// 회피
2
{
3
  std::lock_guard l(mu);
4
  ready = true;
5
}
6
cv.notify_one();
7
// ...
8
std::unique_lock l(mu);
9
cv.wait(l, [&] { return ready; });

1
// Good
2
{
3
  absl::MutexLock l(&mu);
4
  ready = true;
5
}
6
// ...
7
absl::MutexLock l(&mu);
8
mu.Await(absl::Condition(&ready));

1
class Server {
2
  bool ShouldStop() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mu_);
3
  void Run() {
4
    absl::MutexLock l(&mu_);
5
    while (!ShouldStop()) {
6
      mu_.Await(absl::Condition(this, &Server::HasWork));
7
      DoWork();
8
    }
9
  }
10
};

1
const absl::Condition has_work_(this, &Server::HasWork);
2
// ...
3
mu_.Await(has_work_);

1
// 회피
2
mu.Await(absl::Condition(+[] {
3
  other_mu.Lock();   // deadlock 위험
4
  // ...
5
}));

1
// 회피 — DB 쿼리, network 호출 등
2
mu.Await(absl::Condition(+[](DB* db) { return db->RowCount() > 0; }, &db));