Skip to content

Commit 4453050

Browse files
steingransteingran
committed
feat: add distributed counting semaphore with multi-provider support
Implement a distributed counting semaphore abstraction (ISemaphoreService / ISemaphoreProvider) mirroring the existing leader-election design, with providers for PostgreSQL, SQL Server, Redis, Azure Blob Storage, Consul, ZooKeeper, FileSystem, and InMemory. Key design decisions and fixes included in this commit: - AutoStart=false defers acquisition to the caller; StartAsync drives acquisition when true; ExecuteAsync is a no-op stub (BackgroundService requirement) - GetCurrentCountAsync filters by heartbeat age (slotTimeout) so expired holders are excluded from the live count across all providers - PostgreSQL uses pg_advisory_xact_lock(hashtext(semaphoreName)) under ReadCommitted instead of SERIALIZABLE + FOR UPDATE, which deadlocked on empty tables - SQL Server uses sp_getapplock(@resource, 'Exclusive', 'Transaction') under ReadCommitted instead of SERIALIZABLE + range locks, which caused deadlocks when the table was empty - Consul and ZooKeeper apply a post-acquisition count recheck to close the TOCTOU window between the count read and the slot creation - FileSystem provider validates canonical paths to prevent directory traversal via semaphoreName or holderId - Schema and table names are validated against ^[a-zA-Z_][a-zA-Z0-9_]*$ (max 128 chars) before interpolation into SQL to prevent injection - InMemory provider replaced ConcurrentDictionary with plain Dictionary guarded by a single lock, removing redundant thread-safety layering - SemaphoreAcquisition.Dispose() delegates to DisposeAsync() to avoid sync-context deadlocks from direct GetAwaiter().GetResult() on the task - SemaphoreService.DisposeAsync() awaits broadcastTask before completing - UpdateStatus() is a plain void method; the previous async overload only awaited Task.CompletedTask and accepted an unused CancellationToken - HeartbeatTimeout (not the now-removed SlotTimeout) is passed as the expiry window to all provider calls - IAsyncDisposable removed from SemaphoreService class declaration (already satisfied via ISemaphoreService) - CurrentCount decrements are guarded with Math.Max(0, ...) to prevent negative counts on spurious releases - Integration tests: each test instance uses a unique table name and semaphore name (Guid-per-instance); DisposeAsync drops the table; ConcurrentAcquire asserts successCount == maxCount and cross-checks via GetCurrentCountAsync Fixes #100
1 parent 9e5bcf6 commit 4453050

77 files changed

Lines changed: 10529 additions & 78 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎MultiLock.slnx‎

