netengine.cpp

采用完成端口的client/server DLL源码

#include "inetengine.h"
#include "netengine.h"

NetEngineServer* NetEngineServer::m_pMe = NULL;

NetEngineServer::NetEngineServer( MsgFun a_pFun, //消息处理回调函数指针
								  DWORD  a_dwPort //端口
								 )
{
	if( a_pFun ){
		m_pMsgProcessFun = a_pFun;
	}
	else{
		MessageBox( NULL,"a_pFun==NULL","NetEngineServer",NULL );
	}
	m_pMe		= this;

	//初始化WS2.0网络
	WSADATA		t_wsaData;
	WSAStartup(MAKEWORD(2,2),&t_wsaData);

	//初始化网络系统
	this->InitServer( a_dwPort );
}

NetEngineServer::~NetEngineServer( void )
{
	//关闭监听套接字
	closesocket( m_Listen );
	//关闭重叠端口句柄
	CloseHandle( m_hCompletionPort );
	WSACleanup();
}

DWORD WINAPI NetEngineServer::ThreadFunc( LPVOID a_pVoid //附加信息 
							   )
{
	BOOL			t_bResult;
	DWORD			t_dwTranfByte = 0;
	LPOVERLAPPED	t_lpOverlapped;
	KEY*			t_pKey;
	HANDLE			t_hComplete = NetEngineServer::Instance()->GetHandle();

	//进入循环,捕获消息
	for(;;)
	{
		//获取IO状态,阻塞
		t_bResult = GetQueuedCompletionStatus(
			t_hComplete,
			&t_dwTranfByte,
			(LPDWORD)&t_pKey,
			&t_lpOverlapped,
			INFINITE
            );
		if( t_bResult && t_pKey->s_bIsRecv && t_dwTranfByte > 0 ) //接受到消息
		{
			NetEngineServer::Instance()->MsgProcess( t_pKey,NET_MSG_RECV );
			//投递recv
			NetEngineServer::Instance()->PostRecv( t_pKey );
		}		
		else if( t_bResult == false && t_lpOverlapped != NULL )//连接已经断
		{	
			NetEngineServer::Instance()->MsgProcess( t_pKey,NET_MSG_CLOSE );
			NetEngineServer::Instance()->CloseOneSocket( t_pKey );
		}
		else if( t_bResult == false && t_lpOverlapped == NULL )//发生错误
		{
			MessageBox( NULL,"连接发生错误","ThreadFunc",NULL );
			NetEngineServer::Instance()->CloseOneSocket( t_pKey );
		}
	}
	return 0;
}

NetEngineServer* NetEngineServer::Instance( MsgFun a_pFun//消息处理回调函数指针 
										   )
{
	if( m_pMe == NULL ){
		new NetEngineServer( a_pFun );
	}
	return m_pMe;
}

HANDLE NetEngineServer::GetHandle( void )
{
	return m_hCompletionPort;
}

bool NetEngineServer::Listen( void )
{
	KEY*	t_pKey;
	SOCKET	t_AccepteSocket;
	//进入循环,捕获ACCEPT请求
	for(;;)
	{
		//接受新套接字
        if((t_AccepteSocket = WSAAccept(m_Listen, NULL, NULL, NULL, 0))==SOCKET_ERROR )
		{
			MessageBox( NULL,"WSAAccept接受新套接字错误","Listen",NULL );
			return false;
		}

		//设置缓冲区为0
		int t_iZero = 0;
		//if(setsockopt(t_AccepteSocket, SOL_SOCKET,SO_SNDBUF,(char*)&t_iZero,sizeof(t_iZero))==SOCKET_ERROR)
		{
			//MessageBox(NULL,"setsockopt","设置缓冲区错误",NULL);
		}
		//if(setsockopt(t_AccepteSocket, SOL_SOCKET,SO_RCVBUF,(char*)&t_iZero,sizeof(t_iZero))==SOCKET_ERROR)
		{
			//MessageBox(NULL,"setsockopt","设置缓冲区错误",NULL);
		}
		t_pKey = new KEY;
		ZeroMemory(&(t_pKey->s_sRecvOv), sizeof(OVERLAPPED));
		ZeroMemory(&(t_pKey->s_sSendOv), sizeof(OVERLAPPED));
		
		//设置WSASEND后不通知完成端口
		t_pKey->s_sSendOv.hEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
		t_pKey->s_sSendOv.hEvent = (HANDLE)((DWORD)t_pKey->s_sSendOv.hEvent | 0x1);
		t_pKey->s_hSocket = t_AccepteSocket;

		//重叠端口关联
		if( CreateIoCompletionPort(
			(HANDLE)t_AccepteSocket,
			m_hCompletionPort,
			(DWORD)t_pKey,   // No key
			0              // Use default # of threads
            )== NULL )
		{
			MessageBox( NULL,"CreateIoCompletionPort错误","Listen",NULL );
			return false;
		}		

		//投递RECV操作
		t_pKey->s_sRecvBuf.len	= DATA_BUFSIZE;
		t_pKey->s_sRecvBuf.buf	= t_pKey->s_cRBuffer;
		t_pKey->s_sSendBuf.buf  = t_pKey->s_cSBuffer;
		
		this->PostRecv( t_pKey );
	}
	return true;
}

