Windows中的进程间条件变量答案

【问题标题】：inter-process condition variables in WindowsWindows中的进程间条件变量
【发布时间】：2011-08-01 08:15:07
【问题描述】：

我知道我可以使用条件变量来同步线程之间的工作，但是有没有任何类like this (condition variable) 来同步进程之间的工作，提前谢谢

【问题讨论】：

标签： windows winapi windows-7 synchronization

【解决方案1】：

使用一对命名为Semaphore 的对象，一个用于发出信号，一个用作锁。 Windows 上的命名同步对象是自动跨进程的，它会为您处理这部分工作。

这样的课程就可以解决问题。

class InterprocessCondVar {
private: 
    HANDLE  mSem;        // Used to signal waiters
    HANDLE  mLock;       // Semaphore used as inter-process lock 
    int     mWaiters;    // # current waiters 

protected: 

public: 
    InterprocessCondVar(std::string name)
       : mWaiters(0), mLock(NULL), mSem(NULL)
    {
        // NOTE: You'll need a real "security attributes" pointer 
        //   for child processes to see the semaphore! 
        //  "CreateSemaphore" will do nothing but give you the handle if 
        // the semaphore already exists. 
        mSem = CreateSemaphore( NULL, 0, std::numeric_limits<LONG>::max(),  name.c_str()); 

        std::string lockName = name + "_Lock";
        mLock = CreateSemaphore( NULL, 0, 1, lockName.c_str());
        if(!mSem || !mLock) {
            throw std::runtime_exception("Semaphore create failed");
        }
    }
    virtual ~InterprocessCondVar() {
        CloseHandle( mSem);
        CloseHandle( mLock);
    }

    bool Signal();
    bool Broadcast();
    bool Wait(unsigned int waitTimeMs = INFINITE); 
}

一个真正的条件变量提供 3 个调用：
1) "Signal()": 唤醒一个等待线程

bool InterprocessCondVar::Signal() {
    WaitForSingleObject( mLock, INFINITE);           // Lock
    mWaiters--;                                      // Lower wait count
    bool result  = ReleaseSemaphore( mSem, 1, NULL); // Signal 1 waiter 
    ReleaseSemaphore( mLock, 1, NULL);               // Unlock 
    return result;
}

2)“Broadcast()”：唤醒所有线程

 bool InterprocessCondVar::Broadcast() {
    WaitForSingleObject( mLock, INFINITE);                 // Lock 
    bool result = ReleaseSemaphore( mSem, nWaiters, NULL); // Signal all
    mWaiters = 0;                                          // All waiters clear;
    ReleaseSemaphore( mLock, 1, NULL);                     // Unlock 
    return result;       
}

3)“Wait()”：等待信号

bool InterprocessCondVar::Wait(unsigned int waitTimeMs) {
    WaitForSingleObject( mLock, INFINITE);  // Lock
    mWaiters++;                             // Add to wait count
    ReleaseSemaphore( mLock, 1, NULL);      // Unlock 

    // This must be outside the lock
    return (WaitForSingleObject( mSem, waitTimeMs) == WAIT_OBJECT_0);
}

这应该确保 Broadcast() 只唤醒已经在等待的线程和进程，而不是所有未来的。这也是一个非常重量级的对象。对于不需要跨进程存在的 CondVar，我将使用相同的 API 创建一个不同的类，并使用未命名的对象。

【讨论】：

据我所知，您的实现不会跨多个进程工作。您需要将等待线程的计数器放入共享内存中。
另一个注意事项：如果 WaitForSingleObject 由于 WAIT_ABANDONED 或 WAIT_TIMEOUT 而不是 WAIT_OBJECT_0 而返回，您的 mWaiters 变量也会出错。

【解决方案2】：

您可以使用named semaphore 或named mutex。您还可以通过shared memory 在进程之间共享内存。

【讨论】：

【解决方案3】：

对于我正在处理的项目，我需要一个条件变量和互斥锁实现，它可以处理死进程，并且在这种情况下不会导致其他进程陷入死锁。我使用 WIN32 api 提供的本机命名互斥锁实现了互斥锁，因为它们可以通过返回 WAIT_ABANDONED 来指示死进程是否拥有锁。下一个问题是我还需要一个可以跨进程使用的条件变量以及这些互斥锁。我从 user3726672 的建议开始，但很快发现有几个问题，其中计数器变量的状态和信号量的状态最终无效。

在做了一些研究之后，我找到了 Microsoft Research 的一篇论文，它准确地解释了这种情况：Implementing Condition Variables with Semaphores。它为每个线程使用单独的信号量来解决上述问题。

我的最终实现使用了一部分共享内存，我在其中存储了一个线程 ID（等待线程的 ID）的环形缓冲区。然后，这些进程为它们尚未遇到的每个命名信号量/线程 ID 创建自己的句柄并缓存它。然后，信号/广播/等待功能非常简单，并遵循论文中提出的解决方案的想法。如果您的等待操作失败或导致超时，请记住从环形缓冲区中删除您的线程 ID。

对于 Win32 实现，我建议阅读以下文档： Semaphore Objects 和 Using Mutex Objects 描述了实现所需的功能。

替代方案：boost::interprocess 具有一些强大的互斥体仿真支持，但它基于自旋锁并在我们的嵌入式系统上造成非常高的 cpu 负载，这是我们研究自己的实现的最后一个原因。

@user3726672：您能否更新您的帖子以指向此帖子或引用的论文？

最好的问候，迈克尔

更新： 我还查看了 linux/posix 的实现。原来 pthread 已经提供了你需要的一切。只需将 pthread_cond_t 和 pthread_mutex_t 放入某个共享内存中，以便与其他进程共享它，并使用 PTHREAD_PROCESS_SHARED 对其进行初始化。同时在互斥体上设置 PTHREAD_MUTEX_ROBUST。

【讨论】：

【解决方案4】：

是的。您可以为此使用（命名的）互斥锁。使用CreateMutex 创建一个。然后等待它（使用 WaitForSingleObject 之类的函数），并在完成 ReleaseMutex 后释放它。

【讨论】：

【解决方案5】：

作为参考，Boost.Interprocess (documentation for version 1.59) 有条件变量等等。但请注意，在撰写本文时，“Win32 synchronization is too basic”。

【讨论】：