Потребитель-производитель с AutoResetEvent

Question

Я пытаюсь использовать шаблон производителя для обработки и сохранения некоторых данных. Я использую AutoResetEvent для сигнализации между двумя therads здесь это код у меня есть

Здесь функция производитель

public Results[] Evaluate() 
    { 
     processingComplete = false; 
     resultQueue.Clear(); 
     for (int i = 0; i < data.Length; ++i) 
      { 
       if (saveThread.ThreadState == ThreadState.Unstarted) 
        saveThread.Start(); 
       //-.... 
       //Process data 
       // 
       lock (lockobject) 
       { 
        resultQueue.Enqueue(result); 
       } 

       signal.Set(); 
      } 
      processingComplete = true; 
     } 

А вот функция потребительского

private void SaveResults() 
    { 
     Model dataAccess = new Model(); 

     while (!processingComplete || resultQueue.Count > 0) 
     { 
      if (resultQueue.Count == 0) 
       signal.WaitOne(); 
      ModelResults result; 
      lock (lockobject) 

      { 
       result = resultQueue.Dequeue(); 
      } 
      dataAccess.Save(result); 
     } 
     SaveCompleteSignal.Set(); 
    } 

Так что мой вопрос иногда resultQueue.Dequeue() выдает исключение InvalidOperation, потому что очередь пуста. Я не уверен, что я делаю неправильно, не должен сигнал signal.WaitOne(), который блокирует очередь пустым?

Answer 1

Вы проверяете счетчик очереди вне синхронизированного контекста. Поскольку Queue не является потокобезопасным, это может быть проблемой (возможно, в то время как Enqueue находится в процессе Count return 1, но ни один элемент не может быть удален), и это будет серьезно ошибочным, если вы все равно будете использовать более одного потребителя.

Возможно, вы захотите прочитать статьи с резьбой, написанные Джозефом Альбахари, у него также есть good sample for your problem, а также a "better" solution without OS synchronization objects.

Answer 2

У вас возникли проблемы с синхронизацией из-за отсутствия надлежащей блокировки.

Вы должны заблокировать весь доступ к очереди, включая проверку счета.

Кроме того, используя Thread.ThreadState таким образом, это «плохая идея». Из документов MSDN для ThreadState:

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

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

Answer 3

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

public Results[] Evaluate() 
{ 
    processingComplete = false; 
    lock(lockobject) 
    { 
     resultQueue.Clear(); 
    } 
    for (int i = 0; i < data.Length; ++i) 
    { 
     if (saveThread.ThreadState == ThreadState.Unstarted) 
      saveThread.Start(); 
     //-.... 
     //Process data 
     // 
     lock (lockobject) 
     { 
      resultQueue.Enqueue(result); 
     } 

     signal.Set(); 
    } 
    processingComplete = true; 
} 

private void SaveResults() 
{ 
    Model dataAccess = new Model(); 

    while (true) 
    { 
     int count; 

     lock(lockobject) 
     { 
      count = resultQueue.Count; 
     } 
     if (count == 0) 
      signal.WaitOne(); 

     lock(lockobject) 
     { 
      count = resultQueue.Count; 
     } 
     // we got a signal, but queue is empty, processing is complete 
     if (count == 0) 
      break; 

     ModelResults result; 
     lock (lockobject) 
     { 
      result = resultQueue.Dequeue(); 
     } 
     dataAccess.Save(result); 
    } 
    SaveCompleteSignal.Set(); 
}