Skip to content

Commit bb69f84

Browse files
madelsonmadelson
committed
Implement redis RW lock
1 parent 8a3c9f2 commit bb69f84

34 files changed

Lines changed: 1217 additions & 364 deletions

‎DistributedLock.Core/Internal/Helpers.cs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,13 @@ public TTask GetResult()
9595
public void OnCompleted(Action continuation) => this._taskAwaiter.OnCompleted(continuation);
9696
public void UnsafeOnCompleted(Action continuation) => this._taskAwaiter.UnsafeOnCompleted(continuation);
9797
}
98+
99+
public static bool TryGetValue<T>(this T? nullable, out T value)
100+
where T : struct
101+
{
102+
value = nullable.GetValueOrDefault();
103+
return nullable.HasValue;
104+
}
98105
}
99106

100107
/// <summary>

‎DistributedLock.Redis/IRedisSynchronizationPrimitive.cs‎

Lines changed: 0 additions & 23 deletions
This file was deleted.

‎DistributedLock.Redis/Primitives/RedisMutexPrimitive.cs‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
using Medallion.Threading.Internal;
2+
using Medallion.Threading.Redis.RedLock;
3+
using StackExchange.Redis;
4+
using System;
5+
using System.Collections.Generic;
6+
using System.Linq;
7+
using System.Text;
8+
using System.Threading.Tasks;
9+
10+
namespace Medallion.Threading.Redis.Primitives
11+
{
12+
internal class RedisMutexPrimitive : IRedLockAcquirableSynchronizationPrimitive, IRedLockExtensibleSynchronizationPrimitive
13+
{
14+
private readonly RedisKey _key;
15+
private readonly RedisValue _lockId;
16+
private readonly RedLockTimeouts _timeouts;
17+
18+
public RedisMutexPrimitive(RedisKey key, RedisValue lockId, RedLockTimeouts timeouts)
19+
{
20+
this._key = key;
21+
this._lockId = lockId;
22+
this._timeouts = timeouts;
23+
}
24+
25+
public TimeoutValue AcquireTimeout => this._timeouts.AcquireTimeout;
26+
27+
private static readonly RedisScript<RedisMutexPrimitive> ReleaseScript = new RedisScript<RedisMutexPrimitive>(@"
28+
if redis.call('get', @key) == @lockId then
29+
return redis.call('del', @key)
30+
end
31+
return 0",
32+
p => new { key = p._key, lockId = p._lockId }
33+
);
34+
35+
public void Release(IDatabase database, bool fireAndForget) => ReleaseScript.Execute(database, this, fireAndForget);
36+
public Task ReleaseAsync(IDatabaseAsync database, bool fireAndForget) => ReleaseScript.ExecuteAsync(database, this, fireAndForget);
37+
38+
public bool TryAcquire(IDatabase database) =>
39+
database.StringSet(this._key, this._lockId, this._timeouts.Expiry.TimeSpan, When.NotExists, CommandFlags.DemandMaster);
40+
public Task<bool> TryAcquireAsync(IDatabaseAsync database) =>
41+
database.StringSetAsync(this._key, this._lockId, this._timeouts.Expiry.TimeSpan, When.NotExists, CommandFlags.DemandMaster);
42+
43+
private static readonly RedisScript<RedisMutexPrimitive> ExtendScript = new RedisScript<RedisMutexPrimitive>(@"
44+
if redis.call('get', @key) == @lockId then
45+
return redis.call('pexpire', @key, @expiryMillis)
46+
end
47+
return 0",
48+
p => new { key = p._key, lockId = p._lockId, expiryMillis = p._timeouts.Expiry.InMilliseconds }
49+
);
50+
51+
public Task<bool> TryExtendAsync(IDatabaseAsync database) => ExtendScript.ExecuteAsync(database, this).AsBooleanTask();
52+
}
53+
}

‎DistributedLock.Redis/Primitives/RedisReaderWriterLockPrimitives.cs‎

