Vihren Atlas

Atlas › Test

Test_TimedOutWaitingForLock

go.temporal.io/server/common/locks/TestPrioritySemaphoreSuite/Test_TimedOutWaitingForLock

Package
go.temporal.io/server/common/locks
Suite / test hierarchy
TestPrioritySemaphoreSuite/Test_TimedOutWaitingForLock
Test
Test_TimedOutWaitingForLock
Introduced at
priority_semaphore_impl.go ×3 Joint frontier
Covered ranges
13
Covered lines
46
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 46 covered LOC · 13 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.
69 > func NewPrioritySemaphore(n int) *PrioritySemaphoreImpl { priority_semaphore_impl.go
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.
83 > func (s *PrioritySemaphoreImpl) Acquire(ctx context.Context, priority Priority, n int) error { priority_semaphore_impl.go
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
89 > done := ctx.Done() priority_semaphore_impl.go
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()
100 > default: priority_semaphore_impl.go
101 }
102 // Check if acquisition can proceed without waiting
103 > if s.size-s.cur >= n && s.noWaiters(priority) { priority_semaphore_impl.go
104 > // Since we hold s.mu and haven't synchronized since checking done, if priority_semaphore_impl.go
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
113 > if n > s.size { priority_semaphore_impl.go
114 s.mu.Unlock()
115 return ErrRequestTooLarge
116 }
117
118 > ready := make(chan struct{}) priority_semaphore_impl.go
119 > w := waiter{n: n, ready: ready}
120 > elem := s.waitLists[priority].PushBack(w)
121 > s.mu.Unlock()
122 >
123 > select {
124 > case <-done: priority_semaphore_impl.go
125 > s.mu.Lock()
126 > select {
127 case <-ready:
128 // Acquired the semaphore after we were canceled.
130 s.cur -= n
131 s.notifyWaiters()
132 > default: priority_semaphore_impl.go
133 > isFront := s.waitLists[priority].Front() == elem
134 > s.waitLists[priority].Remove(elem)
135 > // If we're at the front and there are extra tokens left, notify other waiters.
136 > if isFront && s.size > s.cur {
137 s.notifyWaiters()
138 }
139 }
140 > s.mu.Unlock() priority_semaphore_impl.go
141 > return ctx.Err()
142
143 case <-ready:
219
220 // noWaiters returns if there is no waiter that has priority higher or equal to lowestPriority.
221 > func (s *PrioritySemaphoreImpl) noWaiters(lowestPriority Priority) bool { priority_semaphore_impl.go
222 > for _, l := range s.waitLists[:lowestPriority+1] {
223 > if l.Len() > 0 {
224 return false
225 }
226 }
227 > return true priority_semaphore_impl.go
228 }