2

由于windows中缺少条件变量（虽然从vis​​ta开始引入，但windows XP和2003不支持），所以在c++中实现线程安全队列并不是很容易。在 Win32 上实现 POSIX 条件变量的策略。我需要的是只使用 CriticalSection 或 Mutex 和 Event 而不使用信号量和条件变量。

我还试图找到一个仅使用 win32 本机 API 的确切实现，但没有运气。所以我自己完成了一个。问题是我不是 100% 确定代码是线程安全的。谁能告诉我好不好？

class CEventSyncQueue
{
public:
    CEventSyncQueue(int nCapacity = -1);
    virtual ~CEventSyncQueue();
    virtual void Put(void* ptr);
    virtual void* Get();
protected:
    int m_nCapacity;
    CPtrList m_list;

    CRITICAL_SECTION m_lock;    
    HANDLE m_hGetEvent;
    HANDLE m_hPutEvent;
};

CEventSyncQueue::CEventSyncQueue(int nCapacity)
{
    m_nCapacity = nCapacity;

    ::InitializeCriticalSection(&m_lock);
    m_hPutEvent = ::CreateEvent(NULL, FALSE, FALSE, NULL);
    m_hGetEvent = ::CreateEvent(NULL, FALSE, FALSE, NULL);
}

CEventSyncQueue::~CEventSyncQueue()
{
    m_list.RemoveAll();

    ::CloseHandle(m_hGetEvent);
    ::CloseHandle(m_hPutEvent);

    ::DeleteCriticalSection(&m_lock);
}

void CEventSyncQueue::Put(void* ptr)
{
    ::EnterCriticalSection(&m_lock);

    while(m_nCapacity > 0 && m_list.GetCount() >= m_nCapacity)
    {
        ::LeaveCriticalSection(&m_lock);

        //wait
        if(::WaitForSingleObject(m_hPutEvent, INFINITE) != WAIT_OBJECT_0)
        {
            ASSERT(FALSE);
        }

        ::EnterCriticalSection(&m_lock);
    }
    if(m_nCapacity > 0)
    {
        ASSERT(m_list.GetCount() < m_nCapacity);
    }
    m_list.AddTail(ptr);

    ::SetEvent(m_hGetEvent);    //notifyAll
    ::LeaveCriticalSection(&m_lock);
}
void* CEventSyncQueue::Get()
{
    ::EnterCriticalSection(&m_lock);

    while(m_list.IsEmpty())
    {
        ::LeaveCriticalSection(&m_lock);

        //wait
        if(::WaitForSingleObject(m_hGetEvent, INFINITE) != WAIT_OBJECT_0)
        {
            ASSERT(FALSE);
        }

        ::EnterCriticalSection(&m_lock);
    }
    ASSERT(!m_list.IsEmpty());
    void* ptr = m_list.RemoveHead();

    ::SetEvent(m_hPutEvent);    //notifyAll
    ::LeaveCriticalSection(&m_lock);

    return ptr;
}
4

3 回答 3

1

在 Windows 中实现线程安全队列很简单。我已经在 Delphi、C++、BCB 等中完成了。

为什么你认为条件变量是必需的？您认为 Windows 消息队列如何工作？

事件是用于 PC 队列的错误原语。最简单/最清晰的方法是使用信号量。

简单的无界生产者-消费者队列。

template <typename T> class PCSqueue{
    CRITICAL_SECTION access;
    deque<T> *objectQueue;
    HANDLE queueSema;
public:
    PCSqueue(){
        objectQueue=new deque<T>;
        InitializeCriticalSection(&access);
        queueSema=CreateSemaphore(NULL,0,MAXINT,NULL);
    };
    void push(T ref){
        EnterCriticalSection(&access);
        objectQueue->push_front(ref);
        LeaveCriticalSection(&access);
        ReleaseSemaphore(queueSema,1,NULL);
    };
    bool pop(T *ref,DWORD timeout){
        if (WAIT_OBJECT_0==WaitForSingleObject(queueSema,timeout)) {
            EnterCriticalSection(&access);
            *ref=objectQueue->back();
            objectQueue->pop_back();
            LeaveCriticalSection(&access);
            return(true);
        }
        else
            return(false);
    };
};

