5
\$\begingroup\$

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 SemaphoreSlim to 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.

\$\endgroup\$
1
\$\begingroup\$

What about RelayComparer name? It sounds more dotnetish :) It could be like this - 17 lines instead of 36 - and could be a way shorter if C# was reasonable :)

class RelayComparer<T> : IEqualityComparer<T>
{
    public RelayComparer(Func<T, T, bool> equals)
        : this(equals, _ => 0)
    {
    }

    public RelayComparer(Func<T, T, bool> equals, Func<T, int> getHashCode)
    {
        _equals = equals ?? throw new ArgumentNullException(nameof(equals));
        _getHashCode = getHashCode ?? throw new ArgumentNullException(nameof(getHashCode));
    }

    readonly Func<T, T, bool> _equals; 
    readonly Func<T, int> _getHashCode;
    public bool Equals(T x, T y) => _equals(x, y);
    public int GetHashCode(T obj) => _getHashCode(obj);
}

Where AsyncScope might have disposing delegate injected to get rid of bidirectional dependency on AsyncDetector:

class AsyncScope : IDisposable
{
    public static readonly IEqualityComparer<AsyncScope> OverlappingComparer = 
        new RelayComparer<AsyncScope>((AsyncScope left, AsyncScope 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 AsyncScope(IStopwatch stopwatch, Action<AsyncScope> dispose)
    {
        _stopwatch = stopwatch;
        _dispose = dispose;
        Start = _stopwatch.Elapsed;
    }

    public void Dispose()
    {
        End = _stopwatch.Elapsed;
        _dispose(this);
    }

    readonly IStopwatch _stopwatch;
    readonly Action<AsyncScope> _dispose;
    public TimeSpan Start { get; }
    public TimeSpan End { get; private set; } 
}

And the AsyncDetector could be twice shorter – please note testability:

public class AsyncDetector
{
    public AsyncDetector()
        : this(new SystemStopwatch())
    {
    }

    public AsyncDetector(IStopwatch stopwatch)
    {
        Runtimes = new ConcurrentBag<AsyncScope>();
        Stopwatch = stopwatch ?? throw new ArgumentNullException(nameof(stopwatch));
        Stopwatch.Start();
    }

    ConcurrentBag<AsyncScope> Runtimes { get; } 
    IStopwatch Stopwatch { get; }

    public IDisposable BeignScope() =>
        new AsyncScope(Stopwatch, s => Runtimes.Add(s));

    public int MaxAsyncDegree => Runtimes
        .GroupBy(t => t, AsyncScope.OverlappingComparer)
        .Select(t => t.Count())
        .Max();

    public IEnumerable<int> AllAsyncDegrees => Runtimes
        .GroupBy(t => t, AsyncScope.OverlappingComparer)
        .Select(t => t.Count());

    public int AsyncGroupCount => Runtimes
        .GroupBy(t => t, AsyncScope.OverlappingComparer)
        .Count();
}

UPDATE I wish we could write the following:

class RelayComparer<T> : IEqualityComparer<T>
{
    public RelayComparer(Func<T, T, bool> equals, Func<T, int> getHashCode)
    {
        Equals = equals ?? throw new ArgumentNullException(nameof(equals));
        GetHashCode = getHashCode ?? throw new ArgumentNullException(nameof(getHashCode));
    }

    public bool Equals(T x, T y) { get; }
    public int GetHashCode(T obj) { get; }
}
\$\endgroup\$
  • \$\begingroup\$ This looks nice. I think I like the name RelayComparer better but I'm afraid it has one issue now that it's so short. It's not longer canonical and does not provide the three default comparisons left, null, right, null, left, right which is why I wrote it in the first place because I didn't want to write them everytime. I also like the idea of AsyncScope now holding the Start and Stop properties ;-) \$\endgroup\$ – t3chb0t Nov 13 '17 at 5:17
  • \$\begingroup\$ Perhaps they are already designing something like the Property-Method hybrid? With all that functional and sugar stuff comming with each release this is quite possible. \$\endgroup\$ – t3chb0t Nov 13 '17 at 5:39
  • \$\begingroup\$ Speaking about I with I could... - I wish it was possible to write something like this: .Select(i => {}).Using(asyncDetector.BeginScope()) so that I don't have to polute each Action with using... I need to do some experiments... \$\endgroup\$ – t3chb0t Nov 13 '17 at 5:49

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.