Lines changed: 150 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,150 @@
1+
using Medallion.Threading.Internal;
2+
using Medallion.Threading.Redis.RedLock;
3+
using StackExchange.Redis;
4+
using System;
5+
using System.Collections.Generic;
6+
using System.Linq;
7+
using System.Text;
8+
using System.Threading.Tasks;
9+
10+
namespace Medallion.Threading.Redis.Primitives
11+
{
12+
internal class RedisReadLockPrimitive : IRedLockAcquirableSynchronizationPrimitive, IRedLockExtensibleSynchronizationPrimitive
13+
{
14+
private readonly RedisValue _lockId = RedLockHelper.CreateLockId();
15+
private readonly RedisKey _readerKey, _writerKey;
16+
private readonly RedLockTimeouts _timeouts;
17+
18+
public RedisReadLockPrimitive(RedisKey readerKey, RedisKey writerKey, RedLockTimeouts timeouts)
19+
{
20+
this._readerKey = readerKey;
21+
this._writerKey = writerKey;
22+
this._timeouts = timeouts;
23+
}
24+
25+
public TimeoutValue AcquireTimeout => this._timeouts.AcquireTimeout;
26+
27+
/// <summary>
28+
/// RELEASE READ
29+
///
30+
/// Just remove our ID from the reader set (noop if it wasn't there or the set DNE)
31+
/// </summary>
32+
private static readonly RedisScript<RedisReadLockPrimitive> ReleaseReadScript = new RedisScript<RedisReadLockPrimitive>(
33+
@"redis.call('srem', @readerKey, @lockId)",
34+
p => new { readerKey = p._readerKey, lockId = p._lockId }
35+
);
36+
37+
public void Release(IDatabase database, bool fireAndForget) => ReleaseReadScript.Execute(database, this, fireAndForget);
38+
public Task ReleaseAsync(IDatabaseAsync database, bool fireAndForget) => ReleaseReadScript.ExecuteAsync(database, this, fireAndForget);
39+
40+
/// <summary>
41+
/// TRY EXTEND READ
42+
///
43+
/// First, check if the reader set exists and our ID is still a member. If not, we fail.
44+
///
45+
/// Then, extend the reader set TTL to be at least our expiry (at least because other readers might be operating with a longer expiry)
46+
/// </summary>
47+
private static readonly RedisScript<RedisReadLockPrimitive> TryExtendReadScript = new RedisScript<RedisReadLockPrimitive>(@"
48+
if redis.call('sismember', @readerKey, @lockId) == 0 then
49+
return 0
50+
end
51+
if readerTtl < @expiryMillis then
52+
redis.call('pexpire', @readerKey, @expiryMillis)
53+
end
54+
return 1",
55+
p => new { readerKey = p._readerKey, lockId = p._lockId, expiryMillis = p._timeouts.Expiry.InMilliseconds }
56+
);
57+
58+
public Task<bool> TryExtendAsync(IDatabaseAsync database) => TryExtendReadScript.ExecuteAsync(database, this).AsBooleanTask();
59+
60+
/// <summary>
61+
/// TRY ACQUIRE READ
62+
///
63+
/// First, check the writer lock value: if it exists then we fail.
64+
///
65+
/// Then, add our ID to the reader set, creating it if it does not exist. Then, extend the TTL
66+
/// of the reader set to be at least our expiry. Return success.
67+
/// </summary>
68+
private static readonly RedisScript<RedisReadLockPrimitive> TryAcquireReadScript = new RedisScript<RedisReadLockPrimitive>($@"
69+
if redis.call('exists', @writerKey) == 1 then
70+
return 0
71+
end
72+
redis.call('sadd', @readerKey, @lockId)
73+
local readerTtl = redis.call('pttl', @readerKey)
74+
if readerTtl < tonumber(@expiryMillis) then
75+
redis.call('pexpire', @readerKey, @expiryMillis)
76+
end
77+
return 1",
78+
p => new { writerKey = p._writerKey, readerKey = p._readerKey, lockId = p._lockId, expiryMillis = p._timeouts.Expiry.InMilliseconds }
79+
);
80+
81+
public Task<bool> TryAcquireAsync(IDatabaseAsync database) => TryAcquireReadScript.ExecuteAsync(database, this).AsBooleanTask();
82+
public bool TryAcquire(IDatabase database) => (bool)TryAcquireReadScript.Execute(database, this);
83+
}
84+
85+
internal class RedisWriterWaitingPrimitive : RedisMutexPrimitive
86+
{
87+
public const string LockIdSuffix = "_WRITERWAITING";
88+
89+
public RedisWriterWaitingPrimitive(RedisKey writerKey, RedisValue baseLockId, RedLockTimeouts timeouts)
90+
: base(writerKey, baseLockId + LockIdSuffix, timeouts)
91+
{
92+
}
93+
}
94+
95+
internal class RedisWriteLockPrimitive : IRedLockAcquirableSynchronizationPrimitive, IRedLockExtensibleSynchronizationPrimitive
96+
{
97+
private readonly RedisKey _readerKey, _writerKey;
98+
private readonly RedisValue _lockId;
99+
private readonly RedLockTimeouts _timeouts;
100+
private readonly RedisMutexPrimitive _mutexPrimitive;
101+
102+
public RedisWriteLockPrimitive(
103+
RedisKey readerKey,
104+
RedisKey writerKey,
105+
RedisValue lockId,
106+
RedLockTimeouts timeouts)
107+
{
108+
this._readerKey = readerKey;
109+
this._writerKey = writerKey;
110+
this._lockId = lockId;
111+
this._timeouts = timeouts;
112+
this._mutexPrimitive = new RedisMutexPrimitive(this._writerKey, this._lockId, this._timeouts);
113+
}
114+
115+
public TimeoutValue AcquireTimeout => this._timeouts.AcquireTimeout;
116+
117+
public void Release(IDatabase database, bool fireAndForget) => this._mutexPrimitive.Release(database, fireAndForget);
118+
public Task ReleaseAsync(IDatabaseAsync database, bool fireAndForget) => this._mutexPrimitive.ReleaseAsync(database, fireAndForget);
119+
120+
/// <summary>
121+
/// TRY ACQUIRE WRITE
122+
///
123+
/// First, check if writerValue exists. If so, fail unless it's our waiting ID.
124+
///
125+
/// Then, check if there are no readers. If so, then set writerValue to our ID and return success. If not, then if the lock
126+
/// has our waiting ID re-up the expiry (avoids the need to extend the writer waiting lock).
127+
///
128+
/// Finally, return failure.
129+
/// </summary>
130+
private static readonly RedisScript<RedisWriteLockPrimitive> TryAcquireWriteScript = new RedisScript<RedisWriteLockPrimitive>($@"
131+
local writerValue = redis.call('get', @writerKey)
132+
if writerValue == false or writerValue == @lockId .. '{RedisWriterWaitingPrimitive.LockIdSuffix}' then
133+
if redis.call('scard', @readerKey) == 0 then
134+
redis.call('set', @writerKey, @lockId, 'px', @expiryMillis)
135+
return 1
136+
end
137+
if writerValue ~= false then
138+
redis.call('pexpire', @writerKey, @expiryMillis)
139+
end
140+
end
141+
return 0",
142+
p => new { writerKey = p._writerKey, readerKey = p._readerKey, lockId = p._lockId, expiryMillis = p._timeouts.Expiry.InMilliseconds }
143+
);
144+
145+
public bool TryAcquire(IDatabase database) => (bool)TryAcquireWriteScript.Execute(database, this);
146+
public Task<bool> TryAcquireAsync(IDatabaseAsync database) => TryAcquireWriteScript.ExecuteAsync(database, this).AsBooleanTask();
147+
148+
public Task<bool> TryExtendAsync(IDatabaseAsync database) => this._mutexPrimitive.TryExtendAsync(database);
149+
}
150+
}

