使用 boost::condition_variable 进行线程同步

Question

我正在做一些关于 C++ 多线程的实验，但我不知道如何解决一个问题。假设我们有线程池，它使用现有线程处理用户请求并在没有可用线程时创建新线程。我创建了 command_queue 线程安全类，它具有 push 和 pop 方法。 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

您需要创建另一个变量（可能是一个信号量），它知道有多少线程正在运行，然后您可以在调用 notify 之前检查它并在需要时创建一个新线程。

另一个更好的选择是在线程超时时不让它们退出。他们应该活着等待通知。而不是在通知超时时退出，而是检查变量以查看程序是否仍在运行或是否“关闭”，如果它仍在运行，则再次开始等待。

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）向玩家发出信号——每个玩家都有一个信号量。在这种情况下，经销商主动销毁空闲线程可能是有意义的，方法是给线程一个kill 我自己 的工作。

Answer 4

现在我找到了一种解决方案，但它并不那么完美。 我有一个名为 free 的 volatile 成员变量——它存储池中的空闲线程数。

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));

同步仍然存在问题，因为如果上下文将在空闲后切换--在thread_function 我们可能会创建一个新线程，我们实际上并不需要，但由于任务队列是线程安全的没问题，这只是不必要的开销。您能提出解决方案吗？您对此有何看法？也许最好让它保持原样，然后在这里再进行一次同步？

Answer 5

另一个想法。您可以查询消息队列的长度。如果时间过长，请创建一个新的工作人员。