Синхронизация потоков с boost :: condition_variable

Question

Я делаю некоторые эксперименты по многопоточности на C++, и я понятия не имею, как решить одну проблему. Допустим, у нас есть пул потоков, который обрабатывает запросы пользователей с использованием существующего потока и создает новый поток, когда нет свободного потока. Я создал класс thread_ safe класса, который имеет методы push и pop. поп ждет, пока очередь пуста и возвращается только тогда, когда команда доступна или время ожидания. Теперь пришло время реализовать пул потоков. Идея состоит в том, чтобы освободить потоки сна в течение некоторого времени и убить нить, если после этого нечего делать. Вот реализация

command_queue::handler_t handler; 
while (handler = tasks.pop(timeout)) 
{ 
    handler(); 
} 

Здесь мы выходим из процедуры потока, если произошел тайм-аут. Это хорошо, но есть проблемы с созданием новых потоков. Допустим, у нас уже есть 2 потока запросов пользователя, они работают в данный момент, но нам нужно выполнить некоторую другую операцию асинхронно. Мы называем

thread_pool::start(some_operation); 

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

if (thread_are_free_threads) // ??? 
    condition.notify_one(); 
else 
    create_thread(thread_proc); 

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

if (!condition.notify_one()) // nobody was notified 
    create_thread(thread_proc); 

Насколько я вижу, нет никакого способа, чтобы проверить это.

Спасибо за ваши ответы.

Answer 1

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

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

Answer 2

Более типичный пул потоков будет выглядеть следующим образом:

Pool::Pool() 
{ 
    runningThreads = 0; 
    actualThreads = 0; 
    finished  = false; 
    jobQue.Init(); 

    mutex.Init(); 
    conditionVariable.Init(); 

    for(int loop=0; loop < threadCount; ++loop) { startThread(threadroutine); } 
} 

Pool::threadroutine() 
{ 

    { 
     // Extra code to count threads sp we can add more if required. 
     RAIILocker doLock(mutex); 
     ++ actualThreads; 
     ++ runningThreads; 
    } 
    while(!finished) 
    { 
     Job job; 
     { 
      RAIILocker doLock(mutex); 

      while(jobQue.empty()) 
      { 
       // This is the key. 
       // Here the thread is suspended (using zero resources) 
       // until some other thread calls the notify_one on the 
       // conditionVariable. At this point exactly one thread is release 
       // and it will start executing as soon as it re-acquires the lock 
       // on the mutex. 
       // 
       -- runningThreads; 
       conditionVariable.wait(mutex); 
       ++ runningThreads; 
      } 
      job = jobQue.getJobAndRemoveFromQue(); 
     } 
     job.execute(); 
    } 
    { 
     // Extra code to count threads sp we can add more if required. 
     RAIILocker doLock(mutex); 
     -- actualThreads; 
     -- runningThreads; 
    } 
} 

Pool::AddJob(Job job) 
{ 
    RAIILocker doLock(mutex); 

    // This is where you would check to see if you need more threads. 
    if (runningThreads == actualThreads) // Plus some other conditions. 
    { 
     // increment both counts. When it waits we decrease the running count. 
     startThread(threadroutine); 
    } 
    jobQue.push_back(job); 
    conditionVariable.notify_one(); // This releases one worker thread 
            // from the call to wait() above. 
            // Note: The worker thread will not start 
            //  until this thread releases the mutex. 
} 

Answer 3

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

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

Answer 4

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

void thread_pool::thread_function() 
{ 
    free++; 
    command_queue::handler_t handler; 
    while (handler = tasks.pop(timeout)) 
    { 
     free--; 
     handler(); 
     free++; 
    } 
    free--; 
} 

, когда я задаю задачу нить я что-то вроде этого

if (free == 0) 
    threads.create_thread(boost::bind(&thread_pool::thread_function, this)); 

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

Answer 5

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