Дублировать исключения с помощью BroadcastBlock в потоке данных TPL

Я пытаюсь использовать поток данных TPL для создания конвейера. Пока все работает нормально, а мой конвейер определяется следующим образом (хотя моя проблема связана только с вещателем, представлениеСуществует, отправлено):

// Define tasks
var productListingBatchBuffer = new BufferBlock<PostSubmissionState>();
var splitFile = new TransformBlock<PostSubmissionState, PostSubmissionState>(s => SplitFile(s));
var saveFile = new TransformBlock<PostSubmissionState, PostSubmissionState>(s => SaveFile(s));
var postSubmission = new TransformBlock<PostSubmissionState, PostSubmissionState>(s => PostSubmission(s));
var broadcaster = new BroadcastBlock<PostSubmissionState>(state => state);
var submissionSucceeded = new ActionBlock<PostSubmissionState>(s => SubmissionSucceeded(s));
var submissionFailed = new ActionBlock<PostSubmissionState>(s => SubmissionFailed(s));

// Link em up
productListingBatchBuffer.LinkTo(splitFile, new DataflowLinkOptions() { PropagateCompletion = true });
splitFile.LinkTo(saveFile, new DataflowLinkOptions() { PropagateCompletion = true });
saveFile.LinkTo(postSubmission, new DataflowLinkOptions() { PropagateCompletion = true });
postSubmission.LinkTo(broadcaster, new DataflowLinkOptions() { PropagateCompletion = true });
broadcaster.LinkTo(submissionSucceeded, new DataflowLinkOptions() { PropagateCompletion = true }, state => state.PostSucceeded);
broadcaster.LinkTo(submissionFailed, new DataflowLinkOptions() { PropagateCompletion = true }, state => !state.PostSucceeded);

Проблема, которую я имею, связана с распространением Исключений. Поскольку мой BroadcastBlock распространяет свое завершение (и, следовательно, любой Fault) на два блока, если исключение действительно происходит, оно распространяется на оба блока. Таким образом, когда я делаю

Task.WaitAll(submissionSucceeded.Completion, submissionFailed.Completion);

Я получаю сводное исключение, содержащее два исключения. Прямо сейчас самое лучшее, что я могу сделать, это отфильтровать их, т.е.:

try
{
    Task.WaitAll(submissionSucceeded.Completion, submissionFailed.Completion);
}
catch (AggregateException ex)
{
    var uniqueExceptions = new AggregateException(ex.Flatten().InnerExceptions.Distinct());
    Console.WriteLine("An exception was thrown.\n{0}", uniqueExceptions.Flatten());
}

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

Ответы

Ответ 1

Я написал пример TPL DataFlow (https://github.com/squideyes/PodFetch), который использует несколько иной подход к завершению и обработке ошибок. Здесь соответствующий код от строки 171 до 201 программы Program.cs:

    scraper.LinkTo(fetcher, link => link != null);
    scraper.LinkTo(DataflowBlock.NullTarget<Link>());

    scraper.HandleCompletion(fetcher);

    Status.Info.Log("Fetching APOD archive list");

    links.ForEach(link => scraper.Post(link));

    scraper.Complete();

    try
    {
        await fetcher.Completion;

        Status.Finished.Log("Fetched: {0:N0}, Skipped: {1:N0}, Errors: {2:N0}, Seconds: {3:N2}",
            fetched, skipped, errored, (DateTime.UtcNow - startedOn).TotalMilliseconds / 1000.0);
    }
    catch (AggregateException errors)
    {
        foreach (var error in errors.InnerExceptions)
            Status.Failure.Log(error.Message);
    }
    catch (TaskCanceledException)
    {
        Status.Cancelled.Log("The process was manually cancelled!");
    }
    catch (Exception error)
    {
        Status.Failure.Log(error.Message);
    }

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

    public static void HandleCompletion(
        this IDataflowBlock source, params IDataflowBlock[] targets)
    {
        source.Completion.ContinueWith(
            task =>
            {
                foreach (var target in targets)
                {
                    if (task.IsFaulted)
                        target.Fault(task.Exception);
                    else
                        target.Complete();
                }
            });
    }

Очень важно, что я вызываю scraper.Complete(), когда я закончил передачу объектов в первый блок в цепочке. Таким образом, метод расширения HandleCompletion имеет дело с продолжением. И, поскольку я жду на приемника (последний блок в цепочке для завершения), легко поймать любые возникающие ошибки в try/catch.