Requeue rather than wake when notifying.

Requeueing avoids contention when the newly-awoken waiter attempts to
acquire the lock held by the notifier.

Bug: 117842465
Change-Id: I78646245b538022012d546df12514f5d43d9f234

2 files changed, 26 insertions, 25 deletions

diff --git a/runtime/base/mutex.cc b/runtime/base/mutex.cc
index b2ddff3f6a..7a9a493a9b 100644
--- a/runtime/base/mutex.cc
+++ b/runtime/base/mutex.cc
@@ -895,40 +895,37 @@ void ConditionVariable::Broadcast(Thread* self) {
   // guard_.AssertExclusiveHeld(self);
   DCHECK_EQ(guard_.GetExclusiveOwnerTid(), SafeGetTid(self));
 #if ART_USE_FUTEXES
-  if (num_waiters_ > 0) {
-    sequence_++;  // Indicate the broadcast occurred.
-    bool done = false;
-    do {
-      int32_t cur_sequence = sequence_.load(std::memory_order_relaxed);
-      // Requeue waiters onto mutex. The waiter holds the contender count on the mutex high ensuring
-      // mutex unlocks will awaken the requeued waiter thread.
-      done = futex(sequence_.Address(), FUTEX_CMP_REQUEUE, 0,
-                   reinterpret_cast<const timespec*>(std::numeric_limits<int32_t>::max()),
-                   guard_.state_.Address(), cur_sequence) != -1;
-      if (!done) {
-        if (errno != EAGAIN && errno != EINTR) {
-          PLOG(FATAL) << "futex cmp requeue failed for " << name_;
-        }
-      }
-    } while (!done);
-  }
+  RequeueWaiters(std::numeric_limits<int32_t>::max());
 #else
   CHECK_MUTEX_CALL(pthread_cond_broadcast, (&cond_));
 #endif
 }
 
-void ConditionVariable::Signal(Thread* self) {
-  DCHECK(self == nullptr || self == Thread::Current());
-  guard_.AssertExclusiveHeld(self);
 #if ART_USE_FUTEXES
+void ConditionVariable::RequeueWaiters(int32_t count) {
   if (num_waiters_ > 0) {
     sequence_++;  // Indicate a signal occurred.
-    // Futex wake 1 waiter who will then come and in contend on mutex. It'd be nice to requeue them
-    // to avoid this, however, requeueing can only move all waiters.
-    int num_woken = futex(sequence_.Address(), FUTEX_WAKE, 1, nullptr, nullptr, 0);
-    // Check something was woken or else we changed sequence_ before they had chance to wait.
-    CHECK((num_woken == 0) || (num_woken == 1));
+    // Move waiters from the condition variable's futex to the guard's futex,
+    // so that they will be woken up when the mutex is released.
+    bool done = futex(sequence_.Address(),
+                      FUTEX_REQUEUE,
+                      /* Threads to wake */ 0,
+                      /* Threads to requeue*/ reinterpret_cast<const timespec*>(count),
+                      guard_.state_.Address(),
+                      0) != -1;
+    if (!done && errno != EAGAIN && errno != EINTR) {
+      PLOG(FATAL) << "futex requeue failed for " << name_;
+    }
   }
+}
+#endif
+
+
+void ConditionVariable::Signal(Thread* self) {
+  DCHECK(self == nullptr || self == Thread::Current());
+  guard_.AssertExclusiveHeld(self);
+#if ART_USE_FUTEXES
+  RequeueWaiters(1);
 #else
   CHECK_MUTEX_CALL(pthread_cond_signal, (&cond_));
 #endif
diff --git a/runtime/base/mutex.h b/runtime/base/mutex.h
index d127d0f01f..7711be9c90 100644
--- a/runtime/base/mutex.h
+++ b/runtime/base/mutex.h
@@ -480,7 +480,9 @@ class ConditionVariable {
   ConditionVariable(const char* name, Mutex& mutex);
   ~ConditionVariable();
 
+  // Requires the mutex to be held.
   void Broadcast(Thread* self);
+  // Requires the mutex to be held.
   void Signal(Thread* self);
   // TODO: No thread safety analysis on Wait and TimedWait as they call mutex operations via their
   //       pointer copy, thereby defeating annotalysis.
@@ -505,6 +507,8 @@ class ConditionVariable {
   // Number of threads that have come into to wait, not the length of the waiters on the futex as
   // waiters may have been requeued onto guard_. Guarded by guard_.
   volatile int32_t num_waiters_;
+
+  void RequeueWaiters(int32_t count);
 #else
   pthread_cond_t cond_;
 #endif