Тайм-аут без выделения времени на задании

Question

Я бы хотел добавить таймаут к ChannelReader.ReadyAsyn c. Вот два решения, которые я нашел:

var cts = new CancellationTokenSource();
cts.CancelAfter(2000);
try {
  var data = chan.ReadAsync(cts.Token);
} catch (OperationCanceledException) {
  // timeout
}

var tasks = new Task[] { Task.Delay(2000), chan.ReadAsync(CancellationToken.None) };
var completedTask = await Task.WhenAny(tasks);
if (completedTask == tasks[0])
  // timeout
else
  var data = ((T)completedTask).Result;

Однако оба эти решения не являются свободными для распределения. Первый выделяет CancellationTokenSource, а второй - таймер в Task.Delay. Есть ли способ сделать подобный код без какого-либо выделения?

EDIT 1: вывод dotTrace при использовании первого решения

Answer 1

Класс ChannelReader<T> имеет метод ReadAllAsync, который предоставляет данные считывателя как IAsyncEnumerable<T>. Ниже приведена перегрузка этого метода, которая также принимает параметр timeout. Этот параметр приводит к тому, что в случае, если считыватель не может испустить какие-либо предметы в течение указанного промежутка времени, выдается TimeoutException.

Для сокращения выделений он использует ту же хитрую технику из Грега ответ , с одним CancellationTokenSource, который перепланируется для отмены после каждой итерации. Подумав немного, я удалил строку CancelAfter(int.MaxValue), потому что в общем случае она скорее вредна, чем полезна, но я могу ошибаться.

public static async IAsyncEnumerable<TSource> ReadAllAsync<TSource>(
    this ChannelReader<TSource> source, TimeSpan timeout,
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    while (true)
    {
        using var cts = new CancellationTokenSource();
        using var linkedCts = CancellationTokenSource
            .CreateLinkedTokenSource(cts.Token, cancellationToken);
        cts.CancelAfter(timeout);
        while (true)
        {
            try
            {
                if (!await source.WaitToReadAsync(linkedCts.Token).ConfigureAwait(false))
                    yield break;
            }
            catch (OperationCanceledException) when (cts.IsCancellationRequested)
            {
                throw new TimeoutException();
            }
            while (source.TryRead(out var item))
            {
                yield return item;
            }
            cts.CancelAfter(timeout);
            if (cts.IsCancellationRequested) break;
        }
    }
}

Как я отмечаю, System.Interactive. Пакет Asyn c включает в себя метод Timeout с подписью, показанной ниже, который можно использовать в сочетании со встроенным ReadAllAsync и предоставлять те же функциональные возможности с вышеприведенной реализацией. Этот метод не оптимизирован для низких распределений.

public static IAsyncEnumerable<TSource> Timeout<TSource>(
    this IAsyncEnumerable<TSource> source, TimeSpan timeout);

Answer 2

Спасибо за ваши ответы, они снова заставили меня задуматься о том, что я искал: повторное использование CancellationTokenSource. После отмены CancellationTokenSource вы не сможете использовать его повторно. Но в моем случае ChannelReader.ReadAsync большую часть времени вернется до срабатывания тайм-аута, поэтому я использовал тот факт, что CancelAfter не воссоздает таймер во второй раз, когда вы вызываете его , чтобы избежать отмены CancellationTokenSource после возврата ChannelReader.ReadAsync.

var timeoutCancellation = new CancellationTokenSource();

while (true)
{
    if (timeoutCancellation.IsCancellationRequested)
    {
        timeoutCancellation.Dispose();
        timeoutCancellation = new CancellationTokenSource();
    }

    T data;
    try
    {
        timeoutCancellation.CancelAfter(2000);
        data = await _queue.Reader.ReadAsync(timeoutCancellation.Token);
        // make sure it doesn't get cancelled so it can be reused in the next iteration
        // Timeout.Infinite won't work because it would delete the underlying timer
        timeoutCancellation.CancelAfter(int.MaxValue);
    }
    catch (OperationCanceledException) // timeout reached
    {
        // handle timeout
        continue;
    }

    // process data
}

Это не без выделения, но значительно уменьшает количество выделенных объектов.