🖥 Как "схлопнуть" вызовы, чтобы вызываемый код не "захлебнулся"?

▶️Итак, сценарий такой. В коде генерируются некие обновления, которые затем обрабатываются неким методом. Обновления могут генерироваться слишком часто, плюс метод обработки может тормозить, в результате обновления не успевают обрабатываться, накапливаются, обработчик потом делает лишнюю работу.

Хотелось бы как-то "накопить" обновления, пока обработчик занят. Причём, сами данные для обновления уже "накапливаются", об этом заботиться не нужно, нужно только разрулить "накопление" запросов к обработчику, чтобы когда обработчик освободиться обрабатывать не все накопившиеся запросы на обновление по отдельности, а за один раз сразу весь пакет "обновлений" обработать.

Вот так происходит сейчас:

async Task UpdateFoo()
{   ...
    await SaveAsync();
}

async Task UpdateBar()
{   ...
    await SaveAsync();
}

async Task UpdateBaz()
{   ...
    await SaveAsync();
}

async Task SaveAsync()
{   // здесь нужно сделать так, 
    // чтобы одновременно обрабатывался только 1 запрос
    // по окончании которого проверялось бы не было ли ещё запросов
    // и если были, то обновление запускалось бы ещё 1 раз (сразу за всё "накопленное")
    // и так пока есть обновления по окончании очередной обработки
}

Как это можно нормально написать? Для того, чтобы зайти один раз — понятно, SemaphoreSlim, видимо (с WaitAsync). А вот остальное как лучше сделать?


▶️Вариант решения.

⏩Общий пример, берём System.Threading.Channels.Channel<T>:

private readonly Channel<T> _channel = Channel<T>.CreateUnbounded();

⏩Пишем метод разрегребания:

private Task _workerTask;

private async Task WorkerAsync()
{
    ChannelReader<T> reader = _channel.Reader;
    List<T> list = new();

    // ждём здесь, если в канале пусто
    while (await reader.WaitToReadAsync())
    {
        // забираем всё что есть
        // ну или можно счётчиком ограничить максимальное количество выгребаемых данных за раз
        while (reader.TryRead(out T data))
        {
            list.Add(data);
        }
        // пачка собрана, погнали. это можно в try-catch завернуть, чтобы воркер не падал
        await UseAsync(list);
        list.Clear();
    }
}

⏩Стартуем воркер:

_workerTask = WorkerAsync();

Чтобы закинуть в канал:

_channel.Writer.TryWrite(x);

⏩Чтобы закрыть канал (насовсем), и чтобы метод WorkerAsync завершился, нужно вызвать:

_channel.Writer.Complete();
await _workerTask;

Вот и все дела 🙃

@csharp_ci