code review 12418043: net: add special netFD mutex

src/pkg/net/fd_mutex.go

 // Copyright 2013 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.
 
 package net
 
 import "sync/atomic"
 
 // fdMutex is a specialized synchronization primitive
-// that manages lifetime of fd and serializes access
+// that manages lifetime of an fd and serializes access
 // to Read and Write methods on netFD.
 type fdMutex struct {
         state uint64
         rsema uint32
         wsema uint32
 }
 
 // fdMutex.state is organized as follows:
 // 1 bit - whether netFD is closed, if set all subsequent lock operations will fail.
 // 1 bit - lock for read operations.
 // 1 bit - lock for write operations.
 // 20 bits - total number of references (read+write+misc).
 // 20 bits - number of outstanding read waiters.
 // 20 bits - number of outstanding write waiters.
 const (
         mutexClosed  = 1 << 0
         mutexRLock   = 1 << 1
         mutexWLock   = 1 << 2
         mutexRef     = 1 << 3
         mutexRefMask = (1<<20 - 1) << 3
         mutexRWait   = 1 << 23
         mutexRMask   = (1<<20 - 1) << 23
         mutexWWait   = 1 << 43
         mutexWMask   = (1<<20 - 1) << 43
 )
 
-// Read operations must do ReadLock/ReadUnlock.
-// Write operations must do WriteLock/WriteUnlock.
-// Misc operations (setsockopt) must do Lock/Unlock.
-// Close operation must do LockAndClose/Unlock.
+// Read operations must do RWLock(true)/RWUnlock(true).
+// Write operations must do RWLock(false)/RWUnlock(false).
+// Misc operations must do Incref/Decref. Misc operations include functions like
+// setsockopt and setDeadline. They need to use Incref/Decref to ensure that
+// they operate on the correct fd in presence of a concurrent Close call
+// (otherwise fd can be closed under their feet).
+// Close operation must do IncrefAndClose/Decref.
 
-// All lock operations return false if fd is closed.
-// All unlock operations return true if fd is closed and it has dropped the last reference.
+// RWLock/Incref return whether fd is open.
+// RWUnlock/Decref return whether fd is closed and there are no remaining references.
 
