code review 4711045: runtime: improve Linux mutex

src/pkg/runtime/linux/thread.c

 // Copyright 2009 The Go Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.
 
 #include "runtime.h"
 #include "defs.h"
 #include "os.h"
 #include "stack.h"
 
 extern SigTab runtime·sigtab[];
 
 // Linux futex.
 //
 //      futexsleep(uint32 *addr, uint32 val)
 //      futexwakeup(uint32 *addr)
 //
 // Futexsleep atomically checks if *addr == val and if so, sleeps on addr.
 // Futexwakeup wakes up one thread sleeping on addr.
 // Futexsleep is allowed to wake up spuriously.
 
 enum
 {
+        MUTEX_UNLOCKED = 0,
+        MUTEX_LOCKED = 1,
+        MUTEX_CONTENDED = 3,
+
+        ACTIVE_SPIN = 4,
+        ACTIVE_SPIN_CNT = 30,
+        PASSIVE_SPIN = 1,
+        TOTAL_SPIN = ACTIVE_SPIN + PASSIVE_SPIN,
+
         FUTEX_WAIT = 0,
         FUTEX_WAKE = 1,
 
         EINTR = 4,
         EAGAIN = 11,
 };
 
 // TODO(rsc): I tried using 1<<40 here but futex woke up (-ETIMEDOUT).
 // I wonder if the timespec that gets to the kernel
 // actually has two 32-bit numbers in it, so that
 // a 64-bit 1<<40 ends up being 0 seconds,
 // 1<<8 nanoseconds.
 static Timespec longtime =
 {
         1<<30,  // 34 years
         0
 };
 
+static Timespec shorttime =
+{
+        0,
+        10000   // 10us
+};
+
 // Atomically,
 //      if(*addr == val) sleep
 // Might be woken up spuriously; that's allowed.
 static void
-futexsleep(uint32 *addr, uint32 val)
+futexsleep(uint32 *addr, uint32 val, Timespec *time)
 {
         // Some Linux kernels have a bug where futex of
         // FUTEX_WAIT returns an internal error code
         // as an errno.  Libpthread ignores the return value
         // here, and so can we: as it says a few lines up,
         // spurious wakeups are allowed.
-        runtime·futex(addr, FUTEX_WAIT, val, &longtime, nil, 0);
+        runtime·futex(addr, FUTEX_WAIT, val, time, nil, 0);
 }
 
-// If any procs are sleeping on addr, wake up at least one.
+// If any procs are sleeping on addr, wake up at most one.

[iant, 2011/07/22 05:38:06]

I think the comment is wrong: looks like it now wakes at most cnt processes.

 static void
-futexwakeup(uint32 *addr)
+futexwakeup(uint32 *addr, uint32 cnt)
 {
         int64 ret;
 
-        ret = runtime·futex(addr, FUTEX_WAKE, 1, nil, nil, 0);
+        ret = runtime·futex(addr, FUTEX_WAKE, cnt, nil, nil, 0);
 
         if(ret >= 0)
                 return;
 
         // I don't know that futex wakeup can return
         // EAGAIN or EINTR, but if it does, it would be
         // safe to loop and call futex again.
-
-        runtime·prints("futexwakeup addr=");
-        runtime·printpointer(addr);
-        runtime·prints(" returned ");
-        runtime·printint(ret);
-        runtime·prints("\n");
+        runtime·printf("futexwakeup addr=%p returned %D\n", addr, ret);
         *(int32*)0x1006 = 0x1006;
 }
 
