diff --git a/source/Nevermore.IntegrationTests/Advanced/ConcurrentAccessFixture.cs b/source/Nevermore.IntegrationTests/Advanced/ConcurrentAccessFixture.cs index 65ecf87f..3fca04f9 100644 --- a/source/Nevermore.IntegrationTests/Advanced/ConcurrentAccessFixture.cs +++ b/source/Nevermore.IntegrationTests/Advanced/ConcurrentAccessFixture.cs @@ -3,6 +3,7 @@ using System.Threading; using System.Threading.Tasks; using FluentAssertions; +using Nevermore.Advanced.Concurrency; using Nevermore.Advanced.Queryable; using Nevermore.IntegrationTests.Model; using Nevermore.IntegrationTests.SetUp; @@ -16,10 +17,15 @@ public class ConcurrentAccessFixture : FixtureWithRelationalStore const int NumberOfDocuments = 256; const int DegreeOfParallelism = NumberOfDocuments; - [Test] - public void ConcurrentAccessDoesNotGoBoom() + [TestCase(ConcurrencyMode.LockOnly)] + [TestCase(ConcurrencyMode.LockAndWarn)] + public void ConcurrentAccessDoesNotGoBoom(ConcurrencyMode concurrencyMode) { NoMonkeyBusiness(); + Configuration.ConcurrencyMode = concurrencyMode; + + var callbackCalled = false; + Configuration.LogConcurrencyWarning = () => callbackCalled = true; var namePrefix = $"{Guid.NewGuid()}-"; @@ -101,12 +107,22 @@ static void ThreadWaitAll(params Thread[] threads) foreach (var thread in threads) thread.Join(); } } + + if (concurrencyMode == ConcurrencyMode.LockAndWarn) + { + callbackCalled.Should().BeTrue(); + } } - [Test] - public async Task AsyncConcurrentAccessDoesNotGoBoom() + [TestCase(ConcurrencyMode.LockOnly)] + [TestCase(ConcurrencyMode.LockAndWarn)] + public async Task AsyncConcurrentAccessDoesNotGoBoom(ConcurrencyMode concurrencyMode) { NoMonkeyBusiness(); + Configuration.ConcurrencyMode = concurrencyMode; + + var callbackCalled = false; + Configuration.LogConcurrencyWarning = () => callbackCalled = true; var namePrefix = $"{Guid.NewGuid()}-"; @@ -161,6 +177,88 @@ await Task.WhenAll( }) .WhenAll(); } + + if (concurrencyMode == ConcurrencyMode.LockAndWarn) + { + callbackCalled.Should().BeTrue(); + } + } + + [TestCase(ConcurrencyMode.NoLocking)] + [TestCase(ConcurrencyMode.WarnOnly)] + public void ConcurrentAccessGoesBoomWhenLockingIsDisabled(ConcurrencyMode concurrencyMode) + { + NoMonkeyBusiness(); + Configuration.ConcurrencyMode = concurrencyMode; + + var callbackCalled = false; + Configuration.LogConcurrencyWarning = () => callbackCalled = true; + + var namePrefix = $"{Guid.NewGuid()}-"; + + // Create a bunch of documents so that we can query for them. + using (var transaction = Store.BeginTransaction()) + { + FluentActions.Invoking( + () => + { + // Only using 10 here because using NumberOfDocuments (256) is too slow + Enumerable.Range(0, 10) + .Select(i => new DocumentWithIdentityId { Name = $"{namePrefix}{i}" }) + .AsParallel() + .WithDegreeOfParallelism(DegreeOfParallelism) + .Select(document => + { + // ReSharper disable once AccessToDisposedClosure + transaction.Insert(document); + return 0; + }) + .ToArray(); + }) + .Should() + .Throw(); + } + + if (concurrencyMode == ConcurrencyMode.WarnOnly) + { + callbackCalled.Should().BeTrue(); + } + } + + [TestCase(ConcurrencyMode.NoLocking)] + [TestCase(ConcurrencyMode.WarnOnly)] + public async Task AsyncConcurrentAccessGoesBoomWhenLockingIsDisabled(ConcurrencyMode concurrencyMode) + { + NoMonkeyBusiness(); + Configuration.ConcurrencyMode = concurrencyMode; + + var callbackCalled = false; + Configuration.LogConcurrencyWarning = () => callbackCalled = true; + + var namePrefix = $"{Guid.NewGuid()}-"; + + // Create a bunch of documents so that we can query for them. + using (var transaction = await Store.BeginWriteTransactionAsync()) + { + await FluentActions.Awaiting( + async () => + { + await Enumerable.Range(0, NumberOfDocuments) + .Select(i => new DocumentWithIdentityId { Name = $"{namePrefix}{i}" }) + // ReSharper disable once AccessToDisposedClosure + .Select(document => transaction.InsertAsync(document)) + .WhenAll(); + await transaction.CommitAsync(); + }) + .Should() + .ThrowExactlyAsync() + .WithMessage("The connection does not support MultipleActiveResultSets."); + } + + if (concurrencyMode == ConcurrencyMode.WarnOnly) + { + callbackCalled.Should().BeTrue(); + } } } } \ No newline at end of file diff --git a/source/Nevermore.Tests/DeadlockAwareLockFixture.cs b/source/Nevermore.Tests/DeadlockAwareLockFixture.cs deleted file mode 100644 index 555bc0a4..00000000 --- a/source/Nevermore.Tests/DeadlockAwareLockFixture.cs +++ /dev/null @@ -1,164 +0,0 @@ -using System; -using System.Threading; -using System.Threading.Tasks; -using Nevermore.Advanced; -using Nito.AsyncEx; -using NUnit.Framework; - -namespace Nevermore.Tests -{ - public class DeadlockAwareLockFixture - { - CancellationToken cancellationToken; - CancellationTokenSource cts; - - [SetUp] - public void SetUp() - { - cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); - cancellationToken = cts.Token; - } - - [TearDown] - public void TearDown() - { - cts?.Dispose(); - } - - [Test] - public void MultipleCallsToWait_ShouldThrow() - { - using var deadlockAwareLock = new DeadlockAwareLock(); - - deadlockAwareLock.Wait(); - - // The second call should immediately throw rather than waiting forever. - Assert.Throws(() => deadlockAwareLock.Wait()); - } - - [Test] - public void MultipleCallsToWaitWithinATask_ShouldThrow() - { - Task.Run(() => - { - // ReSharper disable AccessToDisposedClosure - using var deadlockAwareLock = new DeadlockAwareLock(); - - deadlockAwareLock.Wait(); - - // The second call should immediately throw rather than waiting forever. - Assert.Throws(() => deadlockAwareLock.Wait()); - - // ReSharper restore AccessToDisposedClosure - }, cancellationToken) - .Wait(cancellationToken); - } - - [Test] - public void AcquiringThenReleasingThenAcquiring_ShouldNotThrow() - { - using var deadlockAwareLock = new DeadlockAwareLock(); - - deadlockAwareLock.Wait(); - deadlockAwareLock.Release(); - deadlockAwareLock.Wait(); - } - - [Test] - public void AcquiringThenReleasingThenAcquiringInATask_ShouldNotThrow() - { - Task.Run(() => - { - // ReSharper disable AccessToDisposedClosure - using var deadlockAwareLock = new DeadlockAwareLock(); - - deadlockAwareLock.Wait(); - deadlockAwareLock.Release(); - deadlockAwareLock.Wait(); - - // ReSharper restore AccessToDisposedClosure - }, cancellationToken) - .Wait(cancellationToken); - } - - [Test] - public async Task MultipleCallsToWaitAsync_ShouldThrow() - { - using var deadlockAwareLock = new DeadlockAwareLock(); - - await deadlockAwareLock.WaitAsync(cancellationToken); - - // The second call should immediately throw rather than waiting forever. - Assert.ThrowsAsync(async () => await deadlockAwareLock.WaitAsync(cancellationToken)); - } - - [Test] - public async Task AcquiringAsyncThenReleasingThenAcquiringAsync_ShouldNotThrow() - { - using var deadlockAwareLock = new DeadlockAwareLock(); - - await deadlockAwareLock.WaitAsync(cancellationToken); - deadlockAwareLock.Release(); - await deadlockAwareLock.WaitAsync(cancellationToken); - } - - [Test] - public async Task MultipleTasksContending_ShouldNotThrow() - { - // ReSharper disable AccessToDisposedClosure - using var deadlockAwareLock = new DeadlockAwareLock(); - - // Loop so that we increase the probability that two different tasks are scheduled onto - // the same worker thread. This helps us guarantee that we're not accidentally relying - // on thread IDs or thread locals anywhere. - for (var i = 0; i < 1000; i++) - { - var task0 = Task.Run(async () => - { - await deadlockAwareLock.WaitAsync(cancellationToken); - await Task.Yield(); - deadlockAwareLock.Release(); - }, cancellationToken); - - var task1 = Task.Run(async () => - { - await deadlockAwareLock.WaitAsync(cancellationToken); - await Task.Yield(); - deadlockAwareLock.Release(); - }, cancellationToken); - - await Task.WhenAll(task0, task1); - } - - // ReSharper restore AccessToDisposedClosure - } - - [Test] - public void UsingSyncExtensionMethods_AndReleasingLocksCorrectly_ShouldNotThrow() - { - using var deadlockAwareLock = new DeadlockAwareLock(); - - using (var _ = deadlockAwareLock.Lock()) - { - } - - using (var _ = deadlockAwareLock.Lock()) - { - } - } - - [Test] - public async Task UsingAsyncExtensionMethods_AndReleasingLocksCorrectly_ShouldNotThrow() - { - using var deadlockAwareLock = new DeadlockAwareLock(); - - using (var _ = await deadlockAwareLock.LockAsync(cancellationToken)) - { - } - - using (var _ = await deadlockAwareLock.LockAsync(cancellationToken)) - { - } - } - } -} \ No newline at end of file diff --git a/source/Nevermore/Advanced/Concurrency/ConcurrencyMode.cs b/source/Nevermore/Advanced/Concurrency/ConcurrencyMode.cs new file mode 100644 index 00000000..b6120d01 --- /dev/null +++ b/source/Nevermore/Advanced/Concurrency/ConcurrencyMode.cs @@ -0,0 +1,25 @@ +namespace Nevermore.Advanced.Concurrency +{ + public enum ConcurrencyMode + { + /// + /// When a query is executed in parallel, no locking is performed. + /// + NoLocking, + + /// + /// When a query is executed in parallel, locking is performed. + /// + LockOnly, + + /// + /// When a query is executed in parallel, locking is performed and a warning is logged. + /// + LockAndWarn, + + /// + /// When a query is executed in parallel, a warning is logged. + /// + WarnOnly + } +} \ No newline at end of file diff --git a/source/Nevermore/Advanced/Concurrency/TransactionMutex.cs b/source/Nevermore/Advanced/Concurrency/TransactionMutex.cs new file mode 100644 index 00000000..1e8ec37f --- /dev/null +++ b/source/Nevermore/Advanced/Concurrency/TransactionMutex.cs @@ -0,0 +1,84 @@ +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using Nito.AsyncEx; +using Nito.Disposables; + +namespace Nevermore.Advanced.Concurrency +{ + public class TransactionMutex : IDisposable + { + readonly SemaphoreSlim _loggingSemaphore = new(1, 1); + readonly SemaphoreSlim _lockingSemaphore = new(1, 1); + + readonly ConcurrencyMode _concurrencyMode; + + bool ShouldLog => _concurrencyMode is ConcurrencyMode.LockAndWarn or ConcurrencyMode.WarnOnly; + bool ShouldLock => _concurrencyMode is ConcurrencyMode.LockOnly or ConcurrencyMode.LockAndWarn; + + readonly Action _logCallback; + + public TransactionMutex(ConcurrencyMode concurrencyMode, Action logCallback) + { + _concurrencyMode = concurrencyMode; + _logCallback = logCallback; + } + + public IDisposable Lock() + { + List disposables = new(); + if (ShouldLog) + { + if (_loggingSemaphore.Wait(TimeSpan.Zero)) + { + // Got the lock, no need to log + disposables.Add(new Disposable(() => _loggingSemaphore.Release())); + } + else + { + // Didn't get the lock, something else has it, need to log + _logCallback.Invoke(); + } + } + + if (ShouldLock) + { + disposables.Add(_lockingSemaphore.Lock()); + } + + return CollectionDisposable.Create(disposables); + } + + public async Task LockAsync(CancellationToken cancellationToken) + { + List disposables = new(); + if (ShouldLog) + { + if (await _loggingSemaphore.WaitAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false)) + { + // Got the lock, no need to log + disposables.Add(new Disposable(() => _loggingSemaphore.Release())); + } + else + { + // Didn't get the lock, something else has it, need to log + _logCallback.Invoke(); + } + } + + if (ShouldLock) + { + disposables.Add(await _lockingSemaphore.LockAsync(cancellationToken).ConfigureAwait(false)); + } + + return CollectionDisposable.Create(disposables); + } + + public void Dispose() + { + _loggingSemaphore?.Dispose(); + _lockingSemaphore?.Dispose(); + } + } +} \ No newline at end of file diff --git a/source/Nevermore/Advanced/DeadlockAwareLock.cs b/source/Nevermore/Advanced/DeadlockAwareLock.cs deleted file mode 100644 index 2d863738..00000000 --- a/source/Nevermore/Advanced/DeadlockAwareLock.cs +++ /dev/null @@ -1,75 +0,0 @@ -using System.Runtime.CompilerServices; -using System.Threading; -using System.Threading.Tasks; - -namespace Nevermore.Advanced -{ - /// - /// This class provides a best-effort deadlock detection mechanism. It will identify re-entrant calls from the same - /// task (if there is a task) or the same thread (if there is no task). While it does not _guarantee_ deadlock - /// detection, - /// it does provide a pretty good guarantee that _if_ a DeadlockException is thrown then there was almost certainly - /// going to be a deadlock. In other words: very few false positives; probably some false negatives; better than - /// nothing. - /// - public class DeadlockAwareLock : SemaphoreSlim - { - int? taskWhichHasAcquiredLock; - int? threadWhichHasAcquiredLock; - - public DeadlockAwareLock() : base(1, 1) - { - } - - public new void Wait() - { - AssertNoDeadlock(); - base.Wait(); - RecordLockAcquisition(); - } - - public new async Task WaitAsync(CancellationToken cancellationToken) - { - AssertNoDeadlock(); - await base.WaitAsync(cancellationToken).ConfigureAwait(false); - RecordLockAcquisition(); - } - - public new void Release() - { - threadWhichHasAcquiredLock = null; - taskWhichHasAcquiredLock = null; - base.Release(); - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - void AssertNoDeadlock() - { - if (taskWhichHasAcquiredLock is not null) - { - // If we have a task then we can rely on the task ID. It's not guaranteed (one task can still spawn another) but it's better than nothing. - // If it's a different task which has acquired the lock then there's at least _some_ hope that that task might complete without requiring - // this task to complete. If this task has the lock and is attempting to acquire it again then there's no way out - deadlock. - if (taskWhichHasAcquiredLock == Task.CurrentId) - throw new DeadlockException("This task context has already acquired this lock and has attempted to do so again."); - } - else - { - // If we have no task then our best guess is that we're using sync-only code, which means that the thread ID _should_ be - // a good indicator of the call context. - if (threadWhichHasAcquiredLock is not null) - // If this thread has already acquired a lock and it's trying to do so again then - // it's very unlikely that the first lock will be released, hence deadlock. - if (threadWhichHasAcquiredLock == Thread.CurrentThread.ManagedThreadId) - throw new DeadlockException("This thread has already acquired this lock and has attempted to do so again."); - } - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - void RecordLockAcquisition() - { - threadWhichHasAcquiredLock = Thread.CurrentThread.ManagedThreadId; - taskWhichHasAcquiredLock = Task.CurrentId; - } - } -} \ No newline at end of file diff --git a/source/Nevermore/Advanced/ReadTransaction.cs b/source/Nevermore/Advanced/ReadTransaction.cs index 41831828..6e414b39 100644 --- a/source/Nevermore/Advanced/ReadTransaction.cs +++ b/source/Nevermore/Advanced/ReadTransaction.cs @@ -12,6 +12,7 @@ using System.Threading; using System.Threading.Tasks; using Microsoft.Data.SqlClient; +using Nevermore.Advanced.Concurrency; using Nevermore.Advanced.Queryable; using Nevermore.Advanced.QueryBuilders; using Nevermore.Diagnositcs; @@ -44,7 +45,7 @@ public class ReadTransaction : IReadTransaction, ITransactionDiagnostic DbConnection? connection; protected IUniqueParameterNameGenerator ParameterNameGenerator { get; } = new UniqueParameterNameGenerator(); - protected DeadlockAwareLock DeadlockAwareLock { get; } = new(); + protected TransactionMutex TransactionMutex { get; } // To help track deadlocks readonly List commandTrace; @@ -114,6 +115,7 @@ internal ReadTransaction( registry.Add(this); columnNameResolver = configuration.TableColumnNameResolver(store); + TransactionMutex = new TransactionMutex(configuration.ConcurrencyMode, configuration.LogConcurrencyWarning); } protected DbTransaction? Transaction { get; private set; } @@ -521,8 +523,9 @@ IEnumerable Execute() yield return item; } - return configuration.SupportConcurrentExecution - ? new ThreadSafeEnumerable(Execute, DeadlockAwareLock) + // No point in using the ThreadSafeEnumerable if we're not going to lock. + return configuration.ConcurrencyMode is not ConcurrencyMode.NoLocking + ? new ThreadSafeEnumerable(Execute, TransactionMutex) : Execute(); } @@ -538,8 +541,9 @@ async IAsyncEnumerable Execute() } } - return configuration.SupportConcurrentExecution - ? new ThreadSafeAsyncEnumerable(Execute, DeadlockAwareLock) + // No point in using the ThreadSafeEnumerable if we're not going to lock. + return configuration.ConcurrencyMode is not ConcurrencyMode.NoLocking + ? new ThreadSafeAsyncEnumerable(Execute, TransactionMutex) : Execute(); } @@ -620,8 +624,9 @@ async IAsyncEnumerable Execute() } } - return configuration.SupportConcurrentExecution - ? new ThreadSafeAsyncEnumerable(Execute, DeadlockAwareLock) + // No point in using the ThreadSafeEnumerable if we're not going to lock. + return configuration.ConcurrencyMode is not ConcurrencyMode.NoLocking + ? new ThreadSafeAsyncEnumerable(Execute, TransactionMutex) : Execute(); } @@ -651,8 +656,8 @@ public Task ExecuteNonQueryAsync(string query, CommandParameterValues? args public int ExecuteNonQuery(PreparedCommand preparedCommand) { - using var mutex = configuration.SupportConcurrentExecution - ? DeadlockAwareLock.Lock() + using var mutex = configuration.ConcurrencyMode is not ConcurrencyMode.NoLocking + ? TransactionMutex.Lock() : NoopDisposable.Instance; using var command = CreateCommand(preparedCommand); return command.ExecuteNonQuery(); @@ -660,8 +665,8 @@ public int ExecuteNonQuery(PreparedCommand preparedCommand) public async Task ExecuteNonQueryAsync(PreparedCommand preparedCommand, CancellationToken cancellationToken = default) { - using var mutex = configuration.SupportConcurrentExecution - ? await DeadlockAwareLock.LockAsync(cancellationToken).ConfigureAwait(false) + using var mutex = configuration.ConcurrencyMode is not ConcurrencyMode.NoLocking + ? await TransactionMutex.LockAsync(cancellationToken).ConfigureAwait(false) : NoopDisposable.Instance; using var command = CreateCommand(preparedCommand); return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); @@ -679,8 +684,8 @@ public Task ExecuteScalarAsync(string query, CommandParameterV public TResult ExecuteScalar(PreparedCommand preparedCommand) { - using var mutex = configuration.SupportConcurrentExecution - ? DeadlockAwareLock.Lock() + using var mutex = configuration.ConcurrencyMode is not ConcurrencyMode.NoLocking + ? TransactionMutex.Lock() : NoopDisposable.Instance; using var command = CreateCommand(preparedCommand); var result = command.ExecuteScalar(); @@ -691,8 +696,8 @@ public TResult ExecuteScalar(PreparedCommand preparedCommand) public async Task ExecuteScalarAsync(PreparedCommand preparedCommand, CancellationToken cancellationToken = default) { - using var mutex = configuration.SupportConcurrentExecution - ? await DeadlockAwareLock.LockAsync(cancellationToken).ConfigureAwait(false) + using var mutex = configuration.ConcurrencyMode is not ConcurrencyMode.NoLocking + ? await TransactionMutex.LockAsync(cancellationToken).ConfigureAwait(false) : NoopDisposable.Instance; using var command = CreateCommand(preparedCommand); var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false); @@ -725,8 +730,8 @@ public async Task ExecuteReaderAsync(PreparedCommand preparedComma protected TResult[] ReadResults(PreparedCommand preparedCommand, Func mapper) { - using var mutex = configuration.SupportConcurrentExecution - ? DeadlockAwareLock.Lock() + using var mutex = configuration.ConcurrencyMode is not ConcurrencyMode.NoLocking + ? TransactionMutex.Lock() : NoopDisposable.Instance; using var command = CreateCommand(preparedCommand); @@ -735,8 +740,8 @@ protected TResult[] ReadResults(PreparedCommand preparedCommand, Func ReadResultsAsync(PreparedCommand preparedCommand, Func> mapper, CancellationToken cancellationToken) { - using var mutex = configuration.SupportConcurrentExecution - ? await DeadlockAwareLock.LockAsync(cancellationToken).ConfigureAwait(false) + using var mutex = configuration.ConcurrencyMode is not ConcurrencyMode.NoLocking + ? await TransactionMutex.LockAsync(cancellationToken).ConfigureAwait(false) : NoopDisposable.Instance; using var command = CreateCommand(preparedCommand); @@ -914,7 +919,7 @@ public void Dispose() } TryDispose(TransactionTimer); - TryDispose(DeadlockAwareLock); + TryDispose(TransactionMutex); if (OwnsSqlTransaction) { diff --git a/source/Nevermore/Advanced/ThreadSafeAsyncEnumerable.cs b/source/Nevermore/Advanced/ThreadSafeAsyncEnumerable.cs index 9bb73a34..64d993dc 100644 --- a/source/Nevermore/Advanced/ThreadSafeAsyncEnumerable.cs +++ b/source/Nevermore/Advanced/ThreadSafeAsyncEnumerable.cs @@ -2,6 +2,7 @@ using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; +using Nevermore.Advanced.Concurrency; using Nito.AsyncEx; namespace Nevermore.Advanced @@ -9,21 +10,21 @@ namespace Nevermore.Advanced public class ThreadSafeAsyncEnumerable : IAsyncEnumerable { readonly Func> innerFunc; - readonly DeadlockAwareLock deadlockAwareLock; + readonly TransactionMutex transactionMutex; - public ThreadSafeAsyncEnumerable(IAsyncEnumerable inner, DeadlockAwareLock deadlockAwareLock) : this(() => inner, deadlockAwareLock) + public ThreadSafeAsyncEnumerable(IAsyncEnumerable inner, TransactionMutex transactionMutex) : this(() => inner, transactionMutex) { } - public ThreadSafeAsyncEnumerable(Func> innerFunc, DeadlockAwareLock deadlockAwareLock) + public ThreadSafeAsyncEnumerable(Func> innerFunc, TransactionMutex transactionMutex) { this.innerFunc = innerFunc; - this.deadlockAwareLock = deadlockAwareLock; + this.transactionMutex = transactionMutex; } public async IAsyncEnumerator GetAsyncEnumerator(CancellationToken cancellationToken = new()) { - using var mutex = await deadlockAwareLock.LockAsync(cancellationToken).ConfigureAwait(false); + using var mutex = await transactionMutex.LockAsync(cancellationToken).ConfigureAwait(false); var inner = innerFunc(); await foreach (var item in inner.WithCancellation(cancellationToken).ConfigureAwait(false)) yield return item; } diff --git a/source/Nevermore/Advanced/ThreadSafeEnumerable.cs b/source/Nevermore/Advanced/ThreadSafeEnumerable.cs index dfe9dc3c..1e56ac97 100644 --- a/source/Nevermore/Advanced/ThreadSafeEnumerable.cs +++ b/source/Nevermore/Advanced/ThreadSafeEnumerable.cs @@ -2,29 +2,30 @@ using System.Collections; using System.Collections.Generic; using System.Threading; +using Nevermore.Advanced.Concurrency; namespace Nevermore.Advanced { internal class ThreadSafeEnumerable : IEnumerable { readonly Func> innerFunc; - readonly DeadlockAwareLock deadlockAwareLock; + readonly TransactionMutex transactionMutex; - public ThreadSafeEnumerable(IEnumerable inner, DeadlockAwareLock deadlockAwareLock) : this(() => inner, deadlockAwareLock) + public ThreadSafeEnumerable(IEnumerable inner, TransactionMutex transactionMutex) : this(() => inner, transactionMutex) { } - public ThreadSafeEnumerable(Func> innerFunc, DeadlockAwareLock deadlockAwareLock) + public ThreadSafeEnumerable(Func> innerFunc, TransactionMutex transactionMutex) { this.innerFunc = innerFunc; - this.deadlockAwareLock = deadlockAwareLock; + this.transactionMutex = transactionMutex; } public IEnumerator GetEnumerator() { - deadlockAwareLock.Wait(); + var mutex = transactionMutex.Lock(); var inner = innerFunc(); - return new ThreadSafeEnumerator(inner.GetEnumerator(), () => deadlockAwareLock.Release()); + return new ThreadSafeEnumerator(inner.GetEnumerator(), () => mutex.Dispose()); } IEnumerator IEnumerable.GetEnumerator() diff --git a/source/Nevermore/IRelationalStoreConfiguration.cs b/source/Nevermore/IRelationalStoreConfiguration.cs index 9832c84b..9c8c03af 100644 --- a/source/Nevermore/IRelationalStoreConfiguration.cs +++ b/source/Nevermore/IRelationalStoreConfiguration.cs @@ -1,5 +1,6 @@ using System; using Nevermore.Advanced; +using Nevermore.Advanced.Concurrency; using Nevermore.Advanced.Hooks; using Nevermore.Advanced.InstanceTypeResolvers; using Nevermore.Advanced.ReaderStrategies; @@ -89,12 +90,8 @@ public interface IRelationalStoreConfiguration /// ITableNameResolver TableNameResolver { get; set; } - /// - /// Gets or sets whether concurrent execution of queries is handled by Nevermore. When true, Nevermore will attempt - /// to sequence queries issued concurrently. - /// - /// The default is true. - /// - bool SupportConcurrentExecution { get; set; } + ConcurrencyMode ConcurrencyMode { get; set; } + + Action LogConcurrencyWarning { get; set; } } } \ No newline at end of file diff --git a/source/Nevermore/RelationalStoreConfiguration.cs b/source/Nevermore/RelationalStoreConfiguration.cs index 2b7c3b5e..9dfb5315 100644 --- a/source/Nevermore/RelationalStoreConfiguration.cs +++ b/source/Nevermore/RelationalStoreConfiguration.cs @@ -1,6 +1,7 @@ using System; using Microsoft.Data.SqlClient; using Nevermore.Advanced; +using Nevermore.Advanced.Concurrency; using Nevermore.Advanced.Hooks; using Nevermore.Advanced.InstanceTypeResolvers; using Nevermore.Advanced.ReaderStrategies; @@ -57,7 +58,8 @@ public RelationalStoreConfiguration(Func connectionStringFunc, IPrimaryK TableColumnNameResolver = _ => new SelectAllColumnsTableResolver(); AllowSynchronousOperations = true; - SupportConcurrentExecution = true; + ConcurrencyMode = ConcurrencyMode.LockOnly; + LogConcurrencyWarning = () => { }; QueryLogger = new DefaultQueryLogger(); TransactionLogger = new DefaultTransactionLogger(); @@ -120,7 +122,9 @@ public RelationalStoreConfiguration(Func connectionStringFunc, IPrimaryK public ITableNameResolver TableNameResolver { get; set; } - public bool SupportConcurrentExecution { get; set; } + public ConcurrencyMode ConcurrencyMode { get; set; } + + public Action LogConcurrencyWarning { get; set; } string InitializeConnectionString(string sqlConnectionString) {