/**********************************************
主动式RFID服务器软件！
Windows IOCP（完成端口）对象实现
-------------------------------------------
1. ZhouChang@20080809
文件创建
2. ZhouChang@20080912
功能简化重构
**********************************************/
#include <stdio.h>
#include "winiocp.h"
WinIOCP::WinIOCP()
{
//变量初始化
hListenEventThread = INVALID_HANDLE_VALUE;
IOWorkerThreadHandle = INVALID_HANDLE_VALUE;
//停止服务同步对象
bIOShutdown = true;
hShutdownEvent = INVALID_HANDLE_VALUE;
//服务器Socket
ServerSocket = INVALID_SOCKET;
//IO完成端口
m_hCompletionPort = INVALID_HANDLE_VALUE;
//活动数据统计变量
m_NumberOfActiveConnections = 0;
m_iNumberOfPendlingReads = 0;
//初始化服务器状态
m_ServerStat = IOCPSERVER_CLEANED;
}
WinIOCP::~WinIOCP()
{
if ( m_ServerStat == IOCPSERVER_INITED )
{
CleanIOCP();
}
else if ( m_ServerStat == IOCPSERVER_STARTED)
{
ShutdownIOCPServer();
CleanIOCP();
}
}
bool WinIOCP::InitIOCP(unsigned short port)
{
//构造IOCP服务器框架
int retVal = -1;
//启动Winsock服务
WSADATA wsaData;
ZeroMemory(&wsaData,sizeof(WSADATA));
if( (retVal = WSAStartup(MAKEWORD(2,2), &wsaData)) != 0 )
{
cout << "WSAStartup Failed::Reason Code::"<< retVal << " @" << __FILE__ << ":" << __LINE__ << endl;
return false;
}
//查询工作者线程数目
SYSTEM_INFO sysInfo;
ZeroMemory(&sysInfo,sizeof(SYSTEM_INFO));
GetSystemInfo(&sysInfo);
#if 0 //为以后高性能版本扩展用，XXX
IOThreadsNumber = 1;
#endif
m_iNumberOfPendlingReads = 2; //始终保持一个异步读操作在Socket上读取数据,注意对于IO数据大适合采用较大的数据XXX
//初始化同步对象
InitializeCriticalSection(&m_ClientMapLock);
hShutdownEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
bIOShutdown = false;
//服务器Socket对象建立并且完成bind，listen等相关工作
ServerSocket = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);
if (INVALID_SOCKET == ServerSocket)
{
cout << "WSASocket Failed::Reason Code::"<< WSAGetLastError() << " @" << __FILE__ << ":" << __LINE__ << endl;
return false;
}
const int one = 1; //设置重复绑定，防止上次端口被占用的情况出现
if (setsockopt(ServerSocket, SOL_SOCKET, SO_REUSEADDR, (char *)&one, sizeof(one)) < 0)
{
return false;
}
struct sockaddr_in ServerAddress;
ZeroMemory((char *)&ServerAddress, sizeof(ServerAddress));
ServerAddress.sin_family = AF_INET;
ServerAddress.sin_addr.s_addr = INADDR_ANY;
ServerAddress.sin_port = htons(port);
if (SOCKET_ERROR == bind(ServerSocket, (struct sockaddr *) &ServerAddress, sizeof(ServerAddress)))
{
cout << "bind Failed::Reason Code::"<< WSAGetLastError() << " @" << __FILE__ << ":" << __LINE__ << endl;
return false;
}
//Accept对象事件
ServerSocketEvent = WSACreateEvent();
retVal = WSAEventSelect(ServerSocket,ServerSocketEvent,FD_ACCEPT);
if (retVal != 0)
{
cout << "::WSAEventSelect Failed::Reason Code::"<< WSAGetLastError() << " @" << __FILE__ << ":" << __LINE__ << endl;
return false;
}
//开启监听端口
retVal = listen(ServerSocket, 10);
if( retVal == SOCKET_ERROR )
{
cout << "Server Socket Listen Failed::Reason Code::"<< WSAGetLastError() << " @" << __FILE__ << ":" << __LINE__ << endl;
return false;
}
//初始化IOCP对象, 目前单一IO线程安全版本 XXX
//m_hCompletionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE,NULL,0,IOThreadsNumber);
m_hCompletionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE,NULL,0,1);
if (NULL == m_hCompletionPort)
{
cout << "CreateIoCompletionPort() Failed::Reason::"<< GetLastError() << " @" << __FILE__ << ":" << __LINE__ << endl;
return false;
}
m_ServerStat = IOCPSERVER_INITED;
return true;
}
//释放InitIOCP函数所创建的所有资源
void WinIOCP::CleanIOCP()
{
if ( m_ServerStat != IOCPSERVER_INITED)
{
return;
}
//释放IOCP对象
CloseHandle(m_hCompletionPort);
//释放ServerSocket资源
if ( SOCKET_ERROR != ServerSocket)
{
WSACloseEvent(ServerSocketEvent);
closesocket(ServerSocket);
}
//释放同步资源
DeleteCriticalSection(&m_ClientMapLock);
CloseHandle(hShutdownEvent);
//关闭Winsock库
WSACleanup();
m_ServerStat = IOCPSERVER_CLEANED;
}
bool WinIOCP::StartIOCPServer()
{
DWORD dwThreadId;
//开启Listen,Accept对应的线程
hListenEventThread = CreateThread(NULL,0,ListenEventThread, this, 0, &dwThreadId);
//开启IO工作线程，注意本版本支持单一IO线程
IOWorkerThreadHandle = CreateThread(NULL, 0, IOWorkerThread, (void *)this, 0, &dwThreadId);
m_ServerStat = IOCPSERVER_STARTED;
return true;
}
bool WinIOCP::ShutdownIOCPServer()
{
if ( m_ServerStat != IOCPSERVER_STARTED)
return false;
//0. 首先关闭监听端口线程
SetEvent(hShutdownEvent);
//Let Accept thread go down
WaitForSingleObject(hListenEventThread, INFINITE);
//关闭服务器对应socket对象
closesocket( ServerSocket );
//1. 停止IO线程
DisconnectAll();
//退出所有工作者线程
ShutDownIOWorker();
//2. 关闭相关资源
//关闭IOCP对象
CloseHandle(m_hCompletionPort);
//关闭所有存在的Client，注意Buffer应该在DisconnectAll中处理掉了
FreeAllClients();
m_ServerStat = IOCPSERVER_INITED;
return true;
}
/////////////////////////////////////////////////////////
void WinIOCP::ShutDownIOWorker()
{
DWORD dwExitCode;
bool bIOWorkersRunning = true;
//设置退出事件
if ( bIOShutdown == true)
{
return;
}
bIOShutdown = true;
while(bIOWorkersRunning)
{
// Send Empty Message into CompletionPort so that the threads die.
if(bIOWorkersRunning)
PostQueuedCompletionStatus(m_hCompletionPort, 0, (DWORD) NULL, NULL);
bIOWorkersRunning = false;
if(GetExitCodeThread (IOWorkerThreadHandle, &dwExitCode) && dwExitCode == STILL_ACTIVE)
{
bIOWorkersRunning = true;
continue;
}
}
}
void WinIOCP::AcceptNewClient()
{
sockaddr_in ClientAddress;
int nClientLength = sizeof(ClientAddress);
IOCPClient* pClient = NULL;
SOCKET RemoteSocket = WSAAccept(ServerSocket, (sockaddr*)&ClientAddress, &nClientLength, 0, 0);
if (INVALID_SOCKET == RemoteSocket)
{
cout << "WSAAccept Failed::Reason Code::"<< GetLastError() << " @" << __FILE__ << ":" << __LINE__ << endl;
return;
}
else
{
//分配新的Client对象
pClient = freeClientPool.NewIOCPClient();
if(NULL != pClient)
{
pClient->m_Socket = RemoteSocket;
pClient->m_RemoteAddress = ClientAddress;
pClient->m_ID = RemoteSocket;
//设置SOCKET对象
const char chOpt = 1;
int nErr = setsockopt(RemoteSocket, IPPROTO_TCP, TCP_NODELAY, &chOpt, sizeof(char));
if (nErr == -1)
{
cout << "setsockopt Failed@" << __LINE__ << " Reason Code::"<< GetLastError() << " @" << __FILE__ << ":" << __LINE__ << endl;
goto AcceptNewClient_error;
}
if(AddIOCPClient(pClient))
{
if (!AssociateSocketWitm_hCompletionPort(RemoteSocket, m_hCompletionPort, (DWORD) pClient))
{
//cout << "AssociateSocketWitm_hCompletionPort Failed@" << __LINE__ << " Reason Code::"<< GetLastError() << endl;
DisconnectClient(pClient);
ReleaseClient(pClient);
return;
}
}
IOCPBuffer *pOverlapBuff = freeBufferPool.NewIOCPBuffer();
if(pOverlapBuff != NULL)
{
//触发一次IO完成操作提示，操作类型为IOInitialize
pOverlapBuff->SetOperation(IOInitialize);
BOOL bSuccess = PostQueuedCompletionStatus(m_hCompletionPort, 0, (DWORD) pClient, &pOverlapBuff->m_ol);
if ( (!bSuccess && GetLastError( ) != ERROR_IO_PENDING))
{
ReleaseBuffer(pOverlapBuff);
DisconnectClient(pClient);
cout << "IOInitialize AcceptIncomingClient Error@" << __LINE__ << endl;
ReleaseClient(pClient);
return;
}
}
else
{
cout << "ERROR @ " << __LINE__ << "Could not allocate memory for buffer in Acceptincoming: %d " << WSAGetLastError() << " @" << __FILE__ <<
