FilesExpand file tree

 oracle

/

RedLockRelease.cs

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

103 lines (94 loc) · 4.46 KB

 oracle

/

RedLockRelease.cs

Copy path

Top

File metadata and controls

• Code
• Blame

103 lines (94 loc) · 4.46 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

using Medallion.Threading.Internal;

using StackExchange.Redis;

using System;

using System.Collections.Generic;

using System.Linq;

using System.Text;

using System.Threading.Tasks;

namespace Medallion.Threading.Redis.RedLock

{

internal interface IRedLockReleasableSynchronizationPrimitive

{

Task ReleaseAsync(IDatabaseAsync database, bool fireAndForget);

void Release(IDatabase database, bool fireAndForget);

}

/// <summary>

/// Implements the release operation in the RedLock algorithm. See https://redis.io/topics/distlock

/// </summary>

internal readonly struct RedLockRelease

{

private readonly IRedLockReleasableSynchronizationPrimitive _primitive;

private readonly IReadOnlyDictionary<IDatabase, Task<bool>> _tryAcquireOrRenewTasks;

public RedLockRelease(

IRedLockReleasableSynchronizationPrimitive primitive,

IReadOnlyDictionary<IDatabase, Task<bool>> tryAcquireOrRenewTasks)

{

this._primitive = primitive;

this._tryAcquireOrRenewTasks = tryAcquireOrRenewTasks;

}

public async ValueTask ReleaseAsync()

{

var isSynchronous = SyncViaAsync.IsSynchronous;

var unreleasedTryAcquireOrRenewTasks = this._tryAcquireOrRenewTasks.ToDictionary(kvp => kvp.Key, kvp => kvp.Value);

List<Exception>? releaseExceptions = null;

var successCount = 0;

var faultCount = 0;

var databaseCount = unreleasedTryAcquireOrRenewTasks.Count;

try

{

while (true)

{

var releaseableDatabases = unreleasedTryAcquireOrRenewTasks.Where(kvp => kvp.Value.IsCompleted)

// work through non-faulted tasks first

.OrderByDescending(kvp => kvp.Value.Status == TaskStatus.RanToCompletion)

// then start with failed since no action is required to release those

.ThenBy(kvp => kvp.Value.Status == TaskStatus.RanToCompletion && kvp.Value.Result)

.Select(kvp => kvp.Key)

.ToArray();

foreach (var db in releaseableDatabases)

{

var tryAcquireOrRenewTask = unreleasedTryAcquireOrRenewTasks[db];

unreleasedTryAcquireOrRenewTasks.Remove(db);

if (RedLockHelper.ReturnedFalse(tryAcquireOrRenewTask))

{

// if we failed to acquire, we don't need to release

++successCount;

}

else

{

try

{

if (isSynchronous) { this._primitive.Release(db, fireAndForget: false); }

else { await this._primitive.ReleaseAsync(db, fireAndForget: false).ConfigureAwait(false); }

++successCount;

}

catch (Exception ex)

{

(releaseExceptions ??= new List<Exception>()).Add(ex);

++faultCount;

if (RedLockHelper.HasTooManyFailuresOrFaults(faultCount, databaseCount))

{

throw new AggregateException(releaseExceptions!).Flatten();

}

}

}

if (RedLockHelper.HasSufficientSuccesses(successCount, databaseCount))

{

return;

}

}

// if we haven't released enough yet to be done or certain of success or failure, wait for another to finish

if (isSynchronous) { Task.WaitAny(unreleasedTryAcquireOrRenewTasks.Values.ToArray()); }

else { await Task.WhenAny(unreleasedTryAcquireOrRenewTasks.Values).ConfigureAwait(false); }

}

}

finally // fire and forget the rest

{

foreach (var kvp in unreleasedTryAcquireOrRenewTasks)

{

RedLockHelper.FireAndForgetReleaseUponCompletion(this._primitive, kvp.Key, kvp.Value);

}

}

}

}

}