Правильный способ отменить блокировку потока данных TPL - PullRequest
Новая
       109

Правильный способ отменить блокировку потока данных TPL

5 голосов
/ 29 мая 2020

Я использую блоки TPL для выполнения операции, которая может быть отменена пользователем: у меня есть два варианта, сначала я отменяю весь блок, но не отменяю операцию внутри блока, например: 

_downloadCts = new CancellationTokenSource();

var processBlockV1 = new TransformBlock<int, List<int>>(construct =>
{
    List<int> properties = GetPropertiesMethod(construct );
    var entities = properties
        .AsParallel()
        .Select(DoSometheningWithData)
        .ToList();
    return entities;
}, new ExecutionDataflowBlockOptions() { CancellationToken = _downloadCts.Token });

, а во втором - отменяю внутреннюю операцию, но не сам блок: 

var processBlockV2 = new TransformBlock<int, List<int>>(construct =>
{
    List<int> properties = GetPropertiesMethod(construct);
    var entities = properties
        .AsParallel().WithCancellation(_downloadCts.Token)
        .Select(DoSometheningWithData)
        .ToList();
    return entities;
});

Как я понимаю, первый вариант отменяет весь блок, таким образом закрывая весь конвейер. Мой вопрос в том, будет ли он также отменять внутреннюю операцию и избавляться от всех ресурсов, если они есть (открытые StreamReaders и т. Д.), Или лучше выбрать второй вариант, тогда я сам могу убедиться, что все отменено и очищено, а затем я мог бы использовать какие-то средства (железнодорожное программирование), чтобы спустить поднятый OperationCanceledException вниз по трубе и обрабатывать его там, где я хочу?

1 Ответ

3 голосов
/ 29 мая 2020

Эти две опции не эквивалентны.

  1. Первая опция (CancellationToken = _downloadCts.Token) заставит блок processBlockV1 отбрасывать любые сообщения, которые в данный момент находятся в его буфере (его InputCount изменится на 0), и перестанет принимать новые сообщения (вызов его метода Post неизменно вернет false). Тем не менее, он не остановит обработку сообщений, которые в настоящее время обрабатываются. Они будут полностью обработаны, но не будут распространяться ни на какие связанные блоки. После обработки этих сообщений блок завершится в отмененном состоянии (его свойство Completion.Status станет Canceled).

  2. Второй вариант (отмена внутренних операций) не повлияет на блок в целом. Блоки потока данных допускают любые OperationCanceledException, выброшенные из их функции обработки, и просто игнорируют неисправный элемент и переходят к следующему. Таким образом, после отмены токена все отправленные сообщения все равно будут обрабатываться, и блок продолжит принимать больше. Он просто не будет ничего распространять на связанные блоки, потому что все элементы будут выдавать OperationCanceledException и игнорироваться. В конкретном примере c метод GetPropertiesMethod будет вызываться для всех construct сообщений, и поэтому наложится задержка до завершения блока. Конечное состояние блока будет RanToCompletion.

Важно знать, что блоки Dataflow серьезно относятся к концепции Completion. Они будут ждать, пока все, что они знают, остановится, прежде чем сообщать о завершении. В случае, если вы хотите, чтобы они завершились преждевременно и оставили задачи, которые все еще выполняются, вам придется проделать такие трюки, как упаковка ваших задач отменяемыми оболочками .

...