Skip to content

Commit a69ad8e

Browse files
authored
feat(threading): add AdaptiveSemaphore for dynamic concurrency control (#361)
* feat(threading): add AdaptiveSemaphore for dynamic concurrency control Introduces AdaptiveSemaphore, an async-compatible semaphore that allows adjusting the concurrency limit at runtime to respond to backpressure. - WaitAsync/Wait with IDisposable releaser for using pattern - TryShrinkAsync to reduce concurrency by permanently acquiring a permit - TryGrow to increase concurrency by releasing an additional permit - Configurable minimum and optional maximum bounds - Constructor validation and clamping of initial value - Full test coverage across construction, acquire/release, shrink, grow, and disposal scenarios * test: add rollback limit test for AdaptiveSemaphore on cancellation
1 parent c783288 commit a69ad8e

2 files changed

Lines changed: 543 additions & 0 deletions

File tree

Lines changed: 208 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,208 @@
1+
// MADE Apps licenses this file to you under the MIT license.
2+
// See the LICENSE file in the project root for more information.
3+
4+
namespace MADE.Threading;
5+
6+
/// <summary>
7+
/// Defines a semaphore that allows adjusting the concurrency limit at runtime to respond to backpressure.
8+
/// </summary>
9+
/// <remarks>
10+
/// This is useful for scenarios where the permitted concurrency should change dynamically,
11+
/// such as reducing parallelism when a downstream service returns rate-limit (429) responses
12+
/// or increasing it when conditions improve.
13+
/// <code>
14+
/// var semaphore = new AdaptiveSemaphore(initial: 10, minimum: 1, maximum: 20);
15+
///
16+
/// // Normal usage - acquire and release a permit.
17+
/// using (await semaphore.WaitAsync())
18+
/// {
19+
/// await httpClient.SendAsync(request);
20+
/// }
21+
///
22+
/// // Reduce concurrency on backpressure.
23+
/// await semaphore.TryShrinkAsync();
24+
/// </code>
25+
/// </remarks>
26+
public sealed class AdaptiveSemaphore : IDisposable
27+
{
28+
private readonly int minimum;
29+
private readonly int? maximum;
30+
private readonly SemaphoreSlim semaphore;
31+
private readonly object adjustLock = new();
32+
33+
private int limit;
34+
private bool disposed;
35+
36+
/// <summary>
37+
/// Initializes a new instance of the <see cref="AdaptiveSemaphore"/> class.
38+
/// </summary>
39+
/// <param name="initial">The initial number of concurrent permits.</param>
40+
/// <param name="minimum">The minimum concurrency limit. Defaults to 1.</param>
41+
/// <param name="maximum">The optional maximum concurrency limit.</param>
42+
/// <exception cref="ArgumentOutOfRangeException">Thrown when <paramref name="minimum"/> is less than 1, or <paramref name="maximum"/> is less than <paramref name="minimum"/>.</exception>
43+
public AdaptiveSemaphore(int initial, int minimum = 1, int? maximum = null)
44+
{
45+
ArgumentOutOfRangeException.ThrowIfLessThan(minimum, 1);
46+
47+
if (maximum.HasValue)
48+
{
49+
ArgumentOutOfRangeException.ThrowIfLessThan(maximum.Value, minimum, nameof(maximum));
50+
}
51+
52+
this.minimum = minimum;
53+
this.maximum = maximum;
54+
55+
var bounded = Math.Max(minimum, initial);
56+
if (maximum.HasValue)
57+
{
58+
bounded = Math.Min(bounded, maximum.Value);
59+
}
60+
61+
this.limit = bounded;
62+
this.semaphore = new SemaphoreSlim(bounded);
63+
}
64+
65+
/// <summary>
66+
/// Gets the current concurrency limit.
67+
/// </summary>
68+
public int Limit => this.limit;
69+
70+
/// <summary>
71+
/// Gets the number of permits currently available.
72+
/// </summary>
73+
public int Available => this.semaphore.CurrentCount;
74+
75+
/// <summary>
76+
/// Reduces the concurrency limit by one by permanently acquiring a permit.
77+
/// </summary>
78+
/// <remarks>
79+
/// If the limit is already at the <see cref="minimum"/>, this method does nothing.
80+
/// The acquired permit is not released, effectively reducing the pool of available permits.
81+
/// </remarks>
82+
/// <param name="cancellationToken">The cancellation token.</param>
83+
/// <returns>The new concurrency limit.</returns>
84+
/// <exception cref="ObjectDisposedException">Thrown if the semaphore has been disposed.</exception>
85+
public async Task<int> TryShrinkAsync(CancellationToken cancellationToken = default)
86+
{
87+
ObjectDisposedException.ThrowIf(this.disposed, this);
88+
89+
lock (this.adjustLock)
90+
{
91+
if (this.limit <= this.minimum)
92+
{
93+
return this.limit;
94+
}
95+
96+
this.limit--;
97+
}
98+
99+
try
100+
{
101+
await this.semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
102+
}
103+
catch
104+
{
105+
lock (this.adjustLock)
106+
{
107+
this.limit++;
108+
}
109+
110+
throw;
111+
}
112+
113+
return this.limit;
114+
}
115+
116+
/// <summary>
117+
/// Increases the concurrency limit by one by releasing an additional permit.
118+
/// </summary>
119+
/// <remarks>
120+
/// If the limit is already at the <see cref="maximum"/>, this method does nothing.
121+
/// This is the inverse of <see cref="TryShrinkAsync"/> and should be called when
122+
/// conditions improve and more concurrency is desirable.
123+
/// </remarks>
124+
/// <returns>The new concurrency limit.</returns>
125+
/// <exception cref="ObjectDisposedException">Thrown if the semaphore has been disposed.</exception>
126+
public int TryGrow()
127+
{
128+
ObjectDisposedException.ThrowIf(this.disposed, this);
129+
130+
lock (this.adjustLock)
131+
{
132+
if (this.maximum.HasValue && this.limit >= this.maximum.Value)
133+
{
134+
return this.limit;
135+
}
136+
137+
this.limit++;
138+
139+
try
140+
{
141+
this.semaphore.Release();
142+
}
143+
catch
144+
{
145+
this.limit--;
146+
throw;
147+
}
148+
149+
return this.limit;
150+
}
151+
}
152+
153+
/// <summary>
154+
/// Asynchronously waits to acquire a permit from the semaphore.
155+
/// </summary>
156+
/// <param name="cancellationToken">The cancellation token.</param>
157+
/// <returns>An <see cref="IDisposable"/> that releases the permit when disposed.</returns>
158+
/// <exception cref="ObjectDisposedException">Thrown if the semaphore has been disposed.</exception>
159+
public async Task<IDisposable> WaitAsync(CancellationToken cancellationToken = default)
160+
{
161+
ObjectDisposedException.ThrowIf(this.disposed, this);
162+
await this.semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
163+
return new SemaphoreReleaser(this.semaphore);
164+
}
165+
166+
/// <summary>
167+
/// Synchronously waits to acquire a permit from the semaphore.
168+
/// </summary>
169+
/// <param name="cancellationToken">The cancellation token.</param>
170+
/// <returns>An <see cref="IDisposable"/> that releases the permit when disposed.</returns>
171+
/// <exception cref="ObjectDisposedException">Thrown if the semaphore has been disposed.</exception>
172+
public IDisposable Wait(CancellationToken cancellationToken = default)
173+
{
174+
ObjectDisposedException.ThrowIf(this.disposed, this);
175+
this.semaphore.Wait(cancellationToken);
176+
return new SemaphoreReleaser(this.semaphore);
177+
}
178+
179+
/// <inheritdoc />
180+
public void Dispose()
181+
{
182+
lock (this.adjustLock)
183+
{
184+
if (this.disposed)
185+
{
186+
return;
187+
}
188+
189+
this.semaphore.Dispose();
190+
this.disposed = true;
191+
}
192+
}
193+
194+
private sealed class SemaphoreReleaser : IDisposable
195+
{
196+
private SemaphoreSlim? semaphore;
197+
198+
public SemaphoreReleaser(SemaphoreSlim semaphore)
199+
{
200+
this.semaphore = semaphore;
201+
}
202+
203+
public void Dispose()
204+
{
205+
Interlocked.Exchange(ref this.semaphore, null)?.Release();
206+
}
207+
}
208+
}

0 commit comments

Comments
 (0)