Skip to content

Commit 6ee3eb6

Browse files
madelsonmadelson
authored
Merge pull request #88 from madelson/release-2.0.2
Release 2.0.2
2 parents 7c7fa82 + a60febb commit 6ee3eb6

16 files changed

Lines changed: 142 additions & 62 deletions

‎DistributedLock.Core/DistributedLock.Core.csproj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@
1010
</PropertyGroup>
1111

1212
<PropertyGroup>
13-
<Version>1.0.0</Version>
13+
<Version>1.0.1</Version>
1414
<AssemblyVersion>1.0.0.0</AssemblyVersion>
1515
<Authors>Michael Adelson</Authors>
1616
<Description>Core interfaces and utilities that support the DistributedLock.* family of packages</Description>

‎DistributedLock.Core/Internal/Data/ConnectionMonitor.cs‎

Lines changed: 20 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -248,8 +248,12 @@ private async ValueTask StopOrDisposeAsync(bool isDispose)
248248
this.CloseOrCancelMonitoringHandleRegistrationsNoLock(isCancel: false);
249249

250250
task = this._monitoringWorkerTask;
251-
this._monitorStateChangedTokenSource?.Cancel();
252251

252+
// Note: synchronous cancel here should be safe because we've already set
253+
// the state to disposed above which the monitoring loop will check if it
254+
// takes over the Cancel() thread.
255+
this._monitorStateChangedTokenSource?.Cancel();
256+
253257
// unsubscribe from state change tracking
254258
if (this._stateChangedHandler != null
255259
&& this._weakConnection.TryGetTarget(out var connection))
@@ -304,7 +308,7 @@ private bool StartMonitorWorkerIfNeededNoLock()
304308

305309
// skip if there's nothing to do
306310
if (this._keepaliveCadence.IsInfinite && !this.HasRegisteredMonitoringHandlesNoLock) { return false; }
307-
311+
308312
this._monitorStateChangedTokenSource = new CancellationTokenSource();
309313
// Set up the task as a continuation on the previous task to avoid concurrency in the case where the previous
310314
// one is spinning down. If we change states in rapid succession we could end up with multiple tasks queued up
@@ -318,9 +322,20 @@ private bool StartMonitorWorkerIfNeededNoLock()
318322

319323
private void FireStateChangedNoLock()
320324
{
321-
this._monitorStateChangedTokenSource!.Cancel();
322-
this._monitorStateChangedTokenSource.Dispose();
325+
var monitorStateChangedTokenSource = this._monitorStateChangedTokenSource!;
323326
this._monitorStateChangedTokenSource = new CancellationTokenSource();
327+
// Canceling asynchronously is important because the Cancel() thread can end up
328+
// running continuations inside the monitoring loop (e. g. see
329+
// https://github.com/madelson/DistributedLock/issues/85). Now that we set the new
330+
// token source before canceling the old one we should avoid that particular issue, but
331+
// it is still safer and easier to reason about not to have that happen. This also ensures
332+
// that FireStateChangedNoLock() always returns quickly, even if the monitoring loop
333+
// were to do some synchronous work on the continuation thread.
334+
Task.Run(() =>
335+
{
336+
try { monitorStateChangedTokenSource.Cancel(); }
337+
finally { monitorStateChangedTokenSource.Dispose(); }
338+
});
324339
}
325340

326341
private async Task MonitorWorkerLoop()
@@ -357,7 +372,7 @@ private async Task<bool> DoMonitoringAsync(CancellationToken cancellationToken)
357372
using var _ = await this._connectionLock.AcquireAsync(CancellationToken.None).ConfigureAwait(false);
358373