‎DistributedLock.Redis/Primitives/RedisScript.cs‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
using Medallion.Threading.Redis.RedLock;
2+
using StackExchange.Redis;
3+
using System;
4+
using System.Collections.Generic;
5+
using System.Linq;
6+
using System.Text;
7+
using System.Text.RegularExpressions;
8+
using System.Threading.Tasks;
9+
10+
namespace Medallion.Threading.Redis.Primitives
11+
{
12+
internal class RedisScript<TArgument>
13+
{
14+
private readonly LuaScript _script;
15+
private readonly Func<TArgument, object> _parameters;
16+
17+
public RedisScript(string script, Func<TArgument, object> parameters)
18+
{
19+
this._script = LuaScript.Prepare(RemoveExtraneousWhitespace(script));
20+
this._parameters = parameters;
21+
}
22+
23+
public RedisResult Execute(IDatabase database, TArgument argument, bool fireAndForget = false) =>
24+
this._script.Evaluate(database, this._parameters(argument), flags: RedLockHelper.GetCommandFlags(fireAndForget));
25+
26+
public Task<RedisResult> ExecuteAsync(IDatabaseAsync database, TArgument argument, bool fireAndForget = false) =>
27+
this._script.EvaluateAsync(database, this._parameters(argument), flags: RedLockHelper.GetCommandFlags(fireAndForget));
28+
29+
// send the smallest possible script to the server
30+
private static string RemoveExtraneousWhitespace(string script) => Regex.Replace(script.Trim(), @"\s+", " ");
31+
}
32+
}

