c# - 基于任务的空闲检测

Question

希望限制某些事件之间的间隔并在超过限制时采取措施的情况并不少见。例如，网络对等体之间用于检测另一端是否存活的心跳消息。

在 C# async/await 风格中，可以通过每次心跳到达时替换超时任务来实现:

var client = new TcpClient { ... };
await client.ConnectAsync(...);

Task heartbeatLost = new Task.Delay(HEARTBEAT_LOST_THRESHOLD);
while (...)
{
    Task<int> readTask = client.ReadAsync(buffer, 0, buffer.Length);
    Task first = await Task.WhenAny(heartbeatLost, readTask);
    if (first == readTask) {
        if (ProcessData(buffer, 0, readTask.Result).HeartbeatFound) {
            heartbeatLost = new Task.Delay(HEARTBEAT_LOST_THRESHOLD);
        }
    }
    else if (first == heartbeatLost) {
        TellUserPeerIsDown();
        break;
    }
}

这很方便，但是延迟Task的每个实例都拥有一个Timer，如果很多心跳包到达的时间小于阈值，那就是很多Timer 对象加载线程池。此外，每个 Timer 的完成都会在线程池上运行代码，无论是否有任何继续链接到它。

您不能通过调用 heartbeatLost.Dispose() 释放旧的 Timer；那会给出一个异常(exception)

InvalidOperationException: A task may only be disposed if it is in a completion state

可以创建一个 CancellationTokenSource 并使用它来取消旧的延迟任务，但是当计时器本身具有可重新安排的功能时，创建更多对象来完成此任务似乎不是最佳选择。

集成计时器重新安排的最佳方式是什么，以便代码的结构更像这样？

var client = new TcpClient { ... };
await client.ConnectAsync(...);

var idleTimeout = new TaskDelayedCompletionSource(HEARTBEAT_LOST_THRESHOLD);
Task heartbeatLost = idleTimeout.Task;
while (...)
{
    Task<int> readTask = client.ReadAsync(buffer, 0, buffer.Length);
    Task first = await Task.WhenAny(heartbeatLost, readTask);
    if (first == readTask) {
        if (ProcessData(buffer, 0, readTask.Result).HeartbeatFound) {
            idleTimeout.ResetDelay(HEARTBEAT_LOST_THRESHOLD);
        }
    }
    else if (first == heartbeatLost) {
        TellUserPeerIsDown();
        break;
    }
}

Answer 1

对我来说似乎很简单，你假设的类(class)的名字让你大部分时间都到了那里。您只需要一个 TaskCompletionSource 和一个您不断重置的计时器。

public class TaskDelayedCompletionSource
{
    private TaskCompletionSource<bool> _completionSource;
    private readonly System.Threading.Timer _timer;
    private readonly object _lockObject = new object();

    public TaskDelayedCompletionSource(int interval)
    {
        _completionSource = CreateCompletionSource();
        _timer = new Timer(OnTimerCallback);
        _timer.Change(interval, Timeout.Infinite);
    }

    private static TaskCompletionSource<bool> CreateCompletionSource()
    {
        return new TaskCompletionSource<bool>(TaskCreationOptions.DenyChildAttach | TaskCreationOptions.RunContinuationsAsynchronously | TaskCreationOptions.HideScheduler);
    }

    private void OnTimerCallback(object state)
    {
        //Cache a copy of the completion source before we entier the lock, so we don't complete the wrong source if ResetDelay is in the middle of being called.
        var completionSource = _completionSource;
        lock (_lockObject)
        {
            completionSource.TrySetResult(true);
        }
    }

    public void ResetDelay(int interval)
    {
        lock (_lockObject)
        {
            var oldSource = _completionSource;
            _timer.Change(interval, Timeout.Infinite);
            _completionSource = CreateCompletionSource();
            oldSource.TrySetCanceled();
        }
    }
    public Task Task => _completionSource.Task;
}

这只会创建一个计时器并更新它，任务在计时器触发时完成。

您需要稍微更改代码，因为每次更新结束时间时都会创建一个新的 TaskCompletionSource，您需要将 Task heartbeatLost = idleTimeout.Task; 调用放在 while 循环中。

var client = new TcpClient { ... };
await client.ConnectAsync(...);

var idleTimeout = new TaskDelayedCompletionSource(HEARTBEAT_LOST_THRESHOLD);
while (...)
{
    Task heartbeatLost = idleTimeout.Task;
    Task<int> readTask = client.ReadAsync(buffer, 0, buffer.Length);
    Task first = await Task.WhenAny(heartbeatLost, readTask);
    if (first == readTask) {
        if (ProcessData(buffer, 0, readTask.Result).HeartbeatFound) {
            idleTimeout.ResetDelay(HEARTBEAT_LOST_THRESHOLD);
        }
    }
    else if (first == heartbeatLost) {
        TellUserPeerIsDown();
    }
}

编辑:如果您对完成源的对象创建有所了解(例如，您在游戏引擎中编程，其中 GC 收集是一个很大的问题)，您可以添加额外的逻辑到 OnTimerCallback 和 ResetDelay 以在调用尚未发生并且您确定自己不在 Reset Delay 内时重用完成源。

您可能需要从使用 lock 切换到 SemaphoreSlim 并将回调更改为

    private void OnTimerCallback(object state)
    {
        if(_semaphore.Wait(0))
        {
            _completionSource.TrySetResult(true);
        }
    }

稍后我可能会更新此答案以包含 OnTimerCallback 也包含的内容，但我现在没有时间。