完成端口的网络库（vc源代码）

#include "stdafx.h" #include "SocketServer.h" #include "winsock2.h" #include "afxmt.h" signed Threads; CServer::CServer() { UserData=NULL; ZeroMemory(&acceptInfo,sizeof(acceptInfo)); acceptInfo.hSocket=INVALID_SOCKET; acceptInfo.bOper=&Oper; WSADATA wData; WSAStartup(MAKEWORD(2,2),&wData); Event.ResetEvent; bTerminate=false; hThread=CreateThread(NULL,0,&SendThreadProc,this,0,&lpThread); } CServer::~CServer() { bTerminate=true; DWORD ExitCode; GetExitCodeThread(hThread,&ExitCode); TerminateThread(hThread,ExitCode); CloseHandle(hThread); WSACleanup(); } CServer::SetRecvEvent(OnRecv eRecv) { acceptInfo.RecvEvent=eRecv; } SafeClose(POverlappedEx oi) { if (oi!=NULL) { closesocket(oi->hSocket); delete oi->Worker; //CloseHandle(oi->hCompletion); oi->AInfo->bCManager->cm_Delete(oi->hSocket); } return 0; } DWORD WINAPI ServerWorkThread(LPVOID Param) { PAcceptInfo acceptInfo=(PAcceptInfo)Param; CBOper *MemManager=acceptInfo->bOper; HANDLE hCompletion=acceptInfo->hCompletion; DWORD TransferBytes; DWORD CompletionKey; POverlappedEx oi=NULL; while (true) { //maxs 2005.6.23 /* if(acceptInfo->bStop==true) return 0; */ BOOL hRet=GetQueuedCompletionStatus(hCompletion,&TransferBytes,&CompletionKey,(LPOVERLAPPED*)&oi,INFINITE); if (CompletionKey==0 && TransferBytes==0 && oi==NULL) { break; } if (!hRet) { int Error=WSAGetLastError(); if (oi!=NULL) { SafeClose(oi); MemManager->UnLockMemory((LPVOID)oi); } continue; } if (TransferBytes==0) { if (oi!=NULL) { SafeClose(oi); MemManager->UnLockMemory((LPVOID)oi); } continue; } if (oi==NULL) { continue; } switch (CompletionKey) { case SOCKET_RECV: __try { oi->Worker->Write(oi->buf,TransferBytes); } __finally { Loop: WSABUF Buf; Buf.buf=oi->buf; Buf.len=SocketBuffSize; DWORD RecvBytes=0; DWORD Flags=0; hRet=WSARecv(oi->hSocket,&Buf,1,&RecvBytes,&Flags,(LPWSAOVERLAPPED)oi,NULL); if (hRet==SOCKET_ERROR) { int Error=WSAGetLastError(); if (Error==WSA_IO_PENDING) { continue; } else { SafeClose(oi); MemManager->UnLockMemory((LPVOID)oi); } } /* else { oi->Worker->Write(oi->buf,RecvBytes); goto Loop; } */ break; } } } return 0; } DWORD WINAPI ServerAcceptThread(LPVOID Param) { CServer *Server=(CServer *)Param; CBOper bOper; PAcceptInfo acceptInfo=&Server->acceptInfo; DWORD BuffSize=sizeof(OverlappedEx); bOper.InitMemorys(10,BuffSize,0); acceptInfo->bOper=&bOper; acceptInfo->bCManager=&Server->bCManager; // maxs 2005.6.23 // signed Threads; SYSTEM_INFO sysInfo; GetSystemInfo(&sysInfo); Threads=sysInfo.dwNumberOfProcessors*2+1; acceptInfo->WorkhThread=new HANDLE[Threads]; acceptInfo->WorkTidhread=new DWORD[Threads]; for (int i=0;i<Threads;i++) { acceptInfo->WorkhThread[i]=CreateThread(NULL,0,&ServerWorkThread,acceptInfo,0,&acceptInfo->WorkTidhread[i]); } while (true) { SOCKET hAcceptSocket=accept(acceptInfo->hSocket,NULL,NULL); if (hAcceptSocket==SOCKET_ERROR) { break; } CreateIoCompletionPort((HANDLE)hAcceptSocket,acceptInfo->hCompletion,SOCKET_RECV,0); sockaddr_in Addr; int Len; Len=sizeof(Addr); getpeername(hAcceptSocket,(sockaddr *)&Addr,&Len); acceptInfo->bCManager->cm_Add(inet_ntoa(Addr.sin_addr),"",hAcceptSocket); POverlappedEx oi=(POverlappedEx)bOper.LockMemory(0); if (oi==NULL) { closesocket(hAcceptSocket); continue; } ZeroMemory(oi,sizeof(OverlappedEx)); WSABUF Buf; oi->Worker=new CDynamicManager(Server->UserData); oi->Worker->RecvEvent=acceptInfo->RecvEvent; oi->hSocket=hAcceptSocket; oi->AInfo=acceptInfo; Buf.buf=oi->buf; Buf.len=SocketBuffSize; //WSA_IO_PENDING DWORD RecvBytes=0; DWORD Flags=0; int hRet=WSARecv(hAcceptSocket,&Buf,1,&RecvBytes,&Flags,(LPWSAOVERLAPPED)oi,NULL); if (hRet==SOCKET_ERROR) { int Error=WSAGetLastError(); if (Error!=WSA_IO_PENDING) { //maxs 2005.6.23 SafeClose(oi); acceptInfo->bOper->UnLockMemory((LPVOID)oi); } } } // Sleep(1000); for(i=0;i<Threads;i++) { PostQueuedCompletionStatus(acceptInfo->hCompletion,0,0,NULL); CloseHandle(acceptInfo->WorkhThread[i]); } //maxs 2005.6.23 delete[] acceptInfo->WorkhThread; delete[] acceptInfo->WorkTidhread; CloseHandle(acceptInfo->hCompletion); return 0; } DWORD WINAPI SendThreadProc(LPVOID Param) { CServer *Server=(CServer *)Param; while (!Server->bTerminate) { WaitForSingleObject(Server->Event.m_hObject,INFINITE); PSendInfo Info=NULL; Server->GetSend(&Info); if (Info!=NULL) { int xlen; xlen=Info->len; Server->bCManager.cm_Send(Info->ip,Info->buf,&xlen); GlobalFree(Info->buf); GlobalFree(Info); } } return 0; } CServer::GetSend(PSendInfo *Info) { Lock.Lock(); __try { if (Sends.GetCount()) { *Info=(PSendInfo)Sends.GetHead(); Sends.RemoveHead(); } else { Info=NULL; Event.ResetEvent(); } } __finally { Lock.Unlock(); } } CServer::AddSend(char *ip,char *buf,int *len) { Lock.Lock(); __try { PSendInfo Info; Info=(PSendInfo)GlobalAlloc(GPTR,sizeof(SendInfo)); Info->buf=(char *)GlobalAlloc(GPTR,*len); strcpy(Info->ip,ip); CopyMemory(Info->buf,buf,*len); Info->len=*len; Sends.AddTail(Info); Event.SetEvent(); } __finally { Lock.Unlock(); } } int CServer::Listen() { acceptInfo.bStop=false; acceptInfo.hSocket=WSASocket(AF_INET,SOCK_STREAM,0,NULL,0,WSA_FLAG_OVERLAPPED); if (acceptInfo.hSocket==INVALID_SOCKET) { int Error=WSAGetLastError(); return -1; } SOCKADDR_IN Addr; ZeroMemory(&Addr,sizeof(Addr)); Addr.sin_family=AF_INET; Addr.sin_port=htons(Port); Addr.sin_addr.s_addr=htonl(ADDR_ANY); int hRet=bind(acceptInfo.hSocket,(sockaddr *)&Addr,sizeof(Addr)); if (hRet==SOCKET_ERROR) { return -2; } hRet=listen(acceptInfo.hSocket,SOMAXCONN); if (hRet==SOCKET_ERROR) { return -3; } acceptInfo.hCompletion=CreateIoCompletionPort(INVALID_HANDLE_VALUE,0,0,0); acceptInfo.hThread=CreateThread(NULL,0,ServerAcceptThread,(LPVOID)this,0,&(acceptInfo.idThread)); return 0; } CServer::Close() { acceptInfo.bStop=true; if (acceptInfo.hSocket!=INVALID_SOCKET) { closesocket(acceptInfo.hSocket); CloseHandle(acceptInfo.hThread); acceptInfo.hSocket=NULL; } return 0; }