FilesExpand file tree

 master

/

RedLockExtend.cs

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

96 lines (86 loc) · 3.91 KB

 master

/

RedLockExtend.cs

Copy path

Top

File metadata and controls

• Code
• Blame

96 lines (86 loc) · 3.91 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

using Medallion.Threading.Internal;

using StackExchange.Redis;

using System;

using System.Collections.Generic;

using System.Linq;

using System.Text;

using System.Threading;

using System.Threading.Tasks;

namespace Medallion.Threading.Redis.RedLock

{

internal interface IRedLockExtensibleSynchronizationPrimitive : IRedLockReleasableSynchronizationPrimitive

{

TimeoutValue AcquireTimeout { get; }

Task<bool> TryExtendAsync(IDatabaseAsync database);

}

/// <summary>

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

/// </summary>

internal readonly struct RedLockExtend

{

private readonly IRedLockExtensibleSynchronizationPrimitive _primitive;

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

private readonly CancellationToken _cancellationToken;

public RedLockExtend(

IRedLockExtensibleSynchronizationPrimitive primitive,

Dictionary<IDatabase, Task<bool>> tryAcquireOrRenewTasks,

CancellationToken cancellationToken)

{

this._primitive = primitive;

this._tryAcquireOrRenewTasks = tryAcquireOrRenewTasks;

this._cancellationToken = cancellationToken;

}

public async Task<bool?> TryExtendAsync()

{

Invariant.Require(!SyncViaAsync.IsSynchronous, "should only be called from a background renewal thread which is async");

var incompleteTasks = new HashSet<Task>();

foreach (var kvp in this._tryAcquireOrRenewTasks.ToArray())

{

if (kvp.Value.IsCompleted)

{

incompleteTasks.Add(

this._tryAcquireOrRenewTasks[kvp.Key] = Helpers.SafeCreateTask(

state => state.primitive.TryExtendAsync(state.database),

(primitive: this._primitive, database: kvp.Key)

)

);

}

else

{

// if the previous acquire/renew is still going, just keep waiting for that

incompleteTasks.Add(kvp.Value);

}

}

// For extension we use the same timeout as acquire. This ensures the same min validity time which should be

// sufficient to keep extending

using var timeout = new TimeoutTask(this._primitive.AcquireTimeout, this._cancellationToken);

incompleteTasks.Add(timeout.Task);

var databaseCount = this._tryAcquireOrRenewTasks.Count;

var successCount = 0;

var failCount = 0;

while (true)

{

var completed = await Task.WhenAny(incompleteTasks).ConfigureAwait(false);

if (completed == timeout.Task)

{

await completed.ConfigureAwait(false); // propagate cancellation

return null; // inconclusive

}

if (completed.Status == TaskStatus.RanToCompletion && ((Task<bool>)completed).Result)

{

++successCount;

if (RedLockHelper.HasSufficientSuccesses(successCount, databaseCount)) { return true; }

}

else

{

// note that we treat faulted and failed the same in extend. There's no reason to throw, since

// this is just called by the extend loop. While in theory a fault could indicate some kind of post-success

// failure, most likely it means the db is unreachable and so it is safest to consider it a failure

++failCount;

if (RedLockHelper.HasTooManyFailuresOrFaults(failCount, databaseCount)) { return false; }

}

incompleteTasks.Remove(completed);

}

}

}

}