螺纹安全管道端接

Question

（开始前注意：虽然我的问题很笼统，但我的代码需要使用遗留 Visual Studio 2008 MFC 应用程序编译，并且必须使用 MFC 或 win32 同步，请避免使用 ie boost 或 c++ 11 回答）

我正在尝试实现一个 线程安全管道 （一个具有单个 reader 和一个写入器的队列），我做了以下操作：

template<class T>
class CMultiThreadPipe { 

private:
    HANDLE hSemaphore, hTerminateEvent1, hTerminateEvent2;
    CRITICAL_SECTION listMutex; 
    CList<T*, T*> list;

public:
    CMultiThreadPipe() { 
        InitializeCriticalSection(&listMutex);
        hSemaphore = CreateSemaphore(NULL, 0, LONG_MAX, NULL);
        hTerminateEvent1 = ::CreateEvent(NULL, TRUE, FALSE, NULL); 
        hTerminateEvent2 = ::CreateEvent(NULL, TRUE, FALSE, NULL);
    }

    // pdata must be allocated with new. The dequeueing thread will delete it
    void Enqueue(T* pdata) { 
        EnterCriticalSection(&listMutex);
        list.AddHead(pdata);
        LeaveCriticalSection(&listMutex);
        ReleaseSemaphore(hSemaphore, 1, NULL);
    }

    // if Dequeue returns null it means the pipe was destroyed and no further queue method calls are legal
    // Dequeue caller is responsible to delete the returned instance
    T* Dequeue()
    {
        HANDLE handles[] = { hTerminateEvent1, hSemaphore };
        DWORD waitRes = WaitForMultipleObjects(2, handles, FALSE, INFINITE);
        if (waitRes==WAIT_OBJECT_0) {
            SetEvent(hTerminateEvent2);
            return NULL; // terminated
        }
        EnterCriticalSection(&listMutex);
        T* elem = list.RemoveTail(); 
        LeaveCriticalSection(&listMutex);
        return elem; // handler must delete item
    }

    void Destroy() {
        SetEvent(hTerminateEvent1);
        WaitForSingleObject(hTerminateEvent2, INFINITE);
        EnterCriticalSection(&listMutex);
        POSITION pos = list.GetHeadPosition(); 
        for (int i = 0; i < list.GetCount(); i++) delete list.GetNext(pos); 
        LeaveCriticalSection(&listMutex);
        DeleteCriticalSection(&listMutex);
        CloseHandle(hSemaphore);
    }

    ~CMultiThreadPipe() { 
        Destroy();
    }
};

代码是这样使用的：

class QueueData {
    public:
        QueueData(int i) : m_data(i) {};
        int m_data;
};

UINT DequeueThreadProc(LPVOID dummy);

CMultiThreadedPipe<QueueData>* pPipe = NULL;

void main() {
    pPipe = new CMultiThreadedPipe<QueueData>();
    start new thread running DequeueThreadProc

    int counter=0;
    for (int counter=0; counter<10; counter++)
    {
        pPipe->Enqueue(new QueueData(counter));
        Sleep(300);
    }
    delete pPipe;
}

UINT DequeueThreadProc(LPVOID ignore)
{
    QueueData* queueData;
    while ((queueData = pPipe->Dequeue()) != NULL) {
        delete queueData;
        Sleep(1000);
    };
    return 0;
}

我遇到的问题是终止，在上面的实现中，当管道被销毁时（总是由入队线程），它正在等待出队线程知道它在删除队列之前已经终止。它必须这样做以防止出列线程在管道被销毁后尝试出列的情况。

如果出队线程不继续调用出队，第一个线程将挂在析构函数中，如果出队线程在调用出队之间等待很长时间，第一个线程的析构函数将相应地卡在那里。

我阅读了有关它的各种帖子 none 提到了安全销毁。任何帮助表示赞赏！

Answer 1

对于从多个线程访问的安全销毁对象，您需要对其使用引用计数。在将对象指针传递给新线程之前 - 您增加了对对象的引用。当线程不再使用对象时，或者如果创建线程失败，则减少引用计数。当对象的最后一个引用被释放时——你可以安全地调用对象的析构函数。你不需要在这里等待任何线程。

也用于实现这样的队列 - 在 windows 中存在特殊对象 - 名为 I/O Completion Ports in user space (in kernel space in know as KQUEUE). with this object - implementation will be more efficient and simply - you not need manage self list (CList in your code), synchronize access to it - all this will be done in kernel space for you (PostQueuedCompletionStatus -> KeInsertQueue, GetQueuedCompletionStatus -> KeRemoveQueue)。您只需要创建 iocp，(kqueue) 对象。

