在这 3 种从共享内存读取链表的方法中，为什么第三种最快？

Question

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

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

方法 1：连续对所有链表（称为“通道”）进行循环，查看是否有任何节点已更改为“已填充”：

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));
        }
    }
}

当通道数量较小时，方法 1 给出的延迟非常低。但是当通道数量增加（250K+）时，由于循环遍历所有通道，它变得非常慢。于是我尝试了...

方法2：给每个链表一个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 可能会给出低于 10us 的延迟，而方法 2 会莫名其妙地给出 8ms 的延迟！使用 gettimeofday 看来 update_cursor->state_ 中的更改从服务器视图传播到客户端视图非常慢（我在多核机器上，所以我假设延迟是由于缓存造成的）。所以我尝试了一种混合方法...

方法3：保留更新列表。但连续循环所有通道，并在每次迭代中检查更新列表是否已更新。如果有，请按照推到上面的数字进行操作。如果没有，请检查我们当前迭代到的通道。

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 一样好，但可扩展到大量通道。问题是，我不知道为什么。只是为了打破事情：如果我取消注释“通过更新找到”部分，它会在每个延迟日志消息之间打印。这意味着只能在更新列表中找到东西！所以我不明白这个方法如何比方法 2 更快。

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

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

Update2：尽管有3种方法，但我一次只尝试了1种，因此只有1个服务器和1个客户端在竞争state_成员。

Answer 1

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

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

x86 缓存性能的复杂性并不是我完全熟悉的，但看起来完全合理（至少在理论上），通过让三个不同的线程尽可能地使用 read 来敲击这个内存位置，这似乎是完全合理的（至少在理论上）访问请求大约会在服务器上造成拒绝服务攻击，有时会阻止服务器写入该缓存行几毫秒。

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

您还可以尝试通过在单个核心上运行所有内容来尝试从等式中删除缓存，以便客户端和服务器线程从同一个 L1 缓存中提取内容。

Answer 2

我不知道你是否读过 Herb Sutter 的并发专栏。它们非常有趣，尤其是当您遇到缓存问题时。

实际上，Method2 在这里似乎更好，因为 id 通常小于数据意味着您不必太频繁地往返于主内存（这很费力）。

然而，实际可能发生的情况是，您有这样一行缓存：

Line of cache = [ID1, ID2, ID3, ID4, ...]
                  ^         ^
            client          server

这会产生争用。

以下是 Herb Sutter 的文章：消除虚假共享。基本思想就是人为地增加列表中的 ID，使其完全占据一行缓存。

当您阅读该系列文章时，请查看该系列中的其他文章。也许你会得到一些想法。有一个很好的无锁循环缓冲区，我认为这可以帮助您更新列表:)

Answer 3

我注意到在方法 1 和方法 3 中都有一行 ACQUIRE_MEMORY_BARRIER，我认为这与多线程/竞争条件有关？

不管怎样，方法 2 没有任何睡眠，这意味着下面的代码...

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

将会对处理器造成影响。执行此类生产者/消费者任务的典型方法是使用某种信号量向读者发出更新列表已更改的信号。搜索生产者/消费者多线程应该会给您提供大量示例。这里的主要思想是，这允许线程在等待 update_cursor->state 更改时进入睡眠状态。这可以防止该线程窃取所有 cpu 周期。

Answer 4

答案很难弄清楚，公平地说，我提供的信息很难，但如果有人真正编译了我提供的源代码，他们就有机会;)我说打印了“通过更新列表找到”在每条延迟日志消息之后，但这实际上并不是真的——只有我可以在终端中回滚时才如此。一开始，在没有使用更新列表的情况下发现了大量更新。

问题是，在我在更新列表中设置起点和在每个数据列表中设置起点之间，会存在一些滞后，因为这些操作需要时间。请记住，在整个过程中，列表一直在增长。考虑最简单的情况，我有 2 个数据列表，A 和 B。当我在更新列表中设置起点时，由于列表 A 上有 30 个更新，列表 B 上有 30 个更新，其中恰好有 60 个元素。假设它们已经交替：

A
B
A
B
A // and I start looking at the list here
B

但是当我将更新列表设置到那里之后，B 有大量更新，而 A 没有更新。然后我在每个数据列表中设置了起始位置。我的数据列表起点将在更新激增之后，但更新列表中的起点是在更新激增之前，所以现在我要检查大量更新没有找到他们。上面的混合方法效果最好，因为通过在找不到更新时迭代所有元素，它可以快速缩小更新列表所在位置和数据列表所在位置之间的时间间隙。