2

Может есть какой-то готовый код, чтобы не изобретать велосипеды.

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

Как сейчас:

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

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

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

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

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

6
  • 2
    Producer/consumer? Но его обычно в многопоточном варианте рассматривают...
    – Qwertiy
    Commented 20 мар. в 20:53
  • Вообще, а можно ли их пачкой выполнять? Что если одно из изменений в середине некорректное?
    – Qwertiy
    Commented 20 мар. в 20:55
  • @Qwertiy В какой-то степени да, продюсер-консбюмер, но не хочется на консьюмер специально отдельный поток крутить всё время и хочется как-то "схлопывать" много запросов в одну обработку.
    – CrazyElf
    Commented 20 мар. в 20:56
  • 1
    Думаю, стоит выбрать наиболее полезный ответ и отметить.
    – aepot
    Commented 30 мар. в 19:04
  • 1
    @aepot Я так и недоразобрался пока в вашем ответе, но про каналы думаю всё-равно будет самый полезный, отмечу его
    – CrazyElf
    Commented 1 апр. в 6:46

3 ответа 3

2

Общий пример, берём 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;

Вот и вся магия.


Вот код на основании вашего самоответа, с точно такой же логикой

private readonly Channel<T> _channel = Channel<T>.CreateBounded(1); // максимум один в очереди

private async Task WorkerAsync()
{
    await foreach (var data in _channel.Reader.ReadAllAsync())
    {
        try
        {
            await _producer.SendAsync(data);
        }
        catch (Exception ex)
        {
            _logger.Error(ex, "Ошибка отсылки обновлений");
        }
    } 
}

/// <summary>
/// Отсылка с ограничениями: один поток отсылает, ещё один может ждать в очереди, остальные просто возвращаются
/// </summary>
/// <returns></returns>
private void SendUpdatedData()
{
    if (!channel.Writer.TryWrite(GetData())
    {
        _logger.Debug("Уже есть ожидание отсылки обновления...");
    }
}

Здесь только консьюмера запустить один раз на всё вермя работы приложения, а как его остановить, показано выше - _channel.Writer.Complete()

Task task = WorkerAsync(); // ну или `_ = ...`

Смысл этого таска для того что если у вас приложения завершается, вы например можете вызвать Complete() и через await task; дождаться когда все запросы воркером обработаются, после чего нормально завершить работу. А в предложенном вами решении такой возможности не предусмотрено.

10
  • Погоди, а тот, кто делал await SaveAsync() как узнает, что именно его сохранение выполнено? А то что-то код выгладит в стиле fire and forget без расчёта на await.
    – Qwertiy
    Commented 20 мар. в 22:31
  • 1
    А, типа этого ru.stackoverflow.com/q/549318/178988 поместить внутрь типа T и вернуть такую таску из SaveAsync? Плюсанул.
    – Qwertiy
    Commented 20 мар. в 23:34
  • 1
    @CrazyElf ну создайте канал, запустите воркер, закройте канал когда вам надо. Воркер асинхронный, он не занимает потоков. Когда нет данных, он просто сидит и ждёт асинхронно, когда вы что-то добавите в канал. О каких тасках речь? Это не синхронный код. В каналах там даже изучать особо нечего, это обычный Producer/Consumer примитив. А если не воркер, то надо все равно очередь где-то хранить. Канал это и есть то самое хранилище, при чём потокобезопасное.
    – aepot
    Commented 21 мар. в 7:14
  • 1
    Спасибо за добавление, почитаю, осознаю.
    – CrazyElf
    Commented 25 мар. в 8:03
  • 1
    Не понял, а кто последнее изменение сохранять будет, если оно произойдёт в момент сохранения?
    – Qwertiy
    Commented 25 мар. в 10:34
2

Если все вызовы SaveAsync идут из одного потока, то можно реализовать так:

  1. Помещаем в очередь изменение

  2. Если работа по сохранению уже в процессе, ждём её завершения

  3. Если очередь не пуста

    1. забираем всё из неё
    2. свою задачу сохраняем как ожидающую завершения

    Иначе

    1. ждём завершения сохранения

Если вызовы могут быть многопоточными, то надо офигенно аккуратно расставить локи так, чтобы они покрывали все дайствия, но не время ожидания.


Что-то такое (но ничего не гарантирую, и вообще, писал прямо тут):

object lobj = new();
List<...> changes = new();
Task active = Task.FromResult();

async Task SaveAsync()
{
  Task cur;

  lock (lobj)
  {
    changes.Add(...);
    cur = active;
  }

  await cur;

  lock (lobj)
  {
    if (changes.Count !== 0)
    {
      active = cur = SaveAsyncReal(changes.ToList());
      changes.Clear();
    }
    else
      cur = active;
  }

  await cur;
}

async Task SaveAsyncReal(List<...> diff)
{
  ...
}

Но если сюда ещё и обработку ошибок надо присобачить, то это вообще долго думать надо...

11
  • У автора асинхронный код, какие локи? :)
    – aepot
    Commented 20 мар. в 22:24
  • @aepot, во-первых, а в чём противоречие? Во-вторых, примерно накидал.
    – Qwertiy
    Commented 20 мар. в 22:29
  • В принципе код норм, в стиле JS, где нет всяких Channel<T>.
    – aepot
    Commented 20 мар. в 22:44
  • @aepot, зато не требует лишнего потока)
    – Qwertiy
    Commented 20 мар. в 23:37
  • Я бы в начале заyieldил Task, а локи бы шевелил потом. Но многое зависит от контекста синхронизации. Потоки могут как меняться, так и нет после await, в зависимости от контекста.
    – aepot
    Commented 20 мар. в 23:40
