1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063 |
- #include "StdAfx.h"
- #include "SATTCPServer.h"
- //#include "MainDlg.h"
- #include "SATExecutor.h"
- #include "SATDevices.h"
- // 每一个处理器上产生多少个线程(为了最大限度的提升服务器性能,详见配套文档)
- #define WORKER_THREADS_PER_PROCESSOR 2
- // 同时投递的Accept请求的数量(这个要根据实际的情况灵活设置)
- #define MAX_POST_ACCEPT 10
- // 传递给Worker线程的退出信号
- #define EXIT_CODE NULL
- // 释放指针和句柄资源的宏
- // 释放指针宏
- #define RELEASE(x) {if(x != NULL ){delete x;x=NULL;}}
- // 释放句柄宏
- #define RELEASE_HANDLE(x) {if(x != NULL && x!=INVALID_HANDLE_VALUE){ CloseHandle(x);x = NULL;}}
- // 释放Socket宏
- #define RELEASE_SOCKET(x) {if(x !=INVALID_SOCKET) { closesocket(x);x=INVALID_SOCKET;}}
- CSATTCPServer::CSATTCPServer(void):
- m_nThreads(0),
- m_hShutdownEvent(NULL),
- m_hIOCompletionPort(NULL),
- m_phWorkerThreads(NULL),
- m_strIP(DEFAULT_IP),
- m_nPort(DEFAULT_PORT),
- m_pMain(NULL),
- m_lpfnAcceptEx( NULL ),
- m_pListenContext( NULL )
- {
- }
- CSATTCPServer::~CSATTCPServer(void)
- {
- // 确保资源彻底释放
- this->Stop();
- }
- ///////////////////////////////////////////////////////////////////
- // 工作者线程: 为IOCP请求服务的工作者线程
- // 也就是每当完成端口上出现了完成数据包,就将之取出来进行处理的线程
- ///////////////////////////////////////////////////////////////////
- DWORD WINAPI CSATTCPServer::_WorkerThread(LPVOID lpParam)
- {
- THREADPARAMS_WORKER* pParam = (THREADPARAMS_WORKER*)lpParam;
- CSATTCPServer* pIOCPModel = (CSATTCPServer*)pParam->pIOCPModel;
- int nThreadNo = (int)pParam->nThreadNo;
- pIOCPModel->_ShowMessage(_T("工作者线程启动,ID: %d."),nThreadNo);
- OVERLAPPED *pOverlapped = NULL;
- PER_SOCKET_CONTEXT *pSocketContext = NULL;
- DWORD dwBytesTransfered = 0;
- // 循环处理请求,知道接收到Shutdown信息为止
- while (WAIT_OBJECT_0 != WaitForSingleObject(pIOCPModel->m_hShutdownEvent, 0))
- {
- BOOL bReturn = GetQueuedCompletionStatus(
- pIOCPModel->m_hIOCompletionPort,
- &dwBytesTransfered,
- (PULONG_PTR)&pSocketContext,
- &pOverlapped,
- INFINITE);
- // 如果收到的是退出标志,则直接退出
- if ( EXIT_CODE==(DWORD)pSocketContext )
- {
- pIOCPModel->_ShowMessage(_T("收到的是退出标志,则直接退出,ID: %d."),nThreadNo);
- break;
- }
- // 判断是否出现了错误
- if( !bReturn )
- {
- DWORD dwErr = GetLastError();
- // 显示一下提示信息
- if( !pIOCPModel->HandleError( pSocketContext,dwErr ) )
- {
- break;
- }
- continue;
- }
- else
- {
- // 读取传入的参数
- PER_IO_CONTEXT* pIoContext = CONTAINING_RECORD(pOverlapped, PER_IO_CONTEXT, m_Overlapped);
- // 判断是否有客户端断开了
- if((0 == dwBytesTransfered) && ( RECV_POSTED==pIoContext->m_OpType || SEND_POSTED==pIoContext->m_OpType))
- {
- pIOCPModel->_ShowMessage( _T("客户端 %s:%d 断开连接."),
- inet_ntoa(pSocketContext->m_ClientAddr.sin_addr),
- ntohs(pSocketContext->m_ClientAddr.sin_port) );
- // 释放掉对应的资源
- pIOCPModel->_RemoveContext( pSocketContext );
- continue;
- }
- else
- {
- switch( pIoContext->m_OpType )
- {
- // Accept
- case ACCEPT_POSTED:
- {
- // 为了增加代码可读性,这里用专门的_DoAccept函数进行处理连入请求
- pIOCPModel->_DoAccpet( pSocketContext, pIoContext );
- }
- break;
- // RECV
- case RECV_POSTED:
- {
- // 为了增加代码可读性,这里用专门的_DoRecv函数进行处理接收请求
- pIOCPModel->_DoRecv( pSocketContext,pIoContext );
- }
- break;
- // SEND
- // 这里略过不写了,要不代码太多了,不容易理解,Send操作相对来讲简单一些
- case SEND_POSTED:
- {
- }
- break;
- default:
- // 不应该执行到这里
- TRACE(_T("_WorkThread中的 pIoContext->m_OpType 参数异常.\n"));
- break;
- } //switch
- }//if
- }//if
- }//while
- TRACE(_T("工作者线程 %d 号退出.\n"),nThreadNo);
- // 释放线程参数
- RELEASE(lpParam);
- return 0;
- }
- //====================================================================================
- //
- // 系统初始化和终止
- //
- //====================================================================================
- ////////////////////////////////////////////////////////////////////
- // 初始化WinSock 2.2
- bool CSATTCPServer::LoadSocketLib()
- {
- WSADATA wsaData;
- int nResult;
- nResult = WSAStartup(MAKEWORD(2,2), &wsaData);
- // 错误(一般都不可能出现)
- if (NO_ERROR != nResult)
- {
- this->_ShowMessage(_T("初始化WinSock 2.2失败!\n"));
- return false;
- }
- return true;
- }
- //////////////////////////////////////////////////////////////////
- // 启动服务器
- bool CSATTCPServer::Start(unsigned int port)
- {
- // 初始化线程互斥量
- InitializeCriticalSection(&m_csContextList);
- // 建立系统退出的事件通知
- m_hShutdownEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
- // 初始化IOCP
- if (false == _InitializeIOCP())
- {
- this->_ShowMessage(_T("初始化IOCP失败!\n"));
- return false;
- }
- else
- {
- this->_ShowMessage(_T("\nIOCP初始化完毕\n."));
- }
- // 初始化Socket
- if( false==_InitializeListenSocket(port) )
- {
- this->_ShowMessage(_T("Listen Socket初始化失败!\n"));
- this->_DeInitialize();
- return false;
- }
- else
- {
- this->_ShowMessage(_T("Listen Socket初始化完毕."));
- }
- this->_ShowMessage(_T("系统准备就绪,等候连接....\n"));
- return true;
- }
- ////////////////////////////////////////////////////////////////////
- // 开始发送系统退出消息,退出完成端口和线程资源
- void CSATTCPServer::Stop()
- {
- if( m_pListenContext != NULL && m_pListenContext->m_Socket != INVALID_SOCKET )
- {
- // 激活关闭消息通知
- SetEvent(m_hShutdownEvent);
- for (int i = 0; i < m_nThreads; i++)
- {
- // 通知所有的完成端口操作退出
- PostQueuedCompletionStatus(m_hIOCompletionPort, 0, (DWORD)EXIT_CODE, NULL);
- }
- // 等待所有的客户端资源退出
- WaitForMultipleObjects(m_nThreads, m_phWorkerThreads, TRUE, INFINITE);
- // 清除客户端列表信息
- this->_ClearContextList();
- // 释放其他资源
- this->_DeInitialize();
- this->_ShowMessage(_T("停止监听\n"));
- }
- }
- ////////////////////////////////
- // 初始化完成端口
- bool CSATTCPServer::_InitializeIOCP()
- {
- // 建立第一个完成端口
- m_hIOCompletionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0 );
- if ( NULL == m_hIOCompletionPort)
- {
- this->_ShowMessage(_T("建立完成端口失败!错误代码: %d!\n"), WSAGetLastError());
- return false;
- }
- // 根据本机中的处理器数量,建立对应的线程数
- //m_nThreads = WORKER_THREADS_PER_PROCESSOR * _GetNoOfProcessors();
- m_nThreads = _GetNoOfProcessors();
-
- // 为工作者线程初始化句柄
- m_phWorkerThreads = new HANDLE[m_nThreads];
-
- // 根据计算出来的数量建立工作者线程
- DWORD nThreadID;
- for (int i = 0; i < m_nThreads; i++)
- {
- THREADPARAMS_WORKER* pThreadParams = new THREADPARAMS_WORKER;
- pThreadParams->pIOCPModel = this;
- pThreadParams->nThreadNo = i+1;
- m_phWorkerThreads[i] = ::CreateThread(0, 0, _WorkerThread, (void *)pThreadParams, 0, &nThreadID);
- }
- TRACE(" 建立 _WorkerThread %d 个.\n", m_nThreads );
- return true;
- }
- /////////////////////////////////////////////////////////////////
- // 初始化Socket
- bool CSATTCPServer::_InitializeListenSocket(unsigned int port)
- {
- // AcceptEx 和 GetAcceptExSockaddrs 的GUID,用于导出函数指针
- GUID GuidAcceptEx = WSAID_ACCEPTEX;
- GUID GuidGetAcceptExSockAddrs = WSAID_GETACCEPTEXSOCKADDRS;
- // 服务器地址信息,用于绑定Socket
- struct sockaddr_in ServerAddress;
- // 生成用于监听的Socket的信息
- m_pListenContext = new PER_SOCKET_CONTEXT;
- // 需要使用重叠IO,必须得使用WSASocket来建立Socket,才可以支持重叠IO操作
- m_pListenContext->m_Socket = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);
- if (INVALID_SOCKET == m_pListenContext->m_Socket)
- {
- this->_ShowMessage(_T("初始化Socket失败,错误代码: %d.\n"), WSAGetLastError());
- return false;
- }
- else
- {
- TRACE("WSASocket() 完成.\n");
- }
- // 将Listen Socket绑定至完成端口中
- if( NULL== CreateIoCompletionPort( (HANDLE)m_pListenContext->m_Socket, m_hIOCompletionPort,(DWORD)m_pListenContext, 0))
- {
- this->_ShowMessage(_T("绑定 Listen Socket至完成端口失败!错误代码: %d/n"), WSAGetLastError());
- RELEASE_SOCKET( m_pListenContext->m_Socket );
- return false;
- }
- else
- {
- TRACE("Listen Socket绑定完成端口 完成.\n");
- }
- // 填充地址信息
- ZeroMemory((char *)&ServerAddress, sizeof(ServerAddress));
- ServerAddress.sin_family = AF_INET;
- // 这里可以绑定任何可用的IP地址,或者绑定一个指定的IP地址
- ServerAddress.sin_addr.s_addr = htonl(INADDR_ANY);
- //ServerAddress.sin_addr.s_addr = inet_addr(m_strIP.GetString());
- ServerAddress.sin_port = htons(port);
- // 绑定地址和端口
- if (SOCKET_ERROR == ::bind(m_pListenContext->m_Socket, (struct sockaddr *) &ServerAddress, sizeof(ServerAddress)))
- {
- TRACE("bind() 失败.\n");
- this->_ShowMessage(_T("bind()函数执行错误! 错误代码: %d/n"), WSAGetLastError());
- return false;
- }
- else
- {
- TRACE("bind() 完成.\n");
- }
- // 开始进行监听
- if (SOCKET_ERROR == listen(m_pListenContext->m_Socket,SOMAXCONN))
- {
- this->_ShowMessage(_T("Listen()函数执行出现错误; 错误代码: %d/n"), WSAGetLastError());
- return false;
- }
- else
- {
- TRACE("Listen() 完成.\n");
- }
- // 使用AcceptEx函数,因为这个是属于WinSock2规范之外的微软另外提供的扩展函数
- // 所以需要额外获取一下函数的指针,
- // 获取AcceptEx函数指针
- DWORD dwBytes = 0;
- if(SOCKET_ERROR == WSAIoctl(
- m_pListenContext->m_Socket,
- SIO_GET_EXTENSION_FUNCTION_POINTER,
- &GuidAcceptEx,
- sizeof(GuidAcceptEx),
- &m_lpfnAcceptEx,
- sizeof(m_lpfnAcceptEx),
- &dwBytes,
- NULL,
- NULL))
- {
- this->_ShowMessage(_T("WSAIoctl 未能获取AcceptEx函数指针。错误代码: %d\n"), WSAGetLastError());
- this->_DeInitialize();
- return false;
- }
- // 获取GetAcceptExSockAddrs函数指针,也是同理
- if(SOCKET_ERROR == WSAIoctl(
- m_pListenContext->m_Socket,
- SIO_GET_EXTENSION_FUNCTION_POINTER,
- &GuidGetAcceptExSockAddrs,
- sizeof(GuidGetAcceptExSockAddrs),
- &m_lpfnGetAcceptExSockAddrs,
- sizeof(m_lpfnGetAcceptExSockAddrs),
- &dwBytes,
- NULL,
- NULL))
- {
- this->_ShowMessage(_T("WSAIoctl 未能获取GuidGetAcceptExSockAddrs函数指针。错误代码: %d\n"), WSAGetLastError());
- this->_DeInitialize();
- return false;
- }
- // 为AcceptEx 准备参数,然后投递AcceptEx I/O请求
- for( int i=0;i<MAX_POST_ACCEPT;i++ )
- {
- // 新建一个IO_CONTEXT
- PER_IO_CONTEXT* pAcceptIoContext = m_pListenContext->GetNewIoContext();
- if( false==this->_PostAccept( pAcceptIoContext ) )
- {
- m_pListenContext->RemoveContext(pAcceptIoContext);
- return false;
- }
- }
- this->_ShowMessage( _T("投递 %d 个AcceptEx请求完毕"),MAX_POST_ACCEPT );
- return true;
- }
- ////////////////////////////////////////////////////////////
- // 最后释放掉所有资源
- void CSATTCPServer::_DeInitialize()
- {
- // 删除客户端列表的互斥量
- DeleteCriticalSection(&m_csContextList);
- // 关闭系统退出事件句柄
- RELEASE_HANDLE(m_hShutdownEvent);
- // 释放工作者线程句柄指针
- for( int i=0;i<m_nThreads;i++ )
- {
- RELEASE_HANDLE(m_phWorkerThreads[i]);
- }
-
- RELEASE(m_phWorkerThreads);
- // 关闭IOCP句柄
- RELEASE_HANDLE(m_hIOCompletionPort);
- shutdown(m_pListenContext->m_Socket, SD_BOTH);
- // 先关闭端口;
- RELEASE_SOCKET( m_pListenContext->m_Socket );
- // 关闭监听Socket
- RELEASE(m_pListenContext);
- this->_ShowMessage(_T("释放资源完毕.\n"));
- }
- //====================================================================================
- //
- // 投递完成端口请求
- //
- //====================================================================================
- //////////////////////////////////////////////////////////////////
- // 投递Accept请求
- bool CSATTCPServer::_PostAccept( PER_IO_CONTEXT* pAcceptIoContext )
- {
- ASSERT( INVALID_SOCKET!=m_pListenContext->m_Socket );
- // 准备参数
- DWORD dwBytes = 0;
- pAcceptIoContext->m_OpType = ACCEPT_POSTED;
- WSABUF *p_wbuf = &pAcceptIoContext->m_wsaBuf;
- OVERLAPPED *p_ol = &pAcceptIoContext->m_Overlapped;
-
- // 为以后新连入的客户端先准备好Socket( 这个是与传统accept最大的区别 )
- pAcceptIoContext->m_sockAccept = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
- if( INVALID_SOCKET==pAcceptIoContext->m_sockAccept )
- {
- _ShowMessage(_T("创建用于Accept的Socket失败!错误代码: %d"), WSAGetLastError());
- return false;
- }
- // 投递AcceptEx
- if(FALSE == m_lpfnAcceptEx( m_pListenContext->m_Socket, pAcceptIoContext->m_sockAccept, p_wbuf->buf, p_wbuf->len - ((sizeof(SOCKADDR_IN)+16)*2),
- sizeof(SOCKADDR_IN)+16, sizeof(SOCKADDR_IN)+16, &dwBytes, p_ol))
- {
- if(WSA_IO_PENDING != WSAGetLastError())
- {
- _ShowMessage(_T("投递 AcceptEx 请求失败,错误代码: %d"), WSAGetLastError());
- return false;
- }
- }
- return true;
- }
- ////////////////////////////////////////////////////////////
- // 在有客户端连入的时候,进行处理
- // 流程有点复杂,你要是看不懂的话,就看配套的文档吧....
- // 如果能理解这里的话,完成端口的机制你就消化了一大半了
- // 总之你要知道,传入的是ListenSocket的Context,我们需要复制一份出来给新连入的Socket用
- // 原来的Context还是要在上面继续投递下一个Accept请求
- //
- bool CSATTCPServer::_DoAccpet( PER_SOCKET_CONTEXT* pSocketContext, PER_IO_CONTEXT* pIoContext )
- {
- SOCKADDR_IN* ClientAddr = NULL;
- SOCKADDR_IN* LocalAddr = NULL;
- int remoteLen = sizeof(SOCKADDR_IN), localLen = sizeof(SOCKADDR_IN);
- ///////////////////////////////////////////////////////////////////////////
- // 1. 首先取得连入客户端的地址信息
- // 这个 m_lpfnGetAcceptExSockAddrs 不得了啊~~~~~~
- // 不但可以取得客户端和本地端的地址信息,还能顺便取出客户端发来的第一组数据
- this->m_lpfnGetAcceptExSockAddrs(pIoContext->m_wsaBuf.buf, pIoContext->m_wsaBuf.len - ((sizeof(SOCKADDR_IN)+16)*2),
- sizeof(SOCKADDR_IN)+16, sizeof(SOCKADDR_IN)+16, (LPSOCKADDR*)&LocalAddr, &localLen, (LPSOCKADDR*)&ClientAddr, &remoteLen);
- _RecvProcess(pSocketContext, pIoContext);
- this->_ShowMessage( _T("客户端 %s:%d 连入."), inet_ntoa(ClientAddr->sin_addr), ntohs(ClientAddr->sin_port) );
- //////////////////////////////////////////////////////////////////////////////////////////////////////
- // 2. 这里需要注意,这里传入的这个是ListenSocket上的Context,这个Context我们还需要用于监听下一个连接
- // 所以我还得要将ListenSocket上的Context复制出来一份为新连入的Socket新建一个SocketContext
- PER_SOCKET_CONTEXT* pNewSocketContext = new PER_SOCKET_CONTEXT;
- pNewSocketContext->m_Socket = pIoContext->m_sockAccept;
- memcpy(&(pNewSocketContext->m_ClientAddr), ClientAddr, sizeof(SOCKADDR_IN));
- // 参数设置完毕,将这个Socket和完成端口绑定(这也是一个关键步骤)
- if( false==this->_AssociateWithIOCP( pNewSocketContext ) )
- {
- RELEASE( pNewSocketContext );
- return false;
- }
- ///////////////////////////////////////////////////////////////////////////////////////////////////
- // 3. 继续,建立其下的IoContext,用于在这个Socket上投递第一个Recv数据请求
- PER_IO_CONTEXT* pNewIoContext = pNewSocketContext->GetNewIoContext();
- pNewIoContext->m_OpType = RECV_POSTED;
- pNewIoContext->m_sockAccept = pNewSocketContext->m_Socket;
- // 如果Buffer需要保留,就自己拷贝一份出来
- //memcpy( pNewIoContext->m_szBuffer,pIoContext->m_szBuffer,MAX_BUFFER_LEN );
- // 绑定完毕之后,就可以开始在这个Socket上投递完成请求了
- if( false==this->_PostRecv( pNewIoContext) )
- {
- pNewSocketContext->RemoveContext( pNewIoContext );
- return false;
- }
- /////////////////////////////////////////////////////////////////////////////////////////////////
- // 4. 如果投递成功,那么就把这个有效的客户端信息,加入到ContextList中去(需要统一管理,方便释放资源)
- this->_AddToContextList( pNewSocketContext );
- ////////////////////////////////////////////////////////////////////////////////////////////////
- // 5. 使用完毕之后,把Listen Socket的那个IoContext重置,然后准备投递新的AcceptEx
- pIoContext->ResetBuffer();
- return this->_PostAccept( pIoContext );
- }
- ////////////////////////////////////////////////////////////////////
- // 投递接收数据请求
- bool CSATTCPServer::_PostRecv( PER_IO_CONTEXT* pIoContext )
- {
- // 初始化变量
- DWORD dwFlags = 0;
- DWORD dwBytes = 0;
- WSABUF *p_wbuf = &pIoContext->m_wsaBuf;
- OVERLAPPED *p_ol = &pIoContext->m_Overlapped;
- pIoContext->ResetBuffer();
- pIoContext->m_OpType = RECV_POSTED;
- // 初始化完成后,,投递WSARecv请求
- int nBytesRecv = WSARecv( pIoContext->m_sockAccept, p_wbuf, 1, &dwBytes, &dwFlags, p_ol, NULL );
- // 如果返回值错误,并且错误的代码并非是Pending的话,那就说明这个重叠请求失败了
- if ((SOCKET_ERROR == nBytesRecv) && (WSA_IO_PENDING != WSAGetLastError()))
- {
- this->_ShowMessage(_T("投递第一个WSARecv失败!"));
- return false;
- }
- return true;
- }
- /////////////////////////////////////////////////////////////////
- // 在有接收的数据到达的时候,进行处理
- bool CSATTCPServer::_DoRecv( PER_SOCKET_CONTEXT* pSocketContext, PER_IO_CONTEXT* pIoContext )
- {
- // 先把上一次的数据显示出现,然后就重置状态,发出下一个Recv请求
- SOCKADDR_IN* ClientAddr = &pSocketContext->m_ClientAddr;
- //this->_ShowMessage( _T("收到 %s:%d 信息:%s"),inet_ntoa(ClientAddr->sin_addr), ntohs(ClientAddr->sin_port), pIoContext->m_wsaBuf.buf );
-
- _RecvProcess(pSocketContext, pIoContext);
- // 然后开始投递下一个WSARecv请求
- return _PostRecv( pIoContext );
- }
- /////////////////////////////////////////////////////
- // 将句柄(Socket)绑定到完成端口中
- bool CSATTCPServer::_AssociateWithIOCP( PER_SOCKET_CONTEXT *pContext )
- {
- // 将用于和客户端通信的SOCKET绑定到完成端口中
- HANDLE hTemp = CreateIoCompletionPort((HANDLE)pContext->m_Socket, m_hIOCompletionPort, (DWORD)pContext, 0);
- if (NULL == hTemp)
- {
- this->_ShowMessage(_T("执行CreateIoCompletionPort()出现错误.错误代码:%d"),GetLastError());
- return false;
- }
- return true;
- }
- //====================================================================================
- //
- // ContextList 相关操作
- //
- //====================================================================================
- //////////////////////////////////////////////////////////////
- // 将客户端的相关信息存储到数组中
- void CSATTCPServer::_AddToContextList( PER_SOCKET_CONTEXT *pHandleData )
- {
- EnterCriticalSection(&m_csContextList);
- m_arrayClientContext.Add(pHandleData);
-
- LeaveCriticalSection(&m_csContextList);
- }
- ////////////////////////////////////////////////////////////////
- // 移除某个特定的Context
- void CSATTCPServer::_RemoveContext( PER_SOCKET_CONTEXT *pSocketContext )
- {
- EnterCriticalSection(&m_csContextList);
- for( int i=0;i<m_arrayClientContext.GetCount();i++ )
- {
- if( pSocketContext==m_arrayClientContext.GetAt(i) )
- {
- RELEASE( pSocketContext );
- m_arrayClientContext.RemoveAt(i);
- break;
- }
- }
- LeaveCriticalSection(&m_csContextList);
- }
- ////////////////////////////////////////////////////////////////
- // 清空客户端信息
- void CSATTCPServer::_ClearContextList()
- {
- EnterCriticalSection(&m_csContextList);
- for( int i=0;i<m_arrayClientContext.GetCount();i++ )
- {
- delete m_arrayClientContext.GetAt(i);
- }
- m_arrayClientContext.RemoveAll();
- LeaveCriticalSection(&m_csContextList);
- }
- //====================================================================================
- //
- // 其他辅助函数定义
- //
- //====================================================================================
- ////////////////////////////////////////////////////////////////////
- // 获得本机的IP地址
- CString CSATTCPServer::GetLocalIP()
- {
- // 获得本机主机名
- char hostname[MAX_PATH] = {0};
- gethostname(hostname,MAX_PATH);
- struct hostent FAR* lpHostEnt = gethostbyname(hostname);
- if(lpHostEnt == NULL)
- {
- return DEFAULT_IP;
- }
- // 取得IP地址列表中的第一个为返回的IP(因为一台主机可能会绑定多个IP)
- LPSTR lpAddr = lpHostEnt->h_addr_list[0];
- // 将IP地址转化成字符串形式
- struct in_addr inAddr;
- memmove(&inAddr,lpAddr,4);
- m_strIP = CString( inet_ntoa(inAddr) );
- return m_strIP;
- }
- ///////////////////////////////////////////////////////////////////
- // 获得本机中处理器的数量
- int CSATTCPServer::_GetNoOfProcessors()
- {
- SYSTEM_INFO si;
- GetSystemInfo(&si);
- return si.dwNumberOfProcessors;
- }
- /////////////////////////////////////////////////////////////////////
- // 在主界面中显示提示信息
- void CSATTCPServer::_ShowMessage(const CString szFormat,...) const
- {
- // 根据传入的参数格式化字符串
- CString strMessage;
- va_list arglist;
- // 处理变长参数
- va_start(arglist, szFormat);
- strMessage.FormatV(szFormat,arglist);
- va_end(arglist);
- GLOBAL::WriteTextLog(strMessage);
- }
- #define PAK_LEN sizeof(SATPROTO::DataHeader)
- void CSATTCPServer::_RecvProcess(PER_SOCKET_CONTEXT* pSocketContext, PER_IO_CONTEXT* pIoContext)
- {
- #ifdef _DEBUG
- OutputDebugString("有数据返回\n");
- #endif
- //GLOBAL::WriteTextLog("有数据返回");
- // 小于包头;
- SATPROTO::DataHeader* phead = NULL;
- if (pSocketContext->lastData.size() == 0)
- {
- // 不足包头;
- if (PAK_LEN > pIoContext->m_Overlapped.InternalHigh)
- {
- OutputDebugString("A:不足包头;\n");
- //GLOBAL::WriteTextLog("A:不足包头");
- pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf, pIoContext->m_Overlapped.InternalHigh);
- }
- else
- {
- // 包头充足;
- phead = (SATPROTO::DataHeader*)pIoContext->m_wsaBuf.buf;
- if ( !CheckDataHeader(phead) )
- {
- OutputDebugString("A:包头损坏;\n");
- //GLOBAL::WriteTextLog("A:包头损坏");
- return;
- }
- // 完整的包;
- if (phead->len == pIoContext->m_Overlapped.InternalHigh)
- {
- OutputDebugString("A:完整的包;\n");
- //GLOBAL::WriteTextLog("A:完整的包");
- _TaskProcess(pIoContext, (SATPROTO::Package*)pIoContext->m_wsaBuf.buf);
- }
- // 小包;
- else if (phead->len > pIoContext->m_Overlapped.InternalHigh)
- {
- OutputDebugString("A:小包;\n");
- //GLOBAL::WriteTextLog("A:小包");
- pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf, pIoContext->m_Overlapped.InternalHigh);
- }
- // 超包;
- else if (phead->len < pIoContext->m_Overlapped.InternalHigh)
- {
- OutputDebugString("A:超包;\n");
- //GLOBAL::WriteTextLog("A:超包");
- pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf + phead->len, pIoContext->m_Overlapped.InternalHigh - phead->len);
- _TaskProcess(pIoContext, (SATPROTO::Package*)pIoContext->m_wsaBuf.buf);
- }
- }
- }
- else
- {
- int lastlen = pIoContext->m_Overlapped.InternalHigh;
- if (pSocketContext->lastData.size() >= PAK_LEN)
- {
- phead = (SATPROTO::DataHeader*)pSocketContext->lastData.data();
- if ( !CheckDataHeader(phead) )
- {
- OutputDebugString("C:包头损坏;\n");
- //GLOBAL::WriteTextLog("C:包头损坏");
- pSocketContext->lastData.clear();
- return;
- }
- if (phead->len <= pSocketContext->lastData.size() + pIoContext->m_Overlapped.InternalHigh)
- {
- if ( phead->len <= pSocketContext->lastData.size() )
- {
- OutputDebugString("C:超包;\n");
- //GLOBAL::WriteTextLog("C:超包");
- // 完整包;
- _TaskProcess(pIoContext, (SATPROTO::Package*)pSocketContext->lastData.substr(0, phead->len).data());
- pSocketContext->lastData = pSocketContext->lastData.substr(phead->len);
- pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf, pIoContext->m_Overlapped.InternalHigh);
- }
- else
- {
- OutputDebugString("D:超包;\n");
- //GLOBAL::WriteTextLog("D:超包");
- lastlen = pSocketContext->lastData.size() + pIoContext->m_Overlapped.InternalHigh - phead->len;
- pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf, pIoContext->m_Overlapped.InternalHigh - lastlen);
- // 完整包;
- _TaskProcess(pIoContext, (SATPROTO::Package*)pSocketContext->lastData.data());
- // 剩余包;
- pSocketContext->lastData.clear();
- if (lastlen)
- pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf + pIoContext->m_Overlapped.InternalHigh - lastlen, lastlen);
- }
- }
- else
- {
- OutputDebugString("C:仍不足一个包;\n");
- //GLOBAL::WriteTextLog("C:仍不足一个包");
- pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf, pIoContext->m_Overlapped.InternalHigh);
- }
- }
- else
- {
- // 包头剩余长度;
- int diflen = PAK_LEN - pSocketContext->lastData.size();
- // 仍不足一个包头;
- if ( diflen > pIoContext->m_Overlapped.InternalHigh )
- {
- OutputDebugString("B:仍不足一个包头;\n");
- //GLOBAL::WriteTextLog("B:仍不足一个包头");
- pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf, pIoContext->m_Overlapped.InternalHigh);
- }
- else
- {
- // 拼成完整包头;
- pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf, diflen);
- phead = (SATPROTO::DataHeader*)pSocketContext->lastData.data();
- if ( !CheckDataHeader(phead) )
- {
- OutputDebugString("B:包头损坏;\n");
- //GLOBAL::WriteTextLog("B:包头损坏");
- pSocketContext->lastData.clear();
- return;
- }
- // 完整包;
- if ( phead->len == PAK_LEN + pIoContext->m_Overlapped.InternalHigh - diflen )
- {
- OutputDebugString("B:完整包;\n");
- //GLOBAL::WriteTextLog("B:完整包");
- pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf + diflen, pIoContext->m_Overlapped.InternalHigh - diflen);
- _TaskProcess(pIoContext, (SATPROTO::Package*)pSocketContext->lastData.data());
- pSocketContext->lastData.clear();
- }
- // 小包;
- else if ( phead->len > PAK_LEN + pIoContext->m_Overlapped.InternalHigh - diflen)
- {
- OutputDebugString("B:小包;\n");
- //GLOBAL::WriteTextLog("B:小包");
- pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf + diflen, pIoContext->m_Overlapped.InternalHigh - diflen);
- }
- // 超包;
- else if (phead->len < PAK_LEN + pIoContext->m_Overlapped.InternalHigh - diflen)
- {
- OutputDebugString("B:超包;\n");
- //GLOBAL::WriteTextLog("B:超包");
- // 组完成包;
- pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf + diflen, phead->len - PAK_LEN);
- _TaskProcess(pIoContext, (SATPROTO::Package*)pSocketContext->lastData.data());
- pSocketContext->lastData.clear();
- int last = pIoContext->m_Overlapped.InternalHigh - diflen - phead->len + PAK_LEN;
- if (last)
- {
- pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf + pIoContext->m_Overlapped.InternalHigh - last, last);
- }
- }
- }
- }
- }
- }
- void CSATTCPServer::_TaskProcess(PER_IO_CONTEXT* pIoContext, SATPROTO::Package* pak)
- {
- #ifdef _DEBUG
- OutputDebugString("\n########成功获取数据########\n\n");
- #endif
- //GLOBAL::WriteTextLog("成功获取数据");
- SATPROTO::DataHeader *pHeader = &pak->header;
- if ( !pHeader )
- return;
- if ( pHeader->protocol == 0xAA ) {
- switch ( pHeader->cmd )
- {
- case SATPROTO::CMD_LOGIN:
- case SATPROTO::CMD_LOGOUT:
- {
- // 登录;
- SATPROTO::UserInfo *pLogin = (SATPROTO::UserInfo*)pak->buf;
- GLOBAL::WriteTextLog(_T("User=%s, psw=%s"), pLogin->szUserName, pLogin->szPassword);
- bool bRet = CSATExecutor::GetInstance()->Login(pLogin->szUserName, pLogin->szPassword, pLogin->szActuatorName, !pHeader->cmd);
- // 计算数据包长度;
- long len = sizeof(SATPROTO::DataHeader)+sizeof(SATPROTO::LoginResp);
- byte *pbuff = new byte[len];
- memset(pbuff, 0, len);
- SATPROTO::Package *pSendPak = (SATPROTO::Package *)pbuff;
- pSendPak->header.protocol = 0xAA;
- pSendPak->header.cmd = pHeader->cmd;
- pSendPak->header.len = len;
- #if 0
- // 复制内容到指针中;
- memcpy(pSendPak->buf, &bRet, sizeof(bool));
- char szMessage[MAX_PATH] = {0};
- _stprintf_s(szMessage, _T("登录……"));
- memcpy(pSendPak->buf + sizeof(bool), szMessage, MAX_PATH);
- #else
- const SATHTTP::STLoginResp *pstLoginResp = CSATExecutor::GetInstance()->GetLoginResp();
- SATPROTO::LoginResp *pLoginResp = (SATPROTO::LoginResp*)(pbuff+sizeof(SATPROTO::DataHeader));
- pLoginResp->bStatus = bRet;
- memset(pLoginResp->szMessage, 0, MAX_PATH);
- //_stprintf_s(pLoginResp->szMessage, _T("%s"), pstLoginResp->strMessage.c_str());
- //_tcscpy_s(pLoginResp->szMessage, _T("登录描述……"));
- // 如果使用上方两种方式赋值,会导致'\0'后面的字节不为0;
- memcpy(pLoginResp->szMessage, pstLoginResp->strMessage.c_str(), pstLoginResp->strMessage.size());
- #endif
- // 发送,无须判断是否发送成功;
- send(pIoContext->m_sockAccept, (const char*)pbuff, len, 0);
- // 释放内存;
- delete []pbuff;
- pbuff = NULL;
- }
- break;
- case SATPROTO::CMD_ADD_DEVICE:
- case SATPROTO::CMD_DEL_DEVICE:
- {
- std::string ip = (char*)pak->buf;
- if (pHeader->cmd == SATPROTO::CMD_DEL_DEVICE )
- CSATDevices::DelDevices(ip);
- else
- CSATDevices::AddReticleDevices(ip);
- SATPROTO::DataHeader header;
- header.cmd = pHeader->cmd;
- header.len = sizeof(SATPROTO::DataHeader);
- header.protocol = 0xAA;
- send(pIoContext->m_sockAccept, (const char*)&header, sizeof(SATPROTO::DataHeader), 0);
- }
- break;
- case SATPROTO::CMD_QUERY_DEVICES:
- {
- // 计算数据包长度;
- long len = sizeof(SATPROTO::DataHeader)+sizeof(SATPROTO::DeviceResp);
- byte *pbuff = new byte[len];
- memset(pbuff, 0, len);
- SATPROTO::Package *pSendPak = (SATPROTO::Package *)pbuff;
- pSendPak->header.protocol = 0xAA;
- pSendPak->header.cmd = pHeader->cmd;
- pSendPak->header.len = len;
- // 转换pak->buf为结构体;
- SATPROTO::DeviceResp *pDevResp = (SATPROTO::DeviceResp*)(pbuff+sizeof(SATPROTO::DataHeader));
- memset(pDevResp->ssDevs, 0, SATPROTO::MAX_DEVS*sizeof(SATPROTO::Device));
- // 获取设备列表;
- pDevResp->nSize = CSATDevices::AttachDeviceName2Buffer(pDevResp->ssDevs);
- // 返回给客户端;
- send(pIoContext->m_sockAccept, (const char*)pbuff, len, 0);
- // 释放内存;
- delete []pbuff;
- pbuff = NULL;
- }
- break;
- case SATPROTO::CMD_QUERY_TASK:
- {
- long len = sizeof(SATPROTO::DataHeader) + sizeof(SATPROTO::TaskInfo)*SATPROTO::MAX_TASKS;
- byte *pbuff = new byte[len];
- memset(pbuff, 0, len);
- SATPROTO::Package *pSendPak = (SATPROTO::Package *)pbuff;
- pSendPak->header.protocol = 0xAA;
- pSendPak->header.cmd = pHeader->cmd;
- pSendPak->header.len = len;
- // 转换pak->buf为结构体;
- SATPROTO::TaskInfoResp *pTaskResp = (SATPROTO::TaskInfoResp*)(pbuff+sizeof(SATPROTO::DataHeader));
- memset(pTaskResp->ssTasks, 0, SATPROTO::MAX_TASKS*sizeof(SATPROTO::TaskInfo));
- // 获取设备列表;
- pTaskResp->nSize = CSATExecutor::GetInstance()->AttachTaskInfo2Buffer(pTaskResp->ssTasks);
- // 返回给客户端;
- send(pIoContext->m_sockAccept, (const char*)pbuff, len, 0);
- // 释放内存;
- delete []pbuff;
- pbuff = NULL;
- }
- break;
- default:
- break;
- }
- }
- }
- bool CSATTCPServer::CheckDataHeader(SATPROTO::DataHeader* header)
- {
- if ( header->protocol != 0xAA )
- return false;
- return true;
- }
- /////////////////////////////////////////////////////////////////////
- // 判断客户端Socket是否已经断开,否则在一个无效的Socket上投递WSARecv操作会出现异常
- // 使用的方法是尝试向这个socket发送数据,判断这个socket调用的返回值
- // 因为如果客户端网络异常断开(例如客户端崩溃或者拔掉网线等)的时候,服务器端是无法收到客户端断开的通知的
- bool CSATTCPServer::_IsSocketAlive(SOCKET s)
- {
- int nByteSent=send(s,"",0,0);
- if (-1 == nByteSent) return false;
- return true;
- }
- ///////////////////////////////////////////////////////////////////
- // 显示并处理完成端口上的错误
- bool CSATTCPServer::HandleError( PER_SOCKET_CONTEXT *pContext,const DWORD& dwErr )
- {
- // 如果是超时了,就再继续等吧
- if(WAIT_TIMEOUT == dwErr)
- {
- // 确认客户端是否还活着...
- if( !_IsSocketAlive( pContext->m_Socket) )
- {
- this->_ShowMessage( _T("检测到客户端异常退出!") );
- this->_RemoveContext( pContext );
- return true;
- }
- else
- {
- this->_ShowMessage( _T("网络操作超时!重试中...") );
- return true;
- }
- }
- // 可能是客户端异常退出了
- else if( ERROR_NETNAME_DELETED==dwErr )
- {
- this->_ShowMessage( _T("检测到客户端异常退出!") );
- this->_RemoveContext( pContext );
- return true;
- }
- else
- {
- this->_ShowMessage( _T("完成端口操作出现错误,线程退出。错误代码:%d"),dwErr );
- return false;
- }
- }
|