VB.NET, лучшая практика для сортировки параллельных результатов threadpooling?

Question

Короче говоря, я пытаюсь «сортировать» входящие результаты использования threadpooling по мере их завершения. У меня есть функциональное решение, но в мире нет способа сделать это лучше (он подвержен огромным паузам). И вот я здесь! Я постараюсь поразить пулевые моменты того, что происходит/что должно произойти, а затем опубликовать мое текущее решение.

• Цель кода - получить информацию о файлах в каталоге и затем записать его в текстовый файл.

• У меня есть список (Counter.ListOfFiles), который представляет собой список путей к файлу, отсортированных определенным образом. Это руководство, которое диктует порядок, который мне нужно записать в текстовый файл.

• Я использую threadpool для сбора информации о каждом файле, создаю строковый конструктор со всем текстом, готовым для записи в текстовый файл. Затем я вызываю процедуру (SyncUpdate, inlcluded ниже), отправляю stringbuilder (strBld) из этого потока вместе с именем пути к файлу, который конкретный поток только что написал в stringbuilder о (Xref).

• Процедура включает в себя синхронизацию для хранения всех остальных потоков, пока не найдет поток, передающий правильную информацию. Эта «правильная» информация возникает, когда xref, переданный потоком, соответствует первому элементу в моем списке (FirstListItem). Когда это произойдет, я пишу в текстовый файл, удаляю первый элемент в списке и делаю его снова со следующим потоком.

Способ, которым я пользуюсь монитором, вероятно, невелик, на самом деле у меня мало сомнений в том, что я использую его в наступательной беспристрастной манере. В основном в то время как xref (из потока) <> первый элемент в моем списке, я делаю импульс для монитора. Я изначально использовал monitor.wait, но в конечном итоге просто отказался от попыток отсортировать список, даже если вы используете импульс в другом месте. Возможно, я просто что-то кодировал неловко. В любом случае, я не думаю, что это что-то изменит.

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

Я открыт, чтобы поверить, что я использую неправильный инструмент для работы или просто не использую инструмент, который у меня есть. Я бы предпочел какое-то резьбовое решение (неудивительно, это намного быстрее!). Сегодня я немного бормочу с функциональностью Parallel Task, и многие вещи выглядят многообещающе, но у меня есть еще меньше опыта с этим vs. threadpool, и вы можете видеть, как я злоупотребляю этим! Может быть, что-то с очередью? Вы поняли эту идею. Я бесцельный. Все, что можно было бы предложить, было бы очень оценено. Благодаря! Дайте мне знать, если вам нужна дополнительная информация.

Private Sub SyncUpdateResource(strBld As Object, Xref As String) 
    SyncLock (CType(strBld, StringBuilder)) 
     Dim FirstListitem As String = counter.ListOfFiles.First 
     Do While Xref <> FirstListitem 
      FirstListitem = Counter.ListOfFiles.First 
    'This makes the code much faster for reasons I can only guess at. 
      Thread.Sleep(5) 
      Monitor.PulseAll(CType(strBld, StringBuilder)) 
     Loop 
     Dim strVol As String = Form1.Volname 
     Dim strLFPPath As String = Form1.txtPathDir 
     My.Computer.FileSystem.WriteAllText(strLFPPath & "\" & strVol & ".txt", strBld.ToString, True) 
     Counter.ListOfFiles.Remove(Xref) 
    End SyncLock 
End Sub 

Answer 1

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

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

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

Ваш потребитель вытаскивает вещи из очереди и выводит их. Вы можете увидеть действительно простой пример этого в моей статье Simple Multithreading, Part 2. (Прокрутите вниз до третьего примера, если вас просто интересует код.) Этот пример использует только одного производителя и одного потребителя, но вы можете иметь N производителей, если хотите.

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

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

Я бы использовал кучу, потому что она хорошо работает. .NET Framework не предоставляет кучу, но у меня есть простой, который хорошо подходит для таких заданий. См. A Generic BinaryHeap Class.

Итак, вот как я напишу потребителя (код находится в псевдо-C#, но вы, вероятно, можете легко его преобразовать).

Предполагается, что у вас есть BlockingCollectionsharedQueue, содержащий предметы. Производители помещают предметы в эту очередь. Потребители делают это:

var heap = new BinaryHeap<ItemType>(); 
foreach (var item in sharedQueue.GetConsumingEnumerable()) 
{ 
    if (item.SequenceKey == expectedSequenceKey) 
    { 
     // output this item 
     // then check the heap to see if other items need to be output 
     expectedSequenceKey = expectedSequenceKey + 1; 
     while (heap.Count > 0 && heap.Peek().SequenceKey == expectedSequenceKey) 
     { 
      var heapItem = heap.RemoveRoot(); 
      // output heapItem 
      expectedSequenceKey = expectedSequenceKey + 1; 
     } 
    } 
    else 
    { 
     // item is out of order 
     // put it on the heap 
     heap.Insert(item); 
    } 
} 
// if the heap contains items after everything is processed, 
// then some error occurred. 

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

Если у вас нет бинарной кучи или вы не хотите ее использовать, вы можете сделать то же самое с SortedList<ItemType>. SortedList будет быстрее, чем List, но медленнее, чем BinaryHeap, если количество элементов в списке даже умеренно большое (пара дюжина). Меньше этого, и это, наверное, стирка.

Я знаю, что это много информации. Я рад ответить на любые ваши вопросы.