diff --git a/docs/DistributedLock.MySql.md b/docs/DistributedLock.MySql.md index e6cd264d..d38412d2 100644 --- a/docs/DistributedLock.MySql.md +++ b/docs/DistributedLock.MySql.md @@ -21,7 +21,7 @@ await using (await @lock.AcquireAsync()) MySQL-based locks have been tested against and work with both the [MySQL](https://www.mysql.com/) and [MariaDB](https://mariadb.org/). -MySQL-based locks locks can be constructed with a `connectionString`, an `IDbConnection` or an `IDbTransaction` as a means of connecting to the database. In most cases, using a `connectionString` is preferred because it allows for the library to efficiently multiplex connections under the hood and eliminates the risk that the passed-in `IDbConnection` gets used in a way that disrupts the locking process. Using an `IDbTransaction` is generally equivalent to using an `IDbConnection` (the lock is still connection-scoped), but it allows the lock to participate in an ongoing transaction. **NOTE that since `IDbConnection`/`IDbTransaction` objects are not thread-safe, lock objects constructed with them can only be used by one thread at a time.** +MySQL-based locks locks can be constructed with a `connectionString`, a `DbDataSource` (.NET 7+), an `IDbConnection` or an `IDbTransaction` as a means of connecting to the database. In most cases, using a `connectionString` or a `DbDataSource` is preferred because it allows for the library to efficiently multiplex connections under the hood and eliminates the risk that the passed-in `IDbConnection` gets used in a way that disrupts the locking process. Note that connections are multiplexed per `DbDataSource` instance: two data sources with the same connection string do not share connections. Using an `IDbTransaction` is generally equivalent to using an `IDbConnection` (the lock is still connection-scoped), but it allows the lock to participate in an ongoing transaction. **NOTE that since `IDbConnection`/`IDbTransaction` objects are not thread-safe, lock objects constructed with them can only be used by one thread at a time.** Natively, MySQL's locking functions are case-insensitive with respect to the lock name. Since the DistributedLock library as a whole uses case-sensitive names, lock names containing uppercase characters will be transformed/hashed under the hood (as will empty names or names that are too long). If your program needs to coordinate with other code that is using `GET_LOCK` directly, be sure to express the name in lower case and to pass `exactName: true` when constructing the lock instance (in `exactName` mode, an invalid name will throw an exception rather than silently being transformed into a valid one). diff --git a/docs/DistributedLock.Postgres.md b/docs/DistributedLock.Postgres.md index 6ff94031..fd9f3620 100644 --- a/docs/DistributedLock.Postgres.md +++ b/docs/DistributedLock.Postgres.md @@ -39,7 +39,7 @@ Under the hood, [Postgres advisory locks can be based on either one 64-bit integ - Passing an ASCII string with 0-9 characters, which will be mapped to a `long` based on a custom scheme. - Passing an arbitrary string with the `allowHashing` option set to `true` which will be hashed to a `long`. Note that hashing will only be used if other methods of interpreting the string fail. -In addition to specifying the `key`, Postgres-based locks allow you to specify either a `connectionString`, an `IDbConnection`, or a `DbDataSource` as a means of connecting to the database. In most cases, using a `connectionString` is preferred because it allows for the library to efficiently multiplex connections under the hood and, in the case of `IDbConnection`, eliminates the risk that the passed-in `IDbConnection` gets used in a way that disrupts the locking process. **NOTE that since `IDbConnection` objects are not thread-safe, lock objects constructed with them can only be used by one thread at a time.** +In addition to specifying the `key`, Postgres-based locks allow you to specify either a `connectionString`, an `IDbConnection`, or a `DbDataSource` as a means of connecting to the database. In most cases, using a `connectionString` or a `DbDataSource` is preferred because it allows for the library to efficiently multiplex connections under the hood and eliminates the risk that the passed-in `IDbConnection` gets used in a way that disrupts the locking process. Note that connections are multiplexed per `DbDataSource` instance: two data sources with the same connection string do not share connections. **NOTE that since `IDbConnection` objects are not thread-safe, lock objects constructed with them can only be used by one thread at a time.** ## Options diff --git a/src/DistributedLock.Core/Internal/Data/MultiplexedConnectionLockPool.cs b/src/DistributedLock.Core/Internal/Data/MultiplexedConnectionLockPool.cs index fc968650..76cacd98 100644 --- a/src/DistributedLock.Core/Internal/Data/MultiplexedConnectionLockPool.cs +++ b/src/DistributedLock.Core/Internal/Data/MultiplexedConnectionLockPool.cs @@ -8,31 +8,33 @@ namespace Medallion.Threading.Internal.Data; #else internal #endif - sealed class MultiplexedConnectionLockPool + sealed class MultiplexedConnectionLockPool + where TConnectionSource : notnull { private readonly AsyncLock _lock = AsyncLock.Create(); - private readonly Dictionary> _poolsByConnectionString = []; + private readonly Dictionary> _poolsByConnectionSource; /// - /// The number of times we've called + /// The number of times we've called /// since we last called /// private uint _storeCountSinceLastPrune; /// - /// The number of s stored in + /// The number of s stored in /// private uint _pooledLockCount; - public MultiplexedConnectionLockPool(Func connectionFactory) + public MultiplexedConnectionLockPool(Func connectionFactory, IEqualityComparer? comparer = null) { this.ConnectionFactory = connectionFactory; + this._poolsByConnectionSource = new(comparer); } - internal Func ConnectionFactory { get; } + internal Func ConnectionFactory { get; } public async ValueTask TryAcquireAsync( - string connectionString, + TConnectionSource connectionSource, string name, TimeoutValue timeout, IDbSynchronizationStrategy strategy, @@ -42,7 +44,7 @@ public MultiplexedConnectionLockPool(Func connection { // opportunistic phase: see if we can use a connection that is already holding a lock // to acquire the current lock - var existingLock = await this.GetExistingLockOrDefaultAsync(connectionString).ConfigureAwait(false); + var existingLock = await this.GetExistingLockOrDefaultAsync(connectionSource).ConfigureAwait(false); if (existingLock != null) { var canSafelyDisposeExistingLock = false; @@ -70,12 +72,12 @@ public MultiplexedConnectionLockPool(Func connection finally { // since we took this lock from the pool, always return it to the pool - await this.StoreOrDisposeLockAsync(connectionString, existingLock, shouldDispose: canSafelyDisposeExistingLock).ConfigureAwait(false); + await this.StoreOrDisposeLockAsync(connectionSource, existingLock, shouldDispose: canSafelyDisposeExistingLock).ConfigureAwait(false); } } // normal phase: if we were not able to be opportunistic, ensure that we have a lock - var @lock = new MultiplexedConnectionLock(this.ConnectionFactory(connectionString)); + var @lock = new MultiplexedConnectionLock(this.ConnectionFactory(connectionSource)); MultiplexedConnectionLock.Result? result = null; try { @@ -85,7 +87,7 @@ public MultiplexedConnectionLockPool(Func connection finally { // if we failed to even acquire a result on a brand new lock, then there's definitely no reason to store it - await this.StoreOrDisposeLockAsync(connectionString, @lock, shouldDispose: result?.CanSafelyDispose ?? true).ConfigureAwait(false); + await this.StoreOrDisposeLockAsync(connectionSource, @lock, shouldDispose: result?.CanSafelyDispose ?? true).ConfigureAwait(false); } return result.Value.Handle; @@ -93,11 +95,11 @@ public MultiplexedConnectionLockPool(Func connection @lock.TryAcquireAsync(name, timeout, strategy, keepaliveCadence, cancellationToken, opportunistic); } - private async ValueTask GetExistingLockOrDefaultAsync(string connectionString) + private async ValueTask GetExistingLockOrDefaultAsync(TConnectionSource connectionSource) { using var _ = await this._lock.AcquireAsync(CancellationToken.None).ConfigureAwait(false); - if (this._poolsByConnectionString.TryGetValue(connectionString, out var pool) && pool.Count != 0) + if (this._poolsByConnectionSource.TryGetValue(connectionSource, out var pool) && pool.Count != 0) { --this._pooledLockCount; return pool.Dequeue(); @@ -106,7 +108,7 @@ public MultiplexedConnectionLockPool(Func connection return null; } - private async ValueTask StoreOrDisposeLockAsync(string connectionString, MultiplexedConnectionLock @lock, bool shouldDispose) + private async ValueTask StoreOrDisposeLockAsync(TConnectionSource connectionSource, MultiplexedConnectionLock @lock, bool shouldDispose) { if (shouldDispose) { @@ -121,18 +123,18 @@ private async ValueTask StoreOrDisposeLockAsync(string connectionString, Multipl if (shouldDispose) { // If we're about to dispose the lock, check if it has an empty pool that can be removed from our dictionary. - // By itself this doesn't guarantee cleanup: after a successful acquire we'll have an empty lock left over that won't - // go away unless we use THAT connection string again. To help with this, we have pruning - if (this._poolsByConnectionString.TryGetValue(connectionString, out var pool) && pool.Count == 0) + // By itself this doesn't guarantee cleanup: after a successful acquire we'll have an empty lock left over that won't + // go away unless we use THAT connection source again. To help with this, we have pruning + if (this._poolsByConnectionSource.TryGetValue(connectionSource, out var pool) && pool.Count == 0) { - this._poolsByConnectionString.Remove(connectionString); + this._poolsByConnectionSource.Remove(connectionSource); } } else // otherwise, store the lock { ++this._pooledLockCount; - if (this._poolsByConnectionString.TryGetValue(connectionString, out var existing)) + if (this._poolsByConnectionSource.TryGetValue(connectionSource, out var existing)) { existing.Enqueue(@lock); } @@ -140,7 +142,7 @@ private async ValueTask StoreOrDisposeLockAsync(string connectionString, Multipl { var newPool = new Queue(); newPool.Enqueue(@lock); - this._poolsByConnectionString.Add(connectionString, newPool); + this._poolsByConnectionSource.Add(connectionSource, newPool); } } @@ -160,7 +162,7 @@ private bool IsDueForPruningNoLock() // The whole reason to prune is to avoid memory bloat (connection bloat isn't an issue since we only keep connections // open when needed). So, we don't even consider pruning below a certain storage threshold - var pruningCost = this._pooledLockCount + this._poolsByConnectionString.Count; + var pruningCost = this._pooledLockCount + this._poolsByConnectionSource.Count; return pruningCost > 64 && this._storeCountSinceLastPrune >= pruningCost; } @@ -168,8 +170,8 @@ private async ValueTask PrunePoolsNoLockAsync() { this._storeCountSinceLastPrune = 0; // reset - List? connectionStringsToRemove = null; - foreach (var kvp in this._poolsByConnectionString) + List? connectionSourcesToRemove = null; + foreach (var kvp in this._poolsByConnectionSource) { var pool = kvp.Value; MultiplexedConnectionLock? firstRetainedLock = null; @@ -191,15 +193,15 @@ private async ValueTask PrunePoolsNoLockAsync() if (pool.Count == 0) { - (connectionStringsToRemove ??= new List()).Add(kvp.Key); + (connectionSourcesToRemove ??= new List()).Add(kvp.Key); } } - if (connectionStringsToRemove != null) + if (connectionSourcesToRemove != null) { - foreach (var connectionStringToRemove in connectionStringsToRemove) + foreach (var connectionSourceToRemove in connectionSourcesToRemove) { - this._poolsByConnectionString.Remove(connectionStringToRemove); + this._poolsByConnectionSource.Remove(connectionSourceToRemove); } } } diff --git a/src/DistributedLock.Core/Internal/Data/OptimisticConnectionMultiplexingDbDistributedLock.cs b/src/DistributedLock.Core/Internal/Data/OptimisticConnectionMultiplexingDbDistributedLock.cs index deb72111..ef4cb29a 100644 --- a/src/DistributedLock.Core/Internal/Data/OptimisticConnectionMultiplexingDbDistributedLock.cs +++ b/src/DistributedLock.Core/Internal/Data/OptimisticConnectionMultiplexingDbDistributedLock.cs @@ -8,26 +8,28 @@ namespace Medallion.Threading.Internal.Data; #else internal #endif -sealed class OptimisticConnectionMultiplexingDbDistributedLock : IDbDistributedLock +sealed class OptimisticConnectionMultiplexingDbDistributedLock : IDbDistributedLock + where TConnectionSource : notnull { - private readonly string _name, _connectionString; - private readonly MultiplexedConnectionLockPool _multiplexedConnectionLockPool; + private readonly string _name; + private readonly TConnectionSource _connectionSource; + private readonly MultiplexedConnectionLockPool _multiplexedConnectionLockPool; private readonly TimeoutValue _keepaliveCadence; private readonly IDbDistributedLock _fallbackLock; public OptimisticConnectionMultiplexingDbDistributedLock( - string name, - string connectionString, - MultiplexedConnectionLockPool multiplexedConnectionLockPool, + string name, + TConnectionSource connectionSource, + MultiplexedConnectionLockPool multiplexedConnectionLockPool, TimeoutValue keepaliveCadence) { this._name = name; - this._connectionString = connectionString; + this._connectionSource = connectionSource; this._multiplexedConnectionLockPool = multiplexedConnectionLockPool; this._keepaliveCadence = keepaliveCadence; this._fallbackLock = new DedicatedConnectionOrTransactionDbDistributedLock( - name, - () => this._multiplexedConnectionLockPool.ConnectionFactory(this._connectionString), + name, + () => this._multiplexedConnectionLockPool.ConnectionFactory(this._connectionSource), useTransaction: false, keepaliveCadence: keepaliveCadence ); @@ -44,7 +46,7 @@ public OptimisticConnectionMultiplexingDbDistributedLock( // to an exclusive lock which asks for a long timeout if (!strategy.IsUpgradeable && contextHandle == null) { - return this._multiplexedConnectionLockPool.TryAcquireAsync(this._connectionString, this._name, timeout, strategy, keepaliveCadence: this._keepaliveCadence, cancellationToken); + return this._multiplexedConnectionLockPool.TryAcquireAsync(this._connectionSource, this._name, timeout, strategy, keepaliveCadence: this._keepaliveCadence, cancellationToken); } // otherwise, fall back to our fallback lock diff --git a/src/DistributedLock.MySql/DistributedLock.MySql.csproj b/src/DistributedLock.MySql/DistributedLock.MySql.csproj index ba277a40..5309f1b9 100644 --- a/src/DistributedLock.MySql/DistributedLock.MySql.csproj +++ b/src/DistributedLock.MySql/DistributedLock.MySql.csproj @@ -1,7 +1,7 @@ - netstandard2.0;netstandard2.1;net462 + netstandard2.0;netstandard2.1;net462;net8.0 Medallion.Threading.MySql True 4 @@ -44,6 +44,11 @@ TRACE;DEBUG + + + + + diff --git a/src/DistributedLock.MySql/MySqlDatabaseConnection.cs b/src/DistributedLock.MySql/MySqlDatabaseConnection.cs index f9f6271c..927c21f5 100644 --- a/src/DistributedLock.MySql/MySqlDatabaseConnection.cs +++ b/src/DistributedLock.MySql/MySqlDatabaseConnection.cs @@ -1,6 +1,9 @@ using Medallion.Threading.Internal.Data; using MySqlConnector; using System.Data; +#if NET7_0_OR_GREATER +using System.Data.Common; +#endif namespace Medallion.Threading.MySql; @@ -16,6 +19,13 @@ public MySqlDatabaseConnection(IDbTransaction transaction) { } +#if NET7_0_OR_GREATER + public MySqlDatabaseConnection(DbDataSource dbDataSource) + : base(dbDataSource.CreateConnection(), isExternallyOwned: false) + { + } +#endif + public MySqlDatabaseConnection(string connectionString) : base(new MySqlConnection(connectionString), isExternallyOwned: false) { diff --git a/src/DistributedLock.MySql/MySqlDistributedLock.cs b/src/DistributedLock.MySql/MySqlDistributedLock.cs index f57c0e7e..17dc80b2 100644 --- a/src/DistributedLock.MySql/MySqlDistributedLock.cs +++ b/src/DistributedLock.MySql/MySqlDistributedLock.cs @@ -1,6 +1,9 @@ using Medallion.Threading.Internal; using Medallion.Threading.Internal.Data; using System.Data; +#if NET7_0_OR_GREATER +using System.Data.Common; +#endif using System.Security.Cryptography; using System.Text; @@ -29,9 +32,22 @@ public MySqlDistributedLock(string name, string connectionString, Action + /// Constructs a lock with the given that connects using the provided and + /// . + /// + /// Unless is specified, will be escaped/hashed to ensure name validity. + /// + public MySqlDistributedLock(string name, DbDataSource dbDataSource, Action? options = null, bool exactName = false) + : this(name, exactName, n => CreateInternalLock(n, dbDataSource, options)) + { + } +#endif + /// /// Constructs a lock with the given that connects using the provided . - /// + /// /// Unless is specified, will be escaped/hashed to ensure name validity. /// public MySqlDistributedLock(string name, IDbConnection connection, bool exactName = false) @@ -154,12 +170,28 @@ private static IDbDistributedLock CreateInternalLock(string name, string connect if (useMultiplexing) { - return new OptimisticConnectionMultiplexingDbDistributedLock(name, connectionString, MySqlMultiplexedConnectionLockPool.Instance, keepaliveCadence); + return new OptimisticConnectionMultiplexingDbDistributedLock(name, connectionString, MySqlMultiplexedConnectionLockPool.Instance, keepaliveCadence); } return new DedicatedConnectionOrTransactionDbDistributedLock(name, () => new MySqlDatabaseConnection(connectionString), useTransaction: false, keepaliveCadence); } +#if NET7_0_OR_GREATER + private static IDbDistributedLock CreateInternalLock(string name, DbDataSource dbDataSource, Action? options) + { + if (dbDataSource == null) { throw new ArgumentNullException(nameof(dbDataSource)); } + + var (keepaliveCadence, useMultiplexing) = MySqlConnectionOptionsBuilder.GetOptions(options); + + if (useMultiplexing) + { + return new OptimisticConnectionMultiplexingDbDistributedLock(name, dbDataSource, MySqlMultiplexedConnectionLockPool.DataSourceInstance, keepaliveCadence); + } + + return new DedicatedConnectionOrTransactionDbDistributedLock(name, () => new MySqlDatabaseConnection(dbDataSource), useTransaction: false, keepaliveCadence); + } +#endif + private static IDbDistributedLock CreateInternalLock(string name, IDbConnection connection) { if (connection == null) { throw new ArgumentNullException(nameof(connection)); } diff --git a/src/DistributedLock.MySql/MySqlDistributedSynchronizationProvider.cs b/src/DistributedLock.MySql/MySqlDistributedSynchronizationProvider.cs index 34b078e5..106640d2 100644 --- a/src/DistributedLock.MySql/MySqlDistributedSynchronizationProvider.cs +++ b/src/DistributedLock.MySql/MySqlDistributedSynchronizationProvider.cs @@ -1,4 +1,7 @@ using System.Data; +#if NET7_0_OR_GREATER +using System.Data.Common; +#endif namespace Medallion.Threading.MySql; @@ -19,6 +22,18 @@ public MySqlDistributedSynchronizationProvider(string connectionString, Action new MySqlDistributedLock(name, connectionString, options, exactName); } +#if NET7_0_OR_GREATER + /// + /// Constructs a provider that connects with and . + /// + public MySqlDistributedSynchronizationProvider(DbDataSource dbDataSource, Action? options = null) + { + if (dbDataSource == null) { throw new ArgumentNullException(nameof(dbDataSource)); } + + this._lockFactory = (name, exactName) => new MySqlDistributedLock(name, dbDataSource, options, exactName); + } +#endif + /// /// Constructs a provider that connects with . /// diff --git a/src/DistributedLock.MySql/MySqlMultiplexedConnectionLockPool.cs b/src/DistributedLock.MySql/MySqlMultiplexedConnectionLockPool.cs index 6c32adbc..4bc4f103 100644 --- a/src/DistributedLock.MySql/MySqlMultiplexedConnectionLockPool.cs +++ b/src/DistributedLock.MySql/MySqlMultiplexedConnectionLockPool.cs @@ -1,8 +1,19 @@ -using Medallion.Threading.Internal.Data; +using Medallion.Threading.Internal.Data; +#if NET7_0_OR_GREATER +using System.Data.Common; +#endif namespace Medallion.Threading.MySql; internal static class MySqlMultiplexedConnectionLockPool { - public static readonly MultiplexedConnectionLockPool Instance = new(s => new MySqlDatabaseConnection(s)); + public static readonly MultiplexedConnectionLockPool Instance = new(s => new MySqlDatabaseConnection(s)); + +#if NET7_0_OR_GREATER + /// + /// Pooled by instance: two data sources with the same connection string do not share connections + /// + public static readonly MultiplexedConnectionLockPool DataSourceInstance = + new(dataSource => new MySqlDatabaseConnection(dataSource), ReferenceEqualityComparer.Instance); +#endif } diff --git a/src/DistributedLock.MySql/PublicAPI/net8.0/PublicAPI.Shipped.txt b/src/DistributedLock.MySql/PublicAPI/net8.0/PublicAPI.Shipped.txt new file mode 100644 index 00000000..fd442ade --- /dev/null +++ b/src/DistributedLock.MySql/PublicAPI/net8.0/PublicAPI.Shipped.txt @@ -0,0 +1,3 @@ +#nullable enable +Medallion.Threading.MySql.MySqlDistributedLock.MySqlDistributedLock(string! name, System.Data.Common.DbDataSource! dbDataSource, System.Action? options = null, bool exactName = false) -> void +Medallion.Threading.MySql.MySqlDistributedSynchronizationProvider.MySqlDistributedSynchronizationProvider(System.Data.Common.DbDataSource! dbDataSource, System.Action? options = null) -> void diff --git a/src/DistributedLock.MySql/PublicAPI/net8.0/PublicAPI.Unshipped.txt b/src/DistributedLock.MySql/PublicAPI/net8.0/PublicAPI.Unshipped.txt new file mode 100644 index 00000000..e69de29b diff --git a/src/DistributedLock.Oracle/OracleDistributedLock.cs b/src/DistributedLock.Oracle/OracleDistributedLock.cs index c93231f3..acb5beed 100644 --- a/src/DistributedLock.Oracle/OracleDistributedLock.cs +++ b/src/DistributedLock.Oracle/OracleDistributedLock.cs @@ -71,7 +71,7 @@ internal static IDbDistributedLock CreateInternalLock(string name, string connec if (useMultiplexing) { - return new OptimisticConnectionMultiplexingDbDistributedLock(name, connectionString, OracleMultiplexedConnectionLockPool.Instance, keepaliveCadence); + return new OptimisticConnectionMultiplexingDbDistributedLock(name, connectionString, OracleMultiplexedConnectionLockPool.Instance, keepaliveCadence); } return new DedicatedConnectionOrTransactionDbDistributedLock(name, () => new OracleDatabaseConnection(connectionString), useTransaction: false, keepaliveCadence); diff --git a/src/DistributedLock.Oracle/OracleMultiplexedConnectionLockPool.cs b/src/DistributedLock.Oracle/OracleMultiplexedConnectionLockPool.cs index 31435863..2befa75a 100644 --- a/src/DistributedLock.Oracle/OracleMultiplexedConnectionLockPool.cs +++ b/src/DistributedLock.Oracle/OracleMultiplexedConnectionLockPool.cs @@ -4,5 +4,5 @@ namespace Medallion.Threading.Oracle; internal static class OracleMultiplexedConnectionLockPool { - public static readonly MultiplexedConnectionLockPool Instance = new(s => new OracleDatabaseConnection(s)); + public static readonly MultiplexedConnectionLockPool Instance = new(s => new OracleDatabaseConnection(s)); } diff --git a/src/DistributedLock.Postgres/PostgresDistributedLock.cs b/src/DistributedLock.Postgres/PostgresDistributedLock.cs index 640ed16c..0e9142ae 100644 --- a/src/DistributedLock.Postgres/PostgresDistributedLock.cs +++ b/src/DistributedLock.Postgres/PostgresDistributedLock.cs @@ -36,8 +36,6 @@ public PostgresDistributedLock(PostgresAdvisoryLockKey key, IDbConnection connec /// /// Constructs a lock with the given (effectively the lock name) and , /// and . - /// - /// Not compatible with connection multiplexing. /// public PostgresDistributedLock(PostgresAdvisoryLockKey key, DbDataSource dbDataSource, Action? options = null) : this(key, CreateInternalLock(key, dbDataSource, options)) @@ -68,7 +66,7 @@ internal static IDbDistributedLock CreateInternalLock(PostgresAdvisoryLockKey ke var (keepaliveCadence, useTransaction, useMultiplexing) = PostgresConnectionOptionsBuilder.GetOptions(options); return useMultiplexing - ? new OptimisticConnectionMultiplexingDbDistributedLock(key.ToString(), connectionString, PostgresMultiplexedConnectionLockPool.Instance, keepaliveCadence) + ? new OptimisticConnectionMultiplexingDbDistributedLock(key.ToString(), connectionString, PostgresMultiplexedConnectionLockPool.Instance, keepaliveCadence) : new DedicatedConnectionOrTransactionDbDistributedLock(key.ToString(), () => new PostgresDatabaseConnection(connectionString), useTransaction: useTransaction, keepaliveCadence); } @@ -83,14 +81,10 @@ internal static IDbDistributedLock CreateInternalLock(PostgresAdvisoryLockKey ke { if (dbDataSource == null) { throw new ArgumentNullException(nameof(dbDataSource)); } - // Multiplexing is currently incompatible with DbDataSource (see #238), so default it to false - var originalOptions = options; - options = o => { o.UseMultiplexing(false); originalOptions?.Invoke(o); }; - var (keepaliveCadence, useTransaction, useMultiplexing) = PostgresConnectionOptionsBuilder.GetOptions(options); return useMultiplexing - ? throw new NotSupportedException("Multiplexing is current incompatible with DbDataSource.") + ? new OptimisticConnectionMultiplexingDbDistributedLock(key.ToString(), dbDataSource, PostgresMultiplexedConnectionLockPool.DataSourceInstance, keepaliveCadence) : new DedicatedConnectionOrTransactionDbDistributedLock(key.ToString(), () => new PostgresDatabaseConnection(dbDataSource), useTransaction: useTransaction, keepaliveCadence); } #endif diff --git a/src/DistributedLock.Postgres/PostgresDistributedReaderWriterLock.cs b/src/DistributedLock.Postgres/PostgresDistributedReaderWriterLock.cs index 8a713cc1..78db3b26 100644 --- a/src/DistributedLock.Postgres/PostgresDistributedReaderWriterLock.cs +++ b/src/DistributedLock.Postgres/PostgresDistributedReaderWriterLock.cs @@ -36,8 +36,6 @@ public PostgresDistributedReaderWriterLock(PostgresAdvisoryLockKey key, IDbConne /// /// Constructs a lock with the given (effectively the lock name) and , /// and . - /// - /// Not compatible with connection multiplexing. /// public PostgresDistributedReaderWriterLock(PostgresAdvisoryLockKey key, DbDataSource dbDataSource, Action? options = null) : this(key, PostgresDistributedLock.CreateInternalLock(key, dbDataSource, options)) diff --git a/src/DistributedLock.Postgres/PostgresDistributedSynchronizationProvider.cs b/src/DistributedLock.Postgres/PostgresDistributedSynchronizationProvider.cs index bcb9f600..f0e30d18 100644 --- a/src/DistributedLock.Postgres/PostgresDistributedSynchronizationProvider.cs +++ b/src/DistributedLock.Postgres/PostgresDistributedSynchronizationProvider.cs @@ -39,8 +39,6 @@ public PostgresDistributedSynchronizationProvider(IDbConnection connection) #if NET7_0_OR_GREATER /// /// Constructs a provider which connects to Postgres using the provided and . - /// - /// Not compatible with connection multiplexing. /// public PostgresDistributedSynchronizationProvider(DbDataSource dbDataSource, Action? options = null) { diff --git a/src/DistributedLock.Postgres/PostgresMultiplexedConnectionLockPool.cs b/src/DistributedLock.Postgres/PostgresMultiplexedConnectionLockPool.cs index 7f576747..325abf7a 100644 --- a/src/DistributedLock.Postgres/PostgresMultiplexedConnectionLockPool.cs +++ b/src/DistributedLock.Postgres/PostgresMultiplexedConnectionLockPool.cs @@ -1,8 +1,19 @@ -using Medallion.Threading.Internal.Data; +using Medallion.Threading.Internal.Data; +#if NET7_0_OR_GREATER +using System.Data.Common; +#endif namespace Medallion.Threading.Postgres; internal static class PostgresMultiplexedConnectionLockPool { - public static readonly MultiplexedConnectionLockPool Instance = new(s => new PostgresDatabaseConnection(s)); + public static readonly MultiplexedConnectionLockPool Instance = new(s => new PostgresDatabaseConnection(s)); + +#if NET7_0_OR_GREATER + /// + /// Pooled by instance: two data sources with the same connection string do not share connections + /// + public static readonly MultiplexedConnectionLockPool DataSourceInstance = + new(dataSource => new PostgresDatabaseConnection(dataSource), ReferenceEqualityComparer.Instance); +#endif } diff --git a/src/DistributedLock.SqlServer/SqlDistributedLock.cs b/src/DistributedLock.SqlServer/SqlDistributedLock.cs index e66fbc36..bc9d43dc 100644 --- a/src/DistributedLock.SqlServer/SqlDistributedLock.cs +++ b/src/DistributedLock.SqlServer/SqlDistributedLock.cs @@ -89,7 +89,7 @@ internal static IDbDistributedLock CreateInternalLock(string name, string connec var (keepaliveCadence, useTransaction, useMultiplexing) = SqlConnectionOptionsBuilder.GetOptions(optionsBuilder); return useMultiplexing - ? new OptimisticConnectionMultiplexingDbDistributedLock(name, connectionString, SqlMultiplexedConnectionLockPool.Instance, keepaliveCadence) + ? new OptimisticConnectionMultiplexingDbDistributedLock(name, connectionString, SqlMultiplexedConnectionLockPool.Instance, keepaliveCadence) : new DedicatedConnectionOrTransactionDbDistributedLock(name, () => new SqlDatabaseConnection(connectionString), useTransaction: useTransaction, keepaliveCadence); } diff --git a/src/DistributedLock.SqlServer/SqlMultiplexedConnectionLockPool.cs b/src/DistributedLock.SqlServer/SqlMultiplexedConnectionLockPool.cs index b280c793..b8c51545 100644 --- a/src/DistributedLock.SqlServer/SqlMultiplexedConnectionLockPool.cs +++ b/src/DistributedLock.SqlServer/SqlMultiplexedConnectionLockPool.cs @@ -4,5 +4,5 @@ namespace Medallion.Threading.SqlServer; internal static class SqlMultiplexedConnectionLockPool { - public static readonly MultiplexedConnectionLockPool Instance = new(s => new SqlDatabaseConnection(s)); + public static readonly MultiplexedConnectionLockPool Instance = new(s => new SqlDatabaseConnection(s)); } diff --git a/src/DistributedLock.Tests/Tests/MySql/MySqlDistributedLockTest.cs b/src/DistributedLock.Tests/Tests/MySql/MySqlDistributedLockTest.cs index a706f1f6..f0b2cbd6 100644 --- a/src/DistributedLock.Tests/Tests/MySql/MySqlDistributedLockTest.cs +++ b/src/DistributedLock.Tests/Tests/MySql/MySqlDistributedLockTest.cs @@ -18,6 +18,9 @@ public void TestValidatesConstructorArguments() Assert.Catch(() => new MySqlDistributedLock("a", default(string)!)); Assert.Catch(() => new MySqlDistributedLock("a", default(IDbTransaction)!)); Assert.Catch(() => new MySqlDistributedLock("a", default(IDbConnection)!)); +#if NET7_0_OR_GREATER + Assert.Catch(() => new MySqlDistributedLock("a", default(System.Data.Common.DbDataSource)!)); +#endif Assert.Catch(() => new MySqlDistributedLock(new string('a', MySqlDistributedLock.MaxNameLength + 1), ConnectionString, exactName: true)); Assert.DoesNotThrow(() => new MySqlDistributedLock(new string('a', MySqlDistributedLock.MaxNameLength), ConnectionString, exactName: true)); } @@ -69,4 +72,57 @@ public async Task TestMySqlCommandMustExplicitlyParticipateInTransaction(Type te commandInTransaction.CommandText = "SELECT COUNT(*) FROM distributed_lock.temp"; (await commandInTransaction.ExecuteScalarAsync()).ShouldEqual(2); } + +#if NET7_0_OR_GREATER + [Test] + public async Task TestMultiplexingWithDbDataSourceUsesASharedConnection() + { + var applicationName = UniqueApplicationName(); + await using var dataSource = CreateDataSource(applicationName); + + var lock1 = new MySqlDistributedLock(Guid.NewGuid().ToString(), dataSource); + var lock2 = new MySqlDistributedLock(Guid.NewGuid().ToString(), dataSource); + await using var handle1 = await lock1.AcquireAsync(); + await using var handle2 = await lock2.AcquireAsync(); + + Assert.That(new TestingMySqlDb().CountActiveSessions(applicationName), Is.EqualTo(1), "both locks should share one multiplexed connection"); + } + + [Test] + public async Task TestDbDataSourcePoolIsKeyedByReference() + { + var applicationName = UniqueApplicationName(); + await using var dataSource1 = CreateDataSource(applicationName); + await using var dataSource2 = CreateDataSource(applicationName); + + var lock1 = new MySqlDistributedLock(Guid.NewGuid().ToString(), dataSource1); + var lock2 = new MySqlDistributedLock(Guid.NewGuid().ToString(), dataSource2); + await using var handle1 = await lock1.AcquireAsync(); + await using var handle2 = await lock2.AcquireAsync(); + + Assert.That(new TestingMySqlDb().CountActiveSessions(applicationName), Is.EqualTo(2), "distinct DbDataSource instances with the same connection string should not share connections"); + } + + // DbDataSource uses the same multiplexing flow as connection strings so we don't need exhaustive testing, but we + // want to see mutual exclusion work at least once + [Test] + public async Task TestDbDataSourceConstructorWorks() + { + await using var dataSource = new MySqlDataSource(ConnectionString); + var @lock = new MySqlDistributedLock(Guid.NewGuid().ToString(), dataSource); + await using (await @lock.AcquireAsync()) + { + await using var handle = await @lock.TryAcquireAsync(); + Assert.That(handle, Is.Null); + } + } + + private static string UniqueApplicationName() => $"dbds_test_{Guid.NewGuid():N}"; + + private static MySqlDataSource CreateDataSource(string applicationName) + { + var connectionStringBuilder = new MySqlConnectionStringBuilder(ConnectionString) { ApplicationName = applicationName }; + return new MySqlDataSource(connectionStringBuilder.ConnectionString); + } +#endif } diff --git a/src/DistributedLock.Tests/Tests/MySql/MySqlDistributedSynchronizationProviderTest.cs b/src/DistributedLock.Tests/Tests/MySql/MySqlDistributedSynchronizationProviderTest.cs index ae918f57..5b89394c 100644 --- a/src/DistributedLock.Tests/Tests/MySql/MySqlDistributedSynchronizationProviderTest.cs +++ b/src/DistributedLock.Tests/Tests/MySql/MySqlDistributedSynchronizationProviderTest.cs @@ -2,6 +2,9 @@ using Medallion.Threading.Tests.Data; using NUnit.Framework; using System.Data; +#if NET7_0_OR_GREATER +using System.Data.Common; +#endif namespace Medallion.Threading.Tests.MySql; @@ -13,6 +16,9 @@ public void TestArgumentValidation() Assert.Throws(() => new MySqlDistributedSynchronizationProvider(default(string)!)); Assert.Throws(() => new MySqlDistributedSynchronizationProvider(default(IDbConnection)!)); Assert.Throws(() => new MySqlDistributedSynchronizationProvider(default(IDbTransaction)!)); +#if NET7_0_OR_GREATER + Assert.Throws(() => new MySqlDistributedSynchronizationProvider(default(DbDataSource)!)); +#endif } [Test] diff --git a/src/DistributedLock.Tests/Tests/Postgres/PostgresDistributedLockTest.cs b/src/DistributedLock.Tests/Tests/Postgres/PostgresDistributedLockTest.cs index e8802b8b..cbcb372c 100644 --- a/src/DistributedLock.Tests/Tests/Postgres/PostgresDistributedLockTest.cs +++ b/src/DistributedLock.Tests/Tests/Postgres/PostgresDistributedLockTest.cs @@ -22,14 +22,57 @@ public void TestValidatesConstructorArguments() #if NET7_0_OR_GREATER [Test] - public void TestMultiplexingWithDbDataSourceThrowNotSupportedException() + public async Task TestMultiplexingWithDbDataSourceUsesASharedConnection() { - using var dataSource = new NpgsqlDataSourceBuilder(TestingPostgresDb.DefaultConnectionString).Build(); - Assert.Throws(() => new PostgresDistributedLock(new(0), dataSource, opt => opt.UseMultiplexing())); + var applicationName = UniqueApplicationName(); + await using var dataSource = CreateDataSource(applicationName); + + var lock1 = new PostgresDistributedLock(new(Guid.NewGuid().ToString(), allowHashing: true), dataSource); + var lock2 = new PostgresDistributedLock(new(Guid.NewGuid().ToString(), allowHashing: true), dataSource); + await using var handle1 = await lock1.AcquireAsync(); + await using var handle2 = await lock2.AcquireAsync(); + + Assert.That(await CountSessionsAsync(applicationName), Is.EqualTo(1), "both locks should share one multiplexed connection"); + } + + [Test] + public async Task TestDbDataSourcePoolIsKeyedByReference() + { + var applicationName = UniqueApplicationName(); + await using var dataSource1 = CreateDataSource(applicationName); + await using var dataSource2 = CreateDataSource(applicationName); + + var lock1 = new PostgresDistributedLock(new(Guid.NewGuid().ToString(), allowHashing: true), dataSource1); + var lock2 = new PostgresDistributedLock(new(Guid.NewGuid().ToString(), allowHashing: true), dataSource2); + await using var handle1 = await lock1.AcquireAsync(); + await using var handle2 = await lock2.AcquireAsync(); + + Assert.That(await CountSessionsAsync(applicationName), Is.EqualTo(2), "distinct DbDataSource instances with the same connection string should not share connections"); + } + + [Test] + public async Task TestHandleLostTokenWorksWithDbDataSourceMultiplexing() + { + var applicationName = UniqueApplicationName(); + await using var dataSource = CreateDataSource(applicationName); + + var @lock = new PostgresDistributedLock(new(Guid.NewGuid().ToString(), allowHashing: true), dataSource); + var handle = await @lock.AcquireAsync(); + + using var handleLostEvent = new ManualResetEventSlim(initialState: false); + Assert.That(handle.HandleLostToken.CanBeCanceled, Is.True); // starts monitoring on the multiplexed connection + using var registration = handle.HandleLostToken.Register(handleLostEvent.Set); + + await new TestingPostgresDb().KillSessionsAsync(applicationName, idleSince: null); + + Assert.That(handleLostEvent.Wait(TimeSpan.FromSeconds(10)), Is.True); + + // dispose may throw since the underlying connection is broken + try { handle.Dispose(); } catch { } } - // DbDataSource just calls through to the IDbConnection flow so we don't need exhaustive testing, but we want to - // see it at least work once + // DbDataSource uses the same multiplexing flow as connection strings so we don't need exhaustive testing, but we + // want to see mutual exclusion work at least once [Test] public async Task TestDbDataSourceConstructorWorks() { @@ -41,6 +84,24 @@ public async Task TestDbDataSourceConstructorWorks() Assert.IsNull(handle); } } + + private static string UniqueApplicationName() => $"dbds_test_{Guid.NewGuid():N}"; + + private static NpgsqlDataSource CreateDataSource(string applicationName) + { + var connectionStringBuilder = new NpgsqlConnectionStringBuilder(TestingPostgresDb.DefaultConnectionString) { ApplicationName = applicationName }; + return new NpgsqlDataSourceBuilder(connectionStringBuilder.ConnectionString).Build(); + } + + private static async Task CountSessionsAsync(string applicationName) + { + using var connection = new NpgsqlConnection(TestingPostgresDb.DefaultConnectionString); + await connection.OpenAsync(); + using var command = connection.CreateCommand(); + command.CommandText = "SELECT COUNT(*)::int FROM pg_stat_activity WHERE application_name = @applicationName"; + command.Parameters.AddWithValue("applicationName", applicationName); + return (int)(await command.ExecuteScalarAsync())!; + } #endif [Test]