Lines changed: 15 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -1,21 +1,24 @@
11
<Solution>
22
<Folder Name="/Providers/">
3-
<Project Path="src\Providers\MultiLock.AzureBlobStorage\MultiLock.AzureBlobStorage.csproj" Type="Classic C#" />
4-
<Project Path="src\Providers\MultiLock.Consul\MultiLock.Consul.csproj" Type="Classic C#" />
5-
<Project Path="src\Providers\MultiLock.FileSystem\MultiLock.FileSystem.csproj" Type="Classic C#" />
6-
<Project Path="src\Providers\MultiLock.InMemory\MultiLock.InMemory.csproj" Type="Classic C#" />
7-
<Project Path="src\Providers\MultiLock.PostgreSQL\MultiLock.PostgreSQL.csproj" Type="Classic C#" />
8-
<Project Path="src\Providers\MultiLock.Redis\MultiLock.Redis.csproj" Type="Classic C#" />
9-
<Project Path="src\Providers\MultiLock.SqlServer\MultiLock.SqlServer.csproj" Type="Classic C#" />
10-
<Project Path="src\Providers\MultiLock.ZooKeeper\MultiLock.ZooKeeper.csproj" Type="Classic C#" />
3+
<Project Path="src\Providers\MultiLock.AzureBlobStorage\MultiLock.AzureBlobStorage.csproj" Type="C#" />
4+
<Project Path="src\Providers\MultiLock.Consul\MultiLock.Consul.csproj" Type="C#" />
5+
<Project Path="src\Providers\MultiLock.FileSystem\MultiLock.FileSystem.csproj" Type="C#" />
6+
<Project Path="src\Providers\MultiLock.InMemory\MultiLock.InMemory.csproj" Type="C#" />
7+
<Project Path="src\Providers\MultiLock.PostgreSQL\MultiLock.PostgreSQL.csproj" Type="C#" />
8+
<Project Path="src\Providers\MultiLock.Redis\MultiLock.Redis.csproj" Type="C#" />
9+
<Project Path="src\Providers\MultiLock.SqlServer\MultiLock.SqlServer.csproj" Type="C#" />
10+
<Project Path="src\Providers\MultiLock.ZooKeeper\MultiLock.ZooKeeper.csproj" Type="C#" />
1111
</Folder>
1212
<Folder Name="/Samples/">
13-
<Project Path="samples\MultiLock.MultiProvider\MultiLock.MultiProvider.csproj" Type="Classic C#" />
14-
<Project Path="samples\MultiLock.Sample\MultiLock.Sample.csproj" Type="Classic C#" />
13+
<Project Path="samples/MultiLock.SemaphoreSample/MultiLock.SemaphoreSample.csproj" />
14+
<Project Path="samples\MultiLock.MultiProvider\MultiLock.MultiProvider.csproj" Type="C#" />
15+
<Project Path="samples\MultiLock.Sample\MultiLock.Sample.csproj" Type="C#" />
1516
</Folder>
17+
<Folder Name="/src/" />
18+
<Folder Name="/src/Providers/" />
1619
<Folder Name="/Tests/">
17-
<Project Path="tests\MultiLock.IntegrationTests\MultiLock.IntegrationTests.csproj" Type="Classic C#" />
18-
<Project Path="tests\MultiLock.Tests\MultiLock.Tests.csproj" Type="Classic C#" />
20+
<Project Path="tests\MultiLock.IntegrationTests\MultiLock.IntegrationTests.csproj" Type="C#" />
21+
<Project Path="tests\MultiLock.Tests\MultiLock.Tests.csproj" Type="C#" />
1922
</Folder>
2023
<Folder Name="/_Files/">
2124
<File Path="tests\docker-compose.yml" />

‎README.md‎