DistributedLock.Redis/RedLockAcquire.cs renamed to DistributedLock.Redis/RedLock/RedLockAcquire.cs

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -8,19 +8,26 @@
88
using System.Threading;
99
using System.Threading.Tasks;
1010

11-
namespace Medallion.Threading.Redis
11+
namespace Medallion.Threading.Redis.RedLock
1212
{
13+
internal interface IRedLockAcquirableSynchronizationPrimitive : IRedLockReleasableSynchronizationPrimitive
14+
{
15+
TimeoutValue AcquireTimeout { get; }
16+
Task<bool> TryAcquireAsync(IDatabaseAsync database);
17+
bool TryAcquire(IDatabase database);
18+
}
19+
1320
/// <summary>
1421
/// Implements the acquire operation in the RedLock algorithm. See https://redis.io/topics/distlock
1522
/// </summary>
1623
internal readonly struct RedLockAcquire
1724
{
18-
private readonly IRedisSynchronizationPrimitive _primitive;
25+
private readonly IRedLockAcquirableSynchronizationPrimitive _primitive;
1926
private readonly IReadOnlyList<IDatabase> _databases;
2027
private readonly CancellationToken _cancellationToken;
2128

2229
public RedLockAcquire(
23-
IRedisSynchronizationPrimitive primitive,
30+
IRedLockAcquirableSynchronizationPrimitive primitive,
2431
IReadOnlyList<IDatabase> databases,
2532
CancellationToken cancellationToken)
2633
{
@@ -178,7 +185,7 @@ private async Task<bool> WaitForAcquireAsync(IReadOnlyDictionary<IDatabase, Task
178185
// make sure we didn't time out
179186
if (this._primitive.AcquireTimeout.CompareTo(stopwatch.Elapsed) >= 0)
180187
{
181-
return new Dictionary<IDatabase, Task<bool>> { [database] = Task.FromResult(true) };
188+
return new Dictionary<IDatabase, Task<bool>> { [database] = Task.FromResult(success) };
182189
}
183190

184191
this._primitive.Release(database, fireAndForget: true); // timed out, so release

DistributedLock.Redis/RedLockExtend.cs renamed to DistributedLock.Redis/RedLock/RedLockExtend.cs

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,19 +7,25 @@
77
using System.Threading;
88
using System.Threading.Tasks;
99

10-
namespace Medallion.Threading.Redis
10+
namespace Medallion.Threading.Redis.RedLock
1111
{
12+
internal interface IRedLockExtensibleSynchronizationPrimitive : IRedLockReleasableSynchronizationPrimitive
13+
{
14+
TimeoutValue AcquireTimeout { get; }
15+
Task<bool> TryExtendAsync(IDatabaseAsync database);
16+
}
17+
1218
/// <summary>
1319
/// Implements the extend operation in the RedLock algorithm. See https://redis.io/topics/distlock
1420
/// </summary>
1521
internal readonly struct RedLockExtend
1622
{
17-
private readonly IRedisSynchronizationPrimitive _primitive;
23+
private readonly IRedLockExtensibleSynchronizationPrimitive _primitive;
1824
private readonly Dictionary<IDatabase, Task<bool>> _tryAcquireOrRenewTasks;
1925
private readonly CancellationToken _cancellationToken;
2026

2127
public RedLockExtend(
22-
IRedisSynchronizationPrimitive primitive,
28+
IRedLockExtensibleSynchronizationPrimitive primitive,
2329
Dictionary<IDatabase, Task<bool>> tryAcquireOrRenewTasks,
2430
CancellationToken cancellationToken)
2531
{

0 commit comments

Comments
 (0)