编辑 - 有界队列不会更困难 - 你需要另一个信号来计算空白。我不使用有界队列，但我确信它会没问题 - 一个有 2 个信号量和一个互斥体/CS 的有界队列是标准模式。

编辑：使用 PostMessage() 或 PostThreadMessage() API 调用 - 它们被明确声明为不受“waveOutProc”回调的影响。MSDN 说调用“其他波函数”会导致死锁 - 信号量调用不在该集合中，如果允许 SetEvent() 但 ReleaseSemaphore() 不允许，我确实会感到非常惊讶。事实上，如果允许 SetEvent() 而 ReleaseSemaphore() 在 Windows 中不是 ANYWHERE，我会感到惊讶。

于 2012-07-29T06:20:43.357 回答
0

条件变量？您是指互锁*功能吗？这些已经存在很长时间了 - 我在 Windows 2000 中使用它们。您可以使用它们来构建并发系统，但您仍然需要自己做一些工作。

或者，尝试 OpenMP。要使用它，您需要 Visual Studio 2008 或更高版本。

于 2012-07-30T00:55:01.707 回答
0

再想一想，几乎没有必要显式地实现信号量。相反，只需考虑如何使用事件实现信号量，并以这种方式解决问题。我的第一次尝试使用了手动重置事件，虽然效率低下但显然是正确的，然后我进行了优化。

请注意，我没有调试（甚至编译！）这些代码片段，但它们应该给你正确的想法。这是手动重置版本：

class CEventSyncQueue
{
public:
    CEventSyncQueue(int nCapacity = -1);
    virtual ~CEventSyncQueue();
    virtual void Put(void* ptr);
    virtual void* Get();
protected:
    int m_nCapacity;
    CPtrList m_list;

    CRITICAL_SECTION m_lock;    
    HANDLE m_queue_not_empty;
    HANDLE m_queue_not_full;
};

CEventSyncQueue::CEventSyncQueue(int nCapacity)
{
    m_nCapacity = nCapacity;
    ::InitializeCriticalSection(&m_lock);
    m_queue_not_empty = ::CreateEvent(NULL, TRUE, FALSE, NULL);
    m_queue_not_full = ::CreateEvent(NULL, TRUE, TRUE, NULL);
}

CEventSyncQueue::~CEventSyncQueue()
{
    m_list.RemoveAll();
    ::CloseHandle(m_queue_not_empty);
    ::CloseHandle(m_queue_not_full);
    ::DeleteCriticalSection(&m_lock);
}

void CEventSyncQueue::Put(void* ptr)
{
    bool done = false;
    while (!done)
    {
        // If the queue is full, we must wait until it isn't.
        if(::WaitForSingleObject(m_queue_not_full, INFINITE) != WAIT_OBJECT_0)
        {
            ASSERT(FALSE);
        }

        // However, we might not be the first to respond to the event,
        // so we still need to check whether the queue is full and loop
        // if it is.
        ::EnterCriticalSection(&m_lock);
        if (m_nCapacity <= 0 || m_list.GetCount() < m_nCapacity)
        {
            m_list.AddTail(ptr);
            done = true;
            // The queue is definitely not empty.
            SetEvent(m_queue_not_empty);
            // Check whether the queue is now full.
            if (m_nCapacity > 0 && m_list.GetCount() >= m_nCapacity)
            {
                ResetEvent(m_queue_not_full);
            }
        }
        ::LeaveCriticalSection(&m_lock);
    }
}

