message: Introduce per Waiter lock to reduce contention.
diff: @@ -148,8 +148,31 @@ class ThreadWorkSource {
  "TF_RUN_HANDLER_MAX_RANK_TO_WAKE_UP", kMaxConcurrentHandlers));
  if (max_rank_to_wakeup > 0 &&
  rank_.load(std::memory_order_relaxed) <= max_rank_to_wakeup) {
  + Waiter* w = nullptr;
  + {
  mutex_lock l(waiters_mu_);
  - queue_waiters_.next->cv.notify_one();
  + if (queue_waiters_.next != &queue_waiters_) {
  + // Remove waiter from the LIFO queue
  + w = queue_waiters_.next;
  +
  + CHECK(w->prev != w);
  + CHECK(w->next != w);
  +
  + w->next->prev = w->prev;
  + w->prev->next = w->next;
  +
  + // Use `w->next == &w` to indicate that the waiter has been removed
  + // from the queue.
  + w->next = w;
  + w->prev = w;
  + }
  + }
  + if (w != nullptr) {
  + // We call notify_one() without any locks, so we can miss notifications.
  + // The wake up logic is best effort and a thread will wake in short
  + // period of time in case a notification is missed.
  + w->cv.notify_one();
  + }
  }
  VLOG(3) << "Added " << (is_blocking ? "inter" : "intra") << " work from "
  << traceme_id_.load(std::memory_order_relaxed);
  @@ -163,18 +186,38 @@ class ThreadWorkSource {
  }
  void WaitForWork(int max_sleep_micros) {
  + thread_local Waiter waiter;
  + {
  mutex_lock l(waiters_mu_);
  - Waiter waiter;
  + CHECK_EQ(waiter.next, &waiter);
  + CHECK_EQ(waiter.prev, &waiter);
  +
  // Add waiter to the LIFO queue
  waiter.prev = &queue_waiters_;
  waiter.next = queue_waiters_.next;
  waiter.next->prev = &waiter;
  waiter.prev->next = &waiter;
  + }
  + {
  + mutex_lock l(waiter.mu);
  // Wait on the condition variable
  waiter.cv.wait_for(l, std::chrono::microseconds(max_sleep_micros));
  - // Remove waiter from the LIFO queue
  + }
  +
  + mutex_lock l(waiters_mu_);
  + // Remove waiter from the LIFO queue. Note even when a waiter wakes up due
  + // to a notification we cannot conclude the waiter is not in the queue.
  + // This is due to the fact that a thread preempted right before notifying
  + // may resume after a waiter got re-added.
  + if (waiter.next != &waiter) {
  + CHECK(waiter.prev != &waiter);
  waiter.next->prev = waiter.prev;
  waiter.prev->next = waiter.next;
  + waiter.next = &waiter;
  + waiter.prev = &waiter;
  + } else {
  + CHECK_EQ(waiter.prev, &waiter);
  + }
  }
  int TaskQueueSize(bool is_blocking) {
  @@ -229,7 +272,12 @@ class ThreadWorkSource {
  // queue them in LIFO order rather than the FIFO order used by a single
  // condition variable.
  struct Waiter {
  + Waiter() {
  + next = this;
  + prev = this;
  + }
  condition_variable cv;
  + mutex mu;
  Waiter* next;
  Waiter* prev;
  };
  