Windows核心编程学习九：利用内核对象进行线程同步

1.程序框架

#include "Queue.h"
#include <tchar.h>
#include <windowsx.h>
#include <StrSafe.h>
#include <process.h>
#include "resource.h"

/************************************************************************/

#define chHANDLE_DLGMSG(hWnd, message, fn)                 \
case (message): return (SetDlgMsgResult(hWnd, uMsg,     \
HANDLE_##message((hWnd), (wParam), (lParam), (fn))))

// This macro function calls the C runtime's _beginthreadex function.
// The C runtime library doesn't want to have any reliance on Windows' data
// types such as HANDLE. This means that a Windows programmer needs to cast
// values when using _beginthreadex. Since this is terribly inconvenient,
// I created this macro to perform the casting.
typedef unsigned (__stdcall *PTHREAD_START) (void *);

#define chBEGINTHREADEX(psa, cbStackSize, pfnStartAddr, \
pvParam, dwCreateFlags, pdwThreadId)                 \
((HANDLE)_beginthreadex(                        \
(void *)        (psa),                         \
(unsigned)      (cbStackSize),                 \
(PTHREAD_START) (pfnStartAddr),                \
(void *)        (pvParam),                     \
(unsigned)      (dwCreateFlags),               \
(unsigned *)    (pdwThreadId)))

// Sets the dialog box icons
inline void chSETDLGICONS(HWND hWnd, int idi)
{
SendMessage(hWnd, WM_SETICON, ICON_BIG,  (LPARAM)
LoadIcon((HINSTANCE) GetWindowLongPtr(hWnd, GWLP_HINSTANCE), MAKEINTRESOURCE(idi)));
SendMessage(hWnd, WM_SETICON, ICON_SMALL, (LPARAM)
LoadIcon((HINSTANCE) GetWindowLongPtr(hWnd, GWLP_HINSTANCE), MAKEINTRESOURCE(idi)));
}

BOOL Dlg_OnInitDialog(HWND hWnd, HWND hWndFocus, LPARAM lParam)
{
chSETDLGICONS(hWnd, IDI_QUEUE);

return TRUE;
}

void Dlg_OnCommand(HWND hWnd, int id, HWND hWndCtl, UINT codeNotify)
{
switch (id)
{
case IDCANCEL:
EndDialog(hWnd, id);
break;
}
}

INT_PTR WINAPI Dlg_Proc(HWND hWnd, UINT uMsg, WPARAM wParam, LPARAM lParam)
{
switch (uMsg)
{
chHANDLE_DLGMSG(hWnd, WM_INITDIALOG, Dlg_OnInitDialog);
chHANDLE_DLGMSG(hWnd, WM_COMMAND,	 Dlg_OnCommand);
}

return FALSE;
}

/*************************************************************************/

int WINAPI _tWinMain(HINSTANCE hInstExe, HINSTANCE, PTSTR pszCmdLine, int)
{
DialogBox(hInstExe, MAKEINTRESOURCE(IDD_QUEUE), NULL, Dlg_Proc);

return 0;
}

2.线程安全队列的实现

/*
* File:	CQueue.h
* Time:	2013-07-10
* 描述：	学习《Windows核心编程》
*/

#ifndef _CQUEUE_H_
#define _CQUEUE_H_

#include <windows.h>

class CQueue
{
public:
struct ELEMENT {
int m_nThreadNum;				// 线程号
int m_nRequestNum;				// 请求号
};
typedef ELEMENT* PELEMENT;

private:
PELEMENT	m_pElements;			// 队列元素数组
int			m_nMaxElements;			// 数组长度
HANDLE		m_h[2];					// 两个内核对象，一个互斥量一个信号量
HANDLE		&m_hmtxQ;				// 互斥量对象的引用
HANDLE		&m_hsemNumElements;		// 信号量对象的引用

public:
CQueue(int nMaxElements);
~CQueue();

BOOL Append(PELEMENT pElement, DWORD dwMilliseconds);
BOOL Remove(PELEMENT pElement, DWORD dwMilliseconds);
};

#endif

/************************************************************************/

CQueue::CQueue(int nMaxElements) : m_hmtxQ(m_h[0]), m_hsemNumElements(m_h[1])
{
// 初始化结构数组
m_pElements = (PELEMENT)HeapAlloc(GetProcessHeap(), 0, sizeof(ELEMENT) * nMaxElements);

// 初始化数组长度
m_nMaxElements = nMaxElements;

// 创建互斥量和信号量内核对象
m_hmtxQ = CreateMutex(NULL, FALSE, NULL);
m_hsemNumElements = CreateSemaphore(NULL, 0, nMaxElements, NULL);
}

CQueue::~CQueue()
{
// 清理内核对象和内存
CloseHandle(m_hsemNumElements);
CloseHandle(m_hmtxQ);
HeapFree(GetProcessHeap(), 0, m_pElements);
}

BOOL CQueue::Append(CQueue::PELEMENT pElement, DWORD dwTimeout)
{
BOOL fOk = FALSE;

// 等待互斥量内核对象
DWORD dw = WaitForSingleObject(m_hmtxQ, dwTimeout);

// 返回WAIT_OBJECT_0表示得到了队列的独占访问权
if (dw == WAIT_OBJECT_0) {
LONG lPrevCount;

// 尝试向队列添加新元素
fOk = ReleaseSemaphore(m_hsemNumElements, 1, &lPrevCount);

// fOk为TRUE表示队列没满，可以添加新元素
if (fOk) {
m_pElements[lPrevCount] = *pElement;
}
else {
SetLastError(ERROR_DATABASE_FULL);
}

// 释放互斥量，允许其他线程访问队列
ReleaseMutex(m_hmtxQ);
}
else {
SetLastError(ERROR_TIMEOUT);
}

return fOk;
}

BOOL CQueue::Remove(CQueue::PELEMENT pElement, DWORD dwTimeout)
{
// 确保对队列具有独占访问权，且要求队列中有元素可取
// 等待信号量成功的副作用使它的计数减一，所以不需要显示调用ReleaseSemaphore()
BOOL fOk = (WaitForMultipleObjects(_countof(m_h), m_h, TRUE, dwTimeout) == WAIT_OBJECT_0);

if (fOk)
{
// 获取元素
*pElement = m_pElements[0];

// 取出索引为0 的元素，把数组中剩余元素向前挪一个位置
MoveMemory(&m_pElements[0], &m_pElements[1], sizeof(ELEMENT) * (m_nMaxElements - 1));

// 释放互斥量，允许其他线程访问队列
ReleaseMutex(m_hmtxQ);
}
else
{
SetLastError(ERROR_TIMEOUT);
}

return fOk;
}

3.客户端线程

CQueue				g_q(10);
volatile LONG		g_fShutdown = FALSE;
HWND				g_hWnd;

// Handles to all reader/writer threads
HANDLE g_hThreads[MAXIMUM_WAIT_OBJECTS];

// Number of reader/writer threads
int    g_nNumThreads = 0;

/************************************************************************/

DWORD WINAPI ClientThread(PVOID pvParam)
{
int nThreadNum = PtrToUlong(pvParam);			// 线程号
HWND hWndLB = GetDlgItem(g_hWnd, IDC_CLIENT);

int nRequestNum = 0;
while (1 != InterlockedCompareExchange(&g_fShutdown, 0, 0))
{
nRequestNum++;

TCHAR sz[1024];
CQueue::ELEMENT e = { nThreadNum, nRequestNum };

// 尝试添加元素到队列
if (g_q.Append(&e, 200)) {
StringCchPrintf(sz, _countof(sz), TEXT("客户线程%d 添加元素%d"), nThreadNum, nRequestNum);
}
else {
StringCchPrintf(sz, _countof(sz), TEXT("客户线程%d 添加元素%d失败(%s)"), nThreadNum, nRequestNum,
(GetLastError() == ERROR_TIMEOUT) ? TEXT("超时") : TEXT("队列已满"));
}

ListBox_SetCurSel(hWndLB, ListBox_AddString(hWndLB, sz));
Sleep(2500);
}

return 0;
}

BOOL Dlg_OnInitDialog(HWND hWnd, HWND hWndFocus, LPARAM lParam)
{
chSETDLGICONS(hWnd, IDI_QUEUE);

g_hWnd = hWnd;

DWORD dwThreadID;

// 创建客户端线程
for (int i = 0; i < 4; i++)
g_hThreads[g_nNumThreads++] = chBEGINTHREADEX(NULL, 0, ClientThread, (PVOID)(INT_PTR) i, 0, &dwThreadID);

return TRUE;
}

4.服务器端线程

DWORD WINAPI ServerThread(PVOID pvParam)
{
int nThreadNum = PtrToUlong(pvParam);
HWND hWndLB = GetDlgItem(g_hWnd, IDC_SERVER);

while (1 != InterlockedCompareExchange(&g_fShutdown, 0, 0)) {
TCHAR sz[1024];
CQueue::ELEMENT e;

// 尝试移除元素
if (g_q.Remove(&e, 5000)) {
StringCchPrintf(sz, _countof(sz), TEXT("服务器端线程%d移除客户端线程%d放入的元素%d"), nThreadNum, e.m_nThreadNum, e.m_nRequestNum);
Sleep(2000 * e.m_nThreadNum);
}
else {
StringCchPrintf(sz, _countof(sz), TEXT("服务器端线程%d 没有元素可取"), nThreadNum);
}

ListBox_SetCurSel(hWndLB, ListBox_AddString(hWndLB, sz));
}

return 0;
}

BOOL Dlg_OnInitDialog(HWND hWnd, HWND hWndFocus, LPARAM lParam)
{
chSETDLGICONS(hWnd, IDI_QUEUE);

g_hWnd = hWnd;

DWORD dwThreadID;

// 创建客户端线程
for (int i = 0; i < 4; i++)
g_hThreads[g_nNumThreads++] = chBEGINTHREADEX(NULL, 0, ClientThread, (PVOID)(INT_PTR) i, 0, &dwThreadID);

// 创建读者线程
for (int i = 0; i < 2; i++)
g_hThreads[g_nNumThreads++] = chBEGINTHREADEX(NULL, 0, ServerThread, (PVOID)(INT_PTR) i, 0, &dwThreadID);

return TRUE;
}

int WINAPI _tWinMain(HINSTANCE hInstExe, HINSTANCE, PTSTR pszCmdLine, int)
{
DialogBox(hInstExe, MAKEINTRESOURCE(IDD_QUEUE), NULL, Dlg_Proc);

InterlockedExchange(&g_fShutdown, TRUE);

WaitForMultipleObjects(g_nNumThreads, g_hThreads, TRUE, INFINITE);
while (g_nNumThreads--)
CloseHandle(g_hThreads[g_nNumThreads]);

return 0;
}