Поток данных (библиотека параллельных задач) и асинхронное ожидание

Допустим, я использую блоки потока данных в .NET. Утверждается, что «эта модель потока данных продвигает программирование на основе акторов». Это именно то, что я хочу здесь получить.

Однако, если я обрабатываю сообщения, скажем, из BufferBlock<T> и в обработчике сообщения я решу использовать async/await, это приведет к разветвлению выполнения на текущий поток и рабочий поток для ожидаемой задачи.

Есть ли способ предотвратить одновременное выполнение в действующем/обработчике сообщений здесь?

это нормально, если ожидаемая задача выполняется с использованием неблокирующих операций ввода-вывода с собственными обратными вызовами. Но мне бы очень хотелось, чтобы любой код .NET выполнялся только синхронно.


person Roger Johansson    schedule 30.12.2013    source источник
comment
Было бы намного проще объяснить ответ на конкретном примере. Можете ли вы опубликовать свой код?   -  person Panagiotis Kanavos    schedule 02.01.2014
comment
Забавный факт: это была самая ранняя работа над Akka.NET.   -  person Roger Johansson    schedule 07.12.2016


Ответы (2)


Это очень сильно зависит от того, как именно вы будете обрабатывать эти сообщения.

Если вы используете ActionBlock с async action, не устанавливайте его MaxDegreeOfParallelism (что означает, что используется значение по умолчанию 1) и не связывайте его с BufferBlock, тогда action будет выполняться для одного сообщения за раз, параллелизма не будет .

Если вы обрабатываете сообщения вручную, используя такой цикл:

while (await bufferBlock.OutputAvailableAsync())
{
    var message = await bufferBlock.ReceiveAsync();
    await ProcessMessageAsync(message);
}

Тогда сообщения будут обрабатываться тоже по одному.

Но в обоих случаях это не означает, что сообщение будет обрабатываться одним потоком. Он может обрабатываться несколькими потоками, но не параллельно. Это связано с тем, что после await выполнение может возобновиться в потоке, отличном от того, в котором оно было приостановлено.

Если вы используете какой-либо другой способ обработки сообщений (например, используя описанный выше цикл, но пропуская await перед ProcessMessageAsync()), тогда несколько сообщений могут обрабатываться одновременно.

person svick    schedule 03.01.2014

Вы неправильно понимаете, что делает await. Он ничего не разветвляет, он просто ожидает результата уже асинхронной операции.

Методы, отмеченные ключевым словом async, не становятся автоматически асинхронными. Только когда внутри метода async встречается асинхронная операция, выполнение продолжается асинхронно. Ключевое слово async просто сообщает компилятору, где должно продолжаться выполнение после завершения асинхронной операции.

Ни один поток ThreadPool не теряется и не повреждается во время ожидания, поэтому вам не следует пытаться ограничить, предотвратить или обойти это. На самом деле вы получаете лучшую масштабируемость при использовании асинхронных операций, поскольку потоки ThreadPool, используемые потоком данных TPL, не блокируются в ожидании длительных асинхронных операций, таких как ввод-вывод или вызовы веб-службы.

person Panagiotis Kanavos    schedule 30.12.2013
comment
Хорошо, плохая формулировка с моей стороны, давайте предположим следующий пример, мой актор обрабатывает сообщение из моего буферного блока, в этой обработке мы нажимаем оператор ожидания, теперь выполнение возвращается к вызывающему, верно? остальная часть кода будет выполнена после завершения задачи. разве это не открывает для процессора возможность начать обработку нового сообщения, в то время как ожидаемая задача также выполняется? - person Roger Johansson; 30.12.2013
comment
Не совсем. Блоки действий всегда выполняют свои действия асинхронно, соблюдая параметры выполнения, чтобы ограничить количество одновременных задач. Если вы делаете еще один асинхронный вызов, например HttpClient.GetStringAsync без добавления await выполнение продолжится немедленно. Если вы добавите await, выполнение продолжится только после завершения GetStringAsync. - person Panagiotis Kanavos; 30.12.2013
comment
Это не работает, я пробовал, и блок ожидания выполняется в другом потоке, чем обработчик сообщений. - person Roger Johansson; 01.01.2014
comment
Вот как в целом работает TPL. Что заставляет вас думать, что это проблема или что это не работает? В любом случае выполнение действия началось в потоке ThreadPool. Когда вы вызываете асинхронный метод с await, он продолжается в другом потоке ThreadPool. Когда это закончится, код после await будет выполнен в другом потоке. Любая путаница будет вызвана только в том случае, если вы опустите await: в этом случае и асинхронное действие, и действие ActionBlock могут выполняться одновременно. - person Panagiotis Kanavos; 02.01.2014