完成端口

windows完成端口服务器的另一种实现 (2007-11-09 15:12)

分类： c/c++

几年前为了在windows下实现一个高效简洁的服务器,不得不使用wondows下的完成端口socket编程.

参考了很多的完成端口的代码,始终不是很满意,因此,自己根据相关的原理实现一个.现在公布如下,供需要学习完成端口的同仁们参考.

整个服务由监听线程,工作线程构成.监听线程负责监听连接的到来,工作线程负责通知当有接受数据的到来.

服务头文件:
#pragma once
#include <vector>
using namespace std;
//IO操作类型标志
enum _OP_FLAG
{
OP_NONE,
OP_ACCEPT,
OP_RECV,
OP_SEND
};
//完成键结构
typedef struct tag_TCOMPLETIONPORT
{
OVERLAPPED Overlapped;
WSABUF DataBuf;
CHAR Buffer[1024*400];
INT nOvFlag;
} _TCOMPLETIONPORT, * _LPTCOMPLETIONPORT;

//登录到服务器的客户信息
class CClientInfo
{
public:
// 重叠IO
OVERLAPPED Overlapped; //关联的重叠IO变量
WSABUF DataBuf; //IO的数据缓冲区
_OP_FLAG nOvFlag; //完成状态标志 (op_recv,op_send,op_accept)
SOCKET sock; //客户端套接字
unsigned long total_length; //数据包的总长度
unsigned long pack_len; //当前读取的长度
char RecvBuffer[1024*4]; //接收数据缓冲区
char DealBuffer[1024*4]; //一个通讯包的数据
char Buffer[1024*4]; //发送接收数据缓冲池

int nSendBufferLen; //发送数据大小
bool fIsLogin; //是否登录
int nUserIndex; //客户数据在链表的位置

time_t tConnectTime; //连接到服务器的时间
int sendSpeed; //发送短信的每秒最大速率，默认1条每秒
int LinkCheck; //发送连接检查数据未响应次数，如果超过3次未响应则断开连接
public:
CClientInfo(void)
{
sock=INVALID_SOCKET;
total_length=0;
pack_len=0;
}
~CClientInfo(void)
{
}

}
};

服务实现文件:


/****************************************************************************
///完成端口服务器
///编写人：秦柏林
///
///最后修改日期:2003-06-27
****************************************************************************/
#pragma once
#include <shlwapi.h>
#include <process.h>
#include "ServiceThread.h"


extern std::string LogFileName; //日志文件名

//做日志记录
extern void _stdcall PushLog(std::string LogMsg);
//判断连接IP是否有效
extern int _stdcall IsValidIP(CADOConnection& m_ADO,std::string IP);
//删除登录的客户端的记录
extern void DelLoginUser(CADOConnection& m_ADO,std::string UserID);
//将接收的数据包交给分析线程处理
extern int _stdcall DealRecvDataPacket(CClientInfo* pClient,CADOConnection& m_ADO);
//删除记录
extern void DeleteRecord(std::string OnlySign,std::string TableName,std::string KeyName,CADOConnection& m_ADO);
//数据重发
extern void ReplyClient(CClientInfo* pClient,int replyType,unsigned int lresult/*只对登录回复有用*/,std::string MsgAnswerID);
////////////////////////////////////////////////////////////////////////////////////////////////////////

HANDLE g_hIOCP=NULL; //完成端口句柄

int WaitTime=180; //最大等待客户发送登录数据时间，默认1分钟，超过时间没有收到登陆数据断开连接
int CpuCount=2; //服务务器cpu个数
double SuperTime=180000; //最大超时登录时间默认3分钟
CRITICAL_SECTION Csec_Client; //数据链表操作锁
//当前的客户连接
CWHDynamicArray<CClientInfo> smsClient;

std::string ServerPort="49100"; //服务监听端口

CRITICAL_SECTION Csec_log;
std::deque<std::string> runLog; //系统运行日志队列

std::string selfPath=""; //文件所在路径


///////////////////////////////////////////////////////////////////////////////////////////////

