Здравствуйте, TK, Вы писали:

TK>т.е даже то, что код совершенно не эквивалентный получается тоже не видно? ну

Да видно, конечно, но стартовое сообщение про что было?
Про запустить n конкурентных воркеров и посчитать время, так? Ну вот код именно это и делает. Без обработки частных случаев, да — так они на картину в общем никак не влияют.

С TPL DataFlow, согласен, всё своеобразно. Хорошая новость: его никто и не использует. И альтернативы тоже, нынче в моде готовые решения типа akka.net. Локальное раскидывание задач наотлично с await решается. Ну, или Rx, если нужно быстро сделать и особых требований к производительности нет.

Здравствуйте, Sinix, Вы писали:

S>Да видно, конечно, но стартовое сообщение про что было?
S>Про запустить n конкурентных воркеров и посчитать время, так? Ну вот код именно это и делает. Без обработки частных случаев, да — так они на картину в общем никак не влияют.

В итоге то что вышло?
Один сказал, что воркеры тут не нужны (хотя, это было в условиях), распараллелил Task.Delay/Thread.Sleep, а всю остальную работу (ради чего все и затевалось) сказал, что проще сделать последовательно.
Другой решил, что очередь воркерам не нужна — все проще, когда работа делится поровну и планируется заранее, а что-бы еще проще было вместо того, что-бы посчитать сумму элементов просто вернул их количество.

Все так?

PS
В исходном коде был метод Test:

    private static async Task<int> Test(int iterations, int totalWorkers)
    {
        var buffer = new BufferBlock<int>(new DataflowBlockOptions { BoundedCapacity = 1 });
        var workers = CreateWorkers(buffer, totalWorkers, Reduce);
        var results = MergeResults(workers, Reduce);
        for (var i = 0; i < iterations; ++i)
        {
            await buffer.SendAsync(1);
        }
        buffer.Complete();
        return await results.ReceiveAsync();
    }

Который можно переписать примерно так (просто меняем foreach на IObservable или IEnumerable — не принципиально):

    private static async Task<TResult> Test<TResult>(IObservable<TResult> source, int totalWorkers, Func<IPropagatorBlock<TResult, TResult>> reduce)
    {
        var buffer = new BufferBlock<TResult>(new DataflowBlockOptions { BoundedCapacity = 1 });
        var workers = CreateWorkers(buffer, totalWorkers, reduce);
        var results = MergeResults(workers, reduce);
    source.Subscrive(buffer.AsObserver());
        buffer.Complete();
        return await results.ReceiveAsync();
    }

т.е. количество исходных данных заранее не известно, какая функция будет применяться не известно (функция принимает на вход поток данных, на выходе агрегированый результат), время обработки одного элемента не нормировано.

Enumerable.Range, число итераций, вызов функции с константой N раз — это просто пример для теста. В реальности такого кода не будет.

Итого, чуть сменили сигнатуру на более осмысленную — что надо поменять в ваших реализациях что-бы хоть как-то соотноситься с реальностью?