using System;
using System.Collections.Generic;
using System.Threading.Tasks;
namespace Tapeti.Flow.FlowHelpers
{
///
/// Implementation of an asynchronous locking mechanism.
///
public class LockCollection where T : notnull
{
private readonly Dictionary locks;
///
///
public LockCollection(IEqualityComparer comparer)
{
locks = new Dictionary(comparer);
}
///
/// Waits for and acquires a lock on the specified key. Dispose the returned value to release the lock.
///
///
public Task GetLock(T key)
{
// ReSharper disable once InconsistentlySynchronizedField - by design
var nextLi = new LockItem(locks, key);
try
{
var continueImmediately = false;
lock (locks)
{
if (!locks.TryGetValue(key, out var li))
{
locks.Add(key, nextLi);
continueImmediately = true;
}
else
{
while (li.Next != null)
li = li.Next;
li.Next = nextLi;
}
}
if (continueImmediately)
nextLi.Continue();
}
catch (Exception e)
{
nextLi.Error(e);
}
return nextLi.GetTask();
}
private class LockItem : IDisposable
{
internal volatile LockItem? Next;
private readonly Dictionary locks;
private readonly TaskCompletionSource tcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
private readonly T key;
public LockItem(Dictionary locks, T key)
{
this.locks = locks;
this.key = key;
}
internal void Continue()
{
tcs.TrySetResult(this);
}
internal void Error(Exception e)
{
tcs.SetException(e);
}
internal Task GetTask()
{
return tcs.Task;
}
public void Dispose()
{
lock (locks)
{
if (!locks.TryGetValue(key, out var li))
return;
if (li != this)
{
// Something is wrong (comparer is not stable?), but we cannot lose the completions sources
while (li.Next != null)
li = li.Next;
li.Next = Next;
return;
}
if (Next == null)
{
locks.Remove(key);
return;
}
locks[key] = Next;
}
Next.Continue();
}
}
}
}