sync: improve Mutex to allow successive acquisitions

This implementation allows a goroutine to do successive acquisitions
of a mutex even if there are blocked goroutines.
Moreover, it allows a newcomer goroutine to acquire a mutex ahead of
blocked goroutines (that is, it does not enforce FIFO).
On implementation level it's achieved by separating waiter count and
locked flag.
Benchmark results on HP Z600 (2 x Xeon E5620, 8 HT cores, 2.40GHz)
are as follows (with 4631059 "replace Semacquire/Semrelease implementation"
patch applied):
benchmark                                        old ns/op    new ns/op    delta
sync_test.BenchmarkMutexUncontended                  24.10        25.40   +5.39%
sync_test.BenchmarkMutexUncontended-2                12.00        13.00   +8.33%
sync_test.BenchmarkMutexUncontended-4                 6.06         6.83  +12.71%
sync_test.BenchmarkMutexUncontended-8                 3.63         3.60   -0.83%
sync_test.BenchmarkMutexUncontended-16                2.38         2.49   +4.62%

sync_test.BenchmarkMutex                             25.00        26.40   +5.60%
sync_test.BenchmarkMutex-2                          231.00        49.00  -78.79%
sync_test.BenchmarkMutex-4                          259.00       114.00  -55.98%
sync_test.BenchmarkMutex-8                          641.00       110.00  -82.84%
sync_test.BenchmarkMutex-16                        1380.00        96.30  -93.02%

sync_test.BenchmarkMutexSlack                        24.80        26.20   +5.65%
sync_test.BenchmarkMutexSlack-2                     210.00       106.00  -49.52%
sync_test.BenchmarkMutexSlack-4                     453.00       119.00  -73.73%
sync_test.BenchmarkMutexSlack-8                    1024.00       105.00  -89.75%
sync_test.BenchmarkMutexSlack-16                   1291.00        91.90  -92.88%

sync_test.BenchmarkMutexWork                        796.00       796.00   +0.00%
sync_test.BenchmarkMutexWork-2                      399.00       401.00   +0.50%
sync_test.BenchmarkMutexWork-4                      216.00       212.00   -1.85%
sync_test.BenchmarkMutexWork-8                     1547.00       196.00  -87.33%
sync_test.BenchmarkMutexWork-16                    2754.00       287.00  -89.58%

sync_test.BenchmarkMutexWorkSlack                   792.00       800.00   +1.01%
sync_test.BenchmarkMutexWorkSlack-2                 430.00       420.00   -2.33%
sync_test.BenchmarkMutexWorkSlack-4                 467.00       230.00  -50.75%
sync_test.BenchmarkMutexWorkSlack-8                1860.00       273.00  -85.32%
sync_test.BenchmarkMutexWorkSlack-16               3029.00       294.00  -90.29%

R=rsc
CC=golang-dev
https://golang.org/cl/4631075

diff --git a/src/pkg/sync/mutex.go b/src/pkg/sync/mutex.go
index 13f03cad39452eb4cab48f694ec77114ad1ae006..2d46c89948fe9c3e25633399c68769ad8bdc8ee6 100644 (file)
--- a/src/pkg/sync/mutex.go
+++ b/src/pkg/sync/mutex.go
@@ -17,8 +17,8 @@ import (
 // Mutexes can be created as part of other structures;
 // the zero value for a Mutex is an unlocked mutex.
 type Mutex struct {
-       key  int32
-       sema uint32
+       state int32
+       sema  uint32
 }
 
 // A Locker represents an object that can be locked and unlocked.
@@ -27,15 +27,41 @@ type Locker interface {
        Unlock()
 }
 
+const (
+       mutexLocked = 1 << iota // mutex is locked
+       mutexWoken
+       mutexWaiterShift = iota
+)
+
 // Lock locks m.
 // If the lock is already in use, the calling goroutine
 // blocks until the mutex is available.
 func (m *Mutex) Lock() {
-       if atomic.AddInt32(&m.key, 1) == 1 {
-               // changed from 0 to 1; we hold lock
+       // Fast path: grab unlocked mutex.
+       if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
                return
        }
-       runtime.Semacquire(&m.sema)
+
+       awoke := false
+       for {
+               old := m.state
+               new := old | mutexLocked
+               if old&mutexLocked != 0 {
+                       new = old + 1<<mutexWaiterShift
+               }
+               if awoke {
+                       // The goroutine has been woken from sleep,
+                       // so we need to reset the flag in either case.
+                       new &^= mutexWoken
+               }
+               if atomic.CompareAndSwapInt32(&m.state, old, new) {
+                       if old&mutexLocked == 0 {
+                               break
+                       }
+                       runtime.Semacquire(&m.sema)
+                       awoke = true
+               }
+       }
 }
 
 // Unlock unlocks m.
@@ -45,14 +71,25 @@ func (m *Mutex) Lock() {
 // It is allowed for one goroutine to lock a Mutex and then
 // arrange for another goroutine to unlock it.
 func (m *Mutex) Unlock() {
-       switch v := atomic.AddInt32(&m.key, -1); {
-       case v == 0:
-               // changed from 1 to 0; no contention
-               return
-       case v == -1:
-               // changed from 0 to -1: wasn't locked
-               // (or there are 4 billion goroutines waiting)
+       // Fast path: drop lock bit.
+       new := atomic.AddInt32(&m.state, -mutexLocked)
+       if (new+mutexLocked)&mutexLocked == 0 {
                panic("sync: unlock of unlocked mutex")
        }
-       runtime.Semrelease(&m.sema)
+
+       old := new
+       for {
+               // If there are no waiters or a goroutine has already
+               // been woken or grabbed the lock, no need to wake anyone.
+               if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken) != 0 {
+                       return
+               }
+               // Grab the right to wake someone.
+               new = (old - 1<<mutexWaiterShift) | mutexWoken
+               if atomic.CompareAndSwapInt32(&m.state, old, new) {
+                       runtime.Semrelease(&m.sema)
+                       return
+               }
+               old = m.state
+       }
 }