CThreadManage ThreadManage; //线程管理
bool IsIniVars=false; //全局资源是否被初始化过


//////////////////////////////////////////////////////////////////////////////////////////////
void Lock(LPCRITICAL_SECTION crit_sec)
{
::EnterCriticalSection(crit_sec);
}
void UnLock(LPCRITICAL_SECTION crit_sec)
{
::LeaveCriticalSection(crit_sec);
}
//初始化变量
int IniVariables()
{
int result=0;
TCHAR buf[512]={0};
::GetModuleFileName( NULL,buf, 512 );
selfPath=buf;
int pos=(int)selfPath.find_last_of('\\');
selfPath=selfPath.substr(0,pos);
std::string logPath=selfPath+"\\run_log";
//判断文件夹是否存在
if(!PathFileExists(logPath.c_str()))
{
//创建文件夹
::CreateDirectory(logPath.c_str(),NULL);
}
LogFileName=".\\run_log\\Service";
if(!IsIniVars)
{
////初始化winsock运行环境
WSADATA WSAData = { 0 };
WORD WSA_VERSION=MAKEWORD(2,0);
if ( 0 != WSAStartup(WSA_VERSION, &WSAData ))
{
PushLog("初始化socket环境失败");
//关闭清除
WSACleanup();
PushLog("启动服务失败");
return -1;
}
::InitializeCriticalSection(&Csec_Client);
::InitializeCriticalSection(&Csec_log);

CIniFile theFile(selfPath+"\\Config.ini");
//读取监听端口
ServerPort=theFile.GetParamValue("smsQueue","ServerPort");
//ManagePort=theFile.GetParamValue("smsQueue","ManagePort");
//读取超时设置
std::string strSuperTime=theFile.GetParamValue("smsQueue","SuperTime");
char *stopstring;
SuperTime=strtod(strSuperTime.c_str(),&stopstring);
SuperTime=SuperTime*60*1000;

g_hIOCP = ::CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, (ULONG_PTR)NULL, 0);
//创建完成端口
if (g_hIOCP== NULL)
{
std::string msg="创建完成端口g_hIOCP失败";
PushLog(msg);
return 3;
//创建完成端口失败
}
//cup个数
SYSTEM_INFO sysinfo;
::GetSystemInfo(&sysinfo);
CpuCount=(int)sysinfo.dwNumberOfProcessors;
result=1;
IsIniVars=true;
}
else
{
result=1;
}
return result;
}
//
//通知完成端口退出
void NoticeCompletionExit()
{
//退出端口等待
if(g_hIOCP)
{
::PostQueuedCompletionStatus(g_hIOCP,0,NULL,NULL);
}
}
void CloseSession(int nIndex)
{
if(smsClient.GetData(nIndex))
{
//告诉客户端服务器要关闭
if(smsClient.GetData(nIndex)->sock!=INVALID_SOCKET)
{
ReplyClient(smsClient.GetData(nIndex),0,0,"");
smsClient.GetData(nIndex)->LoginName="";
shutdown(smsClient.GetData(nIndex)->sock,SD_BOTH);
closesocket(smsClient.GetData(nIndex)->sock);
smsClient.GetData(nIndex)->sock=INVALID_SOCKET;
}
smsClient.SetEmptyElement(nIndex);
}
}
void CloseSession(CClientInfo* pSession)
{
if(pSession)
{
if(pSession->sock!=INVALID_SOCKET)
{
ReplyClient(pSession,0,0,"");
pSession->LoginName="";
pSession->fIsLogin=false;
shutdown(pSession->sock,SD_BOTH);
closesocket(pSession->sock);
pSession->sock=INVALID_SOCKET;
}
smsClient.SetEmptyElement(pSession->nUserIndex);
}
}
void CloseClientSession()
{
//关闭所有终端连接
try
{
Lock(&Csec_Client);
int i=0;
for(i=0;i<smsClient.m_nMaxIndex;i++)
{
CloseSession(i);
}
UnLock(&Csec_Client);
}
catch(...)
{
UnLock(&Csec_Client);
}
}

