为什么在这种多线程环境中添加延迟会提高数据吞吐量？

Question

在我的应用程序中，我有两个线程，一个生产者（线程 1）和一个消费者（线程 2）。每个线程都有一个输入和输出接口（实际上是一个指向列表的指针），该接口连接到用作路由器的第三个线程。

生产者写入时，调用memcpy将数据复制到缓冲区中，并将缓冲区推入列表中。同时，路由器线程循环搜索与其连接的所有线程并监视它们的接口，以查看是否有任何线程有数据要发送出去。当它看到线程 1 的列表非空时，它会检查以确定数据用于哪个线程。数据被拼接到目标线程（在本例中为线程 2）的输入列表中，此时线程 2 将 malloc 一些内存，将数据 memcpy 到其中并返回指向这个新区域的指针。

对于我的测试，我正在测量吞吐量，以了解发送 100k 条不同大小的消息需要多长时间。线程 1 发送一定大小的数据，线程 2 读取它并发送回一个小的回复消息，线程 1 读取该消息。这将是一次完整的交换。在第一个测试中，在线程 1 中，我发送了所有 100k 条消息，然后读取了 100k 条回复。在第二个测试中，在线程 1 中，我交替发送消息并等待回复并重复 100k 次。在这两个测试中，线程 2 都在循环读取消息并发送回复。我希望测试 1 具有更高的吞吐量，因为线程应该花费更少的时间等待。但是，它的吞吐量明显低于测试 2。我测量了两个测试用例中单个函数调用（读/写）花费的时间，并且它们在测试 1 中总是花费更长的时间（基于均值和中位数并且没有延迟) 尽管这些数字是相同的数量级。

当我在测试 1 中向线程 1 的发送循环中添加一个不执行任何操作的循环时，我发现这种情况下的吞吐量显着提高，而不是没有延迟。我唯一的猜测是添加延迟会减慢生产者的速度，因此消费者可以吸收数据，从而防止其输入列表变得非常大。我想知道是否可能有其他解释，如果有，我该如何测试它们。

编辑

不幸的是，我自己的代码只是我上面描述的测试，它调用了一个实际执行读/写的库，创建第三个线程等。很难用它做一个最小的例子，因为这个库很复杂而不是矿。我提供了一些伪代码来更详细地说明设置。

int NUM_ITERATIONS = 100000;
int msg_reply = 2; // size of the reply message in words
int msg_size = 512; // indicates 512 64 bit words

void generate(int iterations, int size, interface* out){
    std::vector<long long> vec(size);
    for(int i = 0; i < size; i++)
        vec[i] = (long long) i;

    for(int i = 0; i < iterations; i++)
        out->lib_write((char*) vec.data(), size);
}

void receive(int iterations, int size, interface* in){
    for(int i = 0; i < iterations; i++)
        char* data = in->lib_read(size)

void producer(interface* in, interface* out){
    // test 1
    start = std::chrono::high_resolution_clock::now();
    // write data of size msg_size, NUM_ITERATIONS times to out
    generate(NUM_ITERATIONS, msg_size, out);
    // read data of size msg_reply, NUM_ITERATIONS times from in
    receive(NUM_ITERATIONS, msg_reply, in);
    end = std::chrono::high_resolution_clock::now();
    // using NUM_ITERATIONS, msg_size and time, compute and print throughput to stdout
    print_throughput(end-start, "throughput_0", msg_size);

    // test 2
    start = std::chrono::high_resolution_clock::now();
    for(int j = 0; j < NUM_ITERATIONS; j++){
        generate(1, msg_size, out);
        receive(1, msg_reply, in);
    }
    end = std::chrono::high_resolution_clock::now();
    print_throughput(end-start, "throughput_1", msg_size);
}

void consumer(interface* in, interface* out){
    for(int i = 0; i < 2; i++}{
        for(int j = 0; j < NUM_ITERATIONS; j++){
            receive(1, msg_size, in);
            generate(1, msg_reply, out);
        }
    }
}

对lib_write() 和lib_read() 的调用变得相当复杂。为了详细说明上面的描述，数据被 memcpy 存储到缓冲区中，然后移动到列表中。该接口有一个条件变量成员，写入调用它的notify_one() 方法。第三个线程循环遍历它拥有的所有接口指针并检查它们的列表是否非空。如果是这样，则使用 std::list 中的splice() 方法将数据从一个输出列表拼接到目标的输入列表。同时，消费者调用lib_read()，在接口为空时等待条件变量，然后memcpy将数据放入新区域并返回。

// note: these will not compile as is. Undefined variables are class members

char * interface::lib_read(size_t * _size){

    char * ret;
    {
        std::unique_lock<std::mutex> lock(mutex);
        // packets is an std::list containing the incoming data
        while (packets.empty()) {
            cv.wait(lock);
        }
        curr_read_it = packets.begin();
    }

    size_t buff_size = curr_read_it->size;

    ret = (char *)malloc(buff_size);
    memcpy((char *)ret, (char *)curr_read_it->data, buff_size);
    {
        std::unique_lock<std::mutex> lock(mutex);
        packets.erase(curr_read_it);
        curr_read_it = packets.end();
    }
    return ret;
}

void interface::lib_write(char * data, int size){
    // indicates the destination thread id
    long long header = 1;

    // buffer is a just an array that's max packet sized
    memcpy((char *)buffer.data, &header, sizeof(long long));
    memcpy((char *)buffer.data + sizeof(long long), (char *)data, size * sizeof(long long));

    std::lock_guard<std::mutex> guard(mutex);
    packets.push_back(std::move(buffer));
    cv.notify_one();
}

// this is on thread 3
void route(){
    do{
        // this is a vector containing all the "out" interfaces
        for(int i = 0; i < out_ptrs.size(); i++){
            interface <long long> * _out = out_ptrs[i];
            if(!_out->empty()){
                // this just returns the header id (also locks the mutex)
                long long dest= _out->get_dest();
                // looks up the correct interface based on the id and splices
                // a packet into from _out to the appropriate one. Locks mutex
                in_ptrs[dest_map[dest]]->splice(_out);   
            }
        }

    }while(!done());

Answer 1

我一直在寻找有关哪些因素可能会影响多线程性能以及测试什么的一般建议，以便更好地了解正在发生的事情。

我与其他一些人交谈，我得到的有用的建议是确定操作系统调度是否是问题（这是我怀疑的，但不确定如何测试）。本质上，我使用taskset 和sched_affinity() 来强制应用程序在一个内核或一个内核子集上运行，并查看它们之间的比较以及不受限制的情况。

基于限制，我得到了截然不同的结果，并且可以看到一些趋势，所以我很有信心说这是一个操作系统调度问题。不同的可以在不同的工作负载下产生更好的性能。