Skip to content

Fix issues with deadlock detection #236

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
adam-mccoy wants to merge 3 commits into
base: master
Choose a base branch
from
Draft

Fix issues with deadlock detection #236

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c7b018e
Add actual deadlock test case
adam-mccoy Dec 15, 2022
fe946ce
Do not use extension methods
adam-mccoy Dec 15, 2022
29aa2f1
Fix test
adam-mccoy Dec 15, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 25 additions & 9 deletions source/Nevermore.IntegrationTests/Advanced/ConcurrentAccessFixture.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
using System.Threading;
using System.Threading.Tasks;
using FluentAssertions;
using Nevermore.Advanced;
using Nevermore.Advanced.Queryable;
using Nevermore.IntegrationTests.Model;
using Nevermore.IntegrationTests.SetUp;
Expand All @@ -16,6 +17,22 @@ public class ConcurrentAccessFixture : FixtureWithRelationalStore
const int NumberOfDocuments = 256;
const int DegreeOfParallelism = NumberOfDocuments;

[Test]
public async Task DeadlocksDoGoBoom()
{
NoMonkeyBusiness();

await FluentActions.Awaiting(async () =>
{
using var transaction = await Store.BeginWriteTransactionAsync();

var t1 = transaction.ExecuteNonQueryAsync("WAITFOR DELAY '00:00:02'");
var t2 = transaction.ExecuteNonQueryAsync("WAITFOR DELAY '00:00:02'");

await Task.WhenAll(t1, t2);
}).Should().ThrowAsync<DeadlockException>();
}