void* CEventSyncQueue::Get()
{
    void *result = nullptr;
    while (result == nullptr)
    {
        // If the queue is empty, we must wait until it isn't.
        if(::WaitForSingleObject(m_queue_not_empty, INFINITE) != WAIT_OBJECT_0)
        {
            ASSERT(FALSE);
        }

        // However, we might not be the first to respond to the event,
        // so we still need to check whether the queue is empty and loop
        // if it is.
        ::EnterCriticalSection(&m_lock);
        if (!m_list.IsEmpty())
        {
            result = m_list.RemoveHead();
            ASSERT(result != nullptr);
            // The queue shouldn't be full at this point!
            ASSERT(m_nCapacity <= 0 || m_list.GetCount() < m_nCapacity);
            SetEvent(m_queue_not_full);
            // Check whether the queue is now empty.
            if (m_list.IsEmpty())
            {
                ResetEvent(m_queue_not_empty);
            }
        }
        ::LeaveCriticalSection(&m_lock);
    }
}

这是更高效的自动重置事件版本：

class CEventSyncQueue
{
public:
    CEventSyncQueue(int nCapacity = -1);
    virtual ~CEventSyncQueue();
    virtual void Put(void* ptr);
    virtual void* Get();
protected:
    int m_nCapacity;
    CPtrList m_list;

    CRITICAL_SECTION m_lock;    
    HANDLE m_queue_not_empty;
    HANDLE m_queue_not_full;
};

CEventSyncQueue::CEventSyncQueue(int nCapacity)
{
    m_nCapacity = nCapacity;
    ::InitializeCriticalSection(&m_lock);
    m_queue_not_empty = ::CreateEvent(NULL, FALSE, FALSE, NULL);
    m_queue_not_full = ::CreateEvent(NULL, FALSE, TRUE, NULL);
}

CEventSyncQueue::~CEventSyncQueue()
{
    m_list.RemoveAll();
    ::CloseHandle(m_queue_not_empty);
    ::CloseHandle(m_queue_not_full);
    ::DeleteCriticalSection(&m_lock);
}

void CEventSyncQueue::Put(void* ptr)
{
    if (m_nCapacity <= 0)
    {
        ::EnterCriticalSection(&m_lock);
        m_list.AddTail(ptr);
        SetEvent(m_queue_not_empty);
        ::LeaveCriticalSection(&m_lock);
        return;
    }

    bool done = false;
    while (!done)
    {
        // If the queue is full, we must wait until it isn't.
        if(::WaitForSingleObject(m_queue_not_full, INFINITE) != WAIT_OBJECT_0)
        {
            ASSERT(FALSE);
        }

        // However, under some (rare) conditions we'll get here and find
        // the queue is already full again, so be prepared to loop.
        ::EnterCriticalSection(&m_lock);
        if (m_list.GetCount() < m_nCapacity)
        {
            m_list.AddTail(ptr);
            done = true;
            SetEvent(m_queue_not_empty);
            if (m_list.GetCount() < m_nCapacity)
            {
                SetEvent(m_queue_not_full);
            }
        }
        ::LeaveCriticalSection(&m_lock);
    }
}

void* CEventSyncQueue::Get()
{
    void *result = nullptr;
    while (result == nullptr)
    {
        // If the queue is empty, we must wait until it isn't.
        if(::WaitForSingleObject(m_queue_not_empty, INFINITE) != WAIT_OBJECT_0)
        {
            ASSERT(FALSE);
        }

        // However, under some (rare) conditions we'll get here and find
        // the queue is already empty again, so be prepared to loop.
        ::EnterCriticalSection(&m_lock);
        if (!m_list.IsEmpty())
        {
            result = m_list.RemoveHead();
            ASSERT(result != nullptr);
            // The queue shouldn't be full at this point!
            if (m_nCapacity <= 0) ASSERT(m_list.GetCount() < m_nCapacity);
            SetEvent(m_queue_not_full);
            if (!m_list.IsEmpty())
            {
                SetEvent(m_queue_not_empty);
            }
        }
        ::LeaveCriticalSection(&m_lock);
    }
}
于 2012-07-31T04:53:35.917 回答