Lines changed: 198 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@
1111
[![NuGet - Consul](https://img.shields.io/nuget/v/MultiLock.Consul.svg?label=Consul)](https://www.nuget.org/packages/MultiLock.Consul/)
1212
[![NuGet - ZooKeeper](https://img.shields.io/nuget/v/MultiLock.ZooKeeper.svg?label=ZooKeeper)](https://www.nuget.org/packages/MultiLock.ZooKeeper/)
1313

14-
A comprehensive .NET framework for implementing the Leader Election pattern with support for multiple providers including Azure Blob Storage, SQL Server, Redis, File System, In-Memory, Consul, and ZooKeeper.
14+
A comprehensive .NET framework for implementing **Leader Election** and **Distributed Semaphores** patterns with support for multiple providers including Azure Blob Storage, SQL Server, PostgreSQL, Redis, File System, In-Memory, Consul, and ZooKeeper.
1515

1616
## Requirements
1717

@@ -37,32 +37,25 @@ A comprehensive .NET framework for implementing the Leader Election pattern with
3737
┌────────────────────────────────────────────────────────────────────────────┐
3838
│ CORE FRAMEWORK │
3939
│ │
40-
│ ┌──────────────────────────────────────────────────────────────┐ │
41-
│ │ ILeaderElectionService (Interface) │ │
42-
│ │ • StartAsync() / StopAsync() │ │
43-
│ │ • IsLeader / GetCurrentLeaderAsync() │ │
44-
│ │ • GetLeadershipChangesAsync() │ │
45-
│ └────────────────────────┬─────────────────────────────────────┘ │
46-
│ │ │
47-
│ ▼ │
48-
│ ┌──────────────────────────────────────────────────────────────┐ │
49-
│ │ LeaderElectionService (Implementation) │ │
50-
│ │ • Election Logic │ │
51-
│ │ • Heartbeat Monitoring │ │
52-
│ │ • Event Publishing │ │
53-
│ └────────────────────────┬─────────────────────────────────────┘ │
54-
│ │ │
55-
│ ▼ │
56-
│ ┌──────────────────────────────────────────────────────────────┐ │
57-
│ │ ILeaderElectionProvider (Interface) │ │
58-
│ │ • TryAcquireLeadershipAsync() │ │
59-
│ │ • ReleaseLeadershipAsync() │ │
60-
│ │ • UpdateHeartbeatAsync() │ │
61-
│ └────────────────────────┬─────────────────────────────────────┘ │
62-
│ │ │
63-
└─────────────────────────────┼──────────────────────────────────────────────┘
64-
│
65-
▼
40+
│ ┌─────────────────────────────────┐ ┌─────────────────────────────────┐ │
41+
│ │ ILeaderElectionService │ │ ISemaphoreService │ │
42+
│ │ • StartAsync() / StopAsync() │ │ • AcquireAsync() │ │
43+
│ │ • IsLeader │ │ • TryAcquireAsync() │ │
44+
│ │ • GetLeadershipChangesAsync() │ │ • GetStatusChangesAsync() │ │
45+
│ └───────────────┬─────────────────┘ └───────────────┬─────────────────┘ │
46+
│ │ │ │
47+
│ ▼ ▼ │
48+
│ ┌─────────────────────────────────┐ ┌─────────────────────────────────┐ │
49+
│ │ ILeaderElectionProvider │ │ ISemaphoreProvider │ │
50+
│ │ • TryAcquireLeadershipAsync() │ │ • TryAcquireAsync() │ │
51+
│ │ • ReleaseLeadershipAsync() │ │ • ReleaseAsync() │ │
52+
│ │ • UpdateHeartbeatAsync() │ │ • UpdateHeartbeatAsync() │ │
53+
│ └───────────────┬─────────────────┘ └───────────────┬─────────────────┘ │
54+
│ │ │ │
55+
└──────────────────┼────────────────────────────────────┼────────────────────┘
56+
│ │
57+
└────────────────┬───────────────────┘
58+
▼
6659
┌────────────────────────────────────────────────────────────────────────────┐
6760
│ PROVIDERS │
6861
│ │
@@ -82,8 +75,9 @@ A comprehensive .NET framework for implementing the Leader Election pattern with
8275
## Features
8376

8477
- **Multiple Providers**: Support for various storage backends
85-
- **Distributed Mutex**: First-to-acquire becomes leader strategy
86-
- **Heartbeat Monitoring**: Automatic leader health monitoring and failover
78+
- **Leader Election**: First-to-acquire becomes leader strategy with automatic failover
79+
- **Distributed Semaphores**: Control concurrent access with configurable slot limits
80+
- **Heartbeat Monitoring**: Automatic health monitoring and failover
8781
- **Resilient Design**: Handles transient and persistent failures gracefully
8882
- **Thread-Safe**: All operations are thread-safe
8983
- **Configurable**: Extensive configuration options for timeouts, retry policies, etc.
@@ -602,6 +596,181 @@ Use `GetLeadershipChangesAsync()` to subscribe to these events and optionally fi
602596
- **LockTimeout**: Maximum time to hold a lock during election
603597
- **AutoStart**: Whether to automatically start the election process
604598

599+
## Distributed Semaphores
600+
601+
Distributed semaphores allow you to control concurrent access to a shared resource across multiple instances. Unlike leader election (which allows only one holder), semaphores allow a configurable number of concurrent holders.
602+
603+
### Use Cases
604+
605+
- **Rate Limiting**: Limit concurrent API calls to an external service
606+
- **Resource Pooling**: Control access to a limited pool of resources (database connections, licenses, etc.)
607+
- **Throttling**: Limit concurrent processing of expensive operations
608+
- **Capacity Management**: Ensure only N instances process work simultaneously
609+
610+
### Quick Start
611+
612+
```csharp
613+
using MultiLock;
614+
using MultiLock.InMemory;
615+
616+
var builder = WebApplication.CreateBuilder(args);
617+
618+
// Add semaphore with In-Memory provider (max 5 concurrent holders)
619+
builder.Services.AddSemaphore<InMemorySemaphoreProvider>(options =>
620+
{
621+
options.SemaphoreName = "api-rate-limiter";
622+
options.MaxCount = 5;
623+
options.HeartbeatInterval = TimeSpan.FromSeconds(30);
624+
options.HeartbeatTimeout = TimeSpan.FromSeconds(90);
625+
});
626+
627+
var app = builder.Build();
628+
```
629+
630+
### Using the Semaphore Service
631+
632+
```csharp
633+
public class RateLimitedApiClient
634+
{
635+
private readonly ISemaphoreService _semaphore;
636+
private readonly HttpClient _httpClient;
637+
638+
public RateLimitedApiClient(ISemaphoreService semaphore, HttpClient httpClient)
639+
{
640+
_semaphore = semaphore;
641+
_httpClient = httpClient;
642+
}
643+
644+
public async Task<string> CallExternalApiAsync(CancellationToken cancellationToken)
645+
{
646+
// Block until a slot is available (or cancellationToken is triggered)
647+
await _semaphore.WaitForSlotAsync(cancellationToken);
648+
649+
try
650+
{
651+
// We now hold a slot - make the API call
652+
return await _httpClient.GetStringAsync("/api/data", cancellationToken);
653+
}
654+
finally
655+
{
656+
// Always release the slot, even if the call throws
657+
await _semaphore.ReleaseAsync(cancellationToken);
658+
}
659+
}
660+
661+
public async Task<string?> TryCallExternalApiAsync(CancellationToken cancellationToken)
662+
{
663+
// Try to acquire without blocking; returns false if all slots are taken
664+
bool acquired = await _semaphore.TryAcquireAsync(cancellationToken);
665+
666+
if (!acquired)
667+
return null;
668+
669+
try
670+
{
671+
return await _httpClient.GetStringAsync("/api/data", cancellationToken);
672+
}
673+
finally
674+
{
675+
await _semaphore.ReleaseAsync(cancellationToken);
676+
}
677+
}
678+
}
679+
```
680+
681+
### Monitoring Semaphore Status
682+
683+
```csharp
684+
// Check current status synchronously via the property
685+
SemaphoreStatus status = semaphore.CurrentStatus;
686+
Console.WriteLine($"Holding: {status.IsHolding}");
687+
Console.WriteLine($"Available: {status.AvailableSlots}/{status.MaxCount}");
688+
689+
// Or fetch the latest state including all holders asynchronously
690+
SemaphoreInfo? info = await semaphore.GetSemaphoreInfoAsync(cancellationToken);
691+
if (info != null)
692+
Console.WriteLine($"[Async] Active holders: {info.CurrentCount}/{info.MaxCount}");
693+
694+
// Subscribe to status changes
695+
await foreach (var change in semaphore.GetStatusChangesAsync(cancellationToken))
696+
{
697+
if (change.AcquiredSlot)
698+
{
699+
Console.WriteLine("Acquired a semaphore slot!");
700+
}
701+
else if (change.LostSlot)
702+
{
703+
Console.WriteLine("Lost semaphore slot!");
704+
}
705+
}
706+
```
707+
708+
### Provider-Specific Configuration
709+
710+
All providers support semaphores with the same API:
711+
712+
```csharp
713+
// PostgreSQL
714+
builder.Services.AddPostgreSqlSemaphore(
715+
connectionString: "Host=localhost;Database=MyApp;...",
716+
options => { options.SemaphoreName = "my-semaphore"; options.MaxCount = 10; });
717+
718+
// Redis
719+
builder.Services.AddRedisSemaphore(
720+
connectionString: "localhost:6379",
721+
options => { options.SemaphoreName = "my-semaphore"; options.MaxCount = 10; });
722+
723+
// SQL Server
724+
builder.Services.AddSqlServerSemaphore(
725+
connectionString: "Server=localhost;Database=MyApp;...",
726+
options => { options.SemaphoreName = "my-semaphore"; options.MaxCount = 10; });
727+
728+
// Azure Blob Storage
729+
builder.Services.AddAzureBlobStorageSemaphore(
730+
connectionString: "DefaultEndpointsProtocol=https;...",
731+
options => { options.SemaphoreName = "my-semaphore"; options.MaxCount = 10; });
732+
733+
// Consul
734+
builder.Services.AddConsulSemaphore(
735+
address: "http://localhost:8500",
736+
options => { options.SemaphoreName = "my-semaphore"; options.MaxCount = 10; });
737+
738+
// ZooKeeper
739+
builder.Services.AddZooKeeperSemaphore(
740+
connectionString: "localhost:2181",
741+
options => { options.SemaphoreName = "my-semaphore"; options.MaxCount = 10; });
742+
743+
// File System
744+
builder.Services.AddFileSystemSemaphore(
745+
basePath: "/var/locks",
746+
options => { options.SemaphoreName = "my-semaphore"; options.MaxCount = 10; });
747+
```
748+
749+
### Semaphore Configuration Options
750+
751+
| Option | Description | Default |
752+
|--------|-------------|---------|
753+
| `SemaphoreName` | Unique name for the semaphore | Required |
754+
| `MaxCount` | Maximum concurrent holders | Required |
755+
| `HolderId` | Unique identifier for this holder | Auto-generated |
756+
| `HeartbeatInterval` | How often to send heartbeats | 10 seconds |
757+
| `HeartbeatTimeout` | Time before a holder is considered dead | 30 seconds |
758+
| `AcquisitionInterval` | How often to retry acquiring a slot | 5 seconds |
759+
| `MaxRetryAttempts` | Retry attempts for transient failures | 3 |
760+
| `RetryBaseDelay` | Base delay between retries | 100ms |
761+
| `RetryMaxDelay` | Maximum delay between retries | 5 seconds |
762+
| `AutoStart` | Start acquiring on service start | true |
763+
| `EnableDetailedLogging` | Enable verbose logging | false |
764+
765+
### Semaphore vs Leader Election
766+
767+
| Feature | Leader Election | Semaphore |
768+
|---------|-----------------|-----------|
769+
| Concurrent holders | 1 (exclusive) | N (configurable) |
770+
| Use case | Single leader tasks | Rate limiting, pooling |
771+
| Failover | Automatic re-election | Slot becomes available |
772+
| API | `IsLeader` property | `IsHolding` property |
773+
605774
## Testing
606775

607776
The framework includes comprehensive test coverage:

‎samples/MultiLock.SemaphoreSample/MultiLock.SemaphoreSample.csproj‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
<Project Sdk="Microsoft.NET.Sdk">
2+
3+
<PropertyGroup>
4+
<OutputType>Exe</OutputType>
5+
<TargetFrameworks>net8.0;net10.0</TargetFrameworks>
6+
<ImplicitUsings>enable</ImplicitUsings>
7+
<Nullable>enable</Nullable>
8+
<IsPackable>false</IsPackable>
9+
</PropertyGroup>
10+
11+
<ItemGroup>
12+
<PackageReference Include="Microsoft.Extensions.Hosting" />
13+
<PackageReference Include="Microsoft.Extensions.Logging.Console" />
14+
</ItemGroup>
15+
16+
<ItemGroup>
17+
<ProjectReference Include="..\..\src\MultiLock\MultiLock.csproj" />
18+
<ProjectReference Include="..\..\src\Providers\MultiLock.InMemory\MultiLock.InMemory.csproj" />
19+
<ProjectReference Include="..\..\src\Providers\MultiLock.FileSystem\MultiLock.FileSystem.csproj" />
20+
<ProjectReference Include="..\..\src\Providers\MultiLock.Redis\MultiLock.Redis.csproj" />
21+
<ProjectReference Include="..\..\src\Providers\MultiLock.SqlServer\MultiLock.SqlServer.csproj" />
22+
<ProjectReference Include="..\..\src\Providers\MultiLock.PostgreSQL\MultiLock.PostgreSQL.csproj" />
23+
<ProjectReference Include="..\..\src\Providers\MultiLock.AzureBlobStorage\MultiLock.AzureBlobStorage.csproj" />
24+
<ProjectReference Include="..\..\src\Providers\MultiLock.Consul\MultiLock.Consul.csproj" />
25+
<ProjectReference Include="..\..\src\Providers\MultiLock.ZooKeeper\MultiLock.ZooKeeper.csproj" />
26+
</ItemGroup>
27+
28+
</Project>
29+

0 commit comments

Comments
 (0)