Skip to content

Commit a74571c

Browse files
preardonpreardon
authored
Initial suggestion for Global locking for Timed Services (#3092)
Issue #3075 This is an example of providers to allow locking across resources, included is - In Memory Lock (for when we only want a lock in process) - Azure Blob Lock
1 parent b3346eb commit a74571c

13 files changed

Lines changed: 385 additions & 31 deletions

File tree

‎Brighter.sln‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -329,6 +329,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paramore.Brighter.ServiceAc
329329
EndProject
330330
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paramore.Brighter.Sqlite.Dapper", "src\Paramore.Brighter.Sqlite.Dapper\Paramore.Brighter.Sqlite.Dapper.csproj", "{3384FBF0-5DCB-452D-8288-FAD1D0023089}"
331331
EndProject
332+
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paramore.Brighter.Locking.Azure", "Paramore.Brighter.Locking.Azure\Paramore.Brighter.Locking.Azure.csproj", "{021F3B51-A640-4C0D-9B47-FB4E32DF6715}"
333+
EndProject
332334
Global
333335
GlobalSection(SolutionConfigurationPlatforms) = preSolution
334336
Debug|Any CPU = Debug|Any CPU
@@ -1875,6 +1877,18 @@ Global
18751877
{3384FBF0-5DCB-452D-8288-FAD1D0023089}.Release|Mixed Platforms.Build.0 = Release|Any CPU
18761878
{3384FBF0-5DCB-452D-8288-FAD1D0023089}.Release|x86.ActiveCfg = Release|Any CPU
18771879
{3384FBF0-5DCB-452D-8288-FAD1D0023089}.Release|x86.Build.0 = Release|Any CPU
1880+
{021F3B51-A640-4C0D-9B47-FB4E32DF6715}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
1881+
{021F3B51-A640-4C0D-9B47-FB4E32DF6715}.Debug|Any CPU.Build.0 = Debug|Any CPU
1882+
{021F3B51-A640-4C0D-9B47-FB4E32DF6715}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU
1883+
{021F3B51-A640-4C0D-9B47-FB4E32DF6715}.Debug|Mixed Platforms.Build.0 = Debug|Any CPU
1884+
{021F3B51-A640-4C0D-9B47-FB4E32DF6715}.Debug|x86.ActiveCfg = Debug|Any CPU
1885+
{021F3B51-A640-4C0D-9B47-FB4E32DF6715}.Debug|x86.Build.0 = Debug|Any CPU
1886+
{021F3B51-A640-4C0D-9B47-FB4E32DF6715}.Release|Any CPU.ActiveCfg = Release|Any CPU
1887+
{021F3B51-A640-4C0D-9B47-FB4E32DF6715}.Release|Any CPU.Build.0 = Release|Any CPU
1888+
{021F3B51-A640-4C0D-9B47-FB4E32DF6715}.Release|Mixed Platforms.ActiveCfg = Release|Any CPU
1889+
{021F3B51-A640-4C0D-9B47-FB4E32DF6715}.Release|Mixed Platforms.Build.0 = Release|Any CPU
1890+
{021F3B51-A640-4C0D-9B47-FB4E32DF6715}.Release|x86.ActiveCfg = Release|Any CPU
1891+
{021F3B51-A640-4C0D-9B47-FB4E32DF6715}.Release|x86.Build.0 = Release|Any CPU
18781892
EndGlobalSection
18791893
GlobalSection(SolutionProperties) = preSolution
18801894
HideSolutionNode = FALSE

‎Paramore.Brighter.Locking.Azure/AzureBlobLockingProvider.cs‎

Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,106 @@
1+
using Azure;
2+
using Azure.Storage.Blobs;
3+
using Azure.Storage.Blobs.Specialized;
4+
using Microsoft.Extensions.Logging;
5+
using Paramore.Brighter.Logging;
6+
7+
namespace Paramore.Brighter.Locking.Azure;
8+
9+
public class AzureBlobLockingProvider(AzureBlobLockingProviderOptions options) : IDistributedLock
10+
{
11+
private readonly BlobContainerClient _containerClient = new BlobContainerClient(options.BlobContainerUri, options.TokenCredential);
12+
private readonly ILogger _logger = ApplicationLogging.CreateLogger<AzureBlobLockingProviderOptions>();
13+
14+
private readonly Dictionary<string, string> _leases = new Dictionary<string, string>();
15+
16+
public async Task<bool> ObtainLockAsync(string resource, CancellationToken cancellationToken)
17+
{
18+
var client = GetBlobClient(resource);
19+
20+
// Write if does not exist
21+
if (!await client.ExistsAsync(cancellationToken))
22+
{
23+
await using var emptyStream = new MemoryStream();
24+
await using var writer = new StreamWriter(emptyStream);
25+
await writer.WriteAsync(string.Empty);
26+
await writer.FlushAsync(cancellationToken);
27+
emptyStream.Position = 0;
28+
await client.UploadAsync(emptyStream, cancellationToken: cancellationToken);
29+
}
30+
31+
try
32+
{
33+
var response = await client.GetBlobLeaseClient().AcquireAsync(options.LeaseValidity, cancellationToken: cancellationToken);
34+
_leases.Add(NormaliseResourceName(resource), response.Value.LeaseId);
35+
return true;
36+
}
37+
catch (RequestFailedException e)
38+
{
39+
_logger.LogInformation("Could not Acquire Lease on Blob {LockResourceName}", resource);
40+
return false;
41+
}
42+
}
43+
44+
public bool ObtainLock(string resource)
45+
{
46+
var client = GetBlobClient(resource);
47+
48+
// Write if does not exist
49+
if (!client.Exists())
50+
{
51+
using var emptyStream = new MemoryStream();
52+
using var writer = new StreamWriter(emptyStream);
53+
writer.Write(string.Empty);
54+
writer.Flush();
55+
emptyStream.Position = 0;
56+
client.Upload(emptyStream);
57+
}
58+
59+
try
60+
{
61+
var response = client.GetBlobLeaseClient().Acquire(options.LeaseValidity);
62+
_leases.Add(NormaliseResourceName(resource), response.Value.LeaseId);
63+
return true;
64+
}
65+
catch (RequestFailedException e)
66+
{
67+
_logger.LogInformation("Could not Acquire Lease on Blob {LockResourceName}", resource);
68+
return false;
69+
}
70+
}
71+
72+
public async Task ReleaseLockAsync(string resource, CancellationToken cancellationToken)
73+
{
74+
var client = GetBlobLeaseClientForResource(resource);
75+
if(client == null)
76+
return;
77+
await client.ReleaseAsync(cancellationToken: cancellationToken);
78+
_leases.Remove(NormaliseResourceName(resource));
79+
}
80+
81+
public void ReleaseLock(string resource)
82+
{
83+
var client = GetBlobLeaseClientForResource(resource);
84+
if(client == null)
85+
return;
86+
client.Release();
87+
_leases.Remove(NormaliseResourceName(resource));
88+
}
89+
90+
private BlobLeaseClient? GetBlobLeaseClientForResource(string resource)
91+
{
92+
if (_leases.ContainsKey(NormaliseResourceName(resource)))
93+
return GetBlobClient(resource).GetBlobLeaseClient(_leases[NormaliseResourceName(resource)]);
94+
95+
_logger.LogInformation("No lock found for {LockResourceName}", resource);
96+
return null;
97+
}
98+
99+
private BlobClient GetBlobClient(string resource)
100+
{
101+
var storageLocation = options.StorageLocationFunc.Invoke(NormaliseResourceName(resource));
102+
return _containerClient.GetBlobClient(storageLocation);
103+
}
104+
105+
private static string NormaliseResourceName(string resourceName) => resourceName.ToLower();
106+
}

‎Paramore.Brighter.Locking.Azure/AzureBlobLockingProviderOptions.cs‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
using Azure.Core;
2+
3+
namespace Paramore.Brighter.Locking.Azure;
4+
5+
public class AzureBlobLockingProviderOptions(
6+
Uri blobContainerUri,
7+
TokenCredential tokenCredential
8+
)
9+
{
10+
/// <summary>
11+
/// The URI of the blob container
12+
/// </summary>
13+
public Uri BlobContainerUri { get; init; } = blobContainerUri;
14+
15+
/// <summary>
16+
/// The Credential to use when writing blobs
17+
/// </summary>
18+
public TokenCredential TokenCredential { get; init; } = tokenCredential;
19+
20+
/// <summary>
21+
/// The amount of time before the lease automatically expires
22+
/// </summary>
23+
public TimeSpan LeaseValidity { get; init; } = TimeSpan.FromMinutes(1);
24+
25+
/// <summary>
26+
/// The function to provide the location to store the locks inside of the Blob container
27+
/// </summary>
28+
public Func<string, string> StorageLocationFunc = (resource) => $"lock-{resource}";
29+
}

‎Paramore.Brighter.Locking.Azure/Paramore.Brighter.Locking.Azure.csproj‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
<Project Sdk="Microsoft.NET.Sdk">
2+
3+
<PropertyGroup>
4+
<TargetFramework>net8.0</TargetFramework>
5+
<ImplicitUsings>enable</ImplicitUsings>
6+
<Nullable>enable</Nullable>
7+
<Title>This is the Azure Distributed Locking Provider.</Title>
8+
<Authors>Paul Reardon</Authors>
9+
</PropertyGroup>
10+
11+
<ItemGroup>
12+
<ProjectReference Include="..\src\Paramore.Brighter\Paramore.Brighter.csproj" />
13+
</ItemGroup>
14+
15+
<ItemGroup>
16+
<PackageReference Include="Azure.Storage.Blobs" />
17+
</ItemGroup>
18+
19+
</Project>

‎src/Paramore.Brighter.Extensions.DependencyInjection/ServiceCollectionExtensions.cs‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -206,6 +206,10 @@ public static IBrighterBuilder UseExternalBus(
206206
brighterBuilder.Services.Add(asyncOutboxdescriptor);
207207
}
208208
}
209+
210+
// If no distributed locking service is added, then add the in memory variant
211+
var distributedLock = busConfiguration.DistributedLock ?? new InMemoryLock();
212+
brighterBuilder.Services.AddSingleton(distributedLock);
209213

