Add support for timeouts for locks (#4377)
Frank Wessels
7 years ago
committed by
deekoder
parent
93f126364e
commit
61e0b1454a
@ -0,0 +1,120 @@ |
||||
/* |
||||
* Minio Cloud Storage, (C) 2017 Minio, Inc. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package cmd |
||||
|
||||
import ( |
||||
"sync" |
||||
"sync/atomic" |
||||
"time" |
||||
) |
||||
|
||||
const ( |
||||
dynamicTimeoutIncreaseThresholdPct = 0.33 // Upper threshold for failures in order to increase timeout
|
||||
dynamicTimeoutDecreaseThresholdPct = 0.10 // Lower threshold for failures in order to decrease timeout
|
||||
dynamicTimeoutLogSize = 16 |
||||
maxDuration = time.Duration(1<<63 - 1) |
||||
) |
||||
|
||||
// timeouts that are dynamically adapted based on actual usage results
|
||||
type dynamicTimeout struct { |
||||
timeout int64 |
||||
minimum int64 |
||||
entries int64 |
||||
log [dynamicTimeoutLogSize]time.Duration |
||||
mutex sync.Mutex |
||||
} |
||||
|
||||
// newDynamicTimeout returns a new dynamic timeout initialized with timeout value
|
||||
func newDynamicTimeout(timeout, minimum time.Duration) *dynamicTimeout { |
||||
return &dynamicTimeout{timeout: int64(timeout), minimum: int64(minimum)} |
||||
} |
||||
|
||||
// Timeout returns the current timeout value
|
||||
func (dt *dynamicTimeout) Timeout() time.Duration { |
||||
return time.Duration(atomic.LoadInt64(&dt.timeout)) |
||||
} |
||||
|
||||
// LogSuccess logs the duration of a successful action that
|
||||
// did not hit the timeout
|
||||
func (dt *dynamicTimeout) LogSuccess(duration time.Duration) { |
||||
dt.logEntry(duration) |
||||
} |
||||
|
||||
// LogFailure logs an action that hit the timeout
|
||||
func (dt *dynamicTimeout) LogFailure() { |
||||
dt.logEntry(maxDuration) |
||||
} |
||||
|
||||
// logEntry stores a log entry
|
||||
func (dt *dynamicTimeout) logEntry(duration time.Duration) { |
||||
entries := int(atomic.AddInt64(&dt.entries, 1)) |
||||
index := entries - 1 |
||||
if index < dynamicTimeoutLogSize { |
||||
dt.mutex.Lock() |
||||
dt.log[index] = duration |
||||
dt.mutex.Unlock() |
||||
} |
||||
if entries == dynamicTimeoutLogSize { |
||||
dt.mutex.Lock() |
||||
|
||||
// Make copy on stack in order to call adjust()
|
||||
logCopy := [dynamicTimeoutLogSize]time.Duration{} |
||||
copy(logCopy[:], dt.log[:]) |
||||
|
||||
// reset log entries
|
||||
atomic.StoreInt64(&dt.entries, 0) |
||||
|
||||
dt.mutex.Unlock() |
||||
|
||||
dt.adjust(logCopy) |
||||
} |
||||
} |
||||
|
||||
// adjust changes the value of the dynamic timeout based on the
|
||||
// previous results
|
||||
func (dt *dynamicTimeout) adjust(entries [dynamicTimeoutLogSize]time.Duration) { |
||||
|
||||
failures, average := 0, 0 |
||||
for i := 0; i < len(entries); i++ { |
||||
if entries[i] == maxDuration { |
||||
failures++ |
||||
} else { |
||||
average += int(entries[i]) |
||||
} |
||||
} |
||||
if failures < len(entries) { |
||||
average /= len(entries) - failures |
||||
} |
||||
|
||||
timeOutHitPct := float64(failures) / float64(len(entries)) |
||||
|
||||
if timeOutHitPct > dynamicTimeoutIncreaseThresholdPct { |
||||
// We are hitting the timeout too often, so increase the timeout by 25%
|
||||
timeout := atomic.LoadInt64(&dt.timeout) * 125 / 100 |
||||
atomic.StoreInt64(&dt.timeout, timeout) |
||||
} else if timeOutHitPct < dynamicTimeoutDecreaseThresholdPct { |
||||
// We are hitting the timeout relatively few times, so decrease the timeout
|
||||
average = average * 125 / 100 // Add buffer of 25% on top of average
|
||||
|
||||
timeout := (atomic.LoadInt64(&dt.timeout) + int64(average)) / 2 // Middle between current timeout and average success
|
||||
if timeout < dt.minimum { |
||||
timeout = dt.minimum |
||||
} |
||||
atomic.StoreInt64(&dt.timeout, timeout) |
||||
} |
||||
|
||||
} |
||||
@ -0,0 +1,207 @@ |
||||
/* |
||||
* Minio Cloud Storage, (C) 2017 Minio, Inc. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package cmd |
||||
|
||||
import ( |
||||
"math/rand" |
||||
"testing" |
||||
"time" |
||||
) |
||||
|
||||
func TestDynamicTimeoutSingleIncrease(t *testing.T) { |
||||
|
||||
timeout := newDynamicTimeout(time.Minute, time.Second) |
||||
|
||||
initial := timeout.Timeout() |
||||
|
||||
for i := 0; i < dynamicTimeoutLogSize; i++ { |
||||
timeout.LogFailure() |
||||
} |
||||
|
||||
adjusted := timeout.Timeout() |
||||
|
||||
if initial >= adjusted { |
||||
t.Errorf("Failure to increase timeout, expected %v to be more than %v", adjusted, initial) |
||||
} |
||||
} |
||||
|
||||
func TestDynamicTimeoutDualIncrease(t *testing.T) { |
||||
|
||||
timeout := newDynamicTimeout(time.Minute, time.Second) |
||||
|
||||
initial := timeout.Timeout() |
||||
|
||||
for i := 0; i < dynamicTimeoutLogSize; i++ { |
||||
timeout.LogFailure() |
||||
} |
||||
|
||||
adjusted := timeout.Timeout() |
||||
|
||||
for i := 0; i < dynamicTimeoutLogSize; i++ { |
||||
timeout.LogFailure() |
||||
} |
||||
|
||||
adjustedAgain := timeout.Timeout() |
||||
|
||||
if initial >= adjusted || adjusted >= adjustedAgain { |
||||
t.Errorf("Failure to increase timeout multiple times") |
||||
} |
||||
} |
||||
|
||||
func TestDynamicTimeoutSingleDecrease(t *testing.T) { |
||||
|
||||
timeout := newDynamicTimeout(time.Minute, time.Second) |
||||
|
||||
initial := timeout.Timeout() |
||||
|
||||
for i := 0; i < dynamicTimeoutLogSize; i++ { |
||||
timeout.LogSuccess(20 * time.Second) |
||||
} |
||||
|
||||
adjusted := timeout.Timeout() |
||||
|
||||
if initial <= adjusted { |
||||
t.Errorf("Failure to decrease timeout, expected %v to be less than %v", adjusted, initial) |
||||
} |
||||
} |
||||
|
||||
func TestDynamicTimeoutDualDecrease(t *testing.T) { |
||||
|
||||
timeout := newDynamicTimeout(time.Minute, time.Second) |
||||
|
||||
initial := timeout.Timeout() |
||||
|
||||
for i := 0; i < dynamicTimeoutLogSize; i++ { |
||||
timeout.LogSuccess(20 * time.Second) |
||||
} |
||||
|
||||
adjusted := timeout.Timeout() |
||||
|
||||
for i := 0; i < dynamicTimeoutLogSize; i++ { |
||||
timeout.LogSuccess(20 * time.Second) |
||||
} |
||||
|
||||
adjustedAgain := timeout.Timeout() |
||||
|
||||
if initial <= adjusted || adjusted <= adjustedAgain { |
||||
t.Errorf("Failure to decrease timeout multiple times") |
||||
} |
||||
} |
||||
|
||||
func TestDynamicTimeoutManyDecreases(t *testing.T) { |
||||
|
||||
timeout := newDynamicTimeout(time.Minute, time.Second) |
||||
|
||||
initial := timeout.Timeout() |
||||
|
||||
const successTimeout = 20 * time.Second |
||||
for l := 0; l < 100; l++ { |
||||
for i := 0; i < dynamicTimeoutLogSize; i++ { |
||||
timeout.LogSuccess(successTimeout) |
||||
} |
||||
|
||||
} |
||||
|
||||
adjusted := timeout.Timeout() |
||||
// Check whether eventual timeout is between initial value and success timeout
|
||||
if initial <= adjusted || adjusted <= successTimeout { |
||||
t.Errorf("Failure to decrease timeout appropriately") |
||||
} |
||||
} |
||||
|
||||
func TestDynamicTimeoutHitMinimum(t *testing.T) { |
||||
|
||||
const minimum = 30 * time.Second |
||||
timeout := newDynamicTimeout(time.Minute, minimum) |
||||
|
||||
initial := timeout.Timeout() |
||||
|
||||
const successTimeout = 20 * time.Second |
||||
for l := 0; l < 100; l++ { |
||||
for i := 0; i < dynamicTimeoutLogSize; i++ { |
||||
timeout.LogSuccess(successTimeout) |
||||
} |
||||
} |
||||
|
||||
adjusted := timeout.Timeout() |
||||
// Check whether eventual timeout has hit the minimum value
|
||||
if initial <= adjusted || adjusted != minimum { |
||||
t.Errorf("Failure to decrease timeout appropriately") |
||||
} |
||||
} |
||||
|
||||
func testDynamicTimeoutAdjust(t *testing.T, timeout *dynamicTimeout, f func() float64) { |
||||
|
||||
const successTimeout = 20 * time.Second |
||||
|
||||
for i := 0; i < dynamicTimeoutLogSize; i++ { |
||||
|
||||
rnd := f() |
||||
duration := time.Duration(float64(successTimeout) * rnd) |
||||
|
||||
if duration < 100*time.Millisecond { |
||||
duration = 100 * time.Millisecond |
||||
} |
||||
if duration >= time.Minute { |
||||
timeout.LogFailure() |
||||
} else { |
||||
timeout.LogSuccess(duration) |
||||
} |
||||
} |
||||
} |
||||
|
||||
func TestDynamicTimeoutAdjustExponential(t *testing.T) { |
||||
|
||||
timeout := newDynamicTimeout(time.Minute, time.Second) |
||||
|
||||
rand.Seed(time.Now().UTC().UnixNano()) |
||||
|
||||
initial := timeout.Timeout() |
||||
|
||||
for try := 0; try < 10; try++ { |
||||
|
||||
testDynamicTimeoutAdjust(t, timeout, rand.ExpFloat64) |
||||
|
||||
} |
||||
|
||||
adjusted := timeout.Timeout() |
||||
if initial <= adjusted { |
||||
t.Errorf("Failure to decrease timeout, expected %v to be less than %v", adjusted, initial) |
||||
} |
||||
} |
||||
|
||||
func TestDynamicTimeoutAdjustNormalized(t *testing.T) { |
||||
|
||||
timeout := newDynamicTimeout(time.Minute, time.Second) |
||||
|
||||
rand.Seed(time.Now().UTC().UnixNano()) |
||||
|
||||
initial := timeout.Timeout() |
||||
|
||||
for try := 0; try < 10; try++ { |
||||
|
||||
testDynamicTimeoutAdjust(t, timeout, func() float64 { |
||||
return 1.0 + rand.NormFloat64() |
||||
}) |
||||
|
||||
} |
||||
|
||||
adjusted := timeout.Timeout() |
||||
if initial <= adjusted { |
||||
t.Errorf("Failure to decrease timeout, expected %v to be less than %v", adjusted, initial) |
||||
} |
||||
} |
||||
@ -0,0 +1,121 @@ |
||||
# lsync |
||||
|
||||
Local syncing package with support for timeouts. This package offers both a `sync.Mutex` and `sync.RWMutex` compatible interface. |
||||
|
||||
Additionally it provides `lsync.LFrequentAccess` which uses an atomic load and store of a consistently typed value. This can be usefull for shared data structures that are frequently read but infrequently updated (using an copy-on-write mechanism) without the need for protection with a regular mutex. |
||||
|
||||
### Example of LRWMutex |
||||
|
||||
```go |
||||
// Create RWMutex compatible mutex |
||||
lrwm := NewLRWMutex() |
||||
|
||||
// Try to get lock within timeout |
||||
if !lrwm.GetLock(1000 * time.Millisecond) { |
||||
fmt.Println("Timeout occured") |
||||
return |
||||
} |
||||
|
||||
// Acquired lock, do your stuff ... |
||||
|
||||
lrwm.Unlock() // Release lock |
||||
``` |
||||
|
||||
### Example of LFrequentAccess |
||||
````go |
||||
type Map map[string]string |
||||
|
||||
// Create new LFrequentAccess for type Map |
||||
freqaccess := NewLFrequentAccess(make(Map)) |
||||
|
||||
cur := freqaccess.LockBeforeSet().(Map) // Lock in order to update |
||||
mp := make(Map) // Create new Map |
||||
for k, v := range cur { // Copy over old contents |
||||
mp[k] = v |
||||
} |
||||
mp[key] = val // Add new value |
||||
freqaccess.SetNewCopyAndUnlock(mp) // Exchange old version of map with new version |
||||
|
||||
mpReadOnly := freqaccess.ReadOnlyAccess().(Map) // Get read only access to Map |
||||
fmt.Println(mpReadOnly[key]) // Safe access with no further synchronization |
||||
```` |
||||
|
||||
## Design |
||||
|
||||
The design is pretty straightforward in the sense that `lsync` tries to get a lock in a loop with an exponential [backoff](https://www.awsarchitectureblog.com/2015/03/backoff.html) algorithm. The algorithm is configurable in terms of initial delay and jitter. |
||||
|
||||
If the lock is acquired before the timeout has occured, it will return success to the caller and the caller can proceed as intended. The caller must call `unlock` after the operation that is to be protected has completed in order to release the lock. |
||||
|
||||
When more time has elapsed than the timeout value the lock loop will cancel out and signal back to the caller that the lock has not been acquired. In this case the caller must _not_ call `unlock` since no lock was obtained. Typically it should signal an error back up the call stack so that errors can be dealt with appropriately at the correct level. |
||||
|
||||
Note that this algorithm is not 'real-time' in the sense that it will time out exactly at the timeout value, but instead a (short) while after the timeout has lapsed. It is even possible that (in edge cases) a succesful lock can be returned a very short time after the timeout has lapsed. |
||||
|
||||
## API |
||||
|
||||
#### LMutex |
||||
|
||||
```go |
||||
func (lm *LMutex) Lock() |
||||
func (lm *LMutex) GetLock(timeout time.Duration) (locked bool) |
||||
func (lm *LMutex) Unlock() |
||||
``` |
||||
|
||||
#### LRWMutex |
||||
|
||||
```go |
||||
func (lm *LRWMutex) Lock() |
||||
func (lm *LRWMutex) GetLock(timeout time.Duration) (locked bool) |
||||
func (lm *LRWMutex) RLock() |
||||
func (lm *LRWMutex) GetRLock(timeout time.Duration) (locked bool) |
||||
func (lm *LRWMutex) Unlock() |
||||
func (lm *LRWMutex) RUnlock() |
||||
``` |
||||
|
||||
#### LFrequentAccess |
||||
```go |
||||
func (lm *LFrequentAccess) ReadOnlyAccess() (constReadOnly interface{}) |
||||
func (lm *LFrequentAccess) LockBeforeSet() (constCurVersion interface{}) |
||||
func (lm *LFrequentAccess) SetNewCopyAndUnlock(newCopy interface{}) |
||||
``` |
||||
|
||||
## Benchmarks |
||||
|
||||
### sync.Mutex vs lsync.LMutex |
||||
|
||||
(with `defaultRetryUnit` and `defaultRetryCap` at 10 microsec) |
||||
|
||||
``` |
||||
BenchmarkMutex-8 111 1579 +1322.52% |
||||
BenchmarkMutexSlack-8 120 1033 +760.83% |
||||
BenchmarkMutexWork-8 133 1604 +1106.02% |
||||
BenchmarkMutexWorkSlack-8 137 1038 +657.66% |
||||
``` |
||||
|
||||
(with `defaultRetryUnit` and `defaultRetryCap` at 1 millisec) |
||||
``` |
||||
benchmark old ns/op new ns/op delta |
||||
BenchmarkMutex-8 111 2649 +2286.49% |
||||
BenchmarkMutexSlack-8 120 1719 +1332.50% |
||||
BenchmarkMutexWork-8 133 2637 +1882.71% |
||||
BenchmarkMutexWorkSlack-8 137 1729 +1162.04% |
||||
``` |
||||
|
||||
(with `defaultRetryUnit` and `defaultRetryCap` at 100 millisec) |
||||
|
||||
``` |
||||
benchmark old ns/op new ns/op delta |
||||
BenchmarkMutex-8 111 2649 +2286.49% |
||||
BenchmarkMutexSlack-8 120 2478 +1965.00% |
||||
BenchmarkMutexWork-8 133 2547 +1815.04% |
||||
BenchmarkMutexWorkSlack-8 137 2683 +1858.39% |
||||
``` |
||||
|
||||
### LFrequentAccess |
||||
|
||||
An `lsync.LFrequentAccess` provides an atomic load and store of a consistently typed value. |
||||
|
||||
``` |
||||
benchmark old ns/op new ns/op delta |
||||
BenchmarkLFrequentAccessMap-8 114 4.67 -95.90% |
||||
BenchmarkLFrequentAccessSlice-8 109 5.95 -94.54% |
||||
``` |
||||
@ -0,0 +1,64 @@ |
||||
/* |
||||
* Minio Cloud Storage, (C) 2017 Minio, Inc. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package lsync |
||||
|
||||
import ( |
||||
"sync" |
||||
"sync/atomic" |
||||
) |
||||
|
||||
// LFrequentAccess is a synchronization mechanism for frequently read yet
|
||||
// infrequently updated data structures. It uses a copy-on-write paradigm
|
||||
// for updates to the data.
|
||||
type LFrequentAccess struct { |
||||
state atomic.Value |
||||
writeLock sync.Mutex |
||||
locked bool |
||||
} |
||||
|
||||
// NewLFrequentAccess - initializes a new LFrequentAccess.
|
||||
func NewLFrequentAccess(x interface{}) *LFrequentAccess { |
||||
lm := &LFrequentAccess{} |
||||
lm.state.Store(x) |
||||
return lm |
||||
} |
||||
|
||||
// ReadOnlyAccess returns the data intented for reads without further synchronization
|
||||
func (lm *LFrequentAccess) ReadOnlyAccess() (constReadOnly interface{}) { |
||||
return lm.state.Load() |
||||
} |
||||
|
||||
// LockBeforeSet must be called before updates of the data in order to synchronize
|
||||
// with other potential writers. It returns the current version of the data that
|
||||
// needs to be copied over into a new version.
|
||||
func (lm *LFrequentAccess) LockBeforeSet() (constCurVersion interface{}) { |
||||
lm.writeLock.Lock() |
||||
lm.locked = true |
||||
return lm.state.Load() |
||||
} |
||||
|
||||
// SetNewCopyAndUnlock updates the data with a new modified copy and unlocks
|
||||
// simultaneously. Make sure to call LockBeforeSet beforehand to synchronize
|
||||
// between potential parallel writers (and not lose any updated information).
|
||||
func (lm *LFrequentAccess) SetNewCopyAndUnlock(newCopy interface{}) { |
||||
if !lm.locked { |
||||
panic("SetNewCopyAndUnlock: locked state is false (did you call LockBeforeSet?)") |
||||
} |
||||
lm.state.Store(newCopy) |
||||
lm.locked = false |
||||
lm.writeLock.Unlock() |
||||
} |
||||
@ -0,0 +1,78 @@ |
||||
/* |
||||
* Minio Cloud Storage, (C) 2017 Minio, Inc. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package lsync |
||||
|
||||
import ( |
||||
"sync/atomic" |
||||
"time" |
||||
) |
||||
|
||||
// A LMutex is a mutual exclusion lock with timeouts.
|
||||
type LMutex struct { |
||||
state int64 |
||||
} |
||||
|
||||
// NewLMutex - initializes a new lsync mutex.
|
||||
func NewLMutex() *LMutex { |
||||
return &LMutex{} |
||||
} |
||||
|
||||
// Lock holds a lock on lm.
|
||||
//
|
||||
// If the lock is already in use, the calling go routine
|
||||
// blocks until the mutex is available.
|
||||
func (lm *LMutex) Lock() { |
||||
lm.lockLoop(time.Duration(1<<63 - 1)) |
||||
} |
||||
|
||||
// GetLock tries to get a write lock on lm before the timeout occurs.
|
||||
func (lm *LMutex) GetLock(timeout time.Duration) (locked bool) { |
||||
return lm.lockLoop(timeout) |
||||
} |
||||
|
||||
// lockLoop will acquire either a read or a write lock
|
||||
//
|
||||
// The call will block until the lock is granted using a built-in
|
||||
// timing randomized back-off algorithm to try again until successful
|
||||
func (lm *LMutex) lockLoop(timeout time.Duration) bool { |
||||
doneCh, start := make(chan struct{}), time.Now().UTC() |
||||
defer close(doneCh) |
||||
|
||||
// We timed out on the previous lock, incrementally wait
|
||||
// for a longer back-off time and try again afterwards.
|
||||
for range newRetryTimerSimple(doneCh) { |
||||
|
||||
// Try to acquire the lock.
|
||||
if atomic.CompareAndSwapInt64(&lm.state, NOLOCKS, WRITELOCK) { |
||||
return true |
||||
} else if time.Now().UTC().Sub(start) >= timeout { // Are we past the timeout?
|
||||
break |
||||
} |
||||
// We timed out on the previous lock, incrementally wait
|
||||
// for a longer back-off time and try again afterwards.
|
||||
} |
||||
return false |
||||
} |
||||
|
||||
// Unlock unlocks the lock.
|
||||
//
|
||||
// It is a run-time error if lm is not locked on entry to Unlock.
|
||||
func (lm *LMutex) Unlock() { |
||||
if !atomic.CompareAndSwapInt64(&lm.state, WRITELOCK, NOLOCKS) { |
||||
panic("Trying to Unlock() while no Lock() is active") |
||||
} |
||||
} |
||||
@ -0,0 +1,180 @@ |
||||
/* |
||||
* Minio Cloud Storage, (C) 2017 Minio, Inc. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package lsync |
||||
|
||||
import ( |
||||
"sync" |
||||
"time" |
||||
) |
||||
|
||||
const ( |
||||
WRITELOCK = -1 + iota |
||||
NOLOCKS |
||||
READLOCKS |
||||
) |
||||
|
||||
// A LRWMutex is a mutual exclusion lock with timeouts.
|
||||
type LRWMutex struct { |
||||
state int64 |
||||
m sync.Mutex // Mutex to prevent multiple simultaneous locks
|
||||
} |
||||
|
||||
// NewLRWMutex - initializes a new lsync RW mutex.
|
||||
func NewLRWMutex() *LRWMutex { |
||||
return &LRWMutex{} |
||||
} |
||||
|
||||
// Lock holds a write lock on lm.
|
||||
//
|
||||
// If the lock is already in use, the calling go routine
|
||||
// blocks until the mutex is available.
|
||||
func (lm *LRWMutex) Lock() { |
||||
|
||||
isWriteLock := true |
||||
lm.lockLoop(time.Duration(1<<63-1), isWriteLock) |
||||
} |
||||
|
||||
// GetLock tries to get a write lock on lm before the timeout occurs.
|
||||
func (lm *LRWMutex) GetLock(timeout time.Duration) (locked bool) { |
||||
|
||||
isWriteLock := true |
||||
return lm.lockLoop(timeout, isWriteLock) |
||||
} |
||||
|
||||
// RLock holds a read lock on lm.
|
||||
//
|
||||
// If one or more read lock are already in use, it will grant another lock.
|
||||
// Otherwise the calling go routine blocks until the mutex is available.
|
||||
func (lm *LRWMutex) RLock() { |
||||
|
||||
isWriteLock := false |
||||
lm.lockLoop(time.Duration(1<<63-1), isWriteLock) |
||||
} |
||||
|
||||
// GetRLock tries to get a read lock on lm before the timeout occurs.
|
||||
func (lm *LRWMutex) GetRLock(timeout time.Duration) (locked bool) { |
||||
|
||||
isWriteLock := false |
||||
return lm.lockLoop(timeout, isWriteLock) |
||||
} |
||||
|
||||
// lockLoop will acquire either a read or a write lock
|
||||
//
|
||||
// The call will block until the lock is granted using a built-in
|
||||
// timing randomized back-off algorithm to try again until successful
|
||||
func (lm *LRWMutex) lockLoop(timeout time.Duration, isWriteLock bool) bool { |
||||
doneCh, start := make(chan struct{}), time.Now().UTC() |
||||
defer close(doneCh) |
||||
|
||||
// We timed out on the previous lock, incrementally wait
|
||||
// for a longer back-off time and try again afterwards.
|
||||
for range newRetryTimerSimple(doneCh) { |
||||
|
||||
// Try to acquire the lock.
|
||||
var success bool |
||||
{ |
||||
lm.m.Lock() |
||||
|
||||
if isWriteLock { |
||||
if lm.state == NOLOCKS { |
||||
lm.state = WRITELOCK |
||||
success = true |
||||
} |
||||
} else { |
||||
if lm.state != WRITELOCK { |
||||
lm.state += 1 |
||||
success = true |
||||
} |
||||
} |
||||
|
||||
lm.m.Unlock() |
||||
} |
||||
if success { |
||||
return true |
||||
} |
||||
if time.Now().UTC().Sub(start) >= timeout { // Are we past the timeout?
|
||||
break |
||||
} |
||||
// We timed out on the previous lock, incrementally wait
|
||||
// for a longer back-off time and try again afterwards.
|
||||
} |
||||
return false |
||||
} |
||||
|
||||
// Unlock unlocks the write lock.
|
||||
//
|
||||
// It is a run-time error if lm is not locked on entry to Unlock.
|
||||
func (lm *LRWMutex) Unlock() { |
||||
|
||||
isWriteLock := true |
||||
success := lm.unlock(isWriteLock) |
||||
if !success { |
||||
panic("Trying to Unlock() while no Lock() is active") |
||||
} |
||||
} |
||||
|
||||
// RUnlock releases a read lock held on lm.
|
||||
//
|
||||
// It is a run-time error if lm is not locked on entry to RUnlock.
|
||||
func (lm *LRWMutex) RUnlock() { |
||||
|
||||
isWriteLock := false |
||||
success := lm.unlock(isWriteLock) |
||||
if !success { |
||||
panic("Trying to RUnlock() while no RLock() is active") |
||||
} |
||||
} |
||||
|
||||
func (lm *LRWMutex) unlock(isWriteLock bool) (unlocked bool) { |
||||
lm.m.Lock() |
||||
|
||||
// Try to release lock.
|
||||
if isWriteLock { |
||||
if lm.state == WRITELOCK { |
||||
lm.state = NOLOCKS |
||||
unlocked = true |
||||
} |
||||
} else { |
||||
if lm.state == WRITELOCK || lm.state == NOLOCKS { |
||||
unlocked = false // unlocked called without any active read locks
|
||||
} else { |
||||
lm.state -= 1 |
||||
unlocked = true |
||||
} |
||||
} |
||||
|
||||
lm.m.Unlock() |
||||
return unlocked |
||||
} |
||||
|
||||
// ForceUnlock will forcefully clear a write or read lock.
|
||||
func (lm *LRWMutex) ForceUnlock() { |
||||
lm.m.Lock() |
||||
lm.state = NOLOCKS |
||||
lm.m.Unlock() |
||||
} |
||||
|
||||
// DRLocker returns a sync.Locker interface that implements
|
||||
// the Lock and Unlock methods by calling drw.RLock and drw.RUnlock.
|
||||
func (dm *LRWMutex) DRLocker() sync.Locker { |
||||
return (*drlocker)(dm) |
||||
} |
||||
|
||||
type drlocker LRWMutex |
||||
|
||||
func (dr *drlocker) Lock() { (*LRWMutex)(dr).RLock() } |
||||
func (dr *drlocker) Unlock() { (*LRWMutex)(dr).RUnlock() } |
||||
@ -0,0 +1,142 @@ |
||||
/* |
||||
* Minio Cloud Storage, (C) 2017 Minio, Inc. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package lsync |
||||
|
||||
import ( |
||||
"math/rand" |
||||
"sync" |
||||
"time" |
||||
) |
||||
|
||||
// lockedRandSource provides protected rand source, implements rand.Source interface.
|
||||
type lockedRandSource struct { |
||||
lk sync.Mutex |
||||
src rand.Source |
||||
} |
||||
|
||||
// Int63 returns a non-negative pseudo-random 63-bit integer as an
|
||||
// int64.
|
||||
func (r *lockedRandSource) Int63() (n int64) { |
||||
r.lk.Lock() |
||||
n = r.src.Int63() |
||||
r.lk.Unlock() |
||||
return |
||||
} |
||||
|
||||
// Seed uses the provided seed value to initialize the generator to a
|
||||
// deterministic state.
|
||||
func (r *lockedRandSource) Seed(seed int64) { |
||||
r.lk.Lock() |
||||
r.src.Seed(seed) |
||||
r.lk.Unlock() |
||||
} |
||||
|
||||
// MaxJitter will randomize over the full exponential backoff time
|
||||
const MaxJitter = 1.0 |
||||
|
||||
// NoJitter disables the use of jitter for randomizing the
|
||||
// exponential backoff time
|
||||
const NoJitter = 0.0 |
||||
|
||||
// Global random source for fetching random values.
|
||||
var globalRandomSource = rand.New(&lockedRandSource{ |
||||
src: rand.NewSource(time.Now().UTC().UnixNano()), |
||||
}) |
||||
|
||||
// newRetryTimerJitter creates a timer with exponentially increasing delays
|
||||
// until the maximum retry attempts are reached. - this function is a fully
|
||||
// configurable version, meant for only advanced use cases. For the most part
|
||||
// one should use newRetryTimerSimple and newRetryTimer.
|
||||
func newRetryTimerWithJitter(unit time.Duration, cap time.Duration, jitter float64, doneCh chan struct{}) <-chan int { |
||||
attemptCh := make(chan int) |
||||
|
||||
// normalize jitter to the range [0, 1.0]
|
||||
if jitter < NoJitter { |
||||
jitter = NoJitter |
||||
} |
||||
if jitter > MaxJitter { |
||||
jitter = MaxJitter |
||||
} |
||||
|
||||
// computes the exponential backoff duration according to
|
||||
// https://www.awsarchitectureblog.com/2015/03/backoff.html
|
||||
exponentialBackoffWait := func(attempt int) time.Duration { |
||||
// 1<<uint(attempt) below could overflow, so limit the value of attempt
|
||||
maxAttempt := 30 |
||||
if attempt > maxAttempt { |
||||
attempt = maxAttempt |
||||
} |
||||
//sleep = random_between(0, min(cap, base * 2 ** attempt))
|
||||
sleep := unit * time.Duration(1<<uint(attempt)) |
||||
if sleep > cap { |
||||
sleep = cap |
||||
} |
||||
if jitter != NoJitter { |
||||
sleep -= time.Duration(globalRandomSource.Float64() * float64(sleep) * jitter) |
||||
} |
||||
return sleep |
||||
} |
||||
|
||||
go func() { |
||||
defer close(attemptCh) |
||||
nextBackoff := 0 |
||||
// Channel used to signal after the expiry of backoff wait seconds.
|
||||
var timer *time.Timer |
||||
for { |
||||
select { // Attempts starts.
|
||||
case attemptCh <- nextBackoff: |
||||
nextBackoff++ |
||||
case <-doneCh: |
||||
// Stop the routine.
|
||||
return |
||||
} |
||||
timer = time.NewTimer(exponentialBackoffWait(nextBackoff)) |
||||
// wait till next backoff time or till doneCh gets a message.
|
||||
select { |
||||
case <-timer.C: |
||||
case <-doneCh: |
||||
// stop the timer and return.
|
||||
timer.Stop() |
||||
return |
||||
} |
||||
|
||||
} |
||||
}() |
||||
|
||||
// Start reading..
|
||||
return attemptCh |
||||
} |
||||
|
||||
// Default retry constants.
|
||||
const ( |
||||
defaultRetryUnit = 10 * time.Millisecond // 10 millisecond.
|
||||
defaultRetryCap = 10 * time.Millisecond // 10 millisecond.
|
||||
) |
||||
|
||||
// newRetryTimer creates a timer with exponentially increasing delays
|
||||
// until the maximum retry attempts are reached. - this function provides
|
||||
// resulting retry values to be of maximum jitter.
|
||||
func newRetryTimer(unit time.Duration, cap time.Duration, doneCh chan struct{}) <-chan int { |
||||
return newRetryTimerWithJitter(unit, cap, MaxJitter, doneCh) |
||||
} |
||||
|
||||
// newRetryTimerSimple creates a timer with exponentially increasing delays
|
||||
// until the maximum retry attempts are reached. - this function is a
|
||||
// simpler version with all default values.
|
||||
func newRetryTimerSimple(doneCh chan struct{}) <-chan int { |
||||
return newRetryTimerWithJitter(defaultRetryUnit, defaultRetryCap, MaxJitter, doneCh) |
||||
} |
||||
Loading…
Reference in new issue