[Test]
public void ConcurrentAccessDoesNotGoBoom()
{
Expand Down Expand Up @@ -45,7 +62,7 @@ public void ConcurrentAccessDoesNotGoBoom()
{
ThreadWaitAll(
Enumerable.Range(0, DegreeOfParallelism)
.Select(i =>
.Select(_ =>
{
// ReSharper disable AccessToDisposedClosure
// ReSharper disable ReturnValueOfPureMethodIsNotUsed
Expand Down Expand Up @@ -113,22 +130,21 @@ public async Task AsyncConcurrentAccessDoesNotGoBoom()
// Create a bunch of documents so that we can query for them.
using (var transaction = await Store.BeginWriteTransactionAsync())
{
await Enumerable.Range(0, NumberOfDocuments)
.Select(i => new DocumentWithIdentityId {Name = $"{namePrefix}{i}"})
// ReSharper disable once AccessToDisposedClosure
.Select(document => transaction.InsertAsync(document))
.WhenAll();
var documents = Enumerable.Range(0, NumberOfDocuments)
.Select(i => new DocumentWithIdentityId { Name = $"{namePrefix}{i}" })
.ToArray();

await transaction.InsertManyAsync(documents);
await transaction.CommitAsync();
}

// Now hit it really hard and see if we can provoke a failure.
using (var transaction = await Store.BeginWriteTransactionAsync())
{
await Enumerable.Range(0, DegreeOfParallelism)
.Select(async i =>
.Select(_ => Task.Run(async () =>
{
// ReSharper disable AccessToDisposedClosure
// ReSharper disable ReturnValueOfPureMethodIsNotUsed
var documents = await transaction.Query<DocumentWithIdentityId>()
.Where(x => x.Name.StartsWith(namePrefix))
.ToListAsync();
Expand Down Expand Up @@ -158,7 +174,7 @@ await Task.WhenAll(

// ReSharper restore ReturnValueOfPureMethodIsNotUsed
// ReSharper restore AccessToDisposedClosure
})
}))
.WhenAll();
}
}
Expand Down
93 changes: 66 additions & 27 deletions source/Nevermore/Advanced/ReadTransaction.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@
using Nevermore.Querying.AST;
using Nevermore.TableColumnNameResolvers;
using Nevermore.Transient;
using Nito.AsyncEx;

namespace Nevermore.Advanced
{
Expand Down Expand Up @@ -572,16 +571,30 @@ public Task<int> ExecuteNonQueryAsync(string query, CommandParameterValues? args

public int ExecuteNonQuery(PreparedCommand preparedCommand)
{
using var mutex = DeadlockAwareLock.Lock();
using var command = CreateCommand(preparedCommand);
return command.ExecuteNonQuery();
DeadlockAwareLock.Wait();
try
{
using var command = CreateCommand(preparedCommand);
return command.ExecuteNonQuery();
}
finally
{
DeadlockAwareLock.Release();
}
}

public async Task<int> ExecuteNonQueryAsync(PreparedCommand preparedCommand, CancellationToken cancellationToken = default)
{
using var mutex = await DeadlockAwareLock.LockAsync(cancellationToken);
using var command = CreateCommand(preparedCommand);
return await command.ExecuteNonQueryAsync(cancellationToken);
await DeadlockAwareLock.WaitAsync(cancellationToken);
try
{
using var command = CreateCommand(preparedCommand);
return await command.ExecuteNonQueryAsync(cancellationToken);
}
finally
{
DeadlockAwareLock.Release();
}
}

public TResult ExecuteScalar<TResult>(string query, CommandParameterValues? args = null, RetriableOperation retriableOperation = RetriableOperation.Select, TimeSpan? commandTimeout = null)
Expand All @@ -596,22 +609,36 @@ public Task<TResult> ExecuteScalarAsync<TResult>(string query, CommandParameterV

public TResult ExecuteScalar<TResult>(PreparedCommand preparedCommand)
{
using var mutex = DeadlockAwareLock.Lock();
using var command = CreateCommand(preparedCommand);
var result = command.ExecuteScalar();
if (result == DBNull.Value)
return default!;
return (TResult)result;
DeadlockAwareLock.Wait();
try
{
using var command = CreateCommand(preparedCommand);
var result = command.ExecuteScalar();
if (result == DBNull.Value)
return default!;
return (TResult)result;
}
finally
{
DeadlockAwareLock.Release();
}
}

public async Task<TResult> ExecuteScalarAsync<TResult>(PreparedCommand preparedCommand, CancellationToken cancellationToken = default)
{
using var mutex = await DeadlockAwareLock.LockAsync(cancellationToken);
using var command = CreateCommand(preparedCommand);
var result = await command.ExecuteScalarAsync(cancellationToken);
if (result == DBNull.Value)
return default!;
return (TResult)result;
await DeadlockAwareLock.WaitAsync(cancellationToken);
try
{
using var command = CreateCommand(preparedCommand);
var result = await command.ExecuteScalarAsync(cancellationToken);
if (result == DBNull.Value)
return default!;
return (TResult)result;
}
finally
{
DeadlockAwareLock.Release();
}
}

public DbDataReader ExecuteReader(string query, CommandParameterValues? args = null, TimeSpan? commandTimeout = null)
Expand All @@ -638,18 +665,30 @@ public async Task<DbDataReader> ExecuteReaderAsync(PreparedCommand preparedComma

protected TResult[] ReadResults<TResult>(PreparedCommand preparedCommand, Func<DbDataReader, TResult> mapper)
{
using var mutex = DeadlockAwareLock.Lock();

using var command = CreateCommand(preparedCommand);
return command.ReadResults(mapper);
DeadlockAwareLock.Wait();
try
{
using var command = CreateCommand(preparedCommand);
return command.ReadResults(mapper);
}
finally
{
DeadlockAwareLock.Release();
}
}

protected async Task<TResult[]> ReadResultsAsync<TResult>(PreparedCommand preparedCommand, Func<DbDataReader, Task<TResult>> mapper, CancellationToken cancellationToken)
{
using var mutex = await DeadlockAwareLock.LockAsync(cancellationToken);

using var command = CreateCommand(preparedCommand);
return await command.ReadResultsAsync(mapper, cancellationToken);
await DeadlockAwareLock.WaitAsync(cancellationToken);
try
{
using var command = CreateCommand(preparedCommand);
return await command.ReadResultsAsync(mapper, cancellationToken);
}
finally
{
DeadlockAwareLock.Release();
}
}

PreparedCommand PrepareLoad<TDocument, TKey>(TKey id)
Expand Down
14 changes: 10 additions & 4 deletions source/Nevermore/Advanced/ThreadSafeAsyncEnumerable.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Nito.AsyncEx;

namespace Nevermore.Advanced
{
Expand All @@ -23,9 +22,16 @@ public ThreadSafeAsyncEnumerable(Func<IAsyncEnumerable<T>> innerFunc, DeadlockAw

public async IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = new())
{
using var mutex = await deadlockAwareLock.LockAsync(cancellationToken);
var inner = innerFunc();
await foreach (var item in inner.WithCancellation(cancellationToken)) yield return item;
await deadlockAwareLock.WaitAsync(cancellationToken);
try
{
var inner = innerFunc();
await foreach (var item in inner.WithCancellation(cancellationToken)) yield return item;
}
finally
{
deadlockAwareLock.Release();
}
}
}
}
1 change: 0 additions & 1 deletion source/Nevermore/Advanced/ThreadSafeEnumerable.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
using System;
using System.Collections;
using System.Collections.Generic;
using System.Threading;

namespace Nevermore.Advanced
{
Expand Down