Это работа для TPL Dataflow?

РЕДАКТИРОВАТЬ: Оказалось, что я был очень неправ. TransformBlock возвращает элементы в том же порядке, в котором они пришли, даже если он настроен для параллелизма. Из-за этого код в моем исходном ответе совершенно бесполезен, и вместо него можно использовать обычный TransformBlock.

Исходный ответ:

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

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

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

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

Входной блок должен будет обрабатывать входящие элементы в правильном порядке, поэтому распараллеливания здесь нет. Будет создан новый 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.

Да, TPL Dataflow Библиотека хорошо подходит для этой работы. Он поддерживает все необходимые вам функции: MaxDegreeOfParallelism, BoundedCapacity и _3 _ . Но использование опции BoundedCapacity требует внимания к деталям.

Сначала вы должны убедиться, что вы загрузили первый блок в конвейере с _ 5_. В противном случае, если вы используете Post < / a> и игнорируйте его возвращаемое значение, вы можете потерять сообщения. SendAsync никогда не потеряет сообщения, потому что он асинхронно блокирует вызывающего до тех пор, пока не освободится место для входящего сообщения во внутреннем буфере блока.

public static async Task ProcessAsync(string[] filePaths,
    ConcurrentQueue<MyClass> finalQueue)
{
    var reader = new TransformBlock<string, byte[]>(filePath =>
    {
        byte[] result = ReadBinaryFile(filePath);
        return result;
    }, new ExecutionDataflowBlockOptions()
    {
        MaxDegreeOfParallelism = 1, // This is the default
        BoundedCapacity = 20, // keep memory usage under control
        EnsureOrdered = true // This is also the default
    });

    var deserializer = new TransformBlock<byte[], MyClass>(bytes =>
    {
        MyClass result = Deserialize(bytes);
        return result;
    }, new ExecutionDataflowBlockOptions()
    {
        MaxDegreeOfParallelism = Environment.ProcessorCount,
        BoundedCapacity = 20
    });

    var writer = new ActionBlock<MyClass>(obj =>
    {
        finalQueue.Enqueue(obj);
    });

    reader.LinkTo(deserializer,
        new DataflowLinkOptions() { PropagateCompletion = true });
    PropagateFailure(deserializer, reader); // Link backwards

    deserializer.LinkTo(writer,
        new DataflowLinkOptions() { PropagateCompletion = true });
    PropagateFailure(writer, deserializer); // Link backwards

    foreach (var filePath in filePaths)
    {
        var accepted = await reader.SendAsync(filePath).ConfigureAwait(false);
        if (!accepted) break; // This will happen in case that the block has failed
    }
    reader.Complete(); // This will be ignored if the block has already failed

    await writer.Completion; // This will propagate the first exception that occurred
}

public static async void PropagateFailure(IDataflowBlock block1,
    IDataflowBlock block2)
{
    try { await block1.Completion.ConfigureAwait(false); }
    catch (Exception ex) { block2.Fault(ex); }
}