我正在使用 C++ 开发“Simple Segregated Storage”内存池的无锁版本。

SSS 内存池类似于 slab 分配器:它基本上只是一 block 内存，被分成大小相等的 block ，我们有一个指向第一个可用 block 的空闲列表指针。分配只是将指针向上移动到下一个 block ，释放只是将空闲列表指针设置为释放 block ，并将释放 block 上的“下一个”指针指向自由列表指针的旧值。

所以它基本上是一个单链表。

现在，我正在尝试编写简单隔离存储算法的无锁版本。如果我们假设隔离初始内存块(即创建链表)总是在进入多线程环境之前完成，我们只需要担心分配和释放 block ——在这种情况下问题变得非常类似于无锁单链表，这是一个很好理解的问题。

因此，在我看来，分配和解除分配可以通过使用简单的比较和交换指令以无锁方式轻松完成。

假设我们有以下空闲列表指针:

std::atomic<unsigned char*> m_first

假设我们有一个辅助函数nextof()，它接受一个 block 指针并返回对“下一个指针”的引用。 next 指针实际上嵌入在内存块中 - 因此 nextof 函数如下所示:

unsigned char*& nextof(void* ptr)
{
    return *static_cast<unsigned char**>(ptr);
}

至少 Boost library implements Simple Segregated Storage 是这样的.

无论如何，我对无锁原子 allocate 函数的尝试是:

void* malloc()
{
    unsigned char* first = m_first.load();
    if (!first) return nullptr;

    while (!m_first.compare_exchange_strong(first, nextof(first))) 
    {
        if (!first) break;
    }

    return first;
}

所以 - 这看起来非常简单。我们需要做的就是自动将 m_first 设置为嵌入在 m_first 指向的 block 中的下一个指针。最初，我认为这就像说 m_first.store(nextof(m_first)) 一样简单 - 但不幸的是，我认为这不是原子的，因为我们实际上是在此处进行存储和加载相同的内存位置 - 我们将一个新值存储到 m_first 中，同时加载 m_first 的值以获得它的下一个指针。

所以，看来我们需要进行比较和交换。在上面的代码中，我以原子方式将 m_first 的值加载到局部变量中。然后，在检查 null 之后，我使用 CAS 自动更改 m_first 的值，当且仅当它仍然等于局部变量的值时。如果不是，则将局部变量更新为现在的 m_first，并且我们继续循环，直到我们不被其他线程抢占。

解除分配有点棘手 - 这里我们需要将 m_first 的值更改为解除分配的 block ，并且还将解除分配的 block 上的下一个指针更新为指向m_first 曾经是什么。

我的尝试是:

void free(void* chunk)
{
    unsigned char* first = m_first.load();
    nextof(chunk) = first;

    while (!m_first.compare_exchange_strong(first, static_cast<unsigned char*>(chunk)))
    {
        nextof(chunk) = first;
    }
}

我不太确定我有这个权利，但我认为这是正确的。首先，我以原子方式加载 m_first 的值并将其存储在局部变量中。然后，我将 block-to-be-free'd 上的下一个指针设置为我的 m_first 的本地拷贝。当然，到目前为止，另一个线程可能已经介入并将 m_first 设置为其他内容 - 所以我现在必须执行 CAS 操作来检查 m_first 是否仍然是我所期望的.如果不是，我必须将 next 指针设置为适当的值，然后继续循环。

据我所知，这都是竞争条件安全的，对 malloc 和 free 有一个 CAS 操作。

问题:由于无锁算法很难，我不确定我是否正确。我是否在这里忽略了可能导致竞争条件的事情？

最佳答案

一个老答案，但我想我可能会发表意见。除非我弄错了，否则我相信代码可能会通过 ABA problem 的版本泄漏数据。 .

问题是在将参数 nextof(first) 传递给 malloc 内的原子 compare_exchange_strong 之前对其进行评估。如果我们从 malloc 函数中获取 while 循环:

while (!m_first.compare_exchange_strong(first, nextof(first)))
{
    if (!first) return nullptr;
}

并重写它以在传递给 compare_exchange_strong 之前显式评估 nextof(first)，以使其更明显:

while (true)
{
    auto tmp_next = nextof(first);

    // this is where the chance for ABA exists

    if (m_first.compare_exchange_strong(first, tmp_next))
        break;

    if (!first) return nullptr;
}

然后 m_first 似乎有可能更改为不同的值并在这两行之间返回，另一个线程改变实际的 nextof(m_first)同时:

// I will represent the linked list using letters,
// i.e. m_first -> "A" means that "A" is at the head,
// and "A" -> "B" means that "A" points to "B"

void* malloc()
{
    // at the start, m_first points to "A": m_first -> "A" -> "B" -> "C"
    unsigned char* first = m_first.load();

    // --> thread #2 calls malloc() here and acquires "A", and
    //     removes it, so: m_first -> "B" -> "C"

    if (!first) return nullptr;

    while (true)
    {
        unsigned char * tmp_next = nextof(first);
        // tmp_next is "B" at this moment, because 'first' is still "A"            

        // --> some third thread returns some even older object now,
        //     so m_first -> "M" -> "B" -> "C"

        // --> thread #2 now calls free() to return "A", resulting in
        //     my_first -> "A" -> "M" -> "B" -> "C"

        // m_first is now back to "A", but tmp_next points
        // to old "B" instead of "M", and the following line succeeds
        // and "M" is lost forever: m_first -> "B" -> "C"

        if (m_first.compare_exchange_strong(first, tmp_next))
            break;

        if (!first) return nullptr;
    }

    return first;
}

Wikipedia article on ABA包含一些解决这些问题的建议，例如“标记状态引用”(基本上向这些指针添加类似“交易计数器”的东西，以便您可以检测到第二个“A”与第一个不同)，或“中间我发现实现起来稍微复杂一些(显然是由 John D. Valois in this article 发明的)。

关于c++ - 无锁简单隔离存储算法，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/26665659/