359374
// 1-min increments is kind of an arbitrary choice. We want to avoid this being too short since each time
360-
// we "come up to breathe" that's a waste of resource. We also want to avoid this being too long since
375+
// we "come up to breathe" that's a waste of resources. We also want to avoid this being too long since
361376
// in case people have some kind of monitoring set up for hanging queries
362377
await connection.SleepAsync(
363378
sleepTime: TimeSpan.FromMinutes(1),

‎DistributedLock.Core/Internal/Data/DatabaseConnection.cs‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -112,14 +112,14 @@ private async ValueTask DisposeOrCloseAsync(bool isDispose)
112112
finally
113113
{
114114
#if NETSTANDARD2_1
115-
if (!SyncViaAsync.IsSynchronous && this.InnerConnection is DbConnection dbConnection)
116-
{
117-
await (isDispose ? dbConnection.DisposeAsync() : dbConnection.CloseAsync().AsValueTask()).ConfigureAwait(false);
118-
}
119-
else
120-
{
121-
SyncDisposeConnection();
122-
}
115+
if (!SyncViaAsync.IsSynchronous && this.InnerConnection is DbConnection dbConnection)
116+
{
117+
await (isDispose ? dbConnection.DisposeAsync() : dbConnection.CloseAsync().AsValueTask()).ConfigureAwait(false);
118+
}
119+
else
120+
{
121+
SyncDisposeConnection();
122+
}
123123
#elif NETSTANDARD2_0 || NET461
124124
SyncDisposeConnection();
125125
#else

‎DistributedLock.Core/Internal/Data/MultiplexedConnectionLock.cs‎

Lines changed: 32 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -18,12 +18,20 @@ internal sealed class MultiplexedConnectionLock : IAsyncDisposable
1818
private readonly AsyncLock _mutex = AsyncLock.Create();
1919
private readonly Dictionary<string, TimeoutValue> _heldLocksToKeepaliveCadences = new Dictionary<string, TimeoutValue>();
2020
private readonly DatabaseConnection _connection;
21+
/// <summary>
22+
/// Tracks whether we've successfully opened the connection. We track this explicity instead of just looking at
23+
/// <see cref="DatabaseConnection.CanExecuteQueries"/> because we want to make sure we close() explicitly for every
24+
/// open() and also we want to make sure we do not try to re-open a broken connection.
25+
/// </summary>
26+
private bool _connectionOpened;
2127

2228
public MultiplexedConnectionLock(DatabaseConnection connection)
2329
{
2430
this._connection = connection;
2531
}
2632

33+
private bool IsConnectionBrokenNoLock => this._connectionOpened && !this._connection.CanExecuteQueries;
34+
2735
public async ValueTask<Result> TryAcquireAsync<TLockCookie>(
2836
string name,
2937
TimeoutValue timeout,
@@ -33,8 +41,8 @@ public async ValueTask<Result> TryAcquireAsync<TLockCookie>(
3341
bool opportunistic)
3442
where TLockCookie : class
3543
{
36-
using var mutextHandle = await this._mutex.TryAcquireAsync(opportunistic ? TimeSpan.Zero : Timeout.InfiniteTimeSpan, cancellationToken).ConfigureAwait(false);
37-
if (mutextHandle == null)
44+
using var mutexHandle = await this._mutex.TryAcquireAsync(opportunistic ? TimeSpan.Zero : Timeout.InfiniteTimeSpan, cancellationToken).ConfigureAwait(false);
45+
if (mutexHandle == null)
3846
{
3947
// mutex wasn't free, so just give up
4048
Invariant.Require(opportunistic);
@@ -43,6 +51,10 @@ public async ValueTask<Result> TryAcquireAsync<TLockCookie>(
4351
return new Result(MultiplexedConnectionLockRetry.Retry, canSafelyDispose: false);
4452
}
4553

54+
// This is technically redundant with the similar catch block below, but avoids needing to have
55+
// to attempt a query on a connection that we know is broken.
56+
if (opportunistic && this.IsConnectionBrokenNoLock) { return this.GetAlreadyBrokenResultNoLock(); }
57+
4658
try
4759
{
4860
if (this._heldLocksToKeepaliveCadences.ContainsKey(name))
@@ -53,9 +65,10 @@ public async ValueTask<Result> TryAcquireAsync<TLockCookie>(
5365
return this.GetFailureResultNoLock(isAlreadyHeld: true, opportunistic, timeout);
5466
}
5567

56-
if (!this._connection.CanExecuteQueries)
68+
if (!this._connectionOpened)
5769
{
5870
await this._connection.OpenAsync(cancellationToken).ConfigureAwait(false);
71+
this._connectionOpened = true;
5972
}
6073

6174
var lockCookie = await strategy.TryAcquireAsync(this._connection, name, opportunistic ? TimeSpan.Zero : timeout, cancellationToken).ConfigureAwait(false);
@@ -71,6 +84,11 @@ public async ValueTask<Result> TryAcquireAsync<TLockCookie>(
7184
// shortened the timeout
7285
return this.GetFailureResultNoLock(isAlreadyHeld: false, opportunistic, timeout);
7386
}
87+
// never punish for the connection being broken already (see https://github.com/madelson/DistributedLock/issues/83)
88+
catch when (opportunistic && this.IsConnectionBrokenNoLock)
89+
{
90+
return this.GetAlreadyBrokenResultNoLock();
91+
}
7492
finally
7593
{
7694
await this.CloseConnectionIfNeededNoLockAsync().ConfigureAwait(false);
@@ -90,6 +108,11 @@ public async ValueTask<bool> GetIsInUseAsync()
90108
return mutexHandle == null || this._heldLocksToKeepaliveCadences.Count != 0;
91109
}
92110

111+
private Result GetAlreadyBrokenResultNoLock() =>
112+
// Retry on any already-broken connection to avoid "leaking" the killing or death of connections. We want there to be no observable
113+
// results (other than perf) of multiplexing vs. not.
114+
new Result(MultiplexedConnectionLockRetry.Retry, canSafelyDispose: this._heldLocksToKeepaliveCadences.Count == 0);
115+
93116
private Result GetFailureResultNoLock(bool isAlreadyHeld, bool opportunistic, TimeoutValue timeout)
94117
{
95118
// only opportunistic acquisitions trigger retries
@@ -151,11 +174,13 @@ private async ValueTask ReleaseAsync<TLockCookie>(IDbSynchronizationStrategy<TLo
151174
}
152175
}
153176

154-
private ValueTask CloseConnectionIfNeededNoLockAsync()
177+
private async ValueTask CloseConnectionIfNeededNoLockAsync()
155178
{
156-
return this._heldLocksToKeepaliveCadences.Count == 0 && this._connection.CanExecuteQueries
157-
? this._connection.CloseAsync()
158-
: default;
179+
if (this._connectionOpened && this._heldLocksToKeepaliveCadences.Count == 0)
180+
{
181+
await this._connection.CloseAsync().ConfigureAwait(false);
182+
this._connectionOpened = false;
183+
}
159184
}
160185

161186
private void SetKeepaliveCadenceNoLock()

‎DistributedLock.Core/Internal/Data/MultiplexedConnectionLockPool.cs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,7 @@ public MultiplexedConnectionLockPool(Func<string, DatabaseConnection> connection
5151
{
5252
// opportunistic phase: see if we can use a connection that is already holding a lock
5353
// to acquire the current lock
54-
var existingLock = await this.GetOrCreateLockAsync(connectionString).ConfigureAwait(false);
54+
var existingLock = await this.GetExistingLockOrDefaultAsync(connectionString).ConfigureAwait(false);
5555
if (existingLock != null)
5656
{
5757
var canSafelyDisposeExistingLock = false;
@@ -89,6 +89,7 @@ public MultiplexedConnectionLockPool(Func<string, DatabaseConnection> connection
8989
try
9090
{
9191
result = await TryAcquireAsync(@lock, opportunistic: false).ConfigureAwait(false);
92+
Invariant.Require(result!.Value.Retry == MultiplexedConnectionLockRetry.NoRetry, "Acquire on fresh lock should not recommend a retry");
9293
}
9394
finally
9495
{
@@ -101,7 +102,7 @@ public MultiplexedConnectionLockPool(Func<string, DatabaseConnection> connection
101102
@lock.TryAcquireAsync(name, timeout, strategy, keepaliveCadence, cancellationToken, opportunistic);
102103
}
103104

104-
private async ValueTask<MultiplexedConnectionLock?> GetOrCreateLockAsync(string connectionString)
105+
private async ValueTask<MultiplexedConnectionLock?> GetExistingLockOrDefaultAsync(string connectionString)
105106
{
106107
using var _ = await this._lock.AcquireAsync(CancellationToken.None).ConfigureAwait(false);
107108

‎DistributedLock.Postgres/DistributedLock.Postgres.csproj‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@
1010
</PropertyGroup>
1111

1212
<PropertyGroup>
13-
<Version>1.0.0</Version>
13+
<Version>1.0.1</Version>
1414
<AssemblyVersion>1.0.0.0</AssemblyVersion>
1515
<Authors>Michael Adelson</Authors>
1616
<Description>Provides a distributed lock implementation based on Postgresql</Description>
@@ -41,7 +41,7 @@
4141
</PropertyGroup>
4242

4343
<ItemGroup>
44-
<PackageReference Include="Npgsql" Version="4.1.7" />
44+
<PackageReference Include="Npgsql" Version="5.0.4" />
4545
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All"/>
4646
</ItemGroup>
4747

‎DistributedLock.SqlServer/DistributedLock.SqlServer.csproj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@
1010
</PropertyGroup>
1111

1212
<PropertyGroup>
13-
<Version>1.0.0</Version>
13+
<Version>1.0.1</Version>
1414
<AssemblyVersion>1.0.0.0</AssemblyVersion>
1515
<Authors>Michael Adelson</Authors>
1616
<Description>Provides a distributed lock implementation based on SQL Server</Description>

‎DistributedLock.Tests/AbstractTestCases/Data/ConnectionStringStrategyTestCases.cs‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
using System;
44
using System.Diagnostics;
55
using System.Threading;
6+
using System.Threading.Tasks;
67

78
namespace Medallion.Threading.Tests.Data
89
{
@@ -95,5 +96,29 @@ public void TestKeepaliveDoesNotCreateRaceCondition()
9596
}
9697
});
9798
}
99+
100+
// replicates issue from https://github.com/madelson/DistributedLock/issues/85
101+
[Test]
102+
public async Task TestAccessingHandleLostTokenWhileKeepaliveActiveDoesNotBlock()
103+
{
104+
this._lockProvider.Strategy.KeepaliveCadence = TimeSpan.FromMinutes(5);
105+
106+
var @lock = this._lockProvider.CreateLock(string.Empty);
107+
var handle = await @lock.TryAcquireAsync();
108+
if (handle != null)
109+
{
110+
var accessHandleLostTokenTask = Task.Run(() =>
111+
{
112+
if (handle.HandleLostToken.CanBeCanceled)
113+
{
114+
handle.HandleLostToken.Register(() => { });
115+
}
116+
});
117+
Assert.IsTrue(await accessHandleLostTokenTask.WaitAsync(TimeSpan.FromSeconds(5)));
118+
119+
// do this only on success; on failure we're likely deadlocked and dispose will hang
120+
await handle.DisposeAsync();
121+
}
122+
}
98123
}
99124
}

‎DistributedLock.Tests/AbstractTestCases/Data/MultiplexingConnectionStrategyTestCases.cs‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
using NUnit.Framework;
33
using System;
44
using System.Collections.Generic;
5+
using System.Data.Common;
56
using System.Diagnostics;
67
using System.Runtime.CompilerServices;
78
using System.Threading;
@@ -122,5 +123,31 @@ async Task Test()
122123

123124
string MakeLockName(int i) => $"{nameof(TestHighConcurrencyWithSmallPool)}_{i}";
124125
}
126+
127+
[Test]
128+
public async Task TestBrokenConnectionDoesNotCorruptPool()
129+
{
130+
// This makes sure that for the Semaphore5 lock initial 4 tickets are taken with the default
131+
// application name and therefore won't be killed
132+
this._lockProvider.CreateLock("1");
133+
this._lockProvider.CreateLock("2");
134+
var applicationName = this._lockProvider.Strategy.SetUniqueApplicationName();
135+
136+
var lock1 = this._lockProvider.CreateLock("1");
137+
await using var handle1 = await lock1.AcquireAsync();
138+
139+
// kill the session
140+
await this._lockProvider.Strategy.Db.KillSessionsAsync(applicationName);
141+
142+
var lock2 = this._lockProvider.CreateLock("2");
143+
Assert.DoesNotThrowAsync(async () => await (await lock2.AcquireAsync()).DisposeAsync());
144+
145+
await using var handle2 = await lock2.AcquireAsync();
146+
Assert.DoesNotThrow(() => lock2.TryAcquire()?.Dispose());
147+
148+
Assert.Catch(() => handle1.Dispose());
149+
150+
Assert.DoesNotThrowAsync(async () => await (await lock1.AcquireAsync()).DisposeAsync());
151+
}
125152
}
126153
}

‎DistributedLock.Tests/Infrastructure/Data/ConnectionOptions.cs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,13 +20,13 @@ public sealed class TestingDbConnectionOptions
2020
public DbTransaction? Transaction { get; set; }
2121

2222
public T Create<T>(
23-
Func<string, (bool useMultiplexing, bool useTransaction, TimeSpan keepaliveCadence), T> fromConnectionString,
23+
Func<string, (bool useMultiplexing, bool useTransaction, TimeSpan? keepaliveCadence), T> fromConnectionString,
2424
Func<DbConnection, T> fromConnection,
2525
Func<DbTransaction, T> fromTransaction)
2626
{
2727
if (this.ConnectionString != null)
2828
{
29-
return fromConnectionString(this.ConnectionString, (this.ConnectionStringUseMultiplexing, this.ConnectionStringUseTransaction, this.ConnectionStringKeepaliveCadence ?? Timeout.InfiniteTimeSpan));
29+
return fromConnectionString(this.ConnectionString, (this.ConnectionStringUseMultiplexing, this.ConnectionStringUseTransaction, this.ConnectionStringKeepaliveCadence));
3030
}
3131

3232
if (this.Connection != null)

0 commit comments

Comments
 (0)