Win2000服务器端应用程序开发设计指南-设备I/O及线程间通讯

DWORD SleepEx(
 DWORD dwTimeout,
 BOOL fAlertable);
DWORD WaitForSingleObjectEx(
 HANDLE hObject,
 DWORD dwTimeout,
 BOOL fAlertable);
DWORD WaitForMultipleObjectsEx(
 DWORD cObjects,
 PHANDLE phObjects,
 BOOL fWaitAll,
 DWORD dwTimeout,
 BOOL fAlertable);
BOOL SignalObjectAndWait(
 HANDLE hObjectToSignal,
 HANDLE hObjectToWaitOn,
 DWORD dwMilliseconds,
 BOOL fAlertable);
DWORD MsgWaitForMultipleObjectsEx(
 DWORD nCount,
 PHANDLE pHandles,
 DWORD dwMilliseconds,
 DWORD dwWakeMask,
 DWORD dwFlags);

DWORD QueueUserAPC(
 PAPCFUNC pfnAPC,
 HANDLE hThread,
 ULONG_PTR dwData);

VOID WINAPI APCFunc(ULONG_PTR dwParam);

// APC回呼函数没有做任何事
VOID WINAPI APCFunc(ULONG_PTR dwParam) {
 // 这里没有什么要做的
}
UINT WINAPI ThreadFunc(PVOID pvParam) {
 HANDLE hEvent = (HANDLE) pvParam; // Handle被传递此线程
 // 在可警告的状态中等待以便我们可以被强迫清除自己并离开
 DWORD dw = WaitForSingleObjectEx(hEvent, INFINITE, TRUE);
 if (dw == WAIT_OBJECT_0) {
  // 物件变成通知状态
 }
 if (dw == WAIT_IO_COMPLETION) {
  // QueueUserAPC强迫我们自等待状态中离开
  return(0);   // 线程被乾净地删除
 }
 .
 .
 .
 return(0);
}
void main() {
 HANDLE hEvent = CreateEvent(...);
 HANDLE hThread = (HANDLE)_beginthreadex(NULL, 0,
  ThreadFunc, (PVOID) hEvent, 0, NULL);
 .
 .
 .
 // 强迫第二个线程清除自己并离开
 QueueUserAPC(APCFunc, hThread, NULL);
 WaitForSingleObject(hThread, INFINITE);
 CloseHandle(hThread);
 CloseHandle(hEvent);
}

HANDLE CreateIoCompletionPort( 
 
 HANDLE hfile, 
 
 HANDLE hExistingCompPort, 
 
 ULONG_PTR CompKey, 
 
 DWORD dwNumberOfConcurrentThreads);

HANDLE CreateNewCompletionPort(DWORD dwNumberOfConcurrentThreads) { 
 
 return(CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 
 
  dwNumberOfConcurrentThreads)); 
 
}

LONG g_nThreadsMin;  // 集区中线程的最小数量 
 
LONG g_nThreadsMax;  // 集区中线程的最大数量 
 
LONG g_nThreadsCrnt;  // 集区中线程的当前数量 
 
LONG g_nThreadsBusy;  // 集区中忙碌的线程数量