0

Сам пока сделал на таске и двух семафорах, хотя другие решения тоже интересные, но не совсем подходят под мою задачу. Суть в том, что одно обновление работает, другое ожидает в очереди, все остальные обновления отбрасываются - они излишни, второго ожидающего обновления достаточно.

private void SendUpdatedData()
{
    _ = Task.Run(async () => { await SendAsync(); });
}

/// <summary>
/// Отсылка с ограничениями: один поток отсылает, ещё один может ждать в очереди, остальные просто возвращаются
/// </summary>
/// <returns></returns>
private async Task SendAsync()
{ 
    if (_semaphore2.CurrentCount == 0)
    {
        _logger.Debug("Уже есть ожидание отсылки обновления...");
        return;
    }
    await _semaphore2.WaitAsync();
    try
    {
        await _semaphore1.WaitAsync();
        try
        {
            var data = GetData();
            await _producer.SendAsync(data);
        }
        finally
        {
            _semaphore1.Release();
        }
    }
    catch (Exception ex)
    {
        _logger.Error(ex, "Ошибка отсылки обновлений");
    }
    finally
    {
        _semaphore2.Release();
    }
}

Покритикуйте, пожалуйста. )

10
  • 1
    ` _ = Task.Run(async () => { await SendAsync(); });` так нельзя, но если очень хочется, то это делается так _ = Task.Run(() => SendAsync()); не надо здесь никаких async-await. Манипуляций с семафорами не понял. Также не понял, чем вам каналы не угодили. Если там нужно сделать отлуп вызывающему коду, если очередь переполнена, есть Bounded канал. Channel.CreateBounded<T>(2)
    – aepot
    Commented 25 мар. в 7:43
  • @aepot Спасибо. А почему именно так нельзя? У меня там try/catch ещё на самом деле, поэтому await внтури try/catch. Насчёт CreateBounded почитаю, спасибо! C семафорами я делаю то, что описал - один запрос обрабатывается, второй ждёт в очереди, остальным отлуп.
    – CrazyElf
    Commented 25 мар. в 7:58
  • 1
    Потому что это эквивалент async void. Разница только в принудительном запуске на пуле потоков и без того асинхронного кода.
    – aepot
    Commented 25 мар. в 7:59
  • @aepot, Task.Run(SendAsync) - не? Зачем лямбда?
    – Qwertiy
    Commented 25 мар. в 10:29
  • @Qwertiy можно и так
    – aepot
    Commented 25 мар. в 10:30

Ваш ответ

Нажимая «Отправить ответ», вы соглашаетесь с условиями пользования и подтверждаете, что прочитали политику конфиденциальности.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.