/*
DWORD NetEngineServer::GetNewID( void )
{
	if( m_dwCurID > 9999 ){
		m_dwCurID = 1000;
	}
	
	bool			t_bFind = false;
	CListDWPointer	t_pDW;
	do{
		for( t_pDW=m_cIDList.begin();t_pDW!=m_cIDList.end();t_pDW++ )
		{
			if( *t_pDW == m_dwCurID ){
				t_bFind	= true;
				m_dwCurID++;
				break;
			}
			t_bFind = false;
		}
	}while( t_bFind );
	//插入ID缓冲池
	m_cIDList.insert( m_cIDList.end(),m_dwCurID ); 
	m_dwCurID++;
	return m_dwCurID;
}*/


bool NetEngineServer::CreateWorkerThread( void )
{
	SYSTEM_INFO  t_Sysinfo;
    DWORD        t_dwThreadId;
    DWORD        t_dwThreads;
    DWORD        i;
	
    GetSystemInfo(&t_Sysinfo);
    t_dwThreads = t_Sysinfo.dwNumberOfProcessors * 2;
    for (i=0; i<t_dwThreads; i++)
    {
        HANDLE t_hThread;
        t_hThread = CreateThread(
            NULL, 0, NetEngineServer::ThreadFunc, NULL, 0, &t_dwThreadId);
           //  m_hCompletionPort,0, &t_dwThreadId);
        CloseHandle(t_hThread);
    }
	return true;
}

bool NetEngineServer::PostRecv( KEY* a_pKey )
{
	DWORD t_dwRev,t_dwFlag	= 0;
	a_pKey->s_bIsRecv		= true;
	int i					= 0;
	int err;

	//清空缓冲区
	ZeroMemory( a_pKey->s_cRBuffer,DATA_BUFSIZE );
	ZeroMemory(&(a_pKey->s_sRecvOv), sizeof(OVERLAPPED));
	
	//投递接受消息
	while(++i)
	{
		err = WSARecv( a_pKey->s_hSocket,&a_pKey->s_sRecvBuf,1,&t_dwRev,&t_dwFlag,&a_pKey->s_sRecvOv,NULL );
		if( err != SOCKET_ERROR )
		{
			return true;
		}
		err = WSAGetLastError();
		if(  err == ERROR_IO_PENDING )
		{
			return true;
		}
		if( err == ERROR_INVALID_USER_BUFFER ||
			err == ERROR_NOT_ENOUGH_QUOTA ||
			err == ERROR_NOT_ENOUGH_MEMORY )
		{
			if (i == 5)
			{
				Sleep(50);
				continue;
			}
			MessageBox( NULL,"WSARecv错误","PostRecv",NULL );
			return false;
		}
		break;
	}
	return false;
}

bool NetEngineServer::PostSend( KEY* a_pKey,void* a_pBuffer,DWORD a_dwLen )
{
	DWORD			t_dwByte,t_dwFlag = 0;
	
	//填充发送缓冲区
	memcpy( a_pKey->s_cSBuffer,&a_dwLen,sizeof(DWORD) );
	memcpy( a_pKey->s_cSBuffer+sizeof(DWORD),a_pBuffer,a_dwLen );

	//设置发送SIZE
	a_pKey->s_sSendBuf.len	= a_dwLen + sizeof(DWORD);
	a_pKey->s_bIsRecv		= false;

	/////////////////////
	//ZeroMemory(&(a_pKey->s_sSendOv), sizeof(OVERLAPPED));

	//a_pKey->s_sSendOv.hEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
	//a_pKey->s_sSendOv.hEvent = (HANDLE)((DWORD)a_pKey->s_sSendOv.hEvent | 0x1);
	/////////////////////////
	int err = WSASend(a_pKey->s_hSocket,&a_pKey->s_sSendBuf,1,&t_dwByte,t_dwFlag,&a_pKey->s_sSendOv,NULL );	

	if( err != SOCKET_ERROR )
	{
		return true;
	}
	if( WSAGetLastError() == ERROR_IO_PENDING )
	{
		return true;
	}
	else
	{
		MessageBox( NULL,"WSASend错误","PostSend",NULL );
		return false;
	}
}

bool NetEngineServer::InitServer( DWORD a_dwPort )
{
	sockaddr_in		t_ServerAddress;
	m_Listen = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0,WSA_FLAG_OVERLAPPED);
	
	//地址初始化
	memset(&t_ServerAddress, 0, sizeof(t_ServerAddress));
    t_ServerAddress.sin_family      = AF_INET;
    t_ServerAddress.sin_addr.s_addr = htonl(INADDR_ANY);
    t_ServerAddress.sin_port        = htons(a_dwPort);
	
	//捆绑
	bind(m_Listen,(struct sockaddr *)&t_ServerAddress,sizeof(t_ServerAddress));
	
	//创建IO端口
	m_hCompletionPort = CreateIoCompletionPort(
		INVALID_HANDLE_VALUE,
		NULL,   // No prior port
		0,      // No key
		0       // Use default  # of threads
        );
	if( m_hCompletionPort == NULL )
	{
		MessageBox( NULL,"CreateIoCompletionPort错误","InitServer",NULL );
		return false;
	}
	
	//创建工作线程
	this->CreateWorkerThread();
	
	//监听
	listen( m_Listen, 5 );
	return true;
}

bool NetEngineServer::MsgProcess( KEY* a_pKey,NET_MSG a_eMsg )
{
	if( a_pKey )
	{
		return m_pMsgProcessFun( a_pKey,a_eMsg );
	}
	return false;
}

bool NetEngineServer::CloseOneSocket( KEY* a_pKey )
{
	closesocket( a_pKey->s_hSocket );
	if( a_pKey ){
		delete a_pKey;
	}
	return true;
}