如何在 C++ 多线程中匹配处理时间和接收时间

Question

我正在编写一个 c++ 应用程序，其中每 0.5 秒我将接收 4096 字节的数据。这被处理并且输出将被发送到其他一些应用程序。处理每组数据需要将近 2 秒。 我就是这样做的。 在我的主要功能中，我正在接收数据并将其推送到向量中。 我创建了一个线程，它将始终处理第一个元素并在处理后立即将其删除。下面是我的应用程序接收部分的模拟。

#include<iostream>
#include <unistd.h>
#include <vector>
#include <mutex>
#include <pthread.h>
using namespace std;

struct Student{
    int id;
    int age;
};
vector<Student> dustBin;
pthread_mutex_t lock1;
bool isEven=true;

void *processData(void* arg){
    Student st1;
    while(true)
    {
        if(dustBin.size())
        {       
            printf("front: %d\tSize: %d\n",dustBin.front(),dustBin.size());
            st1 = dustBin.front();
            cout << "Currently Processing ID "<<st1.id<<endl;
            sleep(2);
            pthread_mutex_lock(&lock1);
            dustBin.erase(dustBin.begin());
            cout<<"Deleted"<<endl;
            pthread_mutex_unlock(&lock1);
        }
    }
    return NULL;
}
int main()
{
    pthread_t ptid;
    Student st;
    dustBin.clear();
    pthread_mutex_init(&lock1, NULL);
    pthread_create(&ptid, NULL, &processData, NULL);

    while(true)
    {
        for(int i=0; i<4096; i++)
        {
            st.id = i+1;
            st.age = i+2;
            pthread_mutex_lock(&lock1);
            dustBin.push_back(st);
            printf("Pushed: %d\n",st.id);
            pthread_mutex_unlock(&lock1);
            usleep(500000);
        }
    }

    pthread_join(ptid, NULL);
    pthread_mutex_destroy(&lock1);
}

这段代码的输出是 Output

在此处发布的输出图像中，您可以观察到处理的确切顺序。每 4 次插入它只处理一项。

        Note that the reception time of data <<< processing time.

由于这个原因，我的输入缓冲区增长得非常快。还有一件事是，由于主线程和 processData 线程使用互斥锁，它们相互依赖才能释放锁。由于这个原因，我的传入缓冲区有时会被锁定，从而导致数据丢失。请有人向我建议如何处理此问题或建议我一些方法。

感谢和问候

Vamsi

Answer 1

未定义的行为

读取数据时，必须先锁定才能获取size。

忙着等待

您应该始终避免什么都不做的紧密循环。在这里，如果dustBin 为空，您将立即检查它是否永远使用该内核的 100% 并减慢其他所有内容，耗尽笔记本电脑电池并使其比应有的温度更高。 编写这样的代码非常糟糕！

先学习多线程

您应该阅读一两本关于多线程的书。如果不花时间正确地学习它，正确地做多线程是很难的，而且几乎是不可能的。 C++ Concurrency in Action 强烈推荐用于标准 C++ 多线程。

条件变量

通常你会使用条件变量或某种事件来告诉消费者线程何时添加数据，这样它就不必无用地唤醒以检查是否是这种情况。

由于您有一个典型的生产者/消费者，您应该能够找到很多关于如何做到这一点或特殊容器或其他有助于实现代码的构造的信息。

输出

你的printf 和cout 的东西会对性能产生影响，因为有些东西在锁内，而另一些不在，你可能会得到格式不正确的输出。如果你真的需要输出，第三个线程可能是更好的选择。在任何情况下，您都希望最大限度地减少锁定时间，因此格式化到临时缓冲区可能也是一个好主意。

顺便说一句，标准输出相对较慢，很有可能这甚至可能是您无法快速处理所有数据的原因。

处理率

显然，如果您能够每 0.5 秒产生 4096 字节的数据，但需要 2 秒来处理该数据，那么您就有一个严重的问题。

在此处提问之前，您应该真正考虑在这种情况下您想做什么，因为没有这些信息，我们正在猜测可能的解决方案。

这里有一些可能性：

• 减慢生产者的速度。显然，如果您实时获取数据，这将不起作用。
• 优化消费者（更好的算法、更好的硬件、最佳的并行性……）
• 跳过一些数据

显然，对于性能问题，您应该使用分析器来了解您是否浪费了时间。一旦你知道了这一点，你就会更好地了解在哪里检查以改进你的代码。

花 2 秒处理数据确实很慢，但我们无法为您提供帮助，因为我们不知道您的代码在做什么。

例如，如果您将数据添加到数据库中并且无法跟进，您可能希望将多个插入批处理到单个命令中，以减少通过网络与数据库通信的开销。

另一个例子是，如果您将数据附加到文件中，您可能希望在每次写入之前保持文件打开并积累一些数据。

容器

如果您从头部一个一个地删除项目并且它的大小变得有点大（比如超过 100 个小项目），因为每次其他项目都需要移动，那么向量将不是一个好的选择。

除了按照评论中的建议更改容器外，另一种可能性是使用 2 个向量并交换它们。这样，您将能够减少锁定互斥锁的次数并在不需要锁定的情况下处理许多项目。

如何优化

您应该积累足够的数据（例如 30 秒），停止积累，然后使用该数据测试您的处理速度。如果您无法在不到一半的时间（15 秒）内处理该数据，那么您显然需要以一种或另一种方式提高处理速度。您的消费者之一足够快，然后您可以优化从生产者到消费者的通信。

您必须知道您的瓶颈是 I/O、数据库还是什么，以及某些部分是否可以并行完成。

可能有很多优化可以在你没有显示的代码中完成......

Answer 2

如果您不能足够快地处理消息，则必须丢弃一些消息。

使用固定大小的circular buffer。

那么如果提供者比消费者快，那么旧的条目将被覆盖。

如果你不能跳过一些数据并且你不能足够快地处理它，你就注定要失败。

Answer 3

创建两个const 变量，NBUFFERS 和 NTHREADS，如果您有 16 个内核并且处理速度太慢 4 倍，则最初将它们都设为 8。稍后使用这些值。

创建 NBUFFERS 数据缓冲区，每个缓冲区大到足以容纳 4096 个样本，实际上，只需创建一个大缓冲区并在其中进行偏移以将其划分。

启动 NTHREADS。他们每个人都会不断地等待被告知要处理哪个缓冲区，然后他们将处理它并再次等待另一个缓冲区。

在你的主程序中，进入一个循环，接收数据。将前 4096 个样本接收到第一个缓冲区并通知第一个线程。将第二个 4096 个样本接收到第二个缓冲区并通知第二个线程。

buffer = (buffer + 1) % NBUFFERS
thread = (thread + 1) % NTHREADS

冲洗并重复。由于您有 8 个线程，并且数据每 0.5 秒到达一次，因此每个线程只会每 4 秒获取一个新缓冲区，但只需 2 秒即可清除之前的缓冲区。