DWORD WINAPI ThreadPoolFunc(PVOID pv) { 
 
 // 线程进入集区 
 
 InterlockedIncrement(&g_nThreadsCrnt); 
 
 InterlockedIncrement(&g_nThreadsBusy); 
 
 for (BOOL fStayInPool = TRUE; fStayInPool;) { 
 
  // 线程停止执行并且等待做某些事 
 
  InterlockedDecrement(&m_nThreadsBusy); 
 
  BOOL fOk = GetQueuedCompletionStatus(...); 
 
  DWORD dwIOError = GetLastError(); 
 
  // 线程在做某些事，所以它是忙碌的 
 
  int nThreadsBusy = InterlockedIncrement(&m_nThreadsBusy); 
 
  // 我们应该加另一个线程到集区吗？ 
 
  if (nThreadsBusy == m_nThreadsCrnt) { // 所有的线程都是忙碌的 
 
   if (nThreadsBusy < m_nThreadsMax) { // 集区不是满的 
 
    if (GetCPUUsage() < 75) { // CPU使用率低于75% 
 
     // 增加线程到集区 
 
     CloseHandle(chBEGINTHREADEX(...)); 
 
    } 
 
   } 
 
  } 
 
  if (!fOk && (dwIOError == WAIT_TIMEOUT)) {   // 线程逾时 
 
   if (!ThreadHasIoPending()) { 
 
    // 服务器没有重要的事做，而且这个线程 
 
    // 可能会死掉，因为它没有未完成的I/O请求 
 
    fStayInPool = FALSE; 
 
   } 
 
  } 
 
  if (fOk || (po != NULL)) { 
 
   // 线程醒来处理某些事；处理它 
 
   ... 
 
   if (GetCPUUsage() > 90) { // CPU使用率高于90% 
 
    if (!ThreadHasIoPending()) { // 没有未决定的I/O请求 
 
     if (g_nThreadsCrnt > g_nThreadsMin)) { // 集区在最小量之上 
 
      fStayInPool = FALSE; //从集区中移除线程 
 
     } 
 
    } 
 
   } 
 
  } 
 
 } 
 
 // 线程离开集区 
 
 InterlockedDecrement(&g_nThreadsBusy); 
 
 InterlockedDecrement(&g_nThreadsCurrent); 
 
 return(0); 
 
}

BOOL PostQueuedCompletionStatus( 
 
 HANDLE      hCompPort, 
 
 DWORD       dwNumBytes, 
 
 ULONG_PTR     CompKey,  
 
 OVERLAPPED* pOverlapped);

FileCopy.cpp  

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

模组：FileCopy.cpp  

通告：Copyright (c)2000 Jeffrey Richter  

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

#include "..\CmnHdr.h" // 请参阅附录A  

#include   

#include "..\ClassLib \IOCP.h" //请参阅附录B  

#include "..\ClassLib \EnsureCleanup.h" //请参阅附录B  

#include "Resource.h"  

///////////////////////////////////////////////////////////////////////////////  

// 每个I/O请求需要OVERLAPPED结构及资料缓冲区  

class CIOReq :public OVERLAPPED {  

public:  

 CIOReq() {  

  Internal = InternalHigh = 0;  

  Offset = OffsetHigh = 0;  

  hEvent = NULL;  

  m_nBuffSize = 0;  

  m_pvData = NULL;  

 }  

 ~CIOReq() {  

  if (m_pvData != NULL)  

   VirtualFree(m_pvData, 0, MEM_RELEASE);  

 }  

 BOOL AllocBuffer(SIZE_T nBuffSize) {  

  m_nBuffSize = nBuffSize;  

  m_pvData = VirtualAlloc(NULL, m_nBuffSize, MEM_COMMIT, PAGE_READWRITE);  

  return(m_pvData != NULL);  

 }  

 BOOL Read(HANDLE hDevice, PLARGE_INTEGER pliOffset = NULL) {  

  if (pliOffset != NULL) {  

   Offset = pliOffset->LowPart;  

   OffsetHigh =pliOffset->HighPart;  

  }  

  return(::ReadFile(hDevice, m_pvData, m_nBuffSize, NULL, this));  

 }  

 BOOL Write(HANDLE hDevice, PLARGE_INTEGER pliOffset = NULL) {  

  if (pliOffset != NULL) {  

   Offset = pliOffset->LowPart;  

   OffsetHigh = pliOffset->HighPart;  

  }  

  return(::WriteFile(hDevice, m_pvData, m_nBuffSize, NULL, this));  

 }  

private:  

 SIZE_T m_nBuffSize;  