-func (mu *fdMutex) Lock() bool {
+func (mu *fdMutex) Incref() bool {
         for {
                 old := atomic.LoadUint64(&mu.state)
                 if old&mutexClosed != 0 {
                         return false
                 }
                 new := old + mutexRef
                 if new&mutexRefMask == 0 {
                         panic("net: inconsistent fdMutex")
                 }
                 if atomic.CompareAndSwapUint64(&mu.state, old, new) {
                         return true
                 }
         }
 }
 
-func (mu *fdMutex) LockAndClose() bool {
+func (mu *fdMutex) IncrefAndClose() bool {
         for {
                 old := atomic.LoadUint64(&mu.state)
                 if old&mutexClosed != 0 {
                         return false
                 }
                 // Mark as closed and acquire a reference.
-                new := old | mutexClosed + mutexRef
+                new := (old | mutexClosed) + mutexRef
                 if new&mutexRefMask == 0 {
                         panic("net: inconsistent fdMutex")
                 }
                 // Remove all read and write waiters.
-                new &= ^uint64(mutexRMask | mutexWMask)
+                new &^= mutexRMask | mutexWMask
                 if atomic.CompareAndSwapUint64(&mu.state, old, new) {
                         // Wake all read and write waiters,
                         // they will observe closed flag after wakeup.
                         for old&mutexRMask != 0 {
                                 old -= mutexRWait
                                 runtime_Semrelease(&mu.rsema)
                         }
                         for old&mutexWMask != 0 {
                                 old -= mutexWWait
                                 runtime_Semrelease(&mu.wsema)
                         }
                         return true
                 }
         }
 }
 
-func (mu *fdMutex) Unlock() bool {
+func (mu *fdMutex) Decref() bool {
         for {
                 old := atomic.LoadUint64(&mu.state)
                 if old&mutexRefMask == 0 {
                         panic("net: inconsistent fdMutex")
                 }
                 new := old - mutexRef
                 if atomic.CompareAndSwapUint64(&mu.state, old, new) {
                         return new&(mutexClosed|mutexRef) == mutexClosed
                 }
         }
 }
 
-func (mu *fdMutex) ReadLock() bool {
+func (mu *fdMutex) RWLock(read bool) bool {
+        var mutexBit, mutexWait, mutexMask uint64
+        var mutexSema *uint32
+        if read {
+                mutexBit = mutexRLock
+                mutexWait = mutexRWait
+                mutexMask = mutexRMask
+                mutexSema = &mu.rsema
+        } else {
+                mutexBit = mutexWLock
+                mutexWait = mutexWWait
+                mutexMask = mutexWMask
+                mutexSema = &mu.wsema
+        }
         for {
                 old := atomic.LoadUint64(&mu.state)
                 if old&mutexClosed != 0 {
                         return false
                 }
                 var new uint64
-                if old&mutexRLock == 0 {
-                        // Read lock is free, acquire it.
-                        new = old | mutexRLock + mutexRef
+                if old&mutexBit == 0 {
+                        // Lock is free, acquire it.
+                        new = (old | mutexBit) + mutexRef
                         if new&mutexRefMask == 0 {
                                 panic("net: inconsistent fdMutex")
                         }
                 } else {
-                        // Wait for read lock.
-                        new = old + mutexRWait
-                        if new&mutexRMask == 0 {
+                        // Wait for lock.
+                        new = old + mutexWait
+                        if new&mutexMask == 0 {
                                 panic("net: inconsistent fdMutex")
                         }
                 }
                 if atomic.CompareAndSwapUint64(&mu.state, old, new) {
-                        if old&mutexRLock == 0 {
+                        if old&mutexBit == 0 {
                                 return true
                         }
-                        runtime_Semacquire(&mu.rsema)
-                        // The signaller has substracted mutexRWait.
+                        runtime_Semacquire(mutexSema)
+                        // The signaller has subtracted mutexWait.
                 }
         }
 }
 
-func (mu *fdMutex) ReadUnlock() bool {
+func (mu *fdMutex) RWUnlock(read bool) bool {
+        var mutexBit, mutexWait, mutexMask uint64
+        var mutexSema *uint32
+        if read {
+                mutexBit = mutexRLock
+                mutexWait = mutexRWait
+                mutexMask = mutexRMask
+                mutexSema = &mu.rsema
+        } else {
+                mutexBit = mutexWLock
+                mutexWait = mutexWWait
+                mutexMask = mutexWMask
+                mutexSema = &mu.wsema
+        }
         for {
                 old := atomic.LoadUint64(&mu.state)
-                if old&mutexRLock == 0 || old&mutexRefMask == 0 {
+                if old&mutexBit == 0 || old&mutexRefMask == 0 {
                         panic("net: inconsistent fdMutex")
                 }
-                // Drop read lock, drop reference and wake read waiter if present.
-                new := old&^mutexRLock - mutexRef
-                if old&mutexRMask != 0 {
-                        new -= mutexRWait
+                // Drop lock, drop reference and wake read waiter if present.
+                new := (old &^ mutexBit) - mutexRef
+                if old&mutexMask != 0 {
+                        new -= mutexWait
                 }
                 if atomic.CompareAndSwapUint64(&mu.state, old, new) {
-                        if old&mutexRMask != 0 {
-                                runtime_Semrelease(&mu.rsema)
-                        }
-                        return new&(mutexClosed|mutexRef) == mutexClosed
-                }
-        }
-}
-
-func (mu *fdMutex) WriteLock() bool {
-        for {
-                old := atomic.LoadUint64(&mu.state)
-                if old&mutexClosed != 0 {
-                        return false
-                }
-                var new uint64
-                if old&mutexWLock == 0 {
-                        // Write lock is free, acquire it.
-                        new = old | mutexWLock + mutexRef
-                        if new&mutexRefMask == 0 {
-                                panic("net: inconsistent fdMutex")
-                        }
-                } else {
-                        // Wait for write lock.
-                        new = old + mutexWWait
-                        if new&mutexWMask == 0 {
-                                panic("net: inconsistent fdMutex")
-                        }
-                }
-                if atomic.CompareAndSwapUint64(&mu.state, old, new) {
-                        if old&mutexWLock == 0 {
-                                return true
-                        }
-                        runtime_Semacquire(&mu.wsema)
-                        // The signaller has substracted mutexWWait.
-                }
-        }
-}
-
-func (mu *fdMutex) WriteUnlock() bool {
-        for {
-                old := atomic.LoadUint64(&mu.state)
-                if old&mutexWLock == 0 || old&mutexRefMask == 0 {
-                        panic("net: inconsistent fdMutex")
-                }
-                // Drop write lock, decrement ref and wake write waiter if present.
-                new := old&^mutexWLock - mutexRef
-                if old&mutexWMask != 0 {
-                        new -= mutexWWait
-                }
-                if atomic.CompareAndSwapUint64(&mu.state, old, new) {
-                        if old&mutexWMask != 0 {
-                                runtime_Semrelease(&mu.wsema)
+                        if old&mutexMask != 0 {
+                                runtime_Semrelease(mutexSema)
                         }
                         return new&(mutexClosed|mutexRef) == mutexClosed
                 }
         }
 }
 
 // Implemented in runtime package.
 func runtime_Semacquire(sema *uint32)
 func runtime_Semrelease(sema *uint32)