-
-// Lock and unlock.
-//
-// The lock state is a single 32-bit word that holds
-// a 31-bit count of threads waiting for the lock
-// and a single bit (the low bit) saying whether the lock is held.
-// The uncontended case runs entirely in user space.
-// When contention is detected, we defer to the kernel (futex).
-//
-// A reminder: compare-and-swap runtime·cas(addr, old, new) does
-//      if(*addr == old) { *addr = new; return 1; }
-//      else return 0;
-// but atomically.
-
 static void
 futexlock(Lock *l)
 {
-        uint32 v;
+        uint32 i, v, new, prv;
 
-again:
-        v = l->key;
-        if((v&1) == 0){
-                if(runtime·cas(&l->key, v, v|1)){
-                        // Lock wasn't held; we grabbed it.
-                        return;
+        // Fast path: try to speculatively grab the mutex.
+        v = runtime·xchg(&l->key, MUTEX_LOCKED);
+        if(v == MUTEX_UNLOCKED)
+                return;
+
+        prv = v;
+        for(i = 0;; i++) {
+                for(;;) {
+                        v = l->key;
+                        if(v == MUTEX_UNLOCKED) {
+                                new = prv;
+                        } else if(v == MUTEX_LOCKED) {
+                                if(i != TOTAL_SPIN)
+                                        break;
+                                new = MUTEX_CONTENDED;
+                        } else /*(v == MUTEX_CONTENDED)*/ {
+                                if(i == TOTAL_SPIN)
+                                        break;
+                                new = MUTEX_LOCKED;
+                        }
+                        if(runtime·cas(&l->key, v, new))
+                                break;
                 }
-                goto again;
+
+                if(v == MUTEX_UNLOCKED)
+                        break;
+                if(v == MUTEX_CONTENDED)
+                        prv = MUTEX_CONTENDED;

[iant, 2011/07/22 05:38:06]

I don't understand this.  If v == MUTEX_CONTENDED && i < TOTAL_SPIN, then we
just set l->key to MUTEX_LOCKED.  In that case it seems to me that we now hold
the lock, and should not sleep.

+
+                if(i < ACTIVE_SPIN) {
+                        runtime·procyield(ACTIVE_SPIN_CNT);
+                } else if(i < TOTAL_SPIN) {
+                        runtime·schedyield();
+                } else {
+                        futexsleep(&l->key, MUTEX_CONTENDED, &shorttime);
+                        i = -1;
+                }
         }
-
-        // Lock was held; try to add ourselves to the waiter count.
-        if(!runtime·cas(&l->key, v, v+2))
-                goto again;
-
-        // We're accounted for, now sleep in the kernel.
-        //
-        // We avoid the obvious lock/unlock race because
-        // the kernel won't put us to sleep if l->key has
-        // changed underfoot and is no longer v+2.
-        //
-        // We only really care that (v&1) == 1 (the lock is held),
-        // and in fact there is a futex variant that could
-        // accommodate that check, but let's not get carried away.)
-        futexsleep(&l->key, v+2);
-
-        // We're awake: remove ourselves from the count.
-        for(;;){
-                v = l->key;
-                if(v < 2)
-                        runtime·throw("bad lock key");
-                if(runtime·cas(&l->key, v, v-2))
-                        break;
-        }
-
-        // Try for the lock again.
-        goto again;
 }
 
 static void
 futexunlock(Lock *l)
 {
         uint32 v;
 
-        // Atomically get value and clear lock bit.
-again:
-        v = l->key;
-        if((v&1) == 0)
+        v = runtime·xchg(&l->key, MUTEX_UNLOCKED);
+        if(v == MUTEX_UNLOCKED)
                 runtime·throw("unlock of unlocked lock");
-        if(!runtime·cas(&l->key, v, v&~1))
-                goto again;
-
-        // If there were waiters, wake one.
-        if(v & ~1)
-                futexwakeup(&l->key);
+        if(v == MUTEX_CONTENDED)
+                futexwakeup(&l->key, 1);
 }
 
 void
 runtime·lock(Lock *l)
 {
-        if(m->locks < 0)
-                runtime·throw("lock count");
-        m->locks++;
+        if(m->locks++ < 0)
+                runtime·throw("runtime·lock: lock count");
         futexlock(l);
 }
 
 void
 runtime·unlock(Lock *l)
 {
-        m->locks--;
-        if(m->locks < 0)
-                runtime·throw("lock count");
+        if(--m->locks < 0)
+                runtime·throw("runtime·unlock: lock count");
         futexunlock(l);
 }
 
 void
 runtime·destroylock(Lock*)
 {
 }
 
 
 // One-time notifications.
-//
-// Since the lock/unlock implementation already
-// takes care of sleeping in the kernel, we just reuse it.
-// (But it's a weird use, so it gets its own interface.)
-//
-// We use a lock to represent the event:
-// unlocked == event has happened.
-// Thus the lock starts out locked, and to wait for the
-// event you try to lock the lock.  To signal the event,
-// you unlock the lock.
-
 void
 runtime·noteclear(Note *n)
 {
-        n->lock.key = 0;        // memset(n, 0, sizeof *n)
-        futexlock(&n->lock);
+        n->state = 0;
 }
 
 void
 runtime·notewakeup(Note *n)
 {
-        futexunlock(&n->lock);
+        runtime·xchg(&n->state, 1);
+        futexwakeup(&n->state, 1<<30);
 }
 
 void
 runtime·notesleep(Note *n)
 {
-        futexlock(&n->lock);
-        futexunlock(&n->lock);  // Let other sleepers find out too.
+        while(runtime·atomicload(&n->state) == 0)
+                futexsleep(&n->state, 0, &longtime);
 }
 
 
 // Clone, the Linux rfork.
 enum
 {
         CLONE_VM = 0x100,
         CLONE_FS = 0x200,
         CLONE_FILES = 0x400,
         CLONE_SIGHAND = 0x800,
@@ skipping 79 matching lines @@
                 switch(g->sigcode0) {
                 case FPE_INTDIV:
                         runtime·panicstring("integer divide by zero");
                 case FPE_INTOVF:
                         runtime·panicstring("integer overflow");
                 }
                 runtime·panicstring("floating point error");
         }
         runtime·panicstring(runtime·sigtab[g->sig].name);
 }