Это работа для потока данных TPL?

Я запускаю довольно типичную модель производителя/потребителя для разных задач.

Task1: считывает партии байтов [] из двоичных файлов и запускает новую задачу для каждой коллекции массивов байтов. (операция выполняется для целей управления памятью).

Задача 2-n: это рабочие задачи, и каждый из них работает с переданной коллекцией (из Tasks1) массивов байт и де-сериализует массивы байтов, сортирует их по определенным критериям и затем сохраняет коллекцию результирующих объектов ( каждый массив байтов десериализуется в такой объект) в параллельном словаре.

Задача (n + 1) Я выбрал параллельный словарь, потому что задача этой задачи заключается в объединении тех коллекций, которые хранятся в параллельном словаре в том же порядке, что и в том, как они возникли из Task1. Я достигаю этого, передавая идентификатор collectionID (он имеет тип int и увеличивается для каждой новой коллекции в Task1) вплоть до задачи Task1 до этой задачи. Эта задача в основном проверяет, сохранен ли следующий ожидаемый идентификатор коллекции в параллельном словаре, и если да, вынимает его, добавляет в конечную очередь и проверяет следующую коллекцию в параллельном словаре.

Теперь, из того, что я прочитал, и видео, которое я смотрел, мне кажется, что TPL Dataflow может стать идеальным кандидатом для такой модели производителя/потребителя. Я просто не могу разработать дизайн и, таким образом, начать работу, потому что никогда не работал с TPL Dataflow. С точки зрения пропускной способности и латентности эта библиотека даже до задачи? В настоящее время я обрабатываю 2,5 миллиона байт-массивов и, следовательно, объекты в секунду в результирующих коллекциях. Может ли облегчить поток данных TPL? Меня особенно интересует ответ на следующий вопрос: может ли поток данных TPL сохранять порядок сборочных партий из Task1 при нерестах рабочих задач и повторного слияния их после выполнения рабочими задачами? Это оптимизирует ситуацию? Профилировав всю структуру, я чувствую, что довольно много времени тратится впустую из-за вращения и слишком много одновременных коллекций.

Любые идеи, мысли?

Ответ 1

EDIT: Оказывается, я был очень неправ. TransformBlock возвращает элементы в том же порядке, в каком они вошли, даже если он настроен для parallelism. Из-за этого код в моем исходном ответе полностью бесполезен, и вместо него можно использовать обычный TransformBlock.

Оригинальный ответ:

Насколько мне известно, только одна конструкция parallelism в .Net поддерживает возврат обработанных элементов в том порядке, в котором они вошли: PLINQ с AsOrdered(). Но мне кажется, что PLINQ не соответствует тому, что вам нужно.

TPL Dataflow, с другой стороны, подходит хорошо, я думаю, но у него нет блока, который поддерживал бы parallelism и возвращал элементы в порядке в одно и то же время (TransformBlock поддерживает их оба, но не в то же время). К счастью, блоки потока данных были разработаны с учетом возможности компоновки, поэтому мы можем создать собственный блок, который делает это.

Но сначала мы должны выяснить, как упорядочить результаты. Использование параллельного словаря, как вы и предполагали, наряду с некоторым механизмом синхронизации, безусловно, будет работать. Но я думаю, что есть более простое решение: используйте очередь Task s. В выходной задаче вы удаляете Task, дожидаясь ее завершения (асинхронно), и когда это произойдет, вы отправляете ее результат. Нам по-прежнему нужна некоторая синхронизация для случая, когда очередь пуста, но мы можем получить это бесплатно, если мы выберем, какую очередь использовать умнее.

Итак, общая идея такова: мы пишем IPropagatorBlock с некоторым вводом и некоторым выходом. Самый простой способ создать пользовательский IPropagatorBlock - создать один блок, обрабатывающий вход, другой блок, который производит результаты, и обрабатывает их как один с помощью DataflowBlock.Encapsulate().

Входной блок должен будет обрабатывать входящие элементы в правильном порядке, поэтому там нет распараллеливания. Он создаст новый Task (фактически, TaskCompletionSource, чтобы мы могли позже установить результат Task), добавьте его в очередь, а затем отправьте элемент для обработки вместе с каким-то образом, чтобы установить результат правильного Task. Поскольку нам не нужно связывать этот блок с чем-либо, мы можем использовать ActionBlock.

Выходной блок должен будет принимать Task из очереди, асинхронно ждать их, а затем отправлять их. Но так как у всех блоков есть встроенная в них очередь, а блоки, которые принимают делегаты, имеют асинхронное ожидание, это будет очень просто: new TransformBlock<Task<TOutput>, TOutput>(t => t). Этот блок будет работать как в очереди, так и в качестве выходного блока. Из-за этого нам не нужно иметь дело с какой-либо синхронизацией.

Последняя часть головоломки фактически обрабатывает элементы параллельно. Для этого мы можем использовать еще один ActionBlock, на этот раз с MaxDegreeOfParallelism. Он будет принимать входные данные, обрабатывать их и задавать результат правильного Task в очереди.

Совместим, это может выглядеть так:

public static IPropagatorBlock<TInput, TOutput>
    CreateConcurrentOrderedTransformBlock<TInput, TOutput>(
    Func<TInput, TOutput> transform)
{
    var queue = new TransformBlock<Task<TOutput>, TOutput>(t => t);

    var processor = new ActionBlock<Tuple<TInput, Action<TOutput>>>(
        tuple => tuple.Item2(transform(tuple.Item1)),
        new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded
        });

    var enqueuer = new ActionBlock<TInput>(
        async item =>
        {
            var tcs = new TaskCompletionSource<TOutput>();
            await processor.SendAsync(
                new Tuple<TInput, Action<TOutput>>(item, tcs.SetResult));
            await queue.SendAsync(tcs.Task);
        });

    enqueuer.Completion.ContinueWith(
        _ =>
        {
            queue.Complete();
            processor.Complete();
        });

    return DataflowBlock.Encapsulate(enqueuer, queue);
}

После стольких разговоров, я думаю, это довольно небольшой код.

Кажется, вы очень заботитесь о производительности, поэтому вам может потребоваться тонкая настройка этого кода. Например, имеет смысл установить MaxDegreeOfParallelism блока processor на что-то вроде Environment.ProcessorCount, чтобы избежать переподписки. Кроме того, если задержка более важна, чем пропускная способность для вас, имеет смысл установить MaxMessagesPerTask того же блока на 1 (или другое небольшое число), чтобы при завершении обработки элемента он немедленно отправлялся на выход.

Кроме того, если вы хотите дросселировать входящие элементы, вы можете установить BoundedCapacity в enqueuer.