允许多个线程一次读取给定条件变量，但只有一个线程写入

Posted 2023-02-16

【中文标题】允许多个线程一次读取给定条件变量，但只有一个线程写入

【英文标题】：Allowing multiple threads to read at once given a condition variable, but only one thread writing

【发布时间】：2021-10-12 01:07:30

【问题描述】：

我有一个情况，我有多个线程从仅写入线程 A 的映射中读取。问题是从映射中读取的多个线程都在映射中寻找唯一值以继续，a身份证。

“线程A”

注意：Payload 是一个包含一些信息的简单结构

std::map<unsigned int, int> jobStatus;
std::mutex mutexJobStatus;
std::condition_variable cv;

...
void receiveJobs() 
    while(true) 
        ...
        Payload payload = ...; //receive payload 
        std::lock_guard<std::mutex> lg(mutexJobStatus);
        jobStatus.insert( payload.jobid, payload.value );
        cv.notify_all(); //notify all waiting threads that they can read
    

...

同时在客户端多线程中，线程正在等待

多线程客户端

unsigned int jobId = ...; // item in map this thread is looking for

auto jobTaken = jobStatus.find(jobId);

    std::unique_lock<std::mutex> ul(mutexJobStatus);
    //wait for item to become available in the map
    sced->cv.wait(ul, [jobId, &jobTaken]  jobTaken = jobStatus.find(jobId); return jobTaken != jobStatus.end(); ); 


... 
//continue

当有很多线程读取时，这段代码执行得非常慢。我认为这可能是因为每次读取时，它都会锁定互斥锁，导致过度读取的线程暂停 - 实际上应该允许同时读取多个线程。

我对 C++ 中的多线程相当陌生，我不确定如何解决这个问题。我是否使用了正确的互斥锁/锁/condition_variables？

我将不胜感激有关实现这种并发读取但阻塞写入的最佳方法的任何建议，以使此代码更好地执行。 谢谢

【问题讨论】：

你能不能展示一个minimal reproducible example，而不是孤立的代码片段，每个人都可以剪切/粘贴，完全如图所示，然后得到同样的次优结果？

如果你显示实际代码，那么jobStatus.find(jobId); 是不受保护的，你有 UB，你的代码有多快或多慢并不重要。

无论如何你的方法是有问题的。如果您想获得良好的性能，请使用线程池，其中池中的每个线程都是通用的，即它们不等待特定的作业，而是执行队列中的任何作业。

您是否考虑过将 jobId 设为原子并将其用作条件变量的等待变量？然后做cv.wait(ul, [] return JobId == myId; )，让作者在cv.notify_all()之前做JobId = value

这样，只有应该响应该作业的线程才需要在等待后获取互斥锁。等待时无需搜索地图。我认为在 cv.wait 中搜索地图不好。

【参考方案1】：

因为

unsigned int jobId = ...; // item in map this thread is looking for
std::unique_lock<std::mutex> ul(mutexJobStatus);
auto jobTaken = jobStatus.find(jobId);
//wait for item to become available in the map
sced->cv.wait(ul, [jobId, &jobTaken]  jobTaken = jobStatus.find(jobId); return jobTaken != jobStatus.end(); );
...

有 ... 应该是范围的结尾。

【讨论】：

感谢您的回复。我已经尝试为锁定机制赋予它自己的作用域（我的错我应该在问题中提到这一点 - 现在已经更新它以使用它自己的作用域），这对性能不佳没有影响。

你根本不需要这些东西return jobTaken != jobStatus.end(); 删除所有这些看看。只需在cond_var.wait() 中执行[]return b; 即可。

线程只有在地图中有正确的列表时才能执行它的工作。因此有必要等到它可用 - 所以我无法删除它。

所以不可用哈哈

我认为问题在于，在等待它可用时，它会阻止其他线程读取地图，这会增加延迟。然而实际上，列表变得非常快，并且没有互斥锁，它可以非常快地找到它 - 但是需要互斥锁，否则它不是线程安全的，以防它同时被写入。