阻止两个异步方法与另一个同时运行

Question

我有两个异步函数，我将它们称为ChangeState() 和DoThing()。他们每个人都在等待下游的异步方法。这些是从事件处理程序中调用的，因此它们在执行时不会阻塞任何其他代码。如果调用了ChangeState()，则必须在任何先前的ChangeState() 完成之前，DoThing() 不会执行它的操作。 ChangeState() 可以在它仍在执行时再次调用。在DoThing() 之前开始的任何执行都应该在DoThing() 可以继续之前完成。

反之亦然； ChangeState() 应该等到之前运行的 DoStuff() 完成。

如何在没有死锁危险的情况下实现这一点？

我知道在 lock 语句中不允许等待，这是有充分理由的，这也是我不尝试重新创建该功能的原因。

async void ChangeState(bool state)
{
 //Wait here until any pending DoStuff() is complete.
 await OutsideApi.ChangeState(state);
}

async void DoStuff()
{
 //Wait here until any pending ChangeState() is complete.
 await OutsideApi.DoStuff();
}

Answer 1

根据您的要求，ReaderWriterLock 之类的东西似乎可以帮助您。另外，既然你有async 方法，你应该使用async 锁。不幸的是，.NET 框架本身没有提供等待就绪的ReaderWriterLock 锁。幸运的是，您可以查看 AsyncEx 库或此 article。使用AsyncEx的例子。

var readerWriterLock = new AsyncReaderWriterLock();

async void ChangeState(bool state)
{
    using(await readerWriterLock.ReaderLockAsync())
    {
        await OutsideApi.ChangeState(state);
    }
}

async void DoStuff()
{
    using(await readerWriterLock.WriterLockAsync())
    {
        await OutsideApi.DoStuff();
    }
}

nb 这个方案还是有DoStuff调用不能并发的限制，写锁，但是调用的顺序和ChangeState之前的所有DoStuff都需要finalize反之亦然。（@Scott Chamberlain 提示使用读写器锁）

Answer 2

您可以使用 ManualResetEvent 或 AutoResetEvent 发出一个线程已完成的信号，以便另一个线程可以继续工作。

一些样本可以在here和here找到：

Answer 3

编辑：第一个解决方案不符合要求。

1. 创建自定义锁类。
   此类跟踪从哪个类型（ChangeState 和 DoThing）运行的实例数量，并提供一种检查任务是否可以运行的方法。

        public class CustomLock
        {
            private readonly int[] Running;
            private readonly object _lock;
            public CustomLock(int Count)
            {
                Running = new int[Count];
                _lock = new object();
            }

            public void LockOne(int Task)
            {
                lock (_lock)
                {
                    Running[Task]++;
                }
            }

            public void UnlockOne(int Task)
            {
                lock (_lock)
                {
                    Running[Task]--;
                }
            }

            public bool Locked(int Task)
            {
                lock (_lock)
                {
                    for (int i = 0; i < Running.Length; i++)
                    {
                        if (i != Task && Running[i] != 0)
                            return true;
                    }
                    return false;
                }
            }
        }

1. 更改现有代码。
   ChangeState 将是任务 0，DoStuff 将是任务 1。

private CustomLock Lock = new CustomLock(2); //Create a new instance of the class for 2 tasks

async Task ChangeState(bool state)
{
   while (Lock.Locked(0)) //Wait for the task to get unlocked
      await Task.Delay(10);
   Lock.LockOne(0); //Lock this task
   await OutsideApi.ChangeState(state);
   Lock.UnlockOne(0); //Task finished, unlock one
}

async Task DoStuff()
{
   while (Lock.Locked(1))
      await Task.Delay(10);
   Lock.LockOne(1);
   await OutsideApi.DoStuff();
   Lock.UnlockOne(1);
}

当任何 ChangeState 正在运行时，无需等待即可启动新的，但是当调用 DoStuff 时，它将等到所有 ChangeStates 完成，这也适用于另一种方式。

Answer 4

我为练习创建了一个名为KeyedLock 的同步原语，它允许一次只对一个键进行并发异步操作。所有其他键都排队，稍后分批（按键）解除阻塞。该类旨在像这样使用：

KeyedLock _keyedLock;

async Task ChangeState(bool state)
{
    using (await this._keyedLock.LockAsync("ChangeState"))
    {
        await OutsideApi.ChangeState(state);
    }
}

async Task DoStuff()
{
    using (await this._keyedLock.LockAsync("DoStuff"))
    {
        await OutsideApi.DoStuff();
    }
}

例如下面的调用：

