线程池的 C++ std::thread 停止条件

Question

我正在编写一个程序，它利用线程池在指定扩展名的文件中搜索与正则表达式匹配的文件。

我的线程池如下所示：

for( int i = 0; i < _nThreads; ++i )
    {
            _threads.push_back( thread( &ThreadPool::GrepFunc, this ) );
    }

运行函数如下所示：

void ThreadPool::GrepFunc()
{
    // implement a barrier

while( !_done )
{
    while( !_tasks.empty() )
    {
        fs::path task;
        bool gotTask = false;
        {
            lock_guard<mutex> tl( _taskMutex );
            if( !_tasks.empty() )
            {
                task = _tasks.front();
                _tasks.pop();
                gotTask = true;
            }
        }

        if( gotTask )
        {
            if( std::tr2::sys::is_directory( task ) )
            {
                for( fs::directory_iterator dirIter( task ), endIter; dirIter != endIter; ++dirIter )
                {
                    if( fs::is_directory( dirIter->path() ) )
                    {
                        { lock_guard<mutex> tl( _taskMutex );
                        _tasks.push( dirIter->path() ); }
                    }
                    else
                    {
                        for( auto& e : _args.extensions() )
                        {
                            if( !dirIter->path().extension().compare( e ) )
                            {
                                SearchFile( dirIter->path() );
                            }
                        }
                    }
                }
            }
            else
            {
                for( auto& e : _args.extensions() )
                {
                    if( !task.extension().compare( e ) )
                    {
                        SearchFile( task );
                    }
                }
            }
        }
    }
}
}

本质上，程序从用户那里接收一个初始目录，并将递归地搜索它和所有子目录以查找与扩展名匹配的文件，以寻找正则表达式匹配。我无法弄清楚如何确定何时达到 _done 的停止情况。我需要确保初始目录中的所有目录和文件都已被扫描，并且在我重新加入线程之前，_tasks 中的所有项目都已完成。任何想法都会非常感激。

Answer 1

我建议使用一个线程（可能是产生文件处理线程的同一线程）专门用于递归文件系统搜索匹配文件；它可以将文件添加到工作队列中，文件搜索线程可以从中获取工作。您可以使用条件变量来协调这一点。

如您所见，协调关机有点棘手。文件系统搜索线程完成搜索后，它可以设置一些对工作线程可见的“刚刚完成队列中的内容”标志，然后通知它们全部唤醒并尝试处理另一个文件：如果他们发现文件/工作队列为空他们退出。然后文件系统搜索线程加入所有工作人员。

Answer 2

关于您在 Tony 回答的评论中更新的问题，我建议有两种任务：一种用于递归探索子目录，另一种用于 grep。您需要 SynQueue<TaskBase>、TaskSubDir: TaskBase 和 TaskGrep: TaskBase。 TaskBase 有一个虚拟接口函数Run()。然后线程就可以从SynQueue重复弹出，调用TaskBase::Run()：

1. 如果它有一个TaskSubDir，那么它会找到 给定路径中的子目录和文件： (a) 如果是文件夹，将新的子目录TaskSubDir添加到SynQueue，这样文件夹就可以用线程池递归搜索； (b) 如果它是一个匹配的文件 扩展，然后它将TaskGrep 推送到SynQueue。
2. 如果得到TaskGrep，则执行SearchFile。
3. 如果队列为空，break 退出工作函数。

这样做，您不需要有 2 个队列并在启动 grep 队列之前等待子目录队列完成。

所以回答你的问题：要确定加入条件，你需要做的就是等待所有线程到break 退出工作函数。

最后说明：代码中的第一个 _tasks.empty() 不受互斥体保护，可能会受到竞速条件的影响。我建议您将 mutex 和 cond_var 隐藏在 SynQueue 类中，并添加 SynQueue::empty() 成员函数（受 mutex 保护）。如果您关心效率，您可能需要考虑 无锁队列 来替换 SynQueue。