 PVOID m_pvData;  

};  

///////////////////////////////////////////////////////////////////////////////  

#define BUFFSIZE    (64 * 1024) // I/O缓冲器的大小  

#define MAX_PENDING_IO_REQS 4  // I/O的最大数量  

// 完成识别值，以指出完成的I/O型态  

#define CK_READ 1  

#define CK_WRITE 2  

///////////////////////////////////////////////////////////////////////////////  

BOOL FileCopy(PCTSTR pszFileSrc, PCTSTR pszFileDst) {  

 BOOL fOk = FALSE; // 假定文件复制失败  

 LARGE_INTEGER liFileSizeSrc = {0 }, liFileSizeDst;  

 try {  

  {  

  // 没有缓冲就开启来源文件并撷取文件的大小  

  CEnsureCloseFile hfileSrc = CreateFile(pszFileSrc, GENERIC_READ,  

   FILE_SHARE_READ, NULL, OPEN_EXISTING,  

   FILE_FLAG_NO_BUFFERING | FILE_FLAG_OVERLAPPED, NULL);  

  if (hfileSrc.IsInvalid()) goto leave;  

  // 撷取文件的大小  

  GetFileSizeEx(hfileSrc, &liFileSizeSrc);  

  // 非缓冲的I/O请求区段大小转移  

  // 我使用缓冲器大小的转移，因为它较容易计算  

  liFileSizeDst.QuadPart = chROUNDUP(liFileSizeSrc.QuadPart, BUFFSIZE);  

  // 没有缓冲就开启目标文件并撷取文件的大小  

  CEnsureCloseFile hfileDst = CreateFile(pszFileDst, GENERIC_WRITE,  

   0, NULL, CREATE_ALWAYS,  

   FILE_FLAG_NO_BUFFERING | FILE_FLAG_OVERLAPPED, hfileSrc);  

  if (hfileDst.IsInvalid()) goto leave;  

  // 文件系统同时扩大文件。现在扩大目标文件  

  // 以至于I/O非同步地执行改进了效能  

  SetFilePointerEx(hfileDst, liFileSizeDst, NULL, FILE_BEGIN);  

  SetEndOfFile(hfileDst);  

  // 建立I/O完成端口并且与文件相关联。  

  CIOCP iocp(0);  

  iocp.AssociateDevice(hfileSrc, CK_READ); // 从来源文件读取  

  iocp.AssociateDevice(hfileDst, CK_WRITE); // 写入目标文件  

  // 初始化保持记录变数  

  CIOReq ior[MAX_PENDING_IO_REQS];  

  LARGE_INTEGER liNextReadOffset = { 0 };  

  int nReadsInProgress = 0;  

  int nWritesInProgress = 0;  

  // 最初的文件复制经由模拟写入已经完成来发动  

  //这导致读取操作被发布  

  for (int nIOReq = 0;nIOReq < chDIMOF(ior); nIOReq++) {  

   // 每个I/O请求需要转移的资料缓冲器  

   chVERIFY(ior[nIOReq].AllocBuffer(BUFFSIZE));  

   nWritesInProgress++;  

   iocp.PostStatus(CK_WRITE, 0, &ior[nIOReq]);  

  }  

  // 当未完成的I/O仍然存在时继续执行回圈  

  while ((nReadsInProgress > 0) || (nWritesInProgress > 0)) {  

   // 悬置线程直到I/O完成  

   ULONG_PTR CompKey;  

   DWORD dwNumBytes;  

   CIOReq* pior;  

   iocp.GetStatus(&CompKey, &dwNumBytes, (OVERLAPPED**) &pior,  

    INFINITE);  

   switch (CompKey) {  

   case CK_READ: // 读取完成，写至目的地  

    nReadsInProgress--;  

   pior->Write(hfileDst); // 写入与从来源读取出来相同的位移量  

   nWritesInProgress++;  

   break;  

   case CK_WRITE: // 写入完成，从来源读取  

    nWritesInProgress--;  

    if (liNextReadOffset.QuadPart < liFileSizeDst.QuadPart) {  

     // 不是EOF，读取下一个来源文件的区块资料  

     pior->Read(hfileSrc, &liNextReadOffset);  

     nReadsInProgress++;  

     liNextReadOffset.QuadPart += BUFFSIZE; // 推进来源位移量  

    }  

    break;  

   }  

  }  

  fOk =TRUE;  

  }  

 leave:;  

 }  

 catch (...) {  

 }  

 if (fOk) {  

  // 目标文件大小是分页大小的倍数。开启有缓冲  

  // 的文件以缩减它的大小，使其成为来源文件的大小  

  CEnsureCloseFile hfileDst = CreateFile(pszFileDst, GENERIC_WRITE,  

   0, NULL, OPEN_EXISTING, 0, NULL);  

  if (hfileDst.IsValid()) {  

   SetFilePointerEx(hfileDst, liFileSizeSrc, NULL, FILE_BEGIN);  

   SetEndOfFile(hfileDst);  

  }  

 }  

 return(fOk);  

}  