class CMultiThreadPipe {

public:

    class __declspec(novtable) QueueData {
    public:

        virtual void ProcessItem() = 0;

        virtual ~QueueData()
        {
            DbgPrint("%x: %s<%p>\n", GetCurrentThreadId(), __FUNCTION__, this);
        }

        QueueData()
        {
            DbgPrint("%x: %s<%p>\n", GetCurrentThreadId(), __FUNCTION__, this);
        }
    };

private:
    HANDLE _hIOCP;
    LONG _dwRef;
    ULONG _nThreads;

    void DequeueThreadProc()
    {
        ULONG NumberOfBytesTransferred;
        QueueData* pData;
        OVERLAPPED* pOverlapped;

        while (GetQueuedCompletionStatus(_hIOCP, 
            &NumberOfBytesTransferred, 
            (ULONG_PTR*)&pData, 
            &pOverlapped, INFINITE))
        {
            if (pData)
            {
                pData->ProcessItem();
            }
            else
            {
                break;
            }
        }

        Release();
    }

    __declspec(noreturn) static DWORD CALLBACK _DequeueThreadProc(PVOID pThis)
    {
        reinterpret_cast<CMultiThreadPipe*>(pThis)->DequeueThreadProc();
        FreeLibraryAndExitThread((HMODULE)&__ImageBase, 0);
    }

    ~CMultiThreadPipe()
    {
        if (_hIOCP)
        {
            CloseHandle(_hIOCP);
        }
    }

public:

    CMultiThreadPipe() : _dwRef(1), _hIOCP(0)
    {
    }

    void AddRef()
    {
        InterlockedIncrement(&_dwRef);
    }

    void Release()
    {
        if (!InterlockedDecrement(&_dwRef))
        {
            delete this;
        }
    }

    ULONG Create(DWORD NumberOfDequeueThreads)
    {
        if (_hIOCP = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, NumberOfDequeueThreads))
        {
            ULONG n = 0;
            do 
            {
                HMODULE hModule;
                if (GetModuleHandleExW(GET_MODULE_HANDLE_EX_FLAG_FROM_ADDRESS, (PCWSTR)_DequeueThreadProc, &hModule))
                {
                    AddRef();

                    if (HANDLE hThread = CreateThread(0, 0, _DequeueThreadProc, this, 0, 0))
                    {
                        CloseHandle(hThread);
                        n++;
                    }
                    else
                    {
                        Release();
                        FreeLibrary(hModule);
                    }
                }

            } while (--NumberOfDequeueThreads);

            _nThreads = n;

            return n ? NOERROR : ERROR_GEN_FAILURE;
        }

        return GetLastError();
    }

    ULONG Enqueue(QueueData* pData)
    {
        return PostQueuedCompletionStatus(_hIOCP, 0, (ULONG_PTR)pData, 0) ? NOERROR : GetLastError();
    }

    void Destroy()
    {
        if (ULONG n = _nThreads)
        {
            do 
            {
                PostQueuedCompletionStatus(_hIOCP, 0, 0, 0);
            } while (--n);
        }
    }
};

和用法：

class QueueData : public CMultiThreadPipe::QueueData
{
    int m_data; 

    virtual void ProcessItem()
    {
        DbgPrint("%x: %s<%p>(%u)\n", GetCurrentThreadId(), __FUNCTION__, this, m_data);
        delete this;
    }
public:
    QueueData(int i) : m_data(i) {};
};

void testQueue()
{
    if (CMultiThreadPipe* pPipe = new CMultiThreadPipe)
    {
        if (pPipe->Create(8) == NOERROR)
        {
            int n = 64;

            do 
            {
                if (QueueData* pData = new QueueData(n))
                {
                    if (pPipe->Enqueue(pData))
                    {
                        delete pData;
                    }
                }
            } while (--n);

            pPipe->Destroy();
        }
        pPipe->Release();
    }
}

注意这样的 CMultiThreadPipe 实现 - 工作线程退出时不需要等待。即使您的代码在 dll 中并且您卸载了 dll - 您也不需要等待。每个线程都有自己的对象和模块引用。并在退出时释放它

螺纹安全管道端接

Thread Safe Pipe Termination

c++

mfc

multithreading

win32-process

visual-c++