Правильный способ реализации ThreadPool.RegisterWaitForSingleObject

Question

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

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

resetEvents = new ManualResetEvent[table_seats]; 
      //Spawn 9 threads 
      for (int i = 0; i < table_seats; i++) 
      { 
       resetEvents[i] = new ManualResetEvent(false); 
       //AutoResetEvent ev = new AutoResetEvent(false); 
       RegisteredWaitHandle handle = ThreadPool.RegisterWaitForSingleObject(autoEvent, ObserveSeat, (object)i, 100, false); 
      } 

      //wait for threads to exit 
      WaitHandle.WaitAll(resetEvents); 

Однако не имеет значения, пользуюсь ли я resetEvents [] или ev, и не работает должным образом. Могу ли я реализовать это или я (возможно) недоразумение, как они должны работать.

Спасибо, Р.

Answer 1

Я бы не использовать RegisterWaitForSingleObject для этой цели. Образцы, которые я собираюсь описать здесь, требуют загрузки Reactive Extensions с тех пор, как вы используете .NET v3.5.

Прежде всего, для того, чтобы дождаться всех рабочих элементов от ThreadPool, чтобы использовать класс CountdownEvent. Это намного более элегантно и масштабируемо, чем использование нескольких экземпляров ManualResetEvent. Кроме того, метод WaitHandle.WaitAll ограничен 64 ручками.

var finished = new CountdownEvent(1); 
for (int i = 0; i < table_seats; i++) 
{ 
    finished.AddCount(); 
    ThreadPool.QueueUserWorkItem(ObserveSeat); 
    (state) => 
    { 
     try 
     { 
     ObserveSeat(state); 
     } 
     finally 
     { 
     finished.Signal(); 
     } 
    }, i); 
} 
finished.Signal(); 
finished.Wait(); 

Во-вторых, вы могли бы попробовать вызвать Thread.Sleep(0) после нескольких итераций цикла, чтобы заставить переключение контекста, так что текущий поток уступает другому. Если вы хотите значительно более сложную стратегию координации, используйте класс Barrier. Добавьте еще один параметр в свою ObserveSeat функцию, которая принимает этот механизм синхронизации. Вы можете предоставить его, захватив его в выражении лямбда в приведенном выше коде.

public void ObserveSeat(object state, Barrier barrier) 
{ 
    barrier.AddParticipant(); 
    try 
    { 
    for (int i = 0; i < NUM_ITERATIONS; i++) 
    { 
     if (i % AMOUNT == 0) 
     { 
     // Let the other threads know we are done with this phase and wait 
     // for them to catch up. 
     barrier.SignalAndWait(); 
     } 
     // Perform your work here. 
    } 
    } 
    finally 
    { 
    barrier.RemoveParticipant(); 
    } 
} 

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