如何在工作在同一共享内存区域的两个进程之间共享锁？

Question

我想知道如何做到这一点（使用 C++98）。这是我的场景：我有进程 A 和进程 B。进程 A 在共享内存中分配一个大缓冲区并将其拆分为固定数量的块。然后它使用一系列类似这样的结构来表示每个块：

struct Chunk
{
    Lock lock; //wrapper for pthread_attr_t and pthread_mutex_t
    char* offset; //address of the beginning of this chunk in the shared memory buffer
};

构造时的锁是这样做的：

pthread_mutexattr_init(&attrs);
pthread_mutexattr_setpshared(&attrs, PTHREAD_PROCESS_SHARED);
pthread_mutex_init(&lock,&attrs); //lock is pthread_mutex_t and attrs is pthread_mutexattr_t

调用时的 lock 方法是这样的：

pthread_mutex_lock(&lock);

在将上述“Chunk”创建到共享内存缓冲区的开头时，它使用placement new，如下所示：

char* mem; //pointer to the shared memory
Chunks[i] = new (mem) Chunk; //for i = {1..num chunks}
mem += sizeof(Chunk);

然后它分配偏移量并在其生命周期内继续写入缓冲区的其余部分。每次写入与上述之一相对应的块时，它都会抓住块锁并在完成时释放它。

现在进程 B 启动并将相同的共享内存缓冲区映射到内存中，并尝试像这样检索块：

Chunk** chunks = reinterpret_cast<Chunk**)(mem); //mem being the pointer into the shared memory

然后它通过扫描不同的块来尝试对共享内存进行操作，并在需要时尝试使用锁。

当我运行它时遇到奇怪的崩溃，其中块**是垃圾，我想知道 Lock 是否也可以跨进程工作，或者在上述简单步骤中我是否忽略了任何其他警告？ SHARED pthread attr 是否足够，还是我需要使用完全不同的锁？

Answer 1

当您将共享内存区域拉入进程时，它通常不会位于与访问共享内存的其他进程相同的虚拟地址。因此，您不能只将原始指针存储到共享内存中并期望它们有意义地工作。

因此，在您的情况下，即使 Chunk 在共享内存中，每个块中的 offset 指针在任何其他进程中都没有意义。

一种解决方案是使用从共享内存块开始的偏移量

struct Chunk {
    pthread_mutex_t  lock;
    size_t           offset;
};

char *base; // base address of shared memory
char *mem;  // end of in-use shared memory

Chunk *chunks = reinterpret_cast<Chunk *>(mem);  // pointer to array in shared memory
for (int i = 0; i < num_chunks; i++) {
    // initialize the chunks
    new(mem) Chunk;
    mem += sizeof(Chunk); }
// set the offsets to point at some memory
for (int i = 0; i < num_chunks; i++) {
    chunks[i].offset = mem - base;
    mem += CHUNK_SIZE; // how much memory to allocate for each chunk?
}

现在在 B 中你可以这样做

Chunk *chunks = reinterpret_cast<Chunk *>(base);

但在任一进程中，要访问块的数据，您需要base + chunks[i].offset

或者，您可以使用package that manages the shared memory allocation for you and ensures that it gets mapped at the same address in every process。

Answer 2

除了我的 cmets，我还提供了一个非常基本的 offset_ptr 实现。我同意让一个简单的项目依赖于诸如 boost 之类的东西可能是一种矫枉过正的做法，但即使在达到一定的项目规模并决定切换到一组更严肃的库之前，也值得包装一些关键的东西，比如偏移指针。包装可以帮助您集中您的偏移处理程序代码，并使用断言保护自己。一个不能处理所有情况的简单 offset_ptr 模板仍然比在任何地方复制粘贴的手工编码的偏移量+指针操纵器代码要好得多，它的基本实现是：

template <typename T, typename OffsetInt=ptrdiff_t>
class offset_ptr
{
    template <typename U, typename OI> friend class offset_ptr;
public:
    offset_ptr() : m_Offset(0) {}
    offset_ptr(T* p)
    {
        set_ptr(p);
    }
    offset_ptr(offset_ptr& other)
    {
        set_ptr(other.get_ptr());
    }
    template <typename U, typename OI>
    offset_ptr(offset_ptr<U,OI>& other)
    {
        set_ptr(static_cast<T*>(other.get_ptr()));
    }
    offset_ptr& operator=(T* p)
    {
        set_ptr(p);
        return *this;
    }
    offset_ptr& operator=(offset_ptr& other)
    {
        set_ptr(other.get_ptr());
        return *this;
    }
    template <typename U, typename OI>
    offset_ptr& operator=(offset_ptr<U,OI>& other)
    {
        set_ptr(static_cast<T*>(other.get_ptr()));
        return *this;
    }
    T* operator->()
    {
        assert(m_Offset);
        return get_ptr();
    }
    const T* operator->() const
    {
        assert(m_Offset);
        return get_ptr();
    }
    T& operator*()
    {
        assert(m_Offset);
        return *get_ptr();
    }
    const T& operator*() const
    {
        assert(m_Offset);
        return *get_ptr();
    }
    operator T* ()
    {
        return get_ptr();
    }
    operator const T* () const
    {
        return get_ptr();
    }

private:
    void set_ptr(const T* p)
    {
        m_Offset = p ? OffsetInt((char*)p - (char*)this) : OffsetInt(0);
    }
    T* get_ptr() const
    {
        return m_Offset ? (T*)((char*)this + m_Offset) : (T*)nullptr;
    }

private:
    OffsetInt m_Offset;
};

offset_ptr<int> p;
int x = 5;

struct TestStruct
{
    int member;
    void func()
    {
        printf("%s(%d)\n", __FUNCTION__, member);
    }
};

TestStruct ts;
offset_ptr<TestStruct> pts;

int main()
{
    p = &x;
    *p = 6;
    printf("%d\n", x);

    ts.member = 11;
    if (!pts)
        printf("pts is null\n");
    pts = &ts;
    if (pts)
        pts->func();
    pts = nullptr;
    if (!pts)
        printf("pts is null again\n");

    // this will cause an assert because pts is null
    pts->func();
    return 0;
}

它可能包含一些操作符函数，如果你不习惯实现指针的东西，写起来会很痛苦，但与像 boost 这样的复杂库的完整指针实现相比，这真的很简单，不是吗？它使指针（偏移）操纵器代码更好！没有理由不使用至少这样的包装器！

即使您自己在没有外部库的情况下自行包装锁和其他东西也很有成效，因为在使用裸本机 api 编写/使用锁 2-3-4 次之后，您的代码看起来会更好看如果您使用带有 ctor/destructor/lock()/unlock() 的简单包装器，更不用说可以用断言保护的集中式锁定/解锁代码，有时您可以将调试信息放入锁定类（如 id更容易调试死锁的最后一个储物柜线程...）。

因此，即使您在没有 std 或 boost 的情况下滚动代码，也不要省略包装器。