void UnIniVariables()
{
//CloseClientSession();
if(IsIniVars)
{
::DeleteCriticalSection(&Csec_Client);
::DeleteCriticalSection(&Csec_log);
::CloseHandle(g_hIOCP);
g_hIOCP=NULL;
WSACleanup();
IsIniVars=false;
}
}

///////////////////////////////////////

///发出一次IO读请求
void PostRecv(CClientInfo* pSession,CADOConnection& m_ADO)
{
try
{
DWORD nRecvBytes = 0;
DWORD nFlags = 0;
pSession->nOvFlag=OP_RECV;
pSession->DataBuf.len=1024*4;
memset(&(pSession->Overlapped),0,sizeof(OVERLAPPED));
memset(pSession->Buffer,0,1024*4);
pSession->DataBuf.buf=pSession->Buffer;
pSession->pack_len=pSession->total_length=0;
int result=0;
result=WSARecv(pSession->sock,&(pSession->DataBuf),1,&nRecvBytes, &nFlags, &(pSession->Overlapped), 0 );
////*
if(result==SOCKET_ERROR)
{
int Error=WSAGetLastError();
if(Error!=WSA_IO_PENDING)//出错
{
从登录列表中删除用户
std::string msg="";

CloseSession(pSession);

//*/
}
}
///*/
}
catch(...)
{

}
return ;
}
///
//////////////////////////////////////////////
///
///监听连接线程
///
unsigned int _stdcall AcceptThread(void* Param) //接收短信客户端连接线程
{
CThreadParam* pthread=(CThreadParam*)Param;
std::string msg="";
SOCKET s =socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);//使用TCP协议;
if(s==INVALID_SOCKET)
{
pthread->ThreadEnd();
msg="AcceptThread线程:创建监听的Socket失败";
PushLog(msg);
return 0;
}
try
{
int nPort=49100;//默认服务端口
if(!ServerPort.empty())
{
nPort=atoi(ServerPort.c_str());
}
struct sockaddr_in addr;
addr.sin_family = AF_INET;
addr.sin_port = htons((u_short)nPort);
addr.sin_addr.s_addr= htonl(INADDR_ANY); //针对服务器的所有网卡地址监听

int nLen = sizeof(struct sockaddr_in);//地址长度
SOCKET Accept=INVALID_SOCKET; //接收客户连接
struct sockaddr_in Address; //客户端地址
int optval=1 ; //Socket属性值
unsigned long ul=1;
time_t dtnow; //连接的时间
std::string IP("127.0.0.1"); //连接过来的客户IP
int ooptv=0;
setsockopt(s,SOL_SOCKET,SO_EXCLUSIVEADDRUSE,(char*)&optval,sizeof(optval)); //禁止端口多用
setsockopt(s,SOL_SOCKET,SO_KEEPALIVE,(char*)&optval,sizeof(optval)); //发送保持连接
setsockopt(s,SOL_SOCKET,SO_SNDBUF,(char*)&ooptv,sizeof(ooptv)); //设置发送缓冲区
setsockopt(s,SOL_SOCKET,SO_RCVBUF,(char*)&ooptv,sizeof(ooptv)); //设置接收缓冲区
setsockopt(s,IPPROTO_TCP,TCP_NODELAY,(char*)&optval,sizeof(optval)); //不采用延时算法
setsockopt(s,SOL_SOCKET,SO_DONTLINGER,(char*)&optval,sizeof(optval)); //执行立即关闭
linger lng;
lng.l_linger=0;
lng.l_onoff=1;
setsockopt(s,SOL_SOCKET,SO_LINGER,(char*)&lng,sizeof(lng));

//设置非阻塞模式
if(ioctlsocket(s,FIONBIO,(unsigned long*)&ul)==SOCKET_ERROR)
{
closesocket(s);
s=INVALID_SOCKET;
pthread->ThreadEnd();
msg="AcceptThread线程:设置非阻塞模式失败";
PushLog(msg);
return 0;
}
//禁止重用本地端口
if ((bind(s, (struct sockaddr *)&addr, sizeof( struct sockaddr_in))) == SOCKET_ERROR)
{
closesocket(s);
s=INVALID_SOCKET;
pthread->ThreadEnd();
msg="AcceptThread线程:绑定Socket失败";
PushLog(msg);
return 0;
}
if ((listen(s, 5)) == SOCKET_ERROR)
{
msg="AcceptThread线程:监听Socket失败";
pthread->ThreadEnd();
closesocket(s);
s=INVALID_SOCKET;
PushLog(msg);
return 0;
}
msg.resize(1000);
sprintf_s((char*)msg.c_str(),1000,"AcceptThread线程:开始运行，在端口%d监听连接",nPort);
PushLog(msg);
CADOConnection m_adoConn;
while(!m_adoConn.Open(adoConnString.c_str()))
{
msg="AcceptThread线程:连接数据库失败"+adoConnString;
PushLog(msg);
Sleep(10000);
}
while(!pthread->IsExit())
{
if(!m_adoConn.IsConnect())
{
m_adoConn.Open(adoConnString.c_str());
}
Accept=accept(s,(struct sockaddr *)&Address,&nLen);
if(Accept!=INVALID_SOCKET)
{
//判断对方IP是否是合法的用户，否则直接断开
IP=inet_ntoa(Address.sin_addr);//取得IP地址
if(!IsValidIP(m_adoConn,IP)&&(IP!="127.0.0.1")) //改用com函数中去处理
{
closesocket(Accept);
Accept=INVALID_SOCKET;
Sleep(0);
continue;
}
//一个新的连接
try
{
Lock(&Csec_Client);
int nIndex=-1;
nIndex=smsClient.GetAvailablePosition();
if(nIndex>=0)
{
CClientInfo* pNewUser=smsClient.GetEmptyElement(nIndex);
pNewUser->nUserIndex=nIndex;
pNewUser->IP=IP;
pNewUser->sock=Accept;
pNewUser->fIsLogin=false; //未登录
time(&dtnow);
pNewUser->tConnectTime=dtnow; //连接的时间
pNewUser->LinkCheck=0; //数据连接检查
::CreateIoCompletionPort((HANDLE)pNewUser->sock,g_hIOCP, (ULONG_PTR)pNewUser, 0);//创建一个完成端口关联
//发出一次接收数据的IO请求
PostRecv(pNewUser,m_adoConn);//改用com函数处理
}
else
{
closesocket(Accept);
Accept=INVALID_SOCKET;

}
UnLock(&Csec_Client);
}
catch(...)
{
UnLock(&Csec_Client);
}
}
else//利用空时间检查超时没有登录的客户端,断开它们
{
//检查超时未登录的
Lock(&Csec_Client);
try
{
time_t nt;
time(&nt);
double waiteTime=0;
int i=0;
//使用
for(i=0;i<smsClient.m_nMaxIndex;i++)
{
if(smsClient.IsAvailableElement(i))
{
waiteTime=difftime(nt,smsClient.GetData(i)->tConnectTime);
if(!smsClient.GetData(i)->fIsLogin)
{
//直接断开连接
if(waiteTime>SuperTime)
{
CloseSession(i);
}
Sleep(0);
}
else if(waiteTime>=SuperTime)//发送连接检测数据
{
ReplyClient(smsClient.GetData(i),8,0,"");
smsClient.GetData(i)->LinkCheck=smsClient.GetData(i)->LinkCheck+1;
}
else if(smsClient.GetData(i)->LinkCheck>3)//对于连续3次链路检测没有反应的断开连接
{
CloseSession(i);
}
}
}
UnLock(&Csec_Client);
}
catch(...)
{
UnLock(&Csec_Client);
}
}
Sleep(10);
}
m_adoConn.Close();
shutdown(s,SD_BOTH);
closesocket(s);
s=INVALID_SOCKET;
pthread->ThreadEnd();
msg="AcceptThread线程:正常终止";
PushLog(msg);
}
catch(...)
{
msg="AcceptThread线程:短信服务监听线程异常,10秒后线程自动重起";
PushLog(msg);
if(!pthread->IsExit()&&(pthread!=NULL))
{
//关闭所有连接的客户端
CloseClientSession();
Sleep(10000);
HANDLE hwnd=NULL;
unsigned int tid=0;
::CloseHandle(pthread->hThread);
pthread->hThread=NULL;
hwnd=(HANDLE)_beginthreadex(NULL,0,&AcceptThread,pthread,0,&tid);
if(hwnd)
{
pthread->hThread=hwnd;
}
}
else
{
pthread->ThreadEnd();
}
}
shutdown(s,SD_BOTH);
closesocket(s);
s=INVALID_SOCKET;
_endthreadex(0);
return 0;
}

