WaitForMultipleObjects 饿死答案

【问题标题】：WaitForMultipleObjects starving outWaitForMultipleObjects 饿死
【发布时间】：2016-06-27 18:43:01
【问题描述】：

当我在一组对象上调用 WaitForMultipleObjects 并且另一个线程正在使用 WaitForSingleObject 等待该组中的单个对象时，首选该线程并且调用 WaitForMultipleObjects 的线程永远不会获得 CPU 时间。

看看这个来源：

#include <windows.h>
#include <cstdio>
#include <unordered_map>

HANDLE        hEvt,
              hSema;
bool volatile fReleased;

DWORD WINAPI LockAndReleaseThread( LPVOID lpvThreadParam );

int main()
{
    int const NTHREADS = 10;
    HANDLE    ahWait[2];

    ahWait[0] = ::hEvt  = CreateEvent( NULL, FALSE, TRUE, NULL );
    ahWait[1] = ::hSema = CreateSemaphore( NULL, 0, 1, NULL );
    fReleased = false;

    for( int i = 0; i < NTHREADS; i++ )
        CreateThread( NULL, 0, LockAndReleaseThread, NULL, 0, NULL );

    for( ; ; )
        WaitForMultipleObjects( 2, ahWait, TRUE, INFINITE ),
        std::printf( "main thread is holding lock and received signal\n" ),
        ::fReleased = false,
        SetEvent( ::hEvt );

    return 0;
}

char GetID();

DWORD WINAPI LockAndReleaseThread( LPVOID lpvThreadParam )
{
    for( ; ; )
    {
        WaitForSingleObject( ::hEvt, INFINITE );
        std::printf( "spawned thread with id %c is holding lock\n", (char)GetID() );

        if( !::fReleased )
            ReleaseSemaphore( ::hSema, 1, NULL ),
            ::fReleased = true;

        Sleep( 1000 );
        SetEvent( ::hEvt );
    }

    return 0;
}

char GetID()
{
    static std::unordered_map<DWORD, char> mapTIDsToIDs;
    static char                            nextId = 'A';
    DWORD                                  dwThreadId;

    if( mapTIDsToIDs.find( dwThreadId = GetCurrentThreadId() ) == mapTIDsToIDs.end() )
        return mapTIDsToIDs[dwThreadId] = nextId++;
    else
        return mapTIDsToIDs[dwThreadId];
}

在这段代码中，如果 NTHREADS >= 2，主线程永远不会获得 CPU 时间。我不希望 WaitForSingleObject 或 WaitForMultipleObjects 完全公平，但这在我看来是一个概念缺陷。

如果我替换

WaitForMultipleObjects( 2, ahWait, TRUE, INFINITE ),

与

WaitForSingleObject( ::hSema, INFINITE ),
WaitForSingleObject( ::hEvt, INFINITE ),

主线程获得 CPU 时间。

[编辑1]

即使在 Windows 10 中也没有修复该错误。并且执行 WaitForMultipleObjects 而不是 WaitForSingleObject 并不会改变这种情况。并且生成的线程以循环方式获得 CPU 时间，无论是否有睡眠以及我正在执行 WaitforSingleObject 还是 WaitForMultipleObjects 都无关紧要。

这里使用关键部分的提示对我没有帮助。

我开发了一个条件变量，如果多个线程锁定 condvar 并且一个线程正在等待 condvar 被单选，则可能会发生这种情况：

#include <windows.h>
#include <cassert>
#include <exception>
#include <intrin.h>

#if !defined(NDEBUG)
    #define ONDEBUG(expr) (expr)
#else
    #define ONDEBUG(expr) ((void)0)
#endif

inline
DWORD FastGetCurrentThreadId()
{
#if defined(_M_IX86)
    return __readfsdword( 0x24 );
#elif defined(_M_AMD64)
    return *(DWORD *)(__readgsqword( 0x30 ) + 0x48);
#endif
}

class Exception : public std::exception
{
};

class ResourceException : public Exception
{
};

template <typename TYPE>
inline
TYPE AutoThrowResourceException( TYPE t )
{
    if( !t )
        throw ResourceException();

    return t;
}

class XHANDLE
{
public:
         XHANDLE( HANDLE h = NULL );
         ~XHANDLE();
    void CloseHandle();

public:
    HANDLE h;
};

inline
XHANDLE::XHANDLE( HANDLE h )
{
    this->h = h;
}

inline
XHANDLE::~XHANDLE()
{
    BOOL fClosed;

    if( h && h != INVALID_HANDLE_VALUE )
        fClosed = ::CloseHandle( h ),
        assert(fClosed);
}

inline
void XHANDLE::CloseHandle()
{
    ::CloseHandle( h );
    h = NULL;
}


class CondVar
{
public:
         CondVar();
         ~CondVar();
    void Enter();
    void Wait();
    void Release();
    void ReleaseAll();
    void Leave();

private:
    LONGLONG volatile m_llOwnersAndWaiters;
    DWORD    volatile m_dwRecursionCount;
    DWORD    volatile m_dwOwningThreadId;
    XHANDLE           m_xhEvtEnter,
                      m_xhSemRelease;

private:
    static
    DWORD Owners( LONGLONG llOwnersAndWaiters )
    {
        return (DWORD)llOwnersAndWaiters;
    }