await ChangeState(true);
await DoStuff();
await DoStuff();
await ChangeState(false);
await DoStuff();
await ChangeState(true);

...将按此顺序执行：

ChangeState(true);
ChangeState(false); // concurrently with the above
ChangeState(true); // concurrently with the above
DoStuff(); // after completion of the above
DoStuff(); // concurrently with the above
DoStuff(); // concurrently with the above

KeyedLock 类：

class KeyedLock
{
    private object _currentKey;
    private int _currentCount = 0;
    private WaitingQueue _waitingQueue = new WaitingQueue();
    private readonly object _locker = new object();

    public Task WaitAsync(object key, CancellationToken cancellationToken)
    {
        if (key == null) throw new ArgumentNullException(nameof(key));
        lock (_locker)
        {
            if (_currentKey != null && key != _currentKey)
            {
                var waiter = new TaskCompletionSource<bool>();
                _waitingQueue.Enqueue(new KeyValuePair<object,
                    TaskCompletionSource<bool>>(key, waiter));
                if (cancellationToken != null)
                {
                    cancellationToken.Register(() => waiter.TrySetCanceled());
                }
                return waiter.Task;
            }
            else
            {
                _currentKey = key;
                _currentCount++;
                return cancellationToken.IsCancellationRequested ?
                    Task.FromCanceled(cancellationToken) : Task.FromResult(true);
            }
        }
    }
    public Task WaitAsync(object key) => WaitAsync(key, CancellationToken.None);

    public void Release()
    {
        List<TaskCompletionSource<bool>> tasksToRelease;
        lock (_locker)
        {
            if (_currentCount <= 0) throw new InvalidOperationException();
            _currentCount--;
            if (_currentCount > 0) return;
            _currentKey = null;
            if (_waitingQueue.Count == 0) return;
            var newWaitingQueue = new WaitingQueue();
            tasksToRelease = new List<TaskCompletionSource<bool>>();
            foreach (var entry in _waitingQueue)
            {
                if (_currentKey == null || entry.Key == _currentKey)
                {
                    _currentKey = entry.Key;
                    _currentCount++;
                    tasksToRelease.Add(entry.Value);
                }
                else
                {
                    newWaitingQueue.Enqueue(entry);
                }
            }
            _waitingQueue = newWaitingQueue;
        }
        foreach (var item in tasksToRelease)
        {
            item.TrySetResult(true);
        }
    }
    private class WaitingQueue :
        Queue<KeyValuePair<object, TaskCompletionSource<bool>>>
    { }

    public Task<Releaser> LockAsync(object key,
        CancellationToken cancellationToken)
    {
        var waitTask = this.WaitAsync(key, cancellationToken);
        return waitTask.ContinueWith(
            (_, state) => new Releaser((KeyedLock)state),
            this, cancellationToken,
            TaskContinuationOptions.ExecuteSynchronously,
            TaskScheduler.Default
        );
    }
    public Task<Releaser> LockAsync(object key)
        => LockAsync(key, CancellationToken.None);

    public struct Releaser : IDisposable
    {
        private readonly KeyedLock _parent;
        internal Releaser(KeyedLock parent) { _parent = parent; }
        public void Dispose() { _parent?.Release(); }
    }

}

Answer 5

这似乎很适合一对ReaderWriterLockSlims。

private readonly ReaderWriterLockSlim changeStateLock = new ReaderWriterLockSlim();
private readonly ReaderWriterLockSlim doStuffLock = new ReaderWriterLockSlim();

一个控制对ChangeState的访问，另一个控制对DoStuff的访问。

读取器锁用于表示方法正在执行，写入器锁用于表示另一个方法正在执行。 ReaderWriterLockSlim 允许多次读取，但写入是独占的。

Task.Yield 只是将控制权交还给调用者，因为ReaderWriterLockSlim 的 ar 阻塞。

async Task ChangeState(bool state)
{
    await Task.Yield();

    doStuffLock.EnterWriteLock();

    try
    {
        changeStateLock.EnterReadLock();

        try
        {
            await OutsideApi.ChangeState(state);
        }
        finally
        {
            changeStateLock.ExitReadLock();
        }
    }
    finally
    {
        doStuffLock.ExitWriteLock();
    }
}

async Task DoStuff()
{
    await Task.Yield();

    changeStateLock.EnterWriteLock();

    try
    {
        doStuffLock.EnterReadLock();

        try
        {
            await OutsideApi.DoStuff();
        }
        finally
        {
            doStuffLock.ExitReadLock();
        }
    }
    finally
    {
        changeStateLock.ExitWriteLock();
    }
}