Ускорение потока данных TPL?

Question

Интересно, можно ли оптимизировать следующий код для выполнения быстрее. В настоящее время я представляю максимум около 1,4 миллиона простых сообщений в секунду на довольно простой структуре потока данных. Я знаю, что этот примерный процесс передает/трансформирует сообщения синхронно, однако в настоящее время я тестирую поток данных TPL как возможную замену для своего собственного решения на основе задач и параллельных коллекций. Я знаю, что термины «параллельные» уже предлагают мне работать параллельно, но для текущих целей тестирования я толкал сообщения на своем собственном решении синхронно, и я получаю около 5,1 млн сообщений в секунду. Что мне здесь не хватает, я прочитал TPL Dataflow, был выдвинут как решение с высокой пропускной способностью и низкой задержкой, но до сих пор я должен упускать из виду настройки производительности. Любой, кто мог бы указать мне в правильном направлении, пожалуйста?

class TPLDataFlowExperiments 
{ 
    public TPLDataFlowExperiments() 
    { 
     var buf1 = new BufferBlock<int>(); 

     var transform = new TransformBlock<int, string>(t => 
      { 
       return ""; 
      }); 

     var action = new ActionBlock<string>(s => 
      { 
       //Thread.Sleep(100); 
       //Console.WriteLine(s); 
      }); 

     buf1.LinkTo(transform); 
     transform.LinkTo(action); 

     //Propagate all Completions down the flow 
     buf1.Completion.ContinueWith(t => 
     { 
      transform.Complete(); 
      transform.Completion.ContinueWith(u => 
      { 
       action.Complete(); 
      }); 
     }); 

     Stopwatch watch = new Stopwatch(); 
     watch.Start(); 

     int cap = 10000000; 
     for (int i = 0; i < cap; i++) 
     { 
      buf1.Post(i); 
     } 

     //Mark Buffer as Complete 
     buf1.Complete(); 

     action.Completion.ContinueWith(t => 
      { 
       watch.Stop(); 

       Console.WriteLine("All Blocks finished processing"); 
       Console.WriteLine("Units processed per second: " + cap/watch.ElapsedMilliseconds * 1000); 
      }); 

     Console.ReadLine(); 
    } 
} 

Answer 1

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

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

Но чтобы ответить на ваш вопрос, да, вы не замечаете некоторые улучшения производительности. В частности, SingleProducerConstrained, что означает, что могут использоваться структуры данных с меньшей блокировкой. Если я использую это на обоих блоках (BufferBlock здесь совершенно бесполезен, вы можете безопасно удалить его), скорость увеличивается от примерно 3-4 миллионов единиц в секунду до более чем 5 миллионов на моем компьютере.

Answer 2

Чтобы добавить ответ svick, в тесте используется только один поток обработки для одного блока действий. Таким образом, он проверяет не что иное, как накладные расходы на использование блоков.

DataFlow работает аналогично F # Agents, актерам Scala и реализациям MPI. Каждый блок действий выполняет отдельную задачу за раз, слушая ввод и вывод данных. Ускорение обеспечивается путем разбиения алгоритма по этапам, которые могут выполняться независимо на нескольких ядрах, передавая только сообщения друг другу.

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

Answer 3

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