mirror of
https://github.com/golang/go
synced 2024-10-04 11:11:21 -06:00
8e7072ca83
Change Cond implementation to use a notification list such that waiters can first register for a notification, release the lock, then actually wait. Signalers never have to park anymore. This is intended to address an issue in the previous implementation where Broadcast could fail to signal all waiters. Results of the existing benchmark are below. Original New Diff BenchmarkCond1-48 2000000 745 ns/op 755 +1.3% BenchmarkCond2-48 1000000 1545 ns/op 1532 -0.8% BenchmarkCond4-48 300000 3833 ns/op 3896 +1.6% BenchmarkCond8-48 200000 10049 ns/op 10257 +2.1% BenchmarkCond16-48 100000 21123 ns/op 21236 +0.5% BenchmarkCond32-48 30000 40393 ns/op 41097 +1.7% Fixes #14064 Change-Id: I083466d61593a791a034df61f5305adfb8f1c7f9 Reviewed-on: https://go-review.googlesource.com/18892 Reviewed-by: Dmitry Vyukov <dvyukov@google.com> Reviewed-by: Austin Clements <austin@google.com> Run-TryBot: Caleb Spare <cespare@gmail.com> TryBot-Result: Gobot Gobot <gobot@golang.org>
355 lines
8.7 KiB
Go
355 lines
8.7 KiB
Go
// 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.
|
|
|
|
// Semaphore implementation exposed to Go.
|
|
// Intended use is provide a sleep and wakeup
|
|
// primitive that can be used in the contended case
|
|
// of other synchronization primitives.
|
|
// Thus it targets the same goal as Linux's futex,
|
|
// but it has much simpler semantics.
|
|
//
|
|
// That is, don't think of these as semaphores.
|
|
// Think of them as a way to implement sleep and wakeup
|
|
// such that every sleep is paired with a single wakeup,
|
|
// even if, due to races, the wakeup happens before the sleep.
|
|
//
|
|
// See Mullender and Cox, ``Semaphores in Plan 9,''
|
|
// http://swtch.com/semaphore.pdf
|
|
|
|
package runtime
|
|
|
|
import (
|
|
"runtime/internal/atomic"
|
|
"runtime/internal/sys"
|
|
"unsafe"
|
|
)
|
|
|
|
// Asynchronous semaphore for sync.Mutex.
|
|
|
|
type semaRoot struct {
|
|
lock mutex
|
|
head *sudog
|
|
tail *sudog
|
|
nwait uint32 // Number of waiters. Read w/o the lock.
|
|
}
|
|
|
|
// Prime to not correlate with any user patterns.
|
|
const semTabSize = 251
|
|
|
|
var semtable [semTabSize]struct {
|
|
root semaRoot
|
|
pad [sys.CacheLineSize - unsafe.Sizeof(semaRoot{})]byte
|
|
}
|
|
|
|
//go:linkname sync_runtime_Semacquire sync.runtime_Semacquire
|
|
func sync_runtime_Semacquire(addr *uint32) {
|
|
semacquire(addr, true)
|
|
}
|
|
|
|
//go:linkname net_runtime_Semacquire net.runtime_Semacquire
|
|
func net_runtime_Semacquire(addr *uint32) {
|
|
semacquire(addr, true)
|
|
}
|
|
|
|
//go:linkname sync_runtime_Semrelease sync.runtime_Semrelease
|
|
func sync_runtime_Semrelease(addr *uint32) {
|
|
semrelease(addr)
|
|
}
|
|
|
|
//go:linkname net_runtime_Semrelease net.runtime_Semrelease
|
|
func net_runtime_Semrelease(addr *uint32) {
|
|
semrelease(addr)
|
|
}
|
|
|
|
func readyWithTime(s *sudog, traceskip int) {
|
|
if s.releasetime != 0 {
|
|
s.releasetime = cputicks()
|
|
}
|
|
goready(s.g, traceskip)
|
|
}
|
|
|
|
// Called from runtime.
|
|
func semacquire(addr *uint32, profile bool) {
|
|
gp := getg()
|
|
if gp != gp.m.curg {
|
|
throw("semacquire not on the G stack")
|
|
}
|
|
|
|
// Easy case.
|
|
if cansemacquire(addr) {
|
|
return
|
|
}
|
|
|
|
// Harder case:
|
|
// increment waiter count
|
|
// try cansemacquire one more time, return if succeeded
|
|
// enqueue itself as a waiter
|
|
// sleep
|
|
// (waiter descriptor is dequeued by signaler)
|
|
s := acquireSudog()
|
|
root := semroot(addr)
|
|
t0 := int64(0)
|
|
s.releasetime = 0
|
|
if profile && blockprofilerate > 0 {
|
|
t0 = cputicks()
|
|
s.releasetime = -1
|
|
}
|
|
for {
|
|
lock(&root.lock)
|
|
// Add ourselves to nwait to disable "easy case" in semrelease.
|
|
atomic.Xadd(&root.nwait, 1)
|
|
// Check cansemacquire to avoid missed wakeup.
|
|
if cansemacquire(addr) {
|
|
atomic.Xadd(&root.nwait, -1)
|
|
unlock(&root.lock)
|
|
break
|
|
}
|
|
// Any semrelease after the cansemacquire knows we're waiting
|
|
// (we set nwait above), so go to sleep.
|
|
root.queue(addr, s)
|
|
goparkunlock(&root.lock, "semacquire", traceEvGoBlockSync, 4)
|
|
if cansemacquire(addr) {
|
|
break
|
|
}
|
|
}
|
|
if s.releasetime > 0 {
|
|
blockevent(s.releasetime-t0, 3)
|
|
}
|
|
releaseSudog(s)
|
|
}
|
|
|
|
func semrelease(addr *uint32) {
|
|
root := semroot(addr)
|
|
atomic.Xadd(addr, 1)
|
|
|
|
// Easy case: no waiters?
|
|
// This check must happen after the xadd, to avoid a missed wakeup
|
|
// (see loop in semacquire).
|
|
if atomic.Load(&root.nwait) == 0 {
|
|
return
|
|
}
|
|
|
|
// Harder case: search for a waiter and wake it.
|
|
lock(&root.lock)
|
|
if atomic.Load(&root.nwait) == 0 {
|
|
// The count is already consumed by another goroutine,
|
|
// so no need to wake up another goroutine.
|
|
unlock(&root.lock)
|
|
return
|
|
}
|
|
s := root.head
|
|
for ; s != nil; s = s.next {
|
|
if s.elem == unsafe.Pointer(addr) {
|
|
atomic.Xadd(&root.nwait, -1)
|
|
root.dequeue(s)
|
|
break
|
|
}
|
|
}
|
|
unlock(&root.lock)
|
|
if s != nil {
|
|
readyWithTime(s, 5)
|
|
}
|
|
}
|
|
|
|
func semroot(addr *uint32) *semaRoot {
|
|
return &semtable[(uintptr(unsafe.Pointer(addr))>>3)%semTabSize].root
|
|
}
|
|
|
|
func cansemacquire(addr *uint32) bool {
|
|
for {
|
|
v := atomic.Load(addr)
|
|
if v == 0 {
|
|
return false
|
|
}
|
|
if atomic.Cas(addr, v, v-1) {
|
|
return true
|
|
}
|
|
}
|
|
}
|
|
|
|
func (root *semaRoot) queue(addr *uint32, s *sudog) {
|
|
s.g = getg()
|
|
s.elem = unsafe.Pointer(addr)
|
|
s.next = nil
|
|
s.prev = root.tail
|
|
if root.tail != nil {
|
|
root.tail.next = s
|
|
} else {
|
|
root.head = s
|
|
}
|
|
root.tail = s
|
|
}
|
|
|
|
func (root *semaRoot) dequeue(s *sudog) {
|
|
if s.next != nil {
|
|
s.next.prev = s.prev
|
|
} else {
|
|
root.tail = s.prev
|
|
}
|
|
if s.prev != nil {
|
|
s.prev.next = s.next
|
|
} else {
|
|
root.head = s.next
|
|
}
|
|
s.elem = nil
|
|
s.next = nil
|
|
s.prev = nil
|
|
}
|
|
|
|
// notifyList is a ticket-based notification list used to implement sync.Cond.
|
|
//
|
|
// It must be kept in sync with the sync package.
|
|
type notifyList struct {
|
|
// wait is the ticket number of the next waiter. It is atomically
|
|
// incremented outside the lock.
|
|
wait uint32
|
|
|
|
// notify is the ticket number of the next waiter to be notified. It can
|
|
// be read outside the lock, but is only written to with lock held.
|
|
//
|
|
// Both wait & notify can wrap around, and such cases will be correctly
|
|
// handled as long as their "unwrapped" difference is bounded by 2^31.
|
|
// For this not to be the case, we'd need to have 2^31+ goroutines
|
|
// blocked on the same condvar, which is currently not possible.
|
|
notify uint32
|
|
|
|
// List of parked waiters.
|
|
lock mutex
|
|
head *sudog
|
|
tail *sudog
|
|
}
|
|
|
|
// less checks if a < b, considering a & b running counts that may overflow the
|
|
// 32-bit range, and that their "unwrapped" difference is always less than 2^31.
|
|
func less(a, b uint32) bool {
|
|
return int32(a-b) < 0
|
|
}
|
|
|
|
// notifyListAdd adds the caller to a notify list such that it can receive
|
|
// notifications. The caller must eventually call notifyListWait to wait for
|
|
// such a notification, passing the returned ticket number.
|
|
//go:linkname notifyListAdd sync.runtime_notifyListAdd
|
|
func notifyListAdd(l *notifyList) uint32 {
|
|
// This may be called concurrently, for example, when called from
|
|
// sync.Cond.Wait while holding a RWMutex in read mode.
|
|
return atomic.Xadd(&l.wait, 1) - 1
|
|
}
|
|
|
|
// notifyListWait waits for a notification. If one has been sent since
|
|
// notifyListAdd was called, it returns immediately. Otherwise, it blocks.
|
|
//go:linkname notifyListWait sync.runtime_notifyListWait
|
|
func notifyListWait(l *notifyList, t uint32) {
|
|
lock(&l.lock)
|
|
|
|
// Return right away if this ticket has already been notified.
|
|
if less(t, l.notify) {
|
|
unlock(&l.lock)
|
|
return
|
|
}
|
|
|
|
// Enqueue itself.
|
|
s := acquireSudog()
|
|
s.g = getg()
|
|
s.ticket = t
|
|
s.releasetime = 0
|
|
t0 := int64(0)
|
|
if blockprofilerate > 0 {
|
|
t0 = cputicks()
|
|
s.releasetime = -1
|
|
}
|
|
if l.tail == nil {
|
|
l.head = s
|
|
} else {
|
|
l.tail.next = s
|
|
}
|
|
l.tail = s
|
|
goparkunlock(&l.lock, "semacquire", traceEvGoBlockCond, 3)
|
|
if t0 != 0 {
|
|
blockevent(s.releasetime-t0, 2)
|
|
}
|
|
releaseSudog(s)
|
|
}
|
|
|
|
// notifyListNotifyAll notifies all entries in the list.
|
|
//go:linkname notifyListNotifyAll sync.runtime_notifyListNotifyAll
|
|
func notifyListNotifyAll(l *notifyList) {
|
|
// Fast-path: if there are no new waiters since the last notification
|
|
// we don't need to acquire the lock.
|
|
if atomic.Load(&l.wait) == atomic.Load(&l.notify) {
|
|
return
|
|
}
|
|
|
|
// Pull the list out into a local variable, waiters will be readied
|
|
// outside the lock.
|
|
lock(&l.lock)
|
|
s := l.head
|
|
l.head = nil
|
|
l.tail = nil
|
|
|
|
// Update the next ticket to be notified. We can set it to the current
|
|
// value of wait because any previous waiters are already in the list
|
|
// or will notice that they have already been notified when trying to
|
|
// add themselves to the list.
|
|
atomic.Store(&l.notify, atomic.Load(&l.wait))
|
|
unlock(&l.lock)
|
|
|
|
// Go through the local list and ready all waiters.
|
|
for s != nil {
|
|
next := s.next
|
|
s.next = nil
|
|
readyWithTime(s, 4)
|
|
s = next
|
|
}
|
|
}
|
|
|
|
// notifyListNotifyOne notifies one entry in the list.
|
|
//go:linkname notifyListNotifyOne sync.runtime_notifyListNotifyOne
|
|
func notifyListNotifyOne(l *notifyList) {
|
|
// Fast-path: if there are no new waiters since the last notification
|
|
// we don't need to acquire the lock at all.
|
|
if atomic.Load(&l.wait) == atomic.Load(&l.notify) {
|
|
return
|
|
}
|
|
|
|
lock(&l.lock)
|
|
|
|
// Re-check under the lock if we need to do anything.
|
|
t := l.notify
|
|
if t == atomic.Load(&l.wait) {
|
|
unlock(&l.lock)
|
|
return
|
|
}
|
|
|
|
// Update the next notify ticket number, and try to find the G that
|
|
// needs to be notified. If it hasn't made it to the list yet we won't
|
|
// find it, but it won't park itself once it sees the new notify number.
|
|
atomic.Store(&l.notify, t+1)
|
|
for p, s := (*sudog)(nil), l.head; s != nil; p, s = s, s.next {
|
|
if s.ticket == t {
|
|
n := s.next
|
|
if p != nil {
|
|
p.next = n
|
|
} else {
|
|
l.head = n
|
|
}
|
|
if n == nil {
|
|
l.tail = p
|
|
}
|
|
unlock(&l.lock)
|
|
s.next = nil
|
|
readyWithTime(s, 4)
|
|
return
|
|
}
|
|
}
|
|
unlock(&l.lock)
|
|
}
|
|
|
|
//go:linkname notifyListCheck sync.runtime_notifyListCheck
|
|
func notifyListCheck(sz uintptr) {
|
|
if sz != unsafe.Sizeof(notifyList{}) {
|
|
print("runtime: bad notifyList size - sync=", sz, " runtime=", unsafe.Sizeof(notifyList{}), "\n")
|
|
throw("bad notifyList size")
|
|
}
|
|
}
|