我有一个“服务器”程序，可以更新共享内存中的许多链表以响应外部事件。我希望客户端程序尽快注意到任何列表的更新(最低延迟)。一旦链表的数据被填充并且其下一个指针已设置为有效位置，服务器会将链表节点的 state_ 标记为 FILLED。在此之前，它的 state_ 是 NOT_FILLED_YET。我正在使用内存屏障来确保在内部数据实际准备好之前，客户端不会将 state_ 视为 FILLED(而且它似乎有效，我从未见过损坏数据)。此外，state_ 是易变的，以确保编译器不会解除客户端对其的检查，使其脱离循环。

保持服务器代码完全相同，我想出了 3 种不同的方法让客户端扫描链接列表以查找更改。问题是:为什么第三种方法最快？

方法 1:连续循环遍历所有链表(称为“ channel ”)，查看是否有任何节点已更改为“已填充”:

void method_one()
{
    std::vector<Data*> channel_cursors;
    for(ChannelList::iterator i = channel_list.begin(); i != channel_list.end(); ++i)
    {
        Data* current_item = static_cast<Data*>(i->get(segment)->tail_.get(segment));
        channel_cursors.push_back(current_item);
    }

    while(true)
    {
        for(std::size_t i = 0; i < channel_list.size(); ++i)
        {   
            Data* current_item = channel_cursors[i];

            ACQUIRE_MEMORY_BARRIER;
            if(current_item->state_ == NOT_FILLED_YET) {
                continue;
            }

            log_latency(current_item->tv_sec_, current_item->tv_usec_);

            channel_cursors[i] = static_cast<Data*>(current_item->next_.get(segment));
        }
    }
}

当 channel 数量较少时，方法 1 的延迟非常低。但是当 channel 数量增加(250K+)时，由于遍历所有 channel ，它变得非常慢。所以我尝试...

方法二:给每个链表一个ID。在旁边保留一个单独的“更新列表”。每次更新其中一个链表时，将其 ID 推送到更新列表。现在我们只需要监控单个更新列表，并检查我们从中获取的 ID。

void method_two()
{
    std::vector<Data*> channel_cursors;
    for(ChannelList::iterator i = channel_list.begin(); i != channel_list.end(); ++i)
    {
        Data* current_item = static_cast<Data*>(i->get(segment)->tail_.get(segment));
        channel_cursors.push_back(current_item);
    }

    UpdateID* update_cursor = static_cast<UpdateID*>(update_channel.tail_.get(segment));

    while(true)
    {   
            ACQUIRE_MEMORY_BARRIER;
        if(update_cursor->state_ == NOT_FILLED_YET) {
            continue;
        }

        ::uint32_t update_id = update_cursor->list_id_;

        Data* current_item = channel_cursors[update_id];

        if(current_item->state_ == NOT_FILLED_YET) {
            std::cerr << "This should never print." << std::endl; // it doesn't
            continue;
        }

        log_latency(current_item->tv_sec_, current_item->tv_usec_);

        channel_cursors[update_id] = static_cast<Data*>(current_item->next_.get(segment));
        update_cursor = static_cast<UpdateID*>(update_cursor->next_.get(segment));
    }   
}

方法 2 产生了可怕的延迟。方法 1 可能会产生不到 10 微秒的延迟，而方法 2 会莫名其妙地经常产生 8 毫秒的延迟!使用 gettimeofday 似乎 update_cursor->state_ 的变化从服务器的 View 传播到客户端的 View 非常缓慢(我在多核机器上，所以我假设延迟是由于缓存造成的)。所以我尝试了一种混合方法...

方法三:保留更新列表。但是连续遍历所有 channel ，并在每次迭代中检查更新列表是否已更新。如果有，请使用推到上面的数字。如果没有，请检查我们当前迭代到的 channel 。

void method_three()
{
    std::vector<Data*> channel_cursors;
    for(ChannelList::iterator i = channel_list.begin(); i != channel_list.end(); ++i)
    {
        Data* current_item = static_cast<Data*>(i->get(segment)->tail_.get(segment));
        channel_cursors.push_back(current_item);
    }

    UpdateID* update_cursor = static_cast<UpdateID*>(update_channel.tail_.get(segment));

    while(true)
    {
        for(std::size_t i = 0; i < channel_list.size(); ++i)
        {
            std::size_t idx = i;

            ACQUIRE_MEMORY_BARRIER;
            if(update_cursor->state_ != NOT_FILLED_YET) {
                //std::cerr << "Found via update" << std::endl;
                i--;
                idx = update_cursor->list_id_;
                update_cursor = static_cast<UpdateID*>(update_cursor->next_.get(segment));
            }

            Data* current_item = channel_cursors[idx];

            ACQUIRE_MEMORY_BARRIER;
            if(current_item->state_ == NOT_FILLED_YET) {
                continue;
            }

            found_an_update = true;

            log_latency(current_item->tv_sec_, current_item->tv_usec_);
            channel_cursors[idx] = static_cast<Data*>(current_item->next_.get(segment));
        }
    }
}

此方法的延迟与方法 1 一样好，但可扩展到大量 channel 。问题是，我不知道为什么。只是为了解决问题:如果我取消注释“通过更新找到”部分，它会在每个延迟日志消息之间打印。这意味着只能在更新列表中找到东西!所以我不明白这种方法怎么会比方法2快。

生成随机字符串作为测试数据的完整可编译代码(需要 GCC 和 boost-1.41)位于:http://pastebin.com/0kuzm3Uf

更新:所有 3 种方法都有效地自旋锁定，直到发生更新。不同之处在于他们需要多长时间才能注意到更新已经发生。它们都不断地对处理器征税，所以这并不能解释速度差异。我正在一台 4 核机器上测试，没有其他任何东西在运行，所以服务器和客户端没有什么可以竞争的。我什至制作了一个代码版本，其中更新发出条件信号并让客户端等待该条件——这对任何方法的延迟都没有帮助。

更新2:尽管有3种方法，但我一次只试过1种，所以只有1台服务器和1台客户端在竞争state_成员。

最佳答案

假设:方法 2 以某种方式阻止服务器写入更新。

除了处理器核心本身之外，您可以锤击的其中一件事就是连贯缓存。当您读取给定核心上的值时，该核心上的 L1 缓存必须获得对该缓存行的读取访问权限，这意味着它需要使任何其他缓存对该行的写入访问无效。反之亦然写入一个值。因此，这意味着您不断地在“写入”状态(在服务器核心的缓存中)和“读取”状态(在所有客户端核心的缓存中)之间来回切换缓存行。

x86 缓存性能的复杂性并不是我完全熟悉的东西，但看起来完全合理(至少在理论上)你正在做的事情是让三个不同的线程尽可能多地敲打这个内存位置读取访问请求大约会在服务器上造成拒绝服务攻击，有时会阻止服务器写入该缓存行几毫秒。

您可以做一个实验来检测这一点，方法是查看服务器实际将值写入更新列表所花费的时间，并查看那里是否存在与延迟相对应的延迟。

您也可以尝试从等式中移除缓存的实验，方法是在单个内核上运行所有内容，以便客户端和服务器线程从同一个 L1 缓存中提取内容。

关于c++ - 在这 3 种从共享内存读取链表的方法中，为什么第三快？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/2531572/