异步版本的AutoResetEvent

Question

这是我第二次尝试创建异步版本的AutoResetEvent。一开始我试着使它完全没有锁，但结果证明是不可能的.此实现包含一个锁，但是该锁不是一个宽实例：

/// <summary>
///     Asynchronous version of <see cref="AutoResetEvent" />
/// </summary>
public sealed class AutoResetEventAsync {
    private static readonly Task<bool> Completed = Task.FromResult(true);

    private readonly ConcurrentQueue<TaskCompletionSource<bool>> handlers =
        new ConcurrentQueue<TaskCompletionSource<bool>>();

    private int isSet;

    /// <summary>
    ///     Initializes a new instance of the <see cref="AutoResetEventAsync" /> class with a Boolean value indicating whether to set the intial state to signaled.
    /// </summary>
    /// <param name="initialState">true to set the initial state signaled; false to set the initial state to nonsignaled.</param>
    public AutoResetEventAsync(bool initialState) {
        this.isSet = initialState ? 1 : 0;
    }

    /// <summary>
    ///     Sets the state of the event to signaled, allowing one waiting continuation to proceed.
    /// </summary>
    public void Set() {
        if (!this.TrySet())
            return;

        TaskCompletionSource<bool> handler;

        // Notify first alive handler
        while (this.handlers.TryDequeue(out handler))
            if (CheckIfAlive(handler)) // Flag check
                lock (handler) {
                    if (!CheckIfAlive(handler))
                        continue;

                    if (this.TryReset())
                        handler.SetResult(true);
                    else
                        this.handlers.Enqueue(handler);

                    break;
                }
    }

    /// <summary>
    ///     Try to switch the state to signaled from not signaled
    /// </summary>
    /// <returns>
    ///     true if suceeded, false if failed
    /// </returns>
    private bool TrySet() {
        return Interlocked.CompareExchange(ref this.isSet, 1, 0) == 0;
    }

    /// <summary>
    ///     Waits for a signal asynchronously
    /// </summary>
    public Task WaitAsync() {
        return this.WaitAsync(CancellationToken.None);
    }

    /// <summary>
    ///     Waits for a signal asynchronously
    /// </summary>
    /// <param name="cancellationToken">
    ///     A <see cref="P:System.Threading.Tasks.TaskFactory.CancellationToken" /> to observe while waiting for a signal.
    /// </param>
    /// <exception cref="OperationCanceledException">
    ///     The <paramref name="cancellationToken" /> was canceled.
    /// </exception>
    public Task WaitAsync(CancellationToken cancellationToken) {
        // Short path
        if (this.TryReset())
            return Completed;

        cancellationToken.ThrowIfCancellationRequested();

        // Wait for a signal
        var handler = new TaskCompletionSource<bool>(false);

        this.handlers.Enqueue(handler);

        if (CheckIfAlive(handler)) // Flag check
            lock (handler)
                if (CheckIfAlive(handler) && this.TryReset()) {
                    handler.SetResult(true);
                    return handler.Task;
                }

        cancellationToken.Register(() => {
            if (CheckIfAlive(handler)) // Flag check
                lock (handler)
                    if (CheckIfAlive(handler))
                        handler.SetCanceled();
        });

        return handler.Task;
    }

    private static bool CheckIfAlive(TaskCompletionSource<bool> handler) {
        return handler.Task.Status == TaskStatus.WaitingForActivation;
    }

    private bool TryReset() {
        return Interlocked.CompareExchange(ref this.isSet, 0, 1) == 1;
    }
}

有什么建议吗？

Answer 1

起初，我试图使它完全没有锁，但结果证明是不可能的。这个实现包含一个锁，但是锁不是一个实例。

听到无锁代码可能比使用锁定的代码慢一些，你可能会感到震惊。

无锁代码通常更长，而且肯定要复杂得多。在您完全理解.NET 内存 模型之前，不应该尝试编写无锁代码。在阅读了这些文章之后，乔·达菲的书将是一个很好的起点。我想说，当您能够描述为什么在中，双重检查的锁定中断了.并正确回答这个问题时，您已经准备好编写无锁代码了:它在C#中也被破坏了吗？