    static
    DWORD Waiters( LONGLONG llOwnersAndWaiters )
    {
        return (DWORD)((DWORDLONG)llOwnersAndWaiters >> 32);
    }
};

CondVar::CondVar() :
    m_xhEvtEnter(   AutoThrowResourceException( ::CreateEvent( NULL, FALSE, FALSE, NULL ) ) ),
    m_xhSemRelease( AutoThrowResourceException( ::CreateSemaphore( NULL, 0, 0x7FFFFFFF, NULL ) ) )
{
    m_llOwnersAndWaiters = 0;
    m_dwRecursionCount   = 0;
    m_dwOwningThreadId   = 0;
}

CondVar::~CondVar()
{
}

void CondVar::Enter()
{
    if( m_dwOwningThreadId == FastGetCurrentThreadId() )
    {
        m_dwRecursionCount++;
        return;
    }

    LONGLONG llOwnersAndWaiters = ::InterlockedIncrement64( &m_llOwnersAndWaiters  );

    assert(Owners( llOwnersAndWaiters ) > Waiters( llOwnersAndWaiters ));

    if( (Owners( llOwnersAndWaiters ) - Waiters( llOwnersAndWaiters )) > 1 )
        for( ; ::WaitForSingleObject( m_xhEvtEnter.h, INFINITE ) != WAIT_OBJECT_0; assert(false) );

    ONDEBUG(llOwnersAndWaiters = m_llOwnersAndWaiters);
    assert(Owners( llOwnersAndWaiters ) > Waiters( llOwnersAndWaiters ) && m_dwOwningThreadId == 0);

    m_dwOwningThreadId = FastGetCurrentThreadId();
}

void CondVar::Wait()
{
    LONGLONG llOwnersAndWaiters;
    DWORD    dwSavedRecusionCount;

    ONDEBUG(llOwnersAndWaiters = m_llOwnersAndWaiters);
    assert(Owners( llOwnersAndWaiters ) > Waiters( llOwnersAndWaiters ) && m_dwOwningThreadId == FastGetCurrentThreadId());

    m_dwOwningThreadId   = 0;
    dwSavedRecusionCount = m_dwRecursionCount;
    m_dwRecursionCount   = 0;
    llOwnersAndWaiters   = ::InterlockedAdd64( &m_llOwnersAndWaiters, 0x100000000 );

    if( Owners( llOwnersAndWaiters ) > Waiters( llOwnersAndWaiters ) )
        for( ; !::SetEvent( m_xhEvtEnter.h ); assert(false) );

    HANDLE ahWait[2] = { m_xhEvtEnter.h, m_xhSemRelease.h };

    for( ; ::WaitForMultipleObjects( 2, ahWait, TRUE, INFINITE ) != WAIT_OBJECT_0; assert(false) );

    ONDEBUG(llOwnersAndWaiters = m_llOwnersAndWaiters);
    assert(Owners( llOwnersAndWaiters ) > Waiters( llOwnersAndWaiters ) && m_dwOwningThreadId == 0);

    m_dwOwningThreadId = FastGetCurrentThreadId();
    m_dwRecursionCount = dwSavedRecusionCount;
}

void CondVar::Release()
{
    LONGLONG llOwnersAndWaiters,
             llOwnersAndWaitersPrevOrChanged;

    for( llOwnersAndWaiters = m_llOwnersAndWaiters; Waiters( llOwnersAndWaiters ); llOwnersAndWaiters = llOwnersAndWaitersPrevOrChanged )
    {
        assert(Owners( llOwnersAndWaiters ) > Waiters( llOwnersAndWaiters ) && m_dwOwningThreadId == FastGetCurrentThreadId());

        if( (llOwnersAndWaitersPrevOrChanged = ::InterlockedCompareExchange64( &m_llOwnersAndWaiters, llOwnersAndWaiters - 0x100000000, llOwnersAndWaiters )) == llOwnersAndWaiters )
        {
            for (; !::ReleaseSemaphore( m_xhSemRelease.h, 1, NULL ); assert( false ));
            break;
        }
    }
}

void CondVar::ReleaseAll()
{
    LONGLONG llOwnersAndWaiters,
             llOwnersAndWaitersPrevOrChanged;

    for( llOwnersAndWaiters = m_llOwnersAndWaiters; Waiters( llOwnersAndWaiters ); llOwnersAndWaiters = llOwnersAndWaitersPrevOrChanged )
    {
        assert(Owners( llOwnersAndWaiters ) > Waiters( llOwnersAndWaiters ) && m_dwOwningThreadId == FastGetCurrentThreadId());

        if( (llOwnersAndWaitersPrevOrChanged = ::InterlockedCompareExchange64( &m_llOwnersAndWaiters, llOwnersAndWaiters & 0x0FFFFFFFF, llOwnersAndWaiters )) == llOwnersAndWaiters )
        {
            for (; !::ReleaseSemaphore( m_xhSemRelease.h, (LONG)Waiters( llOwnersAndWaiters ), NULL ); assert( false ));
            break;
        }
    }
}

