共享内存 IPC 同步（无锁）

Question

考虑以下场景：

要求：

• Intel x64 服务器（多个 CPU 插槽 => NUMA）
• Ubuntu 12，GCC 4.6
• 两个进程通过（命名的）共享内存共享大量数据
• 经典的生产者-消费者场景
• 内存排列在一个循环缓冲区中（有 M 个元素）

程序序列（伪代码）：

流程 A（生产者）：

int bufferPos = 0;
while( true )
{
    if( isBufferEmpty( bufferPos ) )
    {
        writeData( bufferPos );
        setBufferFull( bufferPos );

        bufferPos = ( bufferPos + 1 ) % M;
    }
}

流程 B（消费者）：

int bufferPos = 0;
while( true )
{
    if( isBufferFull( bufferPos ) )
    {
        readData( bufferPos );
        setBufferEmpty( bufferPos );

        bufferPos = ( bufferPos + 1 ) % M;
    }
}

现在是一个古老的问题：如何有效地同步它们！？

1. 使用互斥锁保护每个读/写访问
2. 引入“宽限期”，以允许完成写入：读取缓冲区 N 中的数据，此时缓冲区 (N+3) 已被标记为已满（危险，但似乎有效...）
3. ?!?

理想情况下，我想要一些类似于内存屏障的东西，以保证所有先前的读/写在所有 CPU 上都是可见的，类似于：

writeData( i );
MemoryBarrier();

//All data written and visible, set flag
setBufferFull( i );

这样，我只需要监视缓冲区标志，然后就可以安全地读取大数据块。

一般来说，我正在寻找与 Preshing 在此处描述的获取/释放栅栏类似的东西：

http://preshing.com/20130922/acquire-and-release-fences/

（如果我理解正确的话，C++11 原子仅适用于单个进程的线程，而不适用于多个进程。）

但是 GCC 自己的内存屏障（__sync_synchronize 与编译器屏障 asm volatile( "" ::: "memory" ) 可以肯定）似乎没有按预期工作，因为在屏障之后写入变得可见，当我预计它们会完成时。

任何帮助将不胜感激......

顺便说一句：在 Windows 下，使用 volatile 变量（微软特有的行为）可以正常工作...

Answer 1

Boost Interprocess 支持共享内存。

Boost Lockfree 有一个单生产者单消费者队列类型 (spsc_queue)。这基本上就是您所说的循环缓冲区。

这是一个使用此队列以无锁方式传递 IPC 消息（在本例中为 string 类型）的演示。

定义类型

首先，让我们定义我们的类型：

namespace bip = boost::interprocess;
namespace shm
{
    template <typename T>
        using alloc = bip::allocator<T, bip::managed_shared_memory::segment_manager>;

    using char_alloc    =  alloc<char>;
    using shared_string =  bip::basic_string<char, std::char_traits<char>, char_alloc >;
    using string_alloc  =  alloc<shared_string>;

    using ring_buffer = boost::lockfree::spsc_queue<
        shared_string, 
        boost::lockfree::capacity<200> 
        // alternatively, pass
        // boost::lockfree::allocator<string_alloc>
    >;
}

为简单起见，我选择演示运行时大小的 spsc_queue 实现，随机请求 200 个元素的容量。

shared_string typedef 定义了一个字符串，它将透明地从共享内存段分配，因此它们也“神奇地”与其他进程共享。

消费者方面

这是最简单的，所以：

int main()
{
    // create segment and corresponding allocator
    bip::managed_shared_memory segment(bip::open_or_create, "MySharedMemory", 65536);
    shm::string_alloc char_alloc(segment.get_segment_manager());

    shm::ring_buffer *queue = segment.find_or_construct<shm::ring_buffer>("queue")();

这将打开共享内存区域，如果共享队列存在，则定位它。 注意这应该在现实生活中同步。

现在进行实际演示：

while (true)
{
    std::this_thread::sleep_for(std::chrono::milliseconds(10));

    shm::shared_string v(char_alloc);
    if (queue->pop(v))
        std::cout << "Processed: '" << v << "'\n";
}

消费者只是无限地监视队列中的待处理作业，并且每约 10 毫秒处理一个。

生产者方面

生产者端非常相似：

int main()
{
    bip::managed_shared_memory segment(bip::open_or_create, "MySharedMemory", 65536);
    shm::char_alloc char_alloc(segment.get_segment_manager());

    shm::ring_buffer *queue = segment.find_or_construct<shm::ring_buffer>("queue")();

再次，在初始化阶段添加适当的同步。此外，您可能会让生产者在适当的时候负责释放共享内存段。在这个演示中，我只是“让它挂起”。这很适合测试，见下文。

那么，制作人是做什么的呢？

    for (const char* s : { "hello world", "the answer is 42", "where is your towel" })
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(250));
        queue->push({s, char_alloc});
    }
}

是的，生产者产生在大约 750 毫秒内正好 3 条消息，然后退出。

请注意，如果我们这样做（假设一个带有作业控制的 POSIX shell）：

./producer& ./producer& ./producer&
wait

./consumer&

将“立即”打印 3x3 消息，同时让消费者继续运行。正在做

./producer& ./producer& ./producer&

在此之后，将再次实时显示消息“滴入”（以约 250 毫秒的间隔爆发 3 条），因为消费者仍在后台运行

查看完整代码在线查看此要点：https://gist.github.com/sehe/9376856