Выполнение одного и того же ActionBlock несколько раз с неограниченным параллелизмом

Question

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

Я хочу/нужно отправлять сообщения так быстро, как я могу. Я делаю некоторое прототипирования в LINQPad, и это то, что я до сих пор:

// Holds all the messages for my loadMessage ActionBlock to grab its data from 
var bufferBlock = new BufferBlock<string>(); 

// Sends message to where it needs to go as fast as it can. 
var loadMessage = new ActionBlock<string>(msg => 
{ 
    msg.Dump(); 
}, 
new ExecutionDataflowBlockOptions 
{ 
    MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded 
}); 

// Links the blocks together 
bufferBlock.LinkTo(loadMessage); 

// Loads the Buffer 
for (int i = 0; i < 10; i++) 
{ 
    bufferBlock.Post("This is a message"); 
} 

//Calls completion to stop threads 
bufferBlock.Complete(); 
loadMessage.Complete(); 

Проблема заключается в том, что loadMessageBlock не демпинг сообщения в приведенном выше примере. Я искал некоторую информацию с небольшой удачей. Я думаю, что у меня нет чего-то фундаментального для TPL. Я понимаю, что BufferBlock содержит информацию для обработки другими блоками, а ActionBlocked (которая связана с BufferBlock) должна захватывать данные из буфера и делать то, что нужно. После цикла For, который помещает информацию в буфер, завершается завершение завершения, чтобы остановить потоки.

В реализации у меня есть Параллель. Для этого выполняется код внутри моего loadMessage просто отлично. Я просто не могу реализовать TPL, чтобы делать то, что хочу, и я понимаю, что TPL будет быстрее, чем Parallel.For.

Неужели я здесь, как я думаю, это должно работать? Я неправильно использую TPL? Я собираюсь продолжить исследование для ответа, любые указатели будут высоко оценены. Благодаря!

Answer 1

1. То, что я хочу/необходимость, чтобы отправлять сообщения так быстро, как я могу:

   Для достижения этой цели необходимо отправлять/принимать данные одновременно и из блока буфера. Ниже приведен фрагмент кода:

   var bufferBlock = new BufferBlock<string>(); 
   // Write to and read from the message block concurrently. 
   var post01 = Task.Run(() => 
   { 
       // Loads the Buffer 
       for (int i = 0; i < 10; i++) 
       { 
        bufferBlock.Post(string.Format("This is a message {0}",i)); 
       } 
   }); 
   var receive = Task.Run(() => 
   { 
       for (int i = 0; i < 10; i++) 
       { 
        var message = bufferBlock.Receive(); 
        message.Dump(); 
       } 
   }); 
   
   
   Task.WaitAll(post01, receive); 
   

   Более подробно об этом вы можете увидеть на MSDN link

2. я понимаю, что TPL будет быстрее, чем Parallel.For.

   Это неверно, поскольку они используют ту же основную структуру. Они принадлежат к одному пространству имен System.Threading.Tasks

Answer 2

Во-первых, примечание о терминологии: TPL (сокращенно Task Parallel Library) это не то же самое, как TPL DataFlow, это просто подмножество. TPL в целом включает такие вещи, как Parallel.For() и тип Task.

Теперь проблема с вашим кодом заключается в том, что вы слишком быстро завершаете блок loadMessage. После того, как вы вызовете Complete(), блок больше не примет никаких сообщений, поэтому сообщения, которые вы отправили в bufferBlock, никогда не достигнете loadMessage.

Вам необходимо заполнить loadMessage только после того, как bufferBlock отправит все свои сообщения. Какие именно то, что делает PropagateCompletion:

bufferBlock.LinkTo(
    loadMessage, new DataflowLinkOptions { PropagateCompletion = true }); 

// post your data to bufferBlock here 

bufferBlock.Complete(); 
await loadMessage.Completion; 

Кроме того, в этом конкретном случае, bufferBlock не нужен вообще, вы могли бы просто отправил сообщения непосредственно loadMessage.

я понимаю, что TPL будет быстрее, чем Parallel.For

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