Running tasks asynchronously can sometimes by tricky and no matter how careful I am, I sometimes forget some crucial part and my tasks run synchronously. I don't usually notice that until it's too late and performance problems arise because of large amount of data not being processed in parallel/async. It's also difficult to write tests for it.
I thought maybe there is a way to detect if tasks are running asynchronously? My idea was to create the AsyncDetector. It works by running an internal Stopwatch and tracks the two timestamps per action: start & stop. The BeginScope method returns an IDisposable scope that when disposed adds both timestamps to an internal ConcurrentBag. Later when I want to check if tasks were really running async I group all async-scopes by their time intervals and check if any of them overlap. If they do, then I assume they ware running at the same time (at least for a moment).
class AsyncDetector
{
private static readonly IEqualityComparer<(TimeSpan Start, TimeSpan End)> AsyncScopeComparer;
private readonly ConcurrentBag<(TimeSpan Start, TimeSpan End)> _runtimes = new ConcurrentBag<(TimeSpan Start, TimeSpan End)>();
private readonly Stopwatch _stopwatch = Stopwatch.StartNew();
static AsyncDetector()
{
AsyncScopeComparer = AdHocEqualityComparer<(TimeSpan Start, TimeSpan End)>.CreateWithoutHashCode((left, right) =>
{
var a = left.Start.Ticks;
var b = left.End.Ticks;
var c = right.Start.Ticks;
var d = right.End.Ticks;
return
(a <= c && c <= b) ||
(a <= d && d <= b);
});
}
public int MaxAsyncDegree
{
get
{
return
_runtimes
.GroupBy(t => t, AsyncScopeComparer)
.Select(t => t.Count())
.Max();
}
}
public IEnumerable<int> AllAsyncDegrees
{
get
{
return
_runtimes
.GroupBy(t => t, AsyncScopeComparer)
.Select(t => t.Count());
}
}
public int AsyncGroupCount
{
get
{
return
_runtimes
.GroupBy(t => t, AsyncScopeComparer).Count();
}
}
public IDisposable BeignScope()
{
return new AsyncScope(this);
}
private object ToDump() => new { MaxAsyncDegree, AsyncGroupCount };
private class AsyncScope : IDisposable
{
private readonly TimeSpan _start;
private readonly AsyncDetector _asyncDetector;
public AsyncScope(AsyncDetector asyncDetector)
{
_asyncDetector = asyncDetector;
_start = _asyncDetector._stopwatch.Elapsed;
}
public void Dispose()
{
_asyncDetector._runtimes.Add((_start, _asyncDetector._stopwatch.Elapsed));
}
}
}
It calculates the overlap by checking the endpoints of each time interval:
a-----b - task1
c-------d - task2
In case someone wanted to run it, I add the AdHocEqualityComparer that the above class is using:
public class AdHocEqualityComparer<T> : IEqualityComparer<T> { private readonly Func<T, T, bool> _equals; private readonly Func<T, int> _getHashCode; private AdHocEqualityComparer(Func<T, T, bool> equals, Func<T, int> getHashCode) { _equals = equals; _getHashCode = getHashCode; } public static IEqualityComparer<T> CreateWithoutHashCode([NotNull] Func<T, T, bool> equals) { if (equals == null) throw new ArgumentNullException(nameof(@equals)); return Create(equals, _ => 0); } public static IEqualityComparer<T> Create([NotNull] Func<T, T, bool> equals, [NotNull] Func<T, int> getHashCode) { if (equals == null) throw new ArgumentNullException(nameof(equals)); if (getHashCode == null) throw new ArgumentNullException(nameof(getHashCode)); return new AdHocEqualityComparer<T>(equals, getHashCode); } public bool Equals(T x, T y) { if (ReferenceEquals(null, x)) return false; if (ReferenceEquals(null, y)) return false; if (ReferenceEquals(x, y)) return true; return _equals(x, y); } public int GetHashCode(T obj) => _getHashCode(obj); }
I had another implementation before the AsyncDetector that work with thread-ids but it wasn't reliable when working with only async. I post it for reference:
class ParallelityDetector { private readonly object _syncLock; private readonly ObservableCollection<int> _threads; private int _maxThreads = 1; public ParallelityDetector() { _syncLock = new object(); _threads = new ObservableCollection<int>(); _threads.CollectionChanged += (sender, e) => { switch (e.Action) { case NotifyCollectionChangedAction.Add: _maxThreads = Math.Max(_maxThreads, _threads.Distinct().Count()); break; } }; } public int MaxThreadCount => _maxThreads; public void Beign() { lock (_syncLock) _threads.Add(Thread.CurrentThread.ManagedThreadId); } public void End() { lock (_syncLock) _threads.RemoveAt(0); } }
Example
The test code runs four different loops:
Parallel.ForEach
and three different styles of Task.WaitAll:
- One without async, this one runs sequentially
- One with async and without limitations
- One with async but using a
SemaphoreSlimto limit the degree of parallelism
The complete test code:
void Main()
{
var count = 10;
var delay = 500; // in milliseconds
TestParallelForeach(count, delay);
TestWaitAllWithoutAsync(count, delay);
TestWaitAllWithAsync(count, delay);
TestWaitAllWithAsyncAndSemaphoreSlim(count, delay);
}
private static void TestParallelForeach(int count, int delay)
{
var asyncDetector = new AsyncDetector();
Parallel.ForEach(Enumerable.Range(0, count), i =>
{
using (asyncDetector.BeignScope())
{
Thread.Sleep(delay);
PrintThreadId(i);
}
});
asyncDetector.Dump(nameof(TestParallelForeach));
}
private static void TestWaitAllWithoutAsync(int count, int delay)
{
var asyncDetector = new AsyncDetector();
var tasks = Enumerable.Range(0, count).Select(i =>
{
using (asyncDetector.BeignScope())
{
Thread.Sleep(delay);
PrintThreadId(i);
}
return Task.CompletedTask;
});
Task.WaitAll(tasks.ToArray());
asyncDetector.Dump(nameof(TestWaitAllWithoutAsync));
}
private static void TestWaitAllWithAsync(int count, int delay)
{
var asyncDetector = new AsyncDetector();
var tasks = Enumerable.Range(0, count).Select(i => Task.Run(async () =>
{
using (asyncDetector.BeignScope())
{
await Task.Delay(delay);
PrintThreadId(i);
}
return Task.CompletedTask;
}));
Task.WaitAll(tasks.ToArray());
asyncDetector.Dump(nameof(TestWaitAllWithAsync));
}
private static void TestWaitAllWithAsyncAndSemaphoreSlim(int count, int delay)
{
var asyncDetector = new AsyncDetector();
var semaphore = new SemaphoreSlim(Environment.ProcessorCount);
var tasks = Enumerable.Range(0, count).Select(i => Task.Run(async () =>
{
await semaphore.WaitAsync();
using (asyncDetector.BeignScope())
{
await Task.Delay(delay);
PrintThreadId(i);
}
semaphore.Release();
}));
Task.WaitAll(tasks.ToArray());
asyncDetector.Dump(nameof(TestWaitAllWithAsyncAndSemaphoreSlim));
}
private static void PrintThreadId(int item)
{
Console.WriteLine($"{item} [{Thread.CurrentThread.ManagedThreadId}]");
}
Results
As the output shows all cases has been correctly recognized by the AsyncDetector:
0 [10]
2 [13]
3 [12]
1 [11]
4 [7]
8 [13]
5 [11]
6 [10]
7 [12]
9 [7]
TestParallelForeach
MaxAsyncDegree
5
AsyncGroupCount
3
---
0 [12]
1 [12]
2 [12]
3 [12]
4 [12]
5 [12]
6 [12]
7 [12]
8 [12]
9 [12]
TestWaitAllWithoutAsync
MaxAsyncDegree
1
AsyncGroupCount
10
---
9 [10]
6 [10]
5 [10]
4 [10]
3 [10]
2 [10]
1 [10]
0 [10]
8 [7]
7 [11]
TestWaitAllWithAsync
MaxAsyncDegree
10
AsyncGroupCount
1
---
2 [11]
0 [7]
1 [13]
3 [10]
6 [13]
4 [11]
5 [10]
7 [7]
8 [13]
9 [7]
TestWaitAllWithAsyncAndSemaphoreSlim
MaxAsyncDegree
4
AsyncGroupCount
4
I'm not a thread/async expert so this implementation might not be the best one but can you think of anything better that would have the least performance/synchronization hit? I'd like this test to be as invisible as possible.