但是回到重点:在这个实例中，无锁代码比常规锁定要慢得多，因为只有在存在大量争用的情况下，无锁代码才会更快。当您编写async-compatible原语时，您所使用的任何锁只会持续到该方法调用的同步部分--时间非常短。因此，除非您拥有数百个不同的线程--它们同时调用Set或WaitAsync --否则就不会有问题。换句话说，async原语将您“提升”到更高的调度级别--您实际上构建了自己的服务员队列。而您所拥有的任何lock都不会影响真正的争用，它将由该队列表示。

所以，关于实际的代码审查..。

正如@svick正确指出的那样，当您的Set选择重新排队时，代码中存在可能导致饥饿的(良性)竞争状况。通过将“死”TCS实例保存在队列中，您还会占用资源的时间超过了必要的时间。请记住，每个TCS都有一个Task及其所有的连续性，以及它们的lambdas捕获的任何局部变量，等等。

但这并不是破坏交易的因素，只是有点不公平(饥饿)和低效(持有推荐信)。这个实现有一个更险恶和微妙的问题。

当您持有锁时，您正在回拨最终用户代码.这违反了多线程编程的核心规则之一。无论何时您这样做，都存在死锁的可能性。

这很微妙，因为回调并不明显，但在Set和Reset中都有：SetResult和SetCanceled都(同步)调用使用ExecuteSynchronously标志计划的任何任务延续。

Answer 2

你的新代码在我看来是安全的。尽管有一些小问题：

lock (handler)

我认为，当下面的语句超过一行时，应该在lock/if/while之后使用大括号，即使语言不需要它。它使得查看哪个块的结尾更简单。

if (CheckIfAlive(handler)) // Flag check
    lock (handler)
        if (CheckIfAlive(handler) && this.TryReset()) {
            handler.SetResult(true);
            return handler.Task;
        }

我看不出有什么理由在这里再做这种检查了。TryReset()之前已经被检查过几行了。

if (CheckIfAlive(handler)) // Flag check
    lock (handler)
        if (CheckIfAlive(handler))

双重检查锁定是一个危险的做法，因为这是很难得到正确的。我认为您在这里使用它是正确的，但我仍然会避免使用它，除非您知道小的性能增益实际上会对您产生影响。

if (CheckIfAlive(handler))
    handler.SetCanceled();

与其到处调用CheckIfAlive()，不如使用Try*方法。

if (this.TryReset())
    handler.SetResult(true);
else
    this.handlers.Enqueue(handler);

像AutoResetEvent这样的原语通常不能保证严格的FIFO顺序。但我认为他们至少应该保证一定程度的公平。使用您的实现，即使其他项被正常处理，也不太可能永远不会被处理(因为它经常被移到队列的后面)。我认为你应该考虑一下这对你是否有问题。

Answer 3

该代码执行许多cancellationToken.Register()调用而不注销注册。

我会创建一个内部类

private class Handler
{
   /// <summary>
   /// Gets or sets the task completion source.
   /// </summary>
   public TaskCompletionSource<bool> TaskCompletionSource { get; set; }

   /// <summary>
   /// Gets or sets the cancellation token registration.
   /// </summary>
   public CancellationTokenRegistration CancellationTokenRegistration { get; set; }
}

并将其设置为ConcurrentQueue所持有的内容。然后，走吧

var handler = new Handler { TaskCompletionSource = new TaskCompletionSource<bool>(false) };

从那里往下走几行：

lock (handler)
{
    handler.CancellationTokenRegistration = cancellationToken.Register(
        () =>
            {
                if (CheckIfAlive(handler.TaskCompletionSource)) // Flag check
                {
                    lock (handler)
                    {
                        if (CheckIfAlive(handler.TaskCompletionSource))
                        {
                            handler.TaskCompletionSource.SetCanceled();
                        }
                    }
                }
            });
}

最后，在使用脱队列处理程序而不请求的地方洒上handler.CancellationTokenRegistration.Dispose();行(不需要检查null，它是一个结构)。