Atlas › Test
Test_AllThreadsAreWokenUp
Exact test identity: go.temporal.io/server/common/locks/TestPrioritySemaphoreSuite/Test_AllThreadsAreWokenUp
- Package
go.temporal.io/server/common/locks- Suite / test hierarchy
TestPrioritySemaphoreSuite/Test_AllThreadsAreWokenUp- Test
Test_AllThreadsAreWokenUp- Introduced at
- Test_AllThreadsAreWokenUp Frontier kind: Test frontier
- Covered ranges
- 20
- Covered lines
- 75
- Covered files
- 1
Covered source
Expand a file to inspect source; the > gutter marks covered lines.
go.temporal.io/server/common/locks/priority_semaphore_impl.go 75 covered LOC · 20 ranges
Open complete file
67
// maximum combined weight for concurrent access, capable of handling multiple priority levels.
68
// Most of the logic is taken directly from golang's semaphore.Weighted.
70
>
waitLists := make([]*list.List, NumPriorities)
71
>
for i := range waitLists {
72
>
waitLists[i] = list.New()
73
>
}
74
>
return &PrioritySemaphoreImpl{
75
>
size: n,
76
>
waitLists: waitLists,
77
>
}
78
}
79
81
// are available or ctx is done. On success, returns nil. On failure, returns
82
// ctx.Err() and leaves the semaphore unchanged.
84
>
if priority >= NumPriorities {
85
// nolint:forbidigo
86
panic(fmt.Sprintf("semaphore: invalid priority %v, priority must be less than %v", priority, NumPriorities))
87
}
88
90
>
91
>
s.mu.Lock()
92
>
select {
93
case <-done:
94
// ctx becoming done has "happened before" acquiring the semaphore,
98
s.mu.Unlock()
99
return ctx.Err()
101
}
102
// Check if acquisition can proceed without waiting
105
>
// ctx becomes done before we return here, it becoming done must have
106
>
// "happened concurrently" with this call - it cannot "happen before"
107
>
// we return in this branch. So, we're ok to always acquire here.
108
>
s.cur += n
109
>
s.mu.Unlock()
110
>
return nil
111
>
}
112
114
s.mu.Unlock()
115
return ErrRequestTooLarge
116
}
117
119
>
w := waiter{n: n, ready: ready}
120
>
elem := s.waitLists[priority].PushBack(w)
121
>
s.mu.Unlock()
122
>
123
>
select {
124
case <-done:
125
s.mu.Lock()
141
return ctx.Err()
142
144
>
// Acquired the semaphore. Check that ctx isn't already done.
145
>
// We check the done channel instead of calling ctx.Err because we
146
>
// already have the channel, and ctx.Err is O(n) with the nesting
147
>
// depth of ctx.
148
>
select {
149
case <-done:
150
s.Release(n)
151
return ctx.Err()
153
}
155
}
156
}
173
}
174
176
>
s.mu.Lock()
177
>
defer s.mu.Unlock()
178
>
s.cur -= n
179
>
if s.cur < 0 {
180
s.mu.Unlock()
181
panic("semaphore: released more than held")
182
}
184
}
185
187
>
for _, l := range s.waitLists {
188
>
for {
189
>
next := l.Front()
190
>
if next == nil {
191
>
break // No more waiters blocked.
192
}
193
195
>
if !ok {
196
panic("semaphore: failed to cast waiter")
197
}
200
>
// find a waiter with a smaller request), but under load that could cause
201
>
// starvation for large requests; instead, we leave all remaining waiters
202
>
// blocked. For the same reason, we should not wake lower priority waiters.
203
>
//
204
>
// Consider a semaphore used as a read-write lock, with N tokens, N
205
>
// readers, and one writer. Each reader can Acquire(1) to obtain a read
206
>
// lock. The writer can Acquire(N) to obtain a write lock, excluding all
207
>
// of the readers. If we allow the readers to jump ahead in the queue,
208
>
// the writer will starve — there is always one token available for every
209
>
// reader.
210
>
return
211
>
}
212
214
>
l.Remove(next)
215
>
close(w.ready)
216
}
217
}
219
220
// noWaiters returns if there is no waiter that has priority higher or equal to lowestPriority.
222
>
for _, l := range s.waitLists[:lowestPriority+1] {
223
>
if l.Len() > 0 {
224
return false
225
}
226
}
228
}