Ожидание win32 темы

Question

У меня есть полностью поточно-ориентированная структура FIFO (TaskList) для хранения классов задач, нескольких потоков, некоторые из которых создают и хранят задачи, а другие обрабатывают задачи. В классе TaskList есть метод pop_front(), который возвращает первое задание, если оно есть хотя бы. В противном случае возвращается NULL.
Вот пример функции обработки:

TaskList tlist;

unsigned _stdcall ThreadFunction(void * qwe)
{
    Task * task;
    while(!WorkIsOver) // a global bool to end all threads.
    {
        while(task = tlist.pop_front())
        {
            // process Task
        }
    }
    return 0;
}

Моя проблема заключается в том, что иногда в списке задач нет новой задачи, поэтому потоки обработки входят в бесконечный цикл (while(!WorkIsOver)), и нагрузка на процессор увеличивается. Каким-то образом я должен заставить потоки ждать, пока в списке не будет сохранено новое задание. Я думаю о приостановке и возобновлении, но затем мне нужна дополнительная информация о том, какие потоки приостановлены или работают, что усложняет кодирование.

Есть идеи?

PS. Я использую Winapi, а не Boost или TBB для многопоточности. Потому что иногда мне приходится прерывать потоки, которые обрабатываются слишком долго, и немедленно создавать новые. Это очень важно для меня. Пожалуйста, не предлагайте ни одного из этих двух.

Спасибо

Answer 1

Предполагая, что вы разрабатываете это в DevStudio, вы можете получить нужный элемент управления, используя [IO Completion Ports]. Страшное имя, для простого инструмента.

• Сначала создайте порт IOCompletion: CreateIOCompletionPort
• Создайте свой пул рабочих потоков, используя _beginthreadex / CreateThread
• В каждом рабочем потоке реализовать цикл, который вызывает GetQueuedCompletionStatus - Возвращенный ключ lpCompletionKey будет указывать на рабочий элемент для обработки.
• Теперь, когда вы получаете рабочий элемент для обработки: вызывайте PostQueuedCompletionStatus из любого потока - передавая указатель на ваш рабочий элемент в качестве параметра ключа завершения.

Вот и все. 3 вызова API, и вы реализовали механизм пула потоков на основе реализованного в ядре объекта очереди. Каждый вызов PostQueuedCompletionStatus будет автоматически десериализован в поток пула потоков, который блокирует GetQueuedCompletionStatus. Пул рабочих потоков создается и поддерживается вами, поэтому вы можете вызывать TerminateThread для любых рабочих потоков, которые занимают слишком много времени. Еще лучше - в зависимости от того, как оно настроено, ядро ​​будет пробуждать столько потоков, сколько необходимо, чтобы гарантировать, что каждое ядро ​​процессора работает с нагрузкой ~ 100%.

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

Answer 2

Если вы еще не прочитали его, вам следует съесть серию Effective Concurrency Herb Sutter, которая охватывает эту тему и многие другие.

Answer 3

1. Используйте семафор в своей очереди, чтобы указать, есть ли элементы, готовые для обработки.
2. Каждый раз, когда вы добавляете элемент, вызывайте ::ReleaseSemaphore, чтобы увеличить счетчик, связанный с семафором * 1005.*
3. В цикле процесса вашего потока, вызовите ::WaitForSingleObject() для дескриптора вашего объекта семафора - вы можете дать этому ожиданию тайм-аут, чтобы у вас была возможность узнать, что ваш поток должен выйти.В противном случае ваш поток будет разбужен всякий раз, когда есть один или несколько элементов для его обработки, а также имеет приятный побочный эффект уменьшения числа семафоров для вас.

Answer 4

Используйте условные переменные для реализации очереди производителя / потребителя - пример кода здесь .

Если вам требуется поддержка более ранних версий Windows, вы можете использовать условную переменную в Boost.Или вы можете создать свой собственный, скопировав специфичный для Windows код из заголовков Boost, они используют те же API-интерфейсы Win32 под обложками, что и при создании собственного.

Answer 5

Если в TaskList есть какой-то метод wait_until_not_empty, используйте его.Если это не так, то один Sleep (1000) (или какое-либо другое значение) может просто сделать трюк.Правильным решением было бы создать оболочку вокруг TaskList, которая использует дескриптор события auto-reset , чтобы указать, не является ли список пустым.Вам нужно будет заново изобрести текущие методы для pop / push, с новым списком задач, являющимся членом нового класса:

WaitableTaskList::WaitableTaskList()
{
  // task list is empty upon creation
  non_empty_event = CreateEvent(NULL, FALSE, FALSE, NULL);
}

Task* WaitableTaskList::wait_and_pop_front(DWORD timeout)
{
  WaitForSingleObject(non_empty_event, timeout);
  // .. handle error, return NULL on timeout

  Task* result = task_list.pop_front();

  if (!task_list.empty())
    SetEvent(non_empty_event);

  return result;
}

void WaitableTaskList::push_back(Task* item)
{
  task_list.push_back(item);
  SetEvent(non_empty_event);
}

Вы должны добавлять элементы в список задач только с помощью таких методов, как этот wait_and_pop_front().

РЕДАКТИРОВАТЬ: на самом деле это не хорошее решение.Существует способ поднять non_empty_event, даже если список пуст.Ситуация требует 2 потоков, пытающихся выскочить и перечислить 2 элемента.Если список становится пустым между if и SetEvent, у нас будет неправильное состояние.Очевидно, что нам также необходимо реализовать синхронизацию.В этот момент я бы снова пересмотрел простой сон: -)

Answer 6

1. Вы можете использовать пул потоков Windows!
2. Или вы можете использовать API-вызов WaitForSingleObject или WaitForMultipleObjects.
3. Используйте как минимум API-вызов SwitchToThread, когда поток не работает.*

Answer 7

Почему бы просто не использовать существующий пул потоков?Пусть Windows справится со всем этим.