1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
|
// Copyright 2009 The Go Authors. All rights reserved.
// Copyright 2019 Google LLC
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
// This is mostly copied from the standard library's sync/rwmutex.go.
//
// Happens-before relationships indicated to the race detector:
// - Unlock -> Lock (via writerSem)
// - Unlock -> RLock (via readerSem)
// - RUnlock -> Lock (via writerSem)
// - DowngradeLock -> RLock (via readerSem)
package gvsync
import (
"sync"
"sync/atomic"
"unsafe"
)
// DowngradableRWMutex is identical to sync.RWMutex, but adds the DowngradeLock
// method.
type DowngradableRWMutex struct {
w sync.Mutex // held if there are pending writers
writerSem uint32 // semaphore for writers to wait for completing readers
readerSem uint32 // semaphore for readers to wait for completing writers
readerCount int32 // number of pending readers
readerWait int32 // number of departing readers
}
const rwmutexMaxReaders = 1 << 30
// RLock locks rw for reading.
func (rw *DowngradableRWMutex) RLock() {
if RaceEnabled {
RaceDisable()
}
if atomic.AddInt32(&rw.readerCount, 1) < 0 {
// A writer is pending, wait for it.
runtimeSemacquire(&rw.readerSem)
}
if RaceEnabled {
RaceEnable()
RaceAcquire(unsafe.Pointer(&rw.readerSem))
}
}
// RUnlock undoes a single RLock call.
func (rw *DowngradableRWMutex) RUnlock() {
if RaceEnabled {
// TODO: Why does this need to be ReleaseMerge instead of
// Release? IIUC this establishes Unlock happens-before RUnlock, which
// seems unnecessary.
RaceReleaseMerge(unsafe.Pointer(&rw.writerSem))
RaceDisable()
}
if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {
if r+1 == 0 || r+1 == -rwmutexMaxReaders {
panic("RUnlock of unlocked DowngradableRWMutex")
}
// A writer is pending.
if atomic.AddInt32(&rw.readerWait, -1) == 0 {
// The last reader unblocks the writer.
runtimeSemrelease(&rw.writerSem, false)
}
}
if RaceEnabled {
RaceEnable()
}
}
// Lock locks rw for writing.
func (rw *DowngradableRWMutex) Lock() {
if RaceEnabled {
RaceDisable()
}
// First, resolve competition with other writers.
rw.w.Lock()
// Announce to readers there is a pending writer.
r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
// Wait for active readers.
if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
runtimeSemacquire(&rw.writerSem)
}
if RaceEnabled {
RaceEnable()
RaceAcquire(unsafe.Pointer(&rw.writerSem))
}
}
// Unlock unlocks rw for writing.
func (rw *DowngradableRWMutex) Unlock() {
if RaceEnabled {
RaceRelease(unsafe.Pointer(&rw.writerSem))
RaceRelease(unsafe.Pointer(&rw.readerSem))
RaceDisable()
}
// Announce to readers there is no active writer.
r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
if r >= rwmutexMaxReaders {
panic("Unlock of unlocked DowngradableRWMutex")
}
// Unblock blocked readers, if any.
for i := 0; i < int(r); i++ {
runtimeSemrelease(&rw.readerSem, false)
}
// Allow other writers to proceed.
rw.w.Unlock()
if RaceEnabled {
RaceEnable()
}
}
// DowngradeLock atomically unlocks rw for writing and locks it for reading.
func (rw *DowngradableRWMutex) DowngradeLock() {
if RaceEnabled {
RaceRelease(unsafe.Pointer(&rw.readerSem))
RaceDisable()
}
// Announce to readers there is no active writer and one additional reader.
r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders+1)
if r >= rwmutexMaxReaders+1 {
panic("DowngradeLock of unlocked DowngradableRWMutex")
}
// Unblock blocked readers, if any. Note that this loop starts as 1 since r
// includes this goroutine.
for i := 1; i < int(r); i++ {
runtimeSemrelease(&rw.readerSem, false)
}
// Allow other writers to proceed to rw.w.Lock(). Note that they will still
// block on rw.writerSem since at least this reader exists, such that
// DowngradeLock() is atomic with the previous write lock.
rw.w.Unlock()
if RaceEnabled {
RaceEnable()
}
}
//go:linkname runtimeSemacquire sync.runtime_Semacquire
func runtimeSemacquire(s *uint32)
//go:linkname runtimeSemrelease sync.runtime_Semrelease
func runtimeSemrelease(s *uint32, handoff bool)
|