void CondVar::Leave()
{
    LONGLONG llOwnersAndWaiters;
    LONG     lRecursionCount;

    ONDEBUG(llOwnersAndWaiters = m_llOwnersAndWaiters);
    assert(Owners( llOwnersAndWaiters ) > Waiters( llOwnersAndWaiters ) && m_dwOwningThreadId == FastGetCurrentThreadId());

    if( (lRecursionCount = m_dwRecursionCount) != 0 )
    {
        m_dwRecursionCount = lRecursionCount - 1;
        return;
    }

    m_dwOwningThreadId = 0;

    llOwnersAndWaiters = ::InterlockedDecrement64( &m_llOwnersAndWaiters );

    if( Owners( llOwnersAndWaiters ) > Waiters( llOwnersAndWaiters ) )
        for( ; !::SetEvent( m_xhEvtEnter.h ); assert(false) );
}

/* ---------------------------- unit-test ---------------------------- */

#include <cstdio>
#include <cstdlib>
#include <deque>
#include <map>

DWORD WINAPI ThreadFunc( LPVOID lpvThreadParam );

using namespace std;

struct ThreadResult
{
    DWORD dwThreadId;
    DWORD dwCounter;
};

CondVar             cv;
deque<ThreadResult> dqtr;
bool volatile       fStop = false;

int main()
{
    int const              NTHREADS = 16;
    HANDLE                 ahThreads[NTHREADS];
    int                    i;
    std::map<DWORD, DWORD> mTRs;

    for( i = 0; i < NTHREADS; i++ )
        ahThreads[i] = CreateThread( NULL, 0, ThreadFunc, NULL, 0, NULL );

    for( i = 0; i < 10000; i++ )
    {
        ThreadResult tr;

        ::cv.Enter();

        if( ::dqtr.empty() )
            ::cv.Wait();

        tr = ::dqtr.front();
        ::dqtr.pop_front();
        ::cv.Leave();

        printf( "Thread: %08X - Number: %d\n", (unsigned)tr.dwThreadId, (unsigned)tr.dwCounter );

        if( mTRs.find( tr.dwThreadId ) == mTRs.end() )
            mTRs[tr.dwThreadId] = tr.dwCounter;
        else
            assert((mTRs[tr.dwThreadId] + 1) == tr.dwCounter),
            mTRs[tr.dwThreadId] = tr.dwCounter;
    }

    for( ::fStop = true, i = 0; i < NTHREADS; i++ )
        WaitForSingleObject( ahThreads[i], INFINITE );

    return 0;
}

DWORD WINAPI ThreadFunc( LPVOID lpvThreadParam )
{
    ThreadResult tr;

    tr.dwThreadId = FastGetCurrentThreadId();
    tr.dwCounter  = 0;

    for( ; !::fStop; tr.dwCounter++ )
    {
        ::cv.Enter();
        ::dqtr.push_back( tr );
        ::cv.Release();
        Sleep( 100 );
        ::cv.Leave();
    }

    return 0;
}

【问题讨论】：

标签： winapi

【解决方案1】：

我可以重现你的问题，这很奇怪。

对于它的价值，我发现我可以通过专门使用 WaitForMultipleObjects() 来解决它，并包括一个每个线程的虚拟对象，即，

DWORD WINAPI LockAndReleaseThread( LPVOID lpvThreadParam )
{
    HANDLE ahWait[2];
    ahWait[0] = ::hEvt;
    ahWait[1] = CreateEvent(NULL, TRUE, TRUE, NULL);
    for( ; ; )
    {
        WaitForMultipleObjects(2, ahWait, TRUE, INFINITE );
        std::printf( "spawned thread with id %c is holding lock\n", (char)GetID() );

        if( !::fReleased )
            ReleaseSemaphore( ::hSema, 1, NULL ),
            ::fReleased = true;

        Sleep( 1000 );
        SetEvent( ::hEvt );
    }

    return 0;
}

如果子线程共享一个虚拟对象，这似乎不起作用，除非父线程也在等待该对象。因此，虽然它似乎在这种情况下有效，但它可能是一种脆弱的方法。

就个人而言，此时我正在寻找另一种解决方案，但在您的情况下，我真的不知道从哪里开始。（在我的脑海中，我正在考虑一个线程的链接列表，也许还有一些无锁队列。）另一方面，由于您直接从未记录的进程环境块中读取线程 ID，因此可靠性和转发对您而言，兼容性显然不是一个主要问题，因此也许这种解决方法就足够了。 :-)

您还应该注意，Windows 确实有自己的条件变量实现。我不知道你为什么要在这里自己动手。

【讨论】：

@BonitaMontero，嗯，是的，你说的很对。我正在试验的代码并没有像我想象的那样做。当我回到您的代码并仅进行我建议的最小更改时，它停止工作。我有理由相信我最新编辑中显示的代码更改确实有效，这次我更加小心了。