完成端口以及线程池(详细原代码)

#include "StdAfx.h" #include ".\completionioserver.h" #include <windows.h> #include <conio.h> #include <io.h> namespace COMPLETIONIOSERVER_LIB { /************************************************************************/ /*CSimpleFile */ /************************************************************************/ CSimpleFile::CSimpleFile(void) { m_pFile = NULL; m_uOpenFlag = 0; memset(m_pOpenStr,0,3); } CSimpleFile::CSimpleFile( LPCTSTR lpszFileName, UINT nOpenFlags ) { Open(lpszFileName,nOpenFlags); } CSimpleFile::~CSimpleFile(void) { if(m_pFile != NULL) { fclose(m_pFile); } } void CSimpleFile::ConvertFlag(UINT nOpenFlags) { if((nOpenFlags&READ) == READ) { m_pOpenStr[0] = 'r'; if((nOpenFlags&TEXT) == TEXT) { m_pOpenStr[1] = 't'; } if((nOpenFlags&BINARY) == BINARY) { m_pOpenStr[1] = 'b'; } } if((nOpenFlags&WRITE) == WRITE) { m_pOpenStr[0] = 'w'; if((nOpenFlags&TEXT) == TEXT) { m_pOpenStr[1] = 't'; } if((nOpenFlags&BINARY) == BINARY) { m_pOpenStr[1] = 'b'; } } if((nOpenFlags&READWRITE) == READWRITE) { m_pOpenStr[0] = 'r'; m_pOpenStr[1] = '+'; if((nOpenFlags&TEXT) == TEXT) { m_pOpenStr[2] = 't'; } if((nOpenFlags&BINARY) == BINARY) { m_pOpenStr[2] = 'b'; } } if((nOpenFlags&APPEND) == APPEND) { m_pOpenStr[0] = 'a'; m_pOpenStr[1] = '+'; if((nOpenFlags&TEXT) == TEXT) { m_pOpenStr[2] = 't'; } if((nOpenFlags&BINARY) == BINARY) { m_pOpenStr[2] = 'b'; } } } BOOL CSimpleFile::Open( LPCTSTR lpszFileName, UINT nOpenFlags ) { BOOL bRet = FALSE; if(lpszFileName!=NULL) { ConvertFlag(nOpenFlags); try { m_pFile = _tfopen(lpszFileName,m_pOpenStr); if(m_pFile != NULL) { bRet = TRUE; TRACE("打开文件：%s成功\n",lpszFileName); } } catch (...) { TRACE("打开文件:%s失败\n",lpszFileName); } } return bRet; } size_t CSimpleFile::Write( const void* lpBuf, size_t nCount ) { if(m_pFile==NULL) { return -1; } fwrite( lpBuf, sizeof(TCHAR), nCount, m_pFile ); return fflush(m_pFile); } void CSimpleFile::Close() { if(fclose(m_pFile)) { TRACE("关闭文件失败\n"); } } int CSimpleFile::Seek( long lOff, UINT nFrom ) { UINT uFrom = 0; switch(nFrom) { case BEGIN: uFrom = SEEK_SET; break; case END: uFrom = SEEK_END; break; case CURRENT: uFrom = SEEK_CUR; break; default: uFrom = nFrom; } return fseek( m_pFile, lOff, uFrom ); } long CSimpleFile::GetLength() { long fLength = filelength(fileno(m_pFile)); return fLength; } void CSimpleFile::SeekToBegin() { Seek(0,SEEK_SET); } void CSimpleFile::SeekToEnd() { Seek(0,SEEK_END); } size_t CSimpleFile::Read( void* lpBuf, size_t nCount ) { return fread(lpBuf,sizeof(TCHAR),nCount,m_pFile); } int PASCAL CSimpleFile::Remove( LPCTSTR lpszFileName ) { return _tremove(lpszFileName); } long CSimpleFile::GetPosition() { return ftell(m_pFile); } /************************************************************************/ /*CCreateDirectory */ /************************************************************************/ CCreateDirectory::CCreateDirectory(void) { } CCreateDirectory::~CCreateDirectory(void) { } BOOL CCreateDirectory::GetPathName(LPCTSTR lpPathName) { if(lpPathName==NULL) { return false; } TRACE("目录长度：%d",strlen(lpPathName)); LPCTSTR lpPath = lpPathName; while(*lpPath!='\0') { string sFolder; while((*lpPath!='\\')&&(*lpPath!='\0')) { sFolder.push_back(*lpPath); lpPath++; } m_vFolder.push_back(sFolder); if(*lpPath!='\0') { lpPath++; } } return true; } BOOL CCreateDirectory::CreateDirectory( LPCTSTR lpPathName, LPSECURITY_ATTRIBUTES lpSecurityAttributes ) { if(lpPathName==NULL) { return false; } if(!GetPathName(lpPathName)) { return false; } string path; for(int i=0; i<(int)m_vFolder.size(); i++) { if(m_vFolder[i]!=_T("")) { path = path + m_vFolder[i] + _T("\\"); ::CreateDirectory(path.data(),lpSecurityAttributes); } } TRACE("创建目录：%s",lpPathName); m_vFolder.clear(); return true; } /************************************************************************/ /*CSyncLock类的实现 */ /************************************************************************/ CSyncLock::CSyncLock(void) { m_hLock = NULL; } CSyncLock::CSyncLock( LPSECURITY_ATTRIBUTES lpMutexAttributes, BOOL bInitialOwner, LPCTSTR lpName ) { m_hLock = CreateMutex( lpMutexAttributes, // no security attributes bInitialOwner, // initially not owned lpName); // name of mutex if (m_hLock == NULL) { TRACE("CreateMutex失败\n"); } } CSyncLock::~CSyncLock(void) { CloseHandle(m_hLock); } BOOL CSyncLock::Lock( DWORD dwTimeOut ) { BOOL bRet = FALSE; if(m_hLock!=NULL) { if(WaitForSingleObject(m_hLock,dwTimeOut) == WAIT_OBJECT_0) { bRet = TRUE; } } return bRet; } BOOL CSyncLock::Unlock( ) { return ReleaseMutex(m_hLock); } /************************************************************************/ /* BufferPool */ /************************************************************************/ BufferPool::BufferPool(void) { SYSTEM_INFO sInf; GetSystemInfo(&sInf); m_dwPageSize = sInf.dwPageSize; m_nParallel = 0; m_nNumOfBufs = 0; m_nNumOfBufUsed = NULL; m_bRowPosition = NULL; m_nBlockSize = NULL; m_pRowsEvent = NULL; m_pBufRowAddress = NULL; } BufferPool::~BufferPool(void) { BufferFree(); } void BufferPool::DisplayErrorMsg() { LPVOID lpMsgBuf; FormatMessage( FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM | FORMAT_MESSAGE_IGNORE_INSERTS, NULL, GetLastError(), MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT), // Default language (LPTSTR) &lpMsgBuf, 0, NULL ); TRACE("%s\n",(LPTSTR) lpMsgBuf); AfxMessageBox((LPTSTR) lpMsgBuf); LocalFree(lpMsgBuf); } BOOL BufferPool::BufferAlloc(int nNumOfBuf,int nNumOfParallel) { if (nNumOfBuf<1) { #ifdef BUFFER_POOL_DEBUG TRACE("nNumOfBuf<1\n"); #endif return FALSE; } if ((nNumOfBuf%nNumOfParallel)!=0) { #ifdef BUFFER_POOL_DEBUG TRACE("(nNumOfBuf%nNumOfParallel)!=0\n"); #endif return FALSE; } m_nNumOfRowBufs = nNumOfBuf/nNumOfParallel; m_nNumOfBufs = nNumOfBuf; m_nParallel = nNumOfParallel; m_pStartAddress =(PBYTE) VirtualAlloc( //保留虚拟内存 NULL, nNumOfBuf*m_dwPageSize, /*MEM_RESERVE*/MEM_COMMIT|MEM_TOP_DOWN,PAGE_READWRITE ); if (m_pStartAddress==NULL) { #ifdef BUFFER_POOL_DEBUG TRACE("VirtualAlloc Error\n"); #endif return FALSE; } #ifdef BUFFER_POOL_DEBUG TRACE("VirtualAlloc Success\n"); #endif try { m_nNumOfBufUsed = new int[nNumOfParallel]; m_bRowPosition = new std::vector <bool>[nNumOfParallel]; m_nBlockSize = new std::vector <int>[nNumOfParallel]; m_pBufRowAddress = new PBYTE[nNumOfParallel]; m_pRowsEvent = new HANDLE[nNumOfParallel]; for(int i = 0; i<nNumOfParallel; i++) { m_nNumOfBufUsed[i] = 0; m_pBufRowAddress[i] = m_pStartAddress + i*(m_nNumOfRowBufs)*m_dwPageSize; m_bRowPosition[i]