//////////////////////////////////////////////////////
///完成端口的服务工作线程
///负责处理数据的收发操作
unsigned int _stdcall ServerWorkerThread(void* Param) //服务工作线程
{
CThreadParam* pthread=(CThreadParam*)Param;
std::string msg="";
try
{
DWORD dwBytesTransferred = 0;
CClientInfo* pSessionInfo = NULL;
OVERLAPPED * lpPerIoData = NULL;

BOOL bSuccess=FALSE;
int IsPost=1;
while(!pthread->IsExit())
{
bSuccess=::GetQueuedCompletionStatus(g_hIOCP, &dwBytesTransferred, (PULONG_PTR)&pSessionInfo,&lpPerIoData,INFINITE);
if(bSuccess!=0)//完成操作完成
{
if(pSessionInfo)
{
IsPost=1;
if((dwBytesTransferred==0)&&((pSessionInfo->nOvFlag==OP_RECV)||(pSessionInfo->nOvFlag==OP_SEND)))//与完成端口关联的socketg关闭
{
DelLoginUser(m_adoConn,pSessionInfo->LoginName);
if(pSessionInfo->fIsLogin)
{
msg="终端<断开连接>："+pSessionInfo->departName+"，IP："+pSessionInfo->IP+"，登录ID："+pSessionInfo->LoginName;
}
else
{
msg="终端<断开连接>："+pSessionInfo->IP+"断开连接";
}
Lock(&Csec_Client);
try
{
CloseSession(pSessionInfo);
UnLock(&Csec_Client);
}
catch(...)
{
UnLock(&Csec_Client);
}
PushLog(msg);
continue;
}
//接收数据包
Lock(&Csec_Client);
try
{
if(pSessionInfo->nOvFlag==OP_RECV)
{
if((dwBytesTransferred>0))//(pSessionInfo->nOvFlag==OP_RECV)&&
{
//第一次收包
if(pSessionInfo->pack_len==0&&pSessionInfo->total_length==0)
{
memset(pSessionInfo->RecvBuffer,0,1024*4);
if(dwBytesTransferred<4)
{
//继续接收
memcpy(pSessionInfo->RecvBuffer,pSessionInfo->DataBuf.buf,dwBytesTransferred);
pSessionInfo->pack_len=dwBytesTransferred;
}
else
{
//取包的长度
memcpy(&pSessionInfo->total_length,pSessionInfo->DataBuf.buf,4);
pSessionInfo->total_length=ntohl(pSessionInfo->total_length);//包总长
memcpy(pSessionInfo->RecvBuffer,pSessionInfo->DataBuf.buf,dwBytesTransferred);
pSessionInfo->pack_len=dwBytesTransferred;
}
}
else
{
//第一次收包不足4个字节
if(pSessionInfo->pack_len<4)
{
memcpy(pSessionInfo->RecvBuffer+pSessionInfo->pack_len,pSessionInfo->DataBuf.buf,dwBytesTransferred);
pSessionInfo->pack_len=pSessionInfo->pack_len+dwBytesTransferred;
if(pSessionInfo->pack_len>=4)
{
memcpy(&pSessionInfo->total_length,pSessionInfo->RecvBuffer,4);
pSessionInfo->total_length=ntohl(pSessionInfo->total_length);//包总长
}
}
else
{
memcpy(pSessionInfo->RecvBuffer+pSessionInfo->pack_len,pSessionInfo->DataBuf.buf,dwBytesTransferred);
pSessionInfo->pack_len=pSessionInfo->pack_len+dwBytesTransferred;
}
}
DataDeal:
//一次收包完成的操作
if(pSessionInfo->total_length==pSessionInfo->pack_len&&pSessionInfo->pack_len!=0)//一次包接收完成
{
memset(pSessionInfo->DataBuf.buf,0,1024*4);//清零
memset(pSessionInfo->DealBuffer,0,1024*4);
memcpy(pSessionInfo->DealBuffer,pSessionInfo->RecvBuffer,pSessionInfo->total_length);
IsPost=DealRecvDataPacket(pSessionInfo,m_adoConn);//将接收数据交给数据分析处理线程
pSessionInfo->total_length=0;
pSessionInfo->pack_len=0;
}
if(pSessionInfo->total_length<pSessionInfo->pack_len&&pSessionInfo->pack_len!=0&&pSessionInfo->total_length>0)//一次包接收完成,并有超出一条记录的数据包
{
memset(pSessionInfo->DataBuf.buf,0,1024*4);//清零
memset(pSessionInfo->DealBuffer,0,1024*4);
memcpy(pSessionInfo->DealBuffer,pSessionInfo->RecvBuffer,pSessionInfo->total_length);
IsPost=DealRecvDataPacket(pSessionInfo,m_adoConn);//对数据分析处理
char buf[4096]={0};
memset(buf,0,4096);
memcpy(buf,pSessionInfo->RecvBuffer+pSessionInfo->total_length,pSessionInfo->pack_len-pSessionInfo->total_length);
memset(pSessionInfo->RecvBuffer,0,1024*4);
memcpy(pSessionInfo->RecvBuffer,buf,pSessionInfo->pack_len-pSessionInfo->total_length);
pSessionInfo->pack_len=pSessionInfo->pack_len-pSessionInfo->total_length;
if(pSessionInfo->pack_len>=4)
{
memcpy(&pSessionInfo->total_length,pSessionInfo->RecvBuffer,4);
pSessionInfo->total_length=ntohl(pSessionInfo->total_length);//包总长
}
if(pSessionInfo->total_length<=pSessionInfo->pack_len)//如果一次分包未结束，继续分包，直到不足一个结构包
{
goto DataDeal;
}
}
//发出接收IO请求
if(IsPost==1)
{
PostRecv(pSessionInfo,m_adoConn);
}
}
}
UnLock(&Csec_Client);
}
catch(...)
{
UnLock(&Csec_Client);
}
}

if((dwBytesTransferred==0)&&(pSessionInfo==NULL)&&(lpPerIoData==NULL))//收到线程结束信号
{
break;
}
}
else
{
if((lpPerIoData!=NULL)&&pSessionInfo)//偶发错误
{
if(pSessionInfo)
{
Lock(&Csec_Client);
try
{
if(pSessionInfo->fIsLogin)
{
msg="终端<断开连接>："+pSessionInfo->departName+"，IP："+pSessionInfo->IP+"，登录ID："+pSessionInfo->LoginName;
}
else
{
msg="终端<断开连接>："+pSessionInfo->IP+"断开连接";
}
CloseSession(pSessionInfo);
UnLock(&Csec_Client);
PushLog(msg);
}
catch(...)
{
UnLock(&Csec_Client);
}
}
}
}
Sleep(10);
}
m_adoConn.Close();
msg="ServerWorkerThread:正常终止";
PushLog(msg);
pthread->ThreadEnd();
}
catch(...)
{
if((!pthread->IsExit())&&(pthread!=NULL))
{
msg="ServerWorkerThread:出现异常，1秒后自动从新开始";
PushLog(msg);
Sleep(1000);
HANDLE hwnd=NULL;
unsigned int tid=0;
::CloseHandle(pthread->hThread);
pthread->hThread=NULL;
hwnd=(HANDLE)_beginthreadex(NULL,0,&ServerWorkerThread,pthread,0,&tid);
if(hwnd)
{
pthread->hThread=hwnd;
}
}
else
{
pthread->ThreadEnd();
}
}
_endthreadex(0);
return 0;
}