Condvar: Add wait morphing to condvar

This patch adds wait morphing to condvar:

1. condvar->wake*() doesn't wake the thread to take the user mutex. Rather,
it attempts to grab the lock for the sleeping thread, and if the lock is
already taken, move the sleeping thread to wait on the mutex's queue,
without waking the thread up.

2. condvar->wait() now assumes that when it is woken up, it already has
the mutex.

Wait morphing reduces unnecessary context switches, and therefore improves
performance, in two case:

1. The "thundering herd" problem - when there are many threads waiting on
the condvar, if  condvar->wake_all() wakes all of them, all will race to get
the mutex and likely many of them will go back to sleep.

2. The "locked wakeup" problem - when condvar_>wake*() is done with the user
mutex locked (as it is very often does), if we wake a waiter to take the
lock, it may find the lock already held (by the waker) and go back to sleep.

core/condvar.cc

@@ -60,7 +60,15 @@ int condvar_wait(condvar_t *condvar, mutex_t* user_mutex, sched::timer* tmr)
         ret = ETIMEDOUT;
     }
 
-    mutex_lock(user_mutex);
+    if (wr.woken()) {
+        // Our wr was woken. The "wait morphing" protocol used by
+        // condvar_wake*() ensures that this only happens after we got the
+        // user_mutex for ourselves, so no need to mutex_lock() here.
+        user_mutex->receive_lock();
+    } else {
+        mutex_lock(user_mutex);
+    }
+
     return ret;
 }
 
@@ -80,7 +88,10 @@ void condvar_wake_one(condvar_t *condvar)
         if (wr->next == nullptr) {
             condvar->waiters_fifo.newest = nullptr;
         }
-        wr->wake();
+        // Rather than wake the waiter here (wr->wake()) and have it wait
+        // again for the mutex, we do "wait morphing" - have it continue to
+        // sleep until the mutex becomes available.
+        condvar->user_mutex->send_lock(wr);
         // To help the assert() in condvar_wait(), we need to zero saved
         // user_mutex when all concurrent condvar_wait()s are done.
         if (!condvar->waiters_fifo.oldest) {
@@ -100,10 +111,11 @@ void condvar_wake_all(condvar_t *condvar)
     wait_record *wr = condvar->waiters_fifo.oldest;
     // To help the assert() in condvar_wait(), we need to zero saved
     // user_mutex when all concurrent condvar_wait()s are done.
+    auto user_mutex = condvar->user_mutex;
     condvar->user_mutex = nullptr;
     while (wr) {
         auto next_wr = wr->next; // need to save - *wr invalid after wake
-        wr->wake();
+        user_mutex->send_lock(wr);
         wr = next_wr;
     }
     condvar->waiters_fifo.oldest = condvar->waiters_fifo.newest = nullptr;