Vihren Atlas

Atlas › Test

TestLock_Mixed

go.temporal.io/server/common/locks/TestPriorityMutexSuite/TestLock_Mixed

Package
go.temporal.io/server/common/locks
Suite / test hierarchy
TestPriorityMutexSuite/TestLock_Mixed
Test
TestLock_Mixed
Introduced at
priority_mutex_impl.go ×4 Joint frontier
Covered ranges
24
Covered lines
96
Covered files
2

Covered source

Expand a file to inspect source; the > gutter marks covered lines.

go.temporal.io/server/common/locks/priority_mutex_impl.go 58 covered LOC · 19 ranges

Open complete file

29 var _ PriorityMutex = (*PriorityMutexImpl)(nil)
30
31 > func NewPriorityMutex() *PriorityMutexImpl { priority_mutex_impl.go
32 > lock := &sync.Mutex{}
33 > syncCV := NewConditionVariable(lock)
34 > asyncCV := NewConditionVariable(lock)
35 > return &PriorityMutexImpl{
36 > locker: lock,
37 > highCV: syncCV,
38 > lowCV: asyncCV,
39 > highWait: 0,
40 > lowWait: 0,
41 >
42 > lockState: PriorityMutexStateUnlocked,
43 > }
44 > }
45
46 // LockHigh try to lock with high priority, use LockHigh / UnlockHigh pair to lock / unlock
47 func (c *PriorityMutexImpl) LockHigh(
48 ctx context.Context,
49 > ) error { priority_mutex_impl.go
50 > c.locker.Lock()
51 > defer c.locker.Unlock()
52 >
53 > c.highWait += 1
54 > defer func() { c.highWait -= 1 }()
55
56 > for c.lockState != PriorityMutexStateUnlocked && ctx.Err() == nil { priority_mutex_impl.go
57 > c.highCV.Wait(ctx.Done()) priority_mutex_impl.go
58 > }
59
60 > if ctx.Err() != nil { priority_mutex_impl.go
61 return ctx.Err()
62 }
63
64 > c.lockState = PriorityMutexStateLockedByHigh priority_mutex_impl.go
65 > return nil
66 }
67
69 func (c *PriorityMutexImpl) LockLow(
70 ctx context.Context,
71 > ) error { priority_mutex_impl.go
72 > c.locker.Lock()
73 > defer c.locker.Unlock()
74 >
75 > c.lowWait += 1
76 > defer func() { c.lowWait -= 1 }()
77
78 > for c.lockState != PriorityMutexStateUnlocked && ctx.Err() == nil { priority_mutex_impl.go
79 > c.lowCV.Wait(ctx.Done()) priority_mutex_impl.go
80 > }
81
82 > if ctx.Err() != nil { priority_mutex_impl.go
83 return ctx.Err()
84 }
85
86 > c.lockState = PriorityMutexStateLockedByLow priority_mutex_impl.go
87 > return nil
88 }
89
113
114 // UnlockHigh unlock with high priority, use LockHigh / UnlockHigh pair to lock / unlock
115 > func (c *PriorityMutexImpl) UnlockHigh() { priority_mutex_impl.go
116 > c.locker.Lock()
117 > defer c.locker.Unlock()
118 >
119 > if c.lockState != PriorityMutexStateLockedByHigh {
120 panic(fmt.Sprintf("unable to unlock high priority, state: %v\n", c.lockState))
121 }
122
123 > c.lockState = PriorityMutexStateUnlocked priority_mutex_impl.go
124 > c.notify()
125 }
126
127 // UnlockLow unlock with low priority, use LockLow / UnlockLow pair to lock / unlock
128 > func (c *PriorityMutexImpl) UnlockLow() { priority_mutex_impl.go
129 > c.locker.Lock()
130 > defer c.locker.Unlock()
131 >
132 > if c.lockState != PriorityMutexStateLockedByLow {
133 panic(fmt.Sprintf("unable to unlock high priority, state: %v\n", c.lockState))
134 }
135
136 > c.lockState = PriorityMutexStateUnlocked priority_mutex_impl.go
137 > c.notify()
138 }
139
go.temporal.io/server/common/locks/condition_variable_impl.go 38 covered LOC · 5 ranges

Open complete file

18 func NewConditionVariable(
19 lock Locker,
20 > ) *ConditionVariableImpl { condition_variable_impl.go
21 > return &ConditionVariableImpl{
22 > lock: lock,
23 >
24 > chanLock: sync.Mutex{},
25 > channel: newCVChannel(),
26 > }
27 > }
28
29 // Signal wakes one goroutine waiting on this condition variable, if there is any.
30 > func (c *ConditionVariableImpl) Signal() { condition_variable_impl.go
31 > c.chanLock.Lock()
32 > defer c.chanLock.Unlock()
33 >
34 > select {
35 > case c.channel <- struct{}{}:
36 default:
37 // noop
55 func (c *ConditionVariableImpl) Wait(
56 interrupt <-chan struct{},
57 > ) { condition_variable_impl.go
58 >
59 > c.chanLock.Lock()
60 > channel := c.channel
61 > c.chanLock.Unlock()
62 >
63 > // user provided lock must be released after getting the above channel
64 > // so channel to be waited on < release of user lock < acquire of user lock < signal / broadcast
65 > // ```
66 > // That is, if another thread is able to acquire the mutex after the about-to-block
67 > // thread has released it, then a subsequent call to pthread_cond_signal() or
68 > // pthread_cond_broadcast() in that thread behaves as if it were issued after the
69 > // about-to-block thread has blocked.
70 > // ref: https://pubs.opengroup.org/onlinepubs/007908799/xsh/pthread_cond_wait.html
71 > // ```
72 >
73 > c.lock.Unlock()
74 > defer c.lock.Lock()
75 >
76 > select {
77 > case <-channel: condition_variable_impl.go
78 // received signal
79 case <-interrupt: