multithreading - 为什么我在多线程中解析 Google Protocol Buffer 的消息时速度很慢？

Question

我尝试从调用 SerializeToString 生成的二进制文件中解析许多 Google Protocol Buffer 消息。我首先通过调用新函数将所有字节加载到堆内存中。我还有两个数组来存储堆内存中消息的字节开始地址和消息的字节数。
然后我开始通过调用 ParseFromString 来解析消息。我想通过使用多线程来加快过程。
在每个线程中，我传递地址数组和字节计数数组的开始索引和结束索引。

在父进程中。主要代码是:

struct ParsePara
{
    char* str_buffer;
    size_t* buffer_offset;
    size_t* binary_string_length_array;
    size_t start_idx;
    size_t end_idx;
    Flight_Ticket_Info* ticket_info_buffer_array;
};

//Flight_Ticket_Info is class of message
//offset_size is the count of message
ticket_array = new Flight_Ticket_Info[offset_size];
const int max_thread_count = 6;
pthread_t pthread_id_vec[max_thread_count];

CTimer thread_cost;
thread_cost.start();

vector<ParsePara*> para_vec;
const size_t each_count = ceil(float(offset_size) / max_thread_count);
for (size_t k = 0;k < max_thread_count;k++)
{   
    size_t start_idx = each_count * k;
    size_t end_idx = each_count * (k+1);

    if (start_idx >= offset_size)
        break;

    if (end_idx >= offset_size)
        end_idx = offset_size;

    ParsePara* cand_para_ptr = new ParsePara();

    if (!cand_para_ptr)
    {   
        _ERROR_EXIT(0,"[Malloc memory fail.]");
    }   

    cand_para_ptr->str_buffer = m_valdata;//heap memory for storing Bytes of message
    cand_para_ptr->buffer_offset = offset_array;//begin address of each message
    cand_para_ptr->start_idx = start_idx;
    cand_para_ptr->end_idx = end_idx;
    cand_para_ptr->ticket_info_buffer_array = ticket_array;//array to store message
    cand_para_ptr->binary_string_length_array = binary_length_array;//Bytes count of each message

    para_vec.push_back(cand_para_ptr);
}   

for(size_t k = 0 ;k < para_vec.size();k++)
{
    int ret = pthread_create(&pthread_id_vec[k],NULL,parserFlightTicketForMultiThread,para_vec[k]);

    if (0 != ret)
    {
        _ERROR_EXIT(0,"[Error] [create thread fail]");
    }
}

for (size_t k = 0;k < para_vec.size();k++)
{
    pthread_join(pthread_id_vec[k],NULL);
}

在每个线程中，线程函数是:

    void* parserFlightTicketForMultiThread(void* void_para_ptr)
{
    ParsePara* para_ptr = (ParsePara*) void_para_ptr;

    parserFlightTicketForMany(para_ptr->str_buffer,para_ptr->ticket_info_buffer_array,para_ptr->buffer_offset,
            para_ptr->start_idx,para_ptr->end_idx,para_ptr->binary_string_length_array);
}

void parserFlightTicketForMany(const char* str_buffer,Flight_Ticket_Info* ticket_info_buffer_array,
        size_t* buffer_offset,const size_t start_idx,const size_t end_idx,size_t* binary_string_length_array)
{
    printf("start_idx:%d,end_idx:%d\n",start_idx,end_idx);
    for (size_t k = start_idx;k < end_idx;k++)
    {
        if (k % 100000 == 0)
            cout << k << endl;

        size_t cand_offset = buffer_offset[k];
        size_t binary_length = binary_string_length_array[k];
    ticket_info_buffer_array[k].ParseFromString(string(&str_buffer[cand_offset],binary_length-1));
    }
    printf("done %ld %ld\n",start_idx,end_idx);
}

但是多线程成本不止一个线程。
一个线程成本为:40455623ms
我的电脑是8核6线程成本是:131586865ms

任何人都可以帮助我吗？谢谢你!

Answer 1

一些可能的问题 - 您必须尝试确定哪些问题:

Protobuf 解析速度通常受内存带宽而非 CPU 时间的限制，尤其是在输入数据集较大的情况下。在这种情况下，更多线程将无济于事，因为所有内核都在与主内存共享带宽。事实上，让多个核心争夺内存带宽可能会使整体操作变慢。请注意，内存的最大消耗者不是输入字节，而是解析的数据对象——即解析的输出——比编码数据大很多倍。为了改善这个问题，请考虑编写解析循环，以便在解析后立即完全处理每条消息，然后再继续处理文本消息。这样一来，无需分配 k 个 protobuf 对象，您只需为每个线程分配一个 protobuf 对象，并重复使用同一个对象进行解析。这样，对象将(可能)保留在内核的私有(private) L1 缓存中并避免消耗内存带宽；只有输入字节将通过主总线读取。

您如何将数据加载到 RAM 中？你有没有read()成一个大数组还是你mmap() ?在后一种情况下，数据是懒惰地从磁盘读取的——直到你真正尝试解析它才会发生。即使在read()在这种情况下，可能是数据已被换出，从而产生了类似的效果。无论哪种方式，您的线程现在不仅要争夺内存带宽，还要争夺磁盘带宽，这当然要慢得多。让六个线程读取一个大文件的不同部分总体上肯定比让一个线程读取整个文件要慢，因为操作系统针对顺序访问进行了优化。

Protobuf 在解析期间分配内存。许多内存分配器在分配新内存时会锁定。由于你所有的线程都在一个紧密的循环中分配大量的对象，它们将争夺这个锁。确保您使用的是线程友好的内存分配器，例如 Google 的 tcmalloc。请注意，在 parse-consume 循环中重复重用相同的 protobuf 对象，而不是分配大量不同的对象，在这里也会有很大帮助，因为 protobuf 对象会自动为子对象重用内存。

您的代码中可能存在错误，并且在多线程时它可能根本无法达到您的预期。例如，一个错误可能导致所有线程处理相同的数据，而不是不同的数据，并且可能是他们选择的数据恰好更大。确保您正在测试运行单线程与多线程时代码的结果是否完全相同。

简而言之，如果你想要多个内核来让你的代码更快，你不仅要考虑每个内核在做什么，还要考虑每个内核进出什么数据，以及内核之间必须相互通信多少.理想情况下，您希望每个核心都独立运行，而无需与任何人或任何事物交谈；然后你得到最大的并行度。当然，这通常是不可能的，但越接近它越好。

顺便说一句，为您随机优化:

ParseFromString(string(&str_buffer[cand_offset],binary_length-1))

将其替换为:

ParseFromArray(&str_buffer[cand_offset],binary_length-1)

在 std::string 创建复制数据，这会浪费时间(和内存带宽)。 (但这并不能解释为什么线程很慢。)