Repository navigation
Expand file tree
/
Copy pathdrwmutex.go
More file actions
725 lines (617 loc) · 19.7 KB
/
Copy pathdrwmutex.go
File metadata and controls
725 lines (617 loc) · 19.7 KB
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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
// Copyright (c) 2015-2021 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package dsync
import (
"context"
"errors"
"math/rand"
"sort"
"strconv"
"sync"
"time"
"github.com/minio/minio/internal/mcontext"
"github.com/minio/pkg/v2/console"
"github.com/minio/pkg/v2/env"
)
// Indicator if logging is enabled.
var dsyncLog bool
// Retry unit interval
var lockRetryMinInterval time.Duration
var lockRetryBackOff func(*rand.Rand, uint) time.Duration
func init() {
// Check for MINIO_DSYNC_TRACE env variable, if set logging will be enabled for failed REST operations.
dsyncLog = env.Get("_MINIO_DSYNC_TRACE", "0") == "1"
lockRetryMinInterval = 250 * time.Millisecond
if lri := env.Get("_MINIO_LOCK_RETRY_INTERVAL", ""); lri != "" {
v, err := strconv.Atoi(lri)
if err != nil {
panic(err)
}
lockRetryMinInterval = time.Duration(v) * time.Millisecond
}
lockRetryBackOff = backoffWait(
lockRetryMinInterval,
100*time.Millisecond,
5*time.Second,
)
}
func log(format string, data ...interface{}) {
if dsyncLog {
console.Printf(format, data...)
}
}
const (
// dRWMutexAcquireTimeout - default tolerance limit to wait for lock acquisition before.
drwMutexAcquireTimeout = 1 * time.Second // 1 second.
// dRWMutexRefreshTimeout - default timeout for the refresh call
drwMutexRefreshCallTimeout = 5 * time.Second
// dRWMutexUnlockTimeout - default timeout for the unlock call
drwMutexUnlockCallTimeout = 30 * time.Second
// dRWMutexForceUnlockTimeout - default timeout for the unlock call
drwMutexForceUnlockCallTimeout = 30 * time.Second
// dRWMutexRefreshInterval - default the interval between two refresh calls
drwMutexRefreshInterval = 10 * time.Second
drwMutexInfinite = 1<<63 - 1
)
// Timeouts are timeouts for specific operations.
type Timeouts struct {
// Acquire - tolerance limit to wait for lock acquisition before.
Acquire time.Duration
// RefreshCall - timeout for the refresh call
RefreshCall time.Duration
// UnlockCall - timeout for the unlock call
UnlockCall time.Duration
// ForceUnlockCall - timeout for the force unlock call
ForceUnlockCall time.Duration
}
// DefaultTimeouts contains default timeouts.
var DefaultTimeouts = Timeouts{
Acquire: drwMutexAcquireTimeout,
RefreshCall: drwMutexRefreshCallTimeout,
UnlockCall: drwMutexUnlockCallTimeout,
ForceUnlockCall: drwMutexForceUnlockCallTimeout,
}
// A DRWMutex is a distributed mutual exclusion lock.
type DRWMutex struct {
Names []string
writeLocks []string // Array of nodes that granted a write lock
readLocks []string // Array of array of nodes that granted reader locks
rng *rand.Rand
m sync.Mutex // Mutex to prevent multiple simultaneous locks from this node
clnt *Dsync
cancelRefresh context.CancelFunc
refreshInterval time.Duration
lockRetryMinInterval time.Duration
}
// Granted - represents a structure of a granted lock.
type Granted struct {
index int
lockUID string // Locked if set with UID string, unlocked if empty
}
func (g *Granted) isLocked() bool {
return isLocked(g.lockUID)
}
func isLocked(uid string) bool {
return len(uid) > 0
}
// NewDRWMutex - initializes a new dsync RW mutex.
func NewDRWMutex(clnt *Dsync, names ...string) *DRWMutex {
restClnts, _ := clnt.GetLockers()
sort.Strings(names)
return &DRWMutex{
writeLocks: make([]string, len(restClnts)),
readLocks: make([]string, len(restClnts)),
Names: names,
clnt: clnt,
rng: rand.New(&lockedRandSource{src: rand.NewSource(time.Now().UTC().UnixNano())}),
refreshInterval: drwMutexRefreshInterval,
lockRetryMinInterval: lockRetryMinInterval,
}
}
// Lock holds a write lock on dm.
//
// If the lock is already in use, the calling go routine
// blocks until the mutex is available.
func (dm *DRWMutex) Lock(id, source string) {
isReadLock := false
dm.lockBlocking(context.Background(), nil, id, source, isReadLock, Options{
Timeout: drwMutexInfinite,
})
}
// Options lock options.
type Options struct {
Timeout time.Duration
RetryInterval time.Duration
}
// GetLock tries to get a write lock on dm before the timeout elapses.
//
// If the lock is already in use, the calling go routine
// blocks until either the mutex becomes available and return success or
// more time has passed than the timeout value and return false.
func (dm *DRWMutex) GetLock(ctx context.Context, cancel context.CancelFunc, id, source string, opts Options) (locked bool) {
isReadLock := false
return dm.lockBlocking(ctx, cancel, id, source, isReadLock, opts)
}
// RLock holds a read lock on dm.
//
// If one or more read locks are already in use, it will grant another lock.
// Otherwise the calling go routine blocks until the mutex is available.
func (dm *DRWMutex) RLock(id, source string) {
isReadLock := true
dm.lockBlocking(context.Background(), nil, id, source, isReadLock, Options{
Timeout: drwMutexInfinite,
})
}
// GetRLock tries to get a read lock on dm before the timeout elapses.
//
// If one or more read locks are already in use, it will grant another lock.
// Otherwise the calling go routine blocks until either the mutex becomes
// available and return success or more time has passed than the timeout
// value and return false.
func (dm *DRWMutex) GetRLock(ctx context.Context, cancel context.CancelFunc, id, source string, opts Options) (locked bool) {
isReadLock := true
return dm.lockBlocking(ctx, cancel, id, source, isReadLock, opts)
}
// lockBlocking will try to acquire either a read or a write lock
//
// The function will loop using a built-in timing randomized back-off
// algorithm until either the lock is acquired successfully or more
// time has elapsed than the timeout value.
func (dm *DRWMutex) lockBlocking(ctx context.Context, lockLossCallback func(), id, source string, isReadLock bool, opts Options) (locked bool) {
restClnts, _ := dm.clnt.GetLockers()
// Create lock array to capture the successful lockers
locks := make([]string, len(restClnts))
// Add total timeout
ctx, cancel := context.WithTimeout(ctx, opts.Timeout)
defer cancel()
// Tolerance is not set, defaults to half of the locker clients.
tolerance := len(restClnts) / 2
// Quorum is effectively = total clients subtracted with tolerance limit
quorum := len(restClnts) - tolerance
if !isReadLock {
// In situations for write locks, as a special case
// to avoid split brains we make sure to acquire
// quorum + 1 when tolerance is exactly half of the
// total locker clients.
if quorum == tolerance {
quorum++
}
}
log("lockBlocking %s/%s for %#v: lockType readLock(%t), additional opts: %#v, quorum: %d, tolerance: %d, lockClients: %d\n", id, source, dm.Names, isReadLock, opts, quorum, tolerance, len(restClnts))
tolerance = len(restClnts) - quorum
attempt := uint(0)
for {
select {
case <-ctx.Done():
return false
default:
// Try to acquire the lock.
if locked = lock(ctx, dm.clnt, &locks, id, source, isReadLock, tolerance, quorum, dm.Names...); locked {
dm.m.Lock()
// If success, copy array to object
if isReadLock {
copy(dm.readLocks, locks)
} else {
copy(dm.writeLocks, locks)
}
dm.m.Unlock()
log("lockBlocking %s/%s for %#v: granted\n", id, source, dm.Names)
// Refresh lock continuously and cancel if there is no quorum in the lock anymore
dm.startContinousLockRefresh(lockLossCallback, id, source, quorum)
return locked
}
switch {
case opts.RetryInterval < 0:
return false
case opts.RetryInterval > 0:
time.Sleep(opts.RetryInterval)
default:
attempt++
time.Sleep(lockRetryBackOff(dm.rng, attempt))
}
}
}
}
func (dm *DRWMutex) startContinousLockRefresh(lockLossCallback func(), id, source string, quorum int) {
ctx, cancel := context.WithCancel(context.Background())
dm.m.Lock()
dm.cancelRefresh = cancel
dm.m.Unlock()
go func() {
defer cancel()
refreshTimer := time.NewTimer(dm.refreshInterval)
defer refreshTimer.Stop()
for {
select {
case <-ctx.Done():
return
case <-refreshTimer.C:
noQuorum, err := refreshLock(ctx, dm.clnt, id, source, quorum)
if err == nil && noQuorum {
// Clean the lock locally and in remote nodes
forceUnlock(ctx, dm.clnt, id)
// Execute the caller lock loss callback
if lockLossCallback != nil {
lockLossCallback()
}
return
}
refreshTimer.Reset(dm.refreshInterval)
}
}
}()
}
func forceUnlock(ctx context.Context, ds *Dsync, id string) {
ctx, cancel := context.WithTimeout(ctx, ds.Timeouts.ForceUnlockCall)
defer cancel()
restClnts, _ := ds.GetLockers()
args := LockArgs{
UID: id,
}
var wg sync.WaitGroup
for index, c := range restClnts {
wg.Add(1)
// Send refresh request to all nodes
go func(index int, c NetLocker) {
defer wg.Done()
c.ForceUnlock(ctx, args)
}(index, c)
}
wg.Wait()
}
type refreshResult struct {
offline bool
refreshed bool
}
// Refresh the given lock in all nodes, return true to indicate if a lock
// does not exist in enough quorum nodes.
func refreshLock(ctx context.Context, ds *Dsync, id, source string, quorum int) (bool, error) {
restClnts, _ := ds.GetLockers()
// Create buffered channel of size equal to total number of nodes.
ch := make(chan refreshResult, len(restClnts))
var wg sync.WaitGroup
args := LockArgs{
UID: id,
}
for index, c := range restClnts {
wg.Add(1)
// Send refresh request to all nodes
go func(index int, c NetLocker) {
defer wg.Done()
if c == nil {
ch <- refreshResult{offline: true}
return
}
ctx, cancel := context.WithTimeout(ctx, ds.Timeouts.RefreshCall)
defer cancel()
refreshed, err := c.Refresh(ctx, args)
if err != nil {
ch <- refreshResult{offline: true}
log("dsync: Unable to call Refresh failed with %s for %#v at %s\n", err, args, c)
} else {
ch <- refreshResult{refreshed: refreshed}
log("dsync: Refresh returned false for %#v at %s\n", args, c)
}
}(index, c)
}
// Wait until we have either
//
// a) received all refresh responses
// b) received too many refreshed for quorum to be still possible
// c) timed out
//
lockNotFound, lockRefreshed := 0, 0
done := false
for i := 0; i < len(restClnts); i++ {
select {
case refreshResult := <-ch:
if refreshResult.offline {
continue
}
if refreshResult.refreshed {
lockRefreshed++
} else {
lockNotFound++
}
if lockRefreshed >= quorum || lockNotFound > len(restClnts)-quorum {
done = true
}
case <-ctx.Done():
// Refreshing is canceled
return false, ctx.Err()
}
if done {
break
}
}
// We may have some unused results in ch, release them async.
go func() {
wg.Wait()
close(ch)
for range ch {
}
}()
noQuorum := lockNotFound > len(restClnts)-quorum
return noQuorum, nil
}
// lock tries to acquire the distributed lock, returning true or false.
func lock(ctx context.Context, ds *Dsync, locks *[]string, id, source string, isReadLock bool, tolerance, quorum int, names ...string) bool {
for i := range *locks {
(*locks)[i] = ""
}
restClnts, owner := ds.GetLockers()
// Create buffered channel of size equal to total number of nodes.
ch := make(chan Granted, len(restClnts))
var wg sync.WaitGroup
args := LockArgs{
Owner: owner,
UID: id,
Resources: names,
Source: source,
Quorum: quorum,
}
// Combined timeout for the lock attempt.
ctx, cancel := context.WithTimeout(ctx, ds.Timeouts.Acquire)
defer cancel()
// Special context for NetLockers - do not use timeouts.
// Also, pass the trace context info if found for debugging
netLockCtx := context.Background()
tc, ok := ctx.Value(mcontext.ContextTraceKey).(*mcontext.TraceCtxt)
if ok {
netLockCtx = context.WithValue(netLockCtx, mcontext.ContextTraceKey, tc)
}
for index, c := range restClnts {
wg.Add(1)
// broadcast lock request to all nodes
go func(index int, isReadLock bool, c NetLocker) {
defer wg.Done()
g := Granted{index: index}
if c == nil {
log("dsync: nil locker\n")
ch <- g
return
}
var locked bool
var err error
if isReadLock {
if locked, err = c.RLock(netLockCtx, args); err != nil {
log("dsync: Unable to call RLock failed with %s for %#v at %s\n", err, args, c)
}
} else {
if locked, err = c.Lock(netLockCtx, args); err != nil {
log("dsync: Unable to call Lock failed with %s for %#v at %s\n", err, args, c)
}
}
if locked {
g.lockUID = args.UID
}
ch <- g
}(index, isReadLock, c)
}
// Wait until we have either
//
// a) received all lock responses
// b) received too many 'non-'locks for quorum to be still possible
// c) timed out
//
i, locksFailed := 0, 0
done := false
for ; i < len(restClnts); i++ { // Loop until we acquired all locks
select {
case grant := <-ch:
if grant.isLocked() {
// Mark that this node has acquired the lock
(*locks)[grant.index] = grant.lockUID
} else {
locksFailed++
if locksFailed > tolerance {
// We know that we are not going to get the lock anymore,
// so exit out and release any locks that did get acquired
done = true
}
}
case <-ctx.Done():
// Capture timedout locks as failed or took too long
locksFailed++
if locksFailed > tolerance {
// We know that we are not going to get the lock anymore,
// so exit out and release any locks that did get acquired
done = true
}
}
if done {
break
}
}
quorumLocked := checkQuorumLocked(locks, quorum) && locksFailed <= tolerance
if !quorumLocked {
log("dsync: Unable to acquire lock in quorum %#v\n", args)
// Release all acquired locks without quorum.
if !releaseAll(ctx, ds, tolerance, owner, locks, isReadLock, restClnts, names...) {
log("Unable to release acquired locks, these locks will expire automatically %#v\n", args)
}
}
// We may have some unused results in ch,
9798
release them async.
go func() {
wg.Wait()
close(ch)
for grantToBeReleased := range ch {
if grantToBeReleased.isLocked() {
// release abandoned lock
log("Releasing abandoned lock\n")
sendRelease(ctx, ds, restClnts[grantToBeReleased.index],
owner, grantToBeReleased.lockUID, isReadLock, names...)
}
}
}()
return quorumLocked
}
// checkFailedUnlocks determines whether we have sufficiently unlocked all
// resources to ensure no deadlocks for future callers
func checkFailedUnlocks(locks []string, tolerance int) bool {
unlocksFailed := 0
for lockID := range locks {
if isLocked(locks[lockID]) {
unlocksFailed++
}
}
// Unlock failures are higher than tolerance limit
// for this instance of unlocker, we should let the
// caller know that lock is not successfully released
// yet.
if len(locks)-tolerance == tolerance {
// Incase of split brain scenarios where
// tolerance is exactly half of the len(*locks)
// then we need to make sure we have unlocked
// upto tolerance+1 - especially for RUnlock
// to ensure that we don't end up with active
// read locks on the resource after unlocking
// only half of the lockers.
return unlocksFailed >= tolerance
}
return unlocksFailed > tolerance
}
// checkQuorumLocked determines whether we have locked the required quorum of underlying locks or not
func checkQuorumLocked(locks *[]string, quorum int) bool {
count := 0
for _, uid := range *locks {
if isLocked(uid) {
count++
}
}
return count >= quorum
}
// releaseAll releases all locks that are marked as locked
func releaseAll(ctx context.Context, ds *Dsync, tolerance int, owner string, locks *[]string, isReadLock bool, restClnts []NetLocker, names ...string) bool {
var wg sync.WaitGroup
for lockID := range restClnts {
wg.Add(1)
go func(lockID int) {
defer wg.Done()
if sendRelease(ctx, ds, restClnts[lockID], owner, (*locks)[lockID], isReadLock, names...) {
(*locks)[lockID] = ""
}
}(lockID)
}
wg.Wait()
// Return true if releaseAll was successful, otherwise we return 'false'
// to indicate we haven't sufficiently unlocked lockers to avoid deadlocks.
//
// Caller may use this as an indication to call again.
return !checkFailedUnlocks(*locks, tolerance)
}
// Unlock unlocks the write lock.
//
// It is a run-time error if dm is not locked on entry to Unlock.
func (dm *DRWMutex) Unlock(ctx context.Context) {
dm.m.Lock()
dm.cancelRefresh()
dm.m.Unlock()
restClnts, owner := dm.clnt.GetLockers()
// create temp array on stack
locks := make([]string, len(restClnts))
{
dm.m.Lock()
defer dm.m.Unlock()
// Check if minimally a single bool is set in the writeLocks array
lockFound := false
for _, uid := range dm.writeLocks {
if isLocked(uid) {
lockFound = true
break
}
}
if !lockFound {
panic("Trying to Unlock() while no Lock() is active")
}
// Copy write locks to stack array
copy(locks, dm.writeLocks)
}
// Tolerance is not set, defaults to half of the locker clients.
tolerance := len(restClnts) / 2
isReadLock := false
for !releaseAll(ctx, dm.clnt, tolerance, owner, &locks, isReadLock, restClnts, dm.Names...) {
time.Sleep(time.Duration(dm.rng.Float64() * float64(dm.lockRetryMinInterval)))
}
}
// RUnlock releases a read lock held on dm.
//
// It is a run-time error if dm is not locked on entry to RUnlock.
func (dm *DRWMutex) RUnlock(ctx context.Context) {
dm.m.Lock()
dm.cancelRefresh()
dm.m.Unlock()
restClnts, owner := dm.clnt.GetLockers()
// create temp array on stack
locks := make([]string, len(restClnts))
{
dm.m.Lock()
defer dm.m.Unlock()
// Check if minimally a single bool is set in the writeLocks array
lockFound := false
for _, uid := range dm.readLocks {
if isLocked(uid) {
lockFound = true
break
}
}
if !lockFound {
panic("Trying to RUnlock() while no RLock() is active")
}
// Copy write locks to stack array
copy(locks, dm.readLocks)
}
// Tolerance is not set, defaults to half of the locker clients.
tolerance := len(restClnts) / 2
isReadLock := true
for !releaseAll(ctx, dm.clnt, tolerance, owner, &locks, isReadLock, restClnts, dm.Names...) {
time.Sleep(time.Duration(dm.rng.Float64() * float64(dm.lockRetryMinInterval)))
}
}
// sendRelease sends a release message to a node that previously granted a lock
func sendRelease(ctx context.Context, ds *Dsync, c NetLocker, owner string, uid string, isReadLock bool, names ...string) bool {
if c == nil {
log("Unable to call RUnlock failed with %s\n", errors.New("netLocker is offline"))
return false
}
if len(uid) == 0 {
return false
}
args := LockArgs{
Owner: owner,
UID: uid,
Resources: names,
}
netLockCtx, cancel := context.WithTimeout(context.Background(), ds.Timeouts.UnlockCall)
defer cancel()
tc, ok := ctx.Value(mcontext.ContextTraceKey).(*mcontext.TraceCtxt)
if ok {
netLockCtx = context.WithValue(netLockCtx, mcontext.ContextTraceKey, tc)
}
if isReadLock {
if _, err := c.RUnlock(netLockCtx, args); err != nil {
log("dsync: Unable to call RUnlock failed with %s for %#v at %s\n", err, args, c)
return false
}
} else {
if _, err := c.Unlock(netLockCtx, args); err != nil {
log("dsync: Unable to call Unlock failed with %s for %#v at %s\n", err, args, c)
return false
}
}
return true
}