210214
if (busConfiguration.UseRpc)
211215
{

‎src/Paramore.Brighter.Extensions.Hosting/TimedOutboxArchiver.cs‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -11,19 +11,21 @@ namespace Paramore.Brighter.Extensions.Hosting
1111

1212
public class TimedOutboxArchiver(
1313
IServiceScopeFactory serviceScopeFactory,
14+
IDistributedLock distributedLock,
1415
TimedOutboxArchiverOptions options)
1516
: IHostedService, IDisposable
1617
{
1718
private static readonly ILogger s_logger = ApplicationLogging.CreateLogger<TimedOutboxSweeper>();
1819
private Timer _timer;
1920

20-
private static readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
21+
private const string LockingResourceName = "Archiver";
2122

2223
public Task StartAsync(CancellationToken cancellationToken)
2324
{
2425
s_logger.LogInformation("Outbox Archiver Service is starting");
2526

26-
_timer = new Timer(async (e) => await Archive(e, cancellationToken), null, TimeSpan.Zero, TimeSpan.FromSeconds(options.TimerInterval));
27+
_timer = new Timer(async (e) => await Archive(e, cancellationToken), null, TimeSpan.Zero,
28+
TimeSpan.FromSeconds(options.TimerInterval));
2729

2830
return Task.CompletedTask;
2931
}
@@ -44,7 +46,7 @@ public void Dispose()
4446

4547
private async Task Archive(object state, CancellationToken cancellationToken)
4648
{
47-
if (await _semaphore.WaitAsync(TimeSpan.Zero, cancellationToken))
49+
if (await distributedLock.ObtainLockAsync(LockingResourceName, cancellationToken))
4850
{
4951
var scope = serviceScopeFactory.CreateScope();
5052
s_logger.LogInformation("Outbox Archiver looking for messages to Archive");
@@ -60,7 +62,7 @@ private async Task Archive(object state, CancellationToken cancellationToken)
6062
}
6163
finally
6264
{
63-
_semaphore.Release();
65+
await distributedLock.ReleaseLockAsync(LockingResourceName, cancellationToken);
6466
}
6567

6668
s_logger.LogInformation("Outbox Sweeper sleeping");

‎src/Paramore.Brighter.Extensions.Hosting/TimedOutboxSweeper.cs‎

Lines changed: 38 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -14,15 +14,18 @@ namespace Paramore.Brighter.Extensions.Hosting
1414
public class TimedOutboxSweeper : IHostedService, IDisposable
1515
{
1616
private readonly IServiceScopeFactory _serviceScopeFactory;
17+
private readonly IDistributedLock _distributedLock;
1718
private readonly TimedOutboxSweeperOptions _options;
1819
private static readonly ILogger s_logger= ApplicationLogging.CreateLogger<TimedOutboxSweeper>();
1920

2021
private Timer _timer;
21-
//private Timer _timer;
22-
23-
public TimedOutboxSweeper (IServiceScopeFactory serviceScopeFactory, TimedOutboxSweeperOptions options)
22+
private const string LockingResourceName = "OutboxSweeper";
23+
24+
public TimedOutboxSweeper(IServiceScopeFactory serviceScopeFactory, IDistributedLock distributedLock,
25+
TimedOutboxSweeperOptions options)
2426
{
2527
_serviceScopeFactory = serviceScopeFactory;
28+
_distributedLock = distributedLock;
2629
_options = options;
2730
}
2831

@@ -42,36 +45,44 @@ public Task StartAsync(CancellationToken cancellationToken)
4245

4346
private void OnElapsed(object sender, ElapsedEventArgs elapsedEventArgs)
4447
{
45-
s_logger.LogInformation("Outbox Sweeper looking for unsent messages");
46-
47-
var scope = _serviceScopeFactory.CreateScope();
48-
try
48+
if (_distributedLock.ObtainLock(LockingResourceName))
4949
{
50-
IAmACommandProcessor commandProcessor = scope.ServiceProvider.GetService<IAmACommandProcessor>();
50+
s_logger.LogInformation("Outbox Sweeper looking for unsent messages");
5151

52-
var outBoxSweeper = new OutboxSweeper(
53-
millisecondsSinceSent: _options.MinimumMessageAge,
54-
commandProcessor: commandProcessor,
55-
_options.BatchSize,
56-
_options.UseBulk,
57-
_options.Args);
52+
var scope = _serviceScopeFactory.CreateScope();
53+
try
54+
{
55+
IAmACommandProcessor commandProcessor = scope.ServiceProvider.GetService<IAmACommandProcessor>();
5856

59-
if (_options.UseBulk)
60-
outBoxSweeper.SweepAsyncOutbox();
61-
else
62-
outBoxSweeper.Sweep();
63-
}
64-
catch (Exception e)
65-
{
66-
s_logger.LogError(e, "Error while sweeping the outbox.");
67-
throw;
57+
var outBoxSweeper = new OutboxSweeper(
58+
millisecondsSinceSent: _options.MinimumMessageAge,
59+
commandProcessor: commandProcessor,
60+
_options.BatchSize,
61+
_options.UseBulk,
62+
_options.Args);
63+
64+
if (_options.UseBulk)
65+
outBoxSweeper.SweepAsyncOutbox();
66+
else
67+
outBoxSweeper.Sweep();
68+
}
69+
catch (Exception e)
70+
{
71+
s_logger.LogError(e, "Error while sweeping the outbox.");
72+
throw;
73+
}
74+
finally
75+
{
76+
_distributedLock.ReleaseLock(LockingResourceName);
77+
scope.Dispose();
78+
((Timer)sender).Enabled = true;
79+
}
6880
}
69-
finally
81+
else
7082
{
71-
scope.Dispose();
72-
((Timer)sender).Enabled = true;
83+
s_logger.LogWarning("Outbox Sweeper is still running - abandoning attempt.");
7384
}
74-
85+
7586
s_logger.LogInformation("Outbox Sweeper sleeping");
7687
}
7788

‎src/Paramore.Brighter/ExternalBusConfiguration.cs‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -147,6 +147,11 @@ public class ExternalBusConfiguration : IAmExternalBusConfiguration
147147
/// The Outbox we wish to use for messaging
148148
/// </summary>
149149
public IAmAnOutbox Outbox { get; set; }
150+
151+
/// <summary>
152+
/// The Distributed Locking Service
153+
/// </summary>
154+
public IDistributedLock DistributedLock { get; set; }
150155

151156
/// <summary>
152157
/// The maximum amount of messages to deposit into the outbox in one transmissions.

‎src/Paramore.Brighter/IDistributedLock.cs‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
using System.Threading;
2+
using System.Threading.Tasks;
3+
4+
namespace Paramore.Brighter;
5+
6+
public interface IDistributedLock
7+
{
8+
/// <summary>
9+
/// Attempt to obtain a lock on a resource
10+
/// </summary>
11+
/// <param name="resource">The name of the resource to Lock</param>
12+
/// <param name="cancellationToken">The Cancellation Token</param>
13+
/// <returns>True if the lock was obtained</returns>
14+
Task<bool> ObtainLockAsync(string resource, CancellationToken cancellationToken);
15+
16+
/// <summary>
17+
/// Attempt to obtain a lock on a resource
18+
/// </summary>
19+
/// <param name="resource">The name of the resource to Lock</param>
20+
/// <returns>True if the lock was obtained</returns>
21+
bool ObtainLock(string resource);
22+
23+
/// <summary>
24+
/// Release a lock
25+
/// </summary>
26+
/// <param name="resource"></param>
27+
/// <param name="cancellationToken"></param>
28+
/// <returns>Awaitable Task</returns>
29+
Task ReleaseLockAsync(string resource, CancellationToken cancellationToken);
30+
31+
/// <summary>
32+
/// Release a lock
33+
/// </summary>
34+
/// <param name="resource"></param>
35+
/// <returns>Awaitable Task</returns>
36+
void ReleaseLock(string resource);
37+
}

0 commit comments

Comments
 (0)