///////////////////////////////////////////////////////////////////////////////  

BOOL Dlg_OnInitDialog(HWND hwnd, HWND hwndFocus, LPARAM lParam) {  

 chSETDLGICONS(hwnd, IDI_FILECOPY);  

 // 使复制按钮失效，因为还没选择文件  

 EnableWindow(GetDlgItem(hwnd, IDOK), FALSE);  

 return(TRUE);  

}  

///////////////////////////////////////////////////////////////////////////////  

void Dlg_OnCommand(HWND hwnd, int id, HWND hwndCtl, UINT codeNotify) {  

 TCHAR szPathname [_MAX_PATH ];  

 switch (id) {  

 case IDCANCEL:  

  EndDialog(hwnd, id);  

  break;  

 case IDOK:  

  // 复制来源文件到目标文件  

  Static_GetText(GetDlgItem(hwnd, IDC_SRCFILE),  

  szPathname, sizeof(szPathname));  

  SetCursor(LoadCursor(NULL, IDC_WAIT));  

  chMB(FileCopy(szPathname, TEXT("FileCopy.cpy"))  

   ? "File Copy Successful" : "File Copy Failed");  

  break;  

 case IDC_PATHNAME:  

  OPENFILENAME ofn = { OPENFILENAME_SIZE_VERSION_400 };  

  ofn.hwndOwner = hwnd;  

  ofn.lpstrFilter = TEXT("*.*\0 ");  

  lstrcpy(szPathname, TEXT("*.*"));  

  ofn.lpstrFile = szPathname;  

  ofn.nMaxFile = chDIMOF(szPathname);  

  ofn.lpstrTitle = TEXT("Select file to copy");  

  ofn.Flags = OFN_EXPLORER | OFN_FILEMUSTEXIST;  

  BOOL fOk = GetOpenFileName(&ofn);  

  if (fOk) {  

   // 显示来源文件大小给使用者  

   Static_SetText(GetDlgItem(hwnd, IDC_SRCFILE), szPathname);  

   CEnsureCloseFile hfile = CreateFile(szPathname, 0, 0, NULL,  

    OPEN_EXISTING, 0, NULL);  

   if (hfile.IsValid()) {  

    LARGE_INTEGER liFileSize;  

    GetFileSizeEx(hfile, &liFileSize);  

    // 注意：只出现大小的下面（bottom）32位元  

    SetDlgItemInt(hwnd, IDC_SRCFILESIZE, liFileSize.LowPart, FALSE);  

   }  

  }  

  EnableWindow(GetDlgItem(hwnd, IDOK), fOk);  

  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_FILECOPY), NULL, Dlg_Proc);  

 return(0);  

}  
//////////////////////////////// End of File //////////////////////////////////