333 lines
9.9 KiB
C++
333 lines
9.9 KiB
C++
#ifdef _WIN32
|
|
#include "IOCmpltThread.h"
|
|
#include "debug.h"
|
|
#include "MessageDef.h"
|
|
#include "MessageHandle.h"
|
|
|
|
HANDLE g_hCompletionPort;
|
|
|
|
CIOCmpltThread::CIOCmpltThread(unsigned int nStackSize)
|
|
:CThread(nStackSize), m_smProcess("mProcess",1)
|
|
{
|
|
m_hListenSocket = NULL;
|
|
g_hCompletionPort = NULL;
|
|
m_nWorkerThreadNumbers = 0;
|
|
|
|
for(int i=0;i<MAX_PROCESSOR_COUNTER;i++)
|
|
m_hWorkerThread[i] = NULL;
|
|
}
|
|
|
|
void CIOCmpltThread::Main()
|
|
{
|
|
SOCKADDR_IN InternetAddr;
|
|
SOCKET hAcceptSocket;
|
|
SYSTEM_INFO SystemInfo;
|
|
WSADATA wsaData;
|
|
DWORD Ret;
|
|
|
|
if ((Ret = WSAStartup(0x0202, &wsaData)) != 0)
|
|
{
|
|
DebugPrint("WSAStartup failed with error %d\n", Ret);
|
|
return;
|
|
}
|
|
|
|
// Setup an I/O completion port.
|
|
|
|
if ((g_hCompletionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0)) == NULL)
|
|
{
|
|
DebugPrint( "CreateIoCompletionPort failed with error: %d\n", GetLastError());
|
|
return;
|
|
}
|
|
|
|
// Determine how many processors are on the system.
|
|
|
|
GetSystemInfo(&SystemInfo);
|
|
|
|
// Create worker threads based on the number of processors available on the
|
|
// system. Create two worker threads for each processor.
|
|
|
|
m_nWorkerThreadNumbers = SystemInfo.dwNumberOfProcessors*2+2;
|
|
if( m_nWorkerThreadNumbers>MAX_PROCESSOR_COUNTER ){
|
|
m_nWorkerThreadNumbers = MAX_PROCESSOR_COUNTER;
|
|
}
|
|
|
|
for( int i=0; i<m_nWorkerThreadNumbers; i++ )
|
|
{
|
|
// Create a server worker thread and pass the completion port to the thread.
|
|
DWORD ThreadID;
|
|
m_hWorkerThread[i] = CreateThread(NULL, 0, ServerWorkerThread, this, 0, &ThreadID);
|
|
}
|
|
|
|
// Create a listening socket
|
|
|
|
if ((m_hListenSocket = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED)) == INVALID_SOCKET)
|
|
{
|
|
DebugPrint("WSASocket() failed with error %d\n", WSAGetLastError());
|
|
return;
|
|
}
|
|
|
|
InternetAddr.sin_family = AF_INET;
|
|
InternetAddr.sin_addr.s_addr = htonl(INADDR_ANY);
|
|
InternetAddr.sin_port = htons(PORT);
|
|
|
|
if (bind(m_hListenSocket, (PSOCKADDR) &InternetAddr, sizeof(InternetAddr)) == SOCKET_ERROR)
|
|
{
|
|
DebugPrint("bind() failed with error %d\n", WSAGetLastError());
|
|
return;
|
|
}
|
|
|
|
// Prepare socket for listening
|
|
|
|
if (listen(m_hListenSocket, 20) == SOCKET_ERROR)
|
|
{
|
|
DebugPrint("listen() failed with error %d\n", WSAGetLastError());
|
|
return;
|
|
}
|
|
|
|
// Accept connections and assign to the completion port.
|
|
|
|
while(!m_bExit){
|
|
struct sockaddr_in saClient;
|
|
int iClientSize = sizeof(saClient);
|
|
if ((hAcceptSocket = WSAAccept(m_hListenSocket, (SOCKADDR*) &saClient, &iClientSize, NULL, 0)) == SOCKET_ERROR){
|
|
DebugPrint("WSAAccept() failed with error %d\n", WSAGetLastError());
|
|
return;
|
|
}
|
|
|
|
// Create a socket information structure to associate with the socket
|
|
LPCONTEXT_KEY lpContextKey = (LPCONTEXT_KEY)GlobalAlloc(GPTR, sizeof(CONTEXT_KEY));
|
|
if( !lpContextKey ){
|
|
DebugPrint("GlobalAlloc() failed with error %d\n", GetLastError());
|
|
return;
|
|
}
|
|
|
|
// Associate the accepted socket with the original completion port.
|
|
ZeroMemory(&(lpContextKey->overlapped), sizeof(OVERLAPPED));
|
|
lpContextKey->peerIp = saClient.sin_addr.S_un.S_addr;
|
|
lpContextKey->peerPort = ntohs(saClient.sin_port);
|
|
lpContextKey->socket = hAcceptSocket;
|
|
lpContextKey->dataBuf.len = DATA_BUFSIZE;
|
|
lpContextKey->dataBuf.buf = lpContextKey->buffer;
|
|
lpContextKey->iOperation = COMPLETION_KEY_RECV;
|
|
lpContextKey->nTimeCount = 0;
|
|
|
|
if (CreateIoCompletionPort((HANDLE)hAcceptSocket, g_hCompletionPort, (DWORD)lpContextKey, 0) == NULL){
|
|
DebugPrint("CreateIoCompletionPort failed with error %d\n", GetLastError());
|
|
closesocket( lpContextKey->socket );
|
|
GlobalFree( lpContextKey );
|
|
}
|
|
else{
|
|
// Create per I/O socket information structure to associate with the
|
|
// WSARecv call below.
|
|
m_lstContextKeys.push_back( lpContextKey );
|
|
|
|
DWORD RecvBytes;
|
|
DWORD Flags;
|
|
Flags = 0;
|
|
if (WSARecv(hAcceptSocket, &(lpContextKey->dataBuf), 1, &RecvBytes, &Flags,
|
|
&(lpContextKey->overlapped), NULL) == SOCKET_ERROR){
|
|
if (WSAGetLastError() != ERROR_IO_PENDING){
|
|
DebugPrint("1 WSARecv() failed with error %d\n", WSAGetLastError());
|
|
FreeContextKey( lpContextKey );
|
|
continue;
|
|
}
|
|
}
|
|
|
|
|
|
}
|
|
}
|
|
}
|
|
|
|
void CIOCmpltThread::Terminate()
|
|
{
|
|
CThread::Terminate();
|
|
|
|
int i;
|
|
|
|
if(m_hListenSocket)
|
|
{
|
|
closesocket(m_hListenSocket);
|
|
m_hListenSocket = NULL;
|
|
}
|
|
|
|
for( i=0; i<m_nWorkerThreadNumbers; i++ ){
|
|
PostQueuedCompletionStatus(g_hCompletionPort,0,COMPLETION_KEY_SHUTDOWN,NULL);
|
|
}
|
|
|
|
//关闭线程句柄
|
|
for( i=0; i<m_nWorkerThreadNumbers; i++ ){
|
|
if(m_hWorkerThread[i]){
|
|
WaitForSingleObject( m_hWorkerThread[i], INFINITE );
|
|
CloseHandle(m_hWorkerThread[i]);
|
|
}
|
|
}
|
|
|
|
if(g_hCompletionPort)
|
|
{
|
|
CloseHandle(g_hCompletionPort);
|
|
g_hCompletionPort = NULL;
|
|
}
|
|
|
|
if( m_lstContextKeys.size()>0 ){
|
|
CONTEXTKEYLIST::iterator iter = m_lstContextKeys.begin();
|
|
while( iter!=m_lstContextKeys.end() ){
|
|
closesocket( (*iter)->socket );
|
|
GlobalFree( *iter );
|
|
|
|
iter++;
|
|
}
|
|
|
|
m_lstContextKeys.clear();
|
|
}
|
|
}
|
|
|
|
bool CIOCmpltThread::IsValidContextKey( LPCONTEXT_KEY lpContextKey )
|
|
{
|
|
CMutexLock lockProcess( &m_smProcess );
|
|
|
|
if( m_lstContextKeys.size()>0 ){
|
|
CONTEXTKEYLIST::iterator iter = m_lstContextKeys.begin();
|
|
while( iter!=m_lstContextKeys.end() ){
|
|
if( *iter==lpContextKey ){
|
|
return true;
|
|
}
|
|
iter++;
|
|
}
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
void CIOCmpltThread::FreeContextKey( LPCONTEXT_KEY lpContextKey )
|
|
{
|
|
CMutexLock lockProcess( &m_smProcess );
|
|
|
|
if( m_lstContextKeys.size()>0 ){
|
|
CONTEXTKEYLIST::iterator iter = m_lstContextKeys.begin();
|
|
while( iter!=m_lstContextKeys.end() ){
|
|
if( *iter==lpContextKey ){
|
|
m_lstContextKeys.erase( iter );
|
|
closesocket( lpContextKey->socket );
|
|
GlobalFree( lpContextKey );
|
|
|
|
//DebugPrint( "Free client %d .\n", lpContextKey->socket );
|
|
return;
|
|
}
|
|
iter++;
|
|
}
|
|
}
|
|
}
|
|
|
|
void CIOCmpltThread::OnTimer()
|
|
{
|
|
CMutexLock lockProcess( &m_smProcess );
|
|
|
|
if( m_lstContextKeys.size()>0 ){
|
|
CONTEXTKEYLIST::iterator iter = m_lstContextKeys.begin();
|
|
while( iter!=m_lstContextKeys.end() ){
|
|
if( (*iter)->iOperation==COMPLETION_KEY_RECV && ++(*iter)->nTimeCount>=60 ){ //超过1分钟未接收到数据
|
|
shutdown( (*iter)->socket, SD_BOTH );
|
|
DebugPrint( "Client %d shutdown by timeout.\n", (*iter)->socket );
|
|
closesocket( (*iter)->socket );
|
|
(*iter)->iOperation = COMPLETION_KEY_SHUTDOWN;
|
|
}
|
|
|
|
iter++;
|
|
}
|
|
}
|
|
}
|
|
|
|
DWORD WINAPI ServerWorkerThread(LPVOID lpVoid)
|
|
{
|
|
CIOCmpltThread* pThread = (CIOCmpltThread*)lpVoid;
|
|
DWORD dwBytesTransferred = 0;
|
|
LPOVERLAPPED lpOverlapped = NULL;
|
|
LPCONTEXT_KEY lpContextKey = NULL;
|
|
DWORD SendBytes, RecvBytes;
|
|
DWORD Flags;
|
|
BOOL bResult;
|
|
|
|
while(!pThread->IsTerminated()){
|
|
bResult = GetQueuedCompletionStatus(g_hCompletionPort, &dwBytesTransferred,
|
|
(LPDWORD)&lpContextKey, &lpOverlapped, INFINITE);
|
|
if( !bResult ){
|
|
int err = GetLastError();
|
|
DebugPrint("GetQueuedCompletionStatus failed with error %d\n", err);
|
|
|
|
//if( err!=995 ){//&& pThread->IsValidContextKey( lpContextKey ) ){
|
|
pThread->FreeContextKey( lpContextKey );
|
|
//}
|
|
}
|
|
else{
|
|
if( lpOverlapped==NULL && (DWORD)lpContextKey==COMPLETION_KEY_SHUTDOWN ){
|
|
DebugPrint("ServerWorkThread shutdown.\n" );
|
|
return 0;
|
|
}
|
|
|
|
if( lpOverlapped!=NULL && pThread->IsValidContextKey(lpContextKey) ){
|
|
ZeroMemory(&(lpContextKey->overlapped), sizeof(OVERLAPPED));
|
|
if( lpContextKey->iOperation==COMPLETION_KEY_RECV ){
|
|
if( dwBytesTransferred==0 ){
|
|
DebugPrint("Client shutdown.\n" );
|
|
pThread->FreeContextKey( lpContextKey );
|
|
}
|
|
else{
|
|
lpContextKey->nTimeCount = 0; //计数器清零
|
|
|
|
TNETMESSAGE msgRequest;
|
|
if( ReadNetMessage( (unsigned char*)lpContextKey->buffer, dwBytesTransferred, &msgRequest )){
|
|
if( msgRequest.eMessageType==e_versionRequest ){
|
|
char *pb=(char *)&(lpContextKey->peerIp);
|
|
unsigned b1=(unsigned char)(*pb);
|
|
unsigned b2=(unsigned char)(*(pb+1));
|
|
unsigned b3=(unsigned char)(*(pb+2));
|
|
unsigned b4=(unsigned char)(*(pb+3));
|
|
DebugPrint( "CONNECT %d.%d.%d.%d:%d type: TCP\n", b1, b2, b3, b4, lpContextKey->peerPort );
|
|
}
|
|
|
|
TNETMESSAGE msgResponse;
|
|
if( HandleRequest( &msgRequest, &msgResponse, lpContextKey->peerIp, lpContextKey->peerPort ) ){
|
|
int nSize = DATA_BUFSIZE;
|
|
if( WriteNetMessage( (unsigned char*)lpContextKey->buffer, nSize, &msgResponse ) ){
|
|
lpContextKey->iOperation=COMPLETION_KEY_SEND;
|
|
lpContextKey->dataBuf.len = nSize;
|
|
if( WSASend(lpContextKey->socket, &(lpContextKey->dataBuf), 1, &SendBytes, 0,
|
|
&(lpContextKey->overlapped), NULL) == SOCKET_ERROR){
|
|
if( WSAGetLastError() != ERROR_IO_PENDING){
|
|
DebugPrint("WSASend() failed with error %d\n", WSAGetLastError());
|
|
pThread->FreeContextKey( lpContextKey );
|
|
}
|
|
}
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
Flags = 0;
|
|
if (WSARecv(lpContextKey->socket, &(lpContextKey->dataBuf), 1, &RecvBytes, &Flags,
|
|
&(lpContextKey->overlapped), NULL) == SOCKET_ERROR) {
|
|
if( WSAGetLastError()!=ERROR_IO_PENDING ){
|
|
DebugPrint("2 WSARecv() failed with error %d\n", WSAGetLastError());
|
|
pThread->FreeContextKey( lpContextKey );
|
|
}
|
|
}
|
|
}
|
|
}
|
|
else if( lpContextKey->iOperation==COMPLETION_KEY_SEND ){
|
|
lpContextKey->dataBuf.len = DATA_BUFSIZE;
|
|
lpContextKey->iOperation = COMPLETION_KEY_RECV;
|
|
Flags = 0;
|
|
if (WSARecv(lpContextKey->socket, &(lpContextKey->dataBuf), 1, &RecvBytes, &Flags,
|
|
&(lpContextKey->overlapped), NULL) == SOCKET_ERROR) {
|
|
if( WSAGetLastError()!=ERROR_IO_PENDING ){
|
|
DebugPrint("3 WSARecv() failed with error %d\n", WSAGetLastError());
|
|
pThread->FreeContextKey( lpContextKey );
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
#endif |