SATTCPServer.cpp 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992
  1. #include "StdAfx.h"
  2. #include "SATTCPServer.h"
  3. //#include "MainDlg.h"
  4. #include "SATExecutor.h"
  5. // 每一个处理器上产生多少个线程(为了最大限度的提升服务器性能,详见配套文档)
  6. #define WORKER_THREADS_PER_PROCESSOR 2
  7. // 同时投递的Accept请求的数量(这个要根据实际的情况灵活设置)
  8. #define MAX_POST_ACCEPT 10
  9. // 传递给Worker线程的退出信号
  10. #define EXIT_CODE NULL
  11. // 释放指针和句柄资源的宏
  12. // 释放指针宏
  13. #define RELEASE(x) {if(x != NULL ){delete x;x=NULL;}}
  14. // 释放句柄宏
  15. #define RELEASE_HANDLE(x) {if(x != NULL && x!=INVALID_HANDLE_VALUE){ CloseHandle(x);x = NULL;}}
  16. // 释放Socket宏
  17. #define RELEASE_SOCKET(x) {if(x !=INVALID_SOCKET) { closesocket(x);x=INVALID_SOCKET;}}
  18. CSATTCPServer::CSATTCPServer(void):
  19. m_nThreads(0),
  20. m_hShutdownEvent(NULL),
  21. m_hIOCompletionPort(NULL),
  22. m_phWorkerThreads(NULL),
  23. m_strIP(DEFAULT_IP),
  24. m_nPort(DEFAULT_PORT),
  25. m_pMain(NULL),
  26. m_lpfnAcceptEx( NULL ),
  27. m_pListenContext( NULL )
  28. {
  29. }
  30. CSATTCPServer::~CSATTCPServer(void)
  31. {
  32. // 确保资源彻底释放
  33. this->Stop();
  34. }
  35. ///////////////////////////////////////////////////////////////////
  36. // 工作者线程: 为IOCP请求服务的工作者线程
  37. // 也就是每当完成端口上出现了完成数据包,就将之取出来进行处理的线程
  38. ///////////////////////////////////////////////////////////////////
  39. DWORD WINAPI CSATTCPServer::_WorkerThread(LPVOID lpParam)
  40. {
  41. THREADPARAMS_WORKER* pParam = (THREADPARAMS_WORKER*)lpParam;
  42. CSATTCPServer* pIOCPModel = (CSATTCPServer*)pParam->pIOCPModel;
  43. int nThreadNo = (int)pParam->nThreadNo;
  44. pIOCPModel->_ShowMessage(_T("工作者线程启动,ID: %d."),nThreadNo);
  45. OVERLAPPED *pOverlapped = NULL;
  46. PER_SOCKET_CONTEXT *pSocketContext = NULL;
  47. DWORD dwBytesTransfered = 0;
  48. // 循环处理请求,知道接收到Shutdown信息为止
  49. while (WAIT_OBJECT_0 != WaitForSingleObject(pIOCPModel->m_hShutdownEvent, 0))
  50. {
  51. BOOL bReturn = GetQueuedCompletionStatus(
  52. pIOCPModel->m_hIOCompletionPort,
  53. &dwBytesTransfered,
  54. (PULONG_PTR)&pSocketContext,
  55. &pOverlapped,
  56. INFINITE);
  57. // 如果收到的是退出标志,则直接退出
  58. if ( EXIT_CODE==(DWORD)pSocketContext )
  59. {
  60. break;
  61. }
  62. // 判断是否出现了错误
  63. if( !bReturn )
  64. {
  65. DWORD dwErr = GetLastError();
  66. // 显示一下提示信息
  67. if( !pIOCPModel->HandleError( pSocketContext,dwErr ) )
  68. {
  69. break;
  70. }
  71. continue;
  72. }
  73. else
  74. {
  75. // 读取传入的参数
  76. PER_IO_CONTEXT* pIoContext = CONTAINING_RECORD(pOverlapped, PER_IO_CONTEXT, m_Overlapped);
  77. // 判断是否有客户端断开了
  78. if((0 == dwBytesTransfered) && ( RECV_POSTED==pIoContext->m_OpType || SEND_POSTED==pIoContext->m_OpType))
  79. {
  80. pIOCPModel->_ShowMessage( _T("客户端 %s:%d 断开连接."),
  81. inet_ntoa(pSocketContext->m_ClientAddr.sin_addr),
  82. ntohs(pSocketContext->m_ClientAddr.sin_port) );
  83. // 释放掉对应的资源
  84. pIOCPModel->_RemoveContext( pSocketContext );
  85. continue;
  86. }
  87. else
  88. {
  89. switch( pIoContext->m_OpType )
  90. {
  91. // Accept
  92. case ACCEPT_POSTED:
  93. {
  94. // 为了增加代码可读性,这里用专门的_DoAccept函数进行处理连入请求
  95. pIOCPModel->_DoAccpet( pSocketContext, pIoContext );
  96. }
  97. break;
  98. // RECV
  99. case RECV_POSTED:
  100. {
  101. // 为了增加代码可读性,这里用专门的_DoRecv函数进行处理接收请求
  102. pIOCPModel->_DoRecv( pSocketContext,pIoContext );
  103. }
  104. break;
  105. // SEND
  106. // 这里略过不写了,要不代码太多了,不容易理解,Send操作相对来讲简单一些
  107. case SEND_POSTED:
  108. {
  109. }
  110. break;
  111. default:
  112. // 不应该执行到这里
  113. TRACE(_T("_WorkThread中的 pIoContext->m_OpType 参数异常.\n"));
  114. break;
  115. } //switch
  116. }//if
  117. }//if
  118. }//while
  119. TRACE(_T("工作者线程 %d 号退出.\n"),nThreadNo);
  120. // 释放线程参数
  121. RELEASE(lpParam);
  122. return 0;
  123. }
  124. //====================================================================================
  125. //
  126. // 系统初始化和终止
  127. //
  128. //====================================================================================
  129. ////////////////////////////////////////////////////////////////////
  130. // 初始化WinSock 2.2
  131. bool CSATTCPServer::LoadSocketLib()
  132. {
  133. WSADATA wsaData;
  134. int nResult;
  135. nResult = WSAStartup(MAKEWORD(2,2), &wsaData);
  136. // 错误(一般都不可能出现)
  137. if (NO_ERROR != nResult)
  138. {
  139. this->_ShowMessage(_T("初始化WinSock 2.2失败!\n"));
  140. return false;
  141. }
  142. return true;
  143. }
  144. //////////////////////////////////////////////////////////////////
  145. // 启动服务器
  146. bool CSATTCPServer::Start(unsigned int port)
  147. {
  148. // 初始化线程互斥量
  149. InitializeCriticalSection(&m_csContextList);
  150. // 建立系统退出的事件通知
  151. m_hShutdownEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
  152. // 初始化IOCP
  153. if (false == _InitializeIOCP())
  154. {
  155. this->_ShowMessage(_T("初始化IOCP失败!\n"));
  156. return false;
  157. }
  158. else
  159. {
  160. this->_ShowMessage(_T("\nIOCP初始化完毕\n."));
  161. }
  162. // 初始化Socket
  163. if( false==_InitializeListenSocket(port) )
  164. {
  165. this->_ShowMessage(_T("Listen Socket初始化失败!\n"));
  166. this->_DeInitialize();
  167. return false;
  168. }
  169. else
  170. {
  171. this->_ShowMessage(_T("Listen Socket初始化完毕."));
  172. }
  173. this->_ShowMessage(_T("系统准备就绪,等候连接....\n"));
  174. return true;
  175. }
  176. ////////////////////////////////////////////////////////////////////
  177. // 开始发送系统退出消息,退出完成端口和线程资源
  178. void CSATTCPServer::Stop()
  179. {
  180. if( m_pListenContext!=NULL && m_pListenContext->m_Socket!=INVALID_SOCKET )
  181. {
  182. // 激活关闭消息通知
  183. SetEvent(m_hShutdownEvent);
  184. for (int i = 0; i < m_nThreads; i++)
  185. {
  186. // 通知所有的完成端口操作退出
  187. PostQueuedCompletionStatus(m_hIOCompletionPort, 0, (DWORD)EXIT_CODE, NULL);
  188. }
  189. // 等待所有的客户端资源退出
  190. WaitForMultipleObjects(m_nThreads, m_phWorkerThreads, TRUE, INFINITE);
  191. // 清除客户端列表信息
  192. this->_ClearContextList();
  193. // 释放其他资源
  194. this->_DeInitialize();
  195. this->_ShowMessage(_T("停止监听\n"));
  196. }
  197. }
  198. ////////////////////////////////
  199. // 初始化完成端口
  200. bool CSATTCPServer::_InitializeIOCP()
  201. {
  202. // 建立第一个完成端口
  203. m_hIOCompletionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0 );
  204. if ( NULL == m_hIOCompletionPort)
  205. {
  206. this->_ShowMessage(_T("建立完成端口失败!错误代码: %d!\n"), WSAGetLastError());
  207. return false;
  208. }
  209. // 根据本机中的处理器数量,建立对应的线程数
  210. m_nThreads = WORKER_THREADS_PER_PROCESSOR * _GetNoOfProcessors();
  211. // 为工作者线程初始化句柄
  212. m_phWorkerThreads = new HANDLE[m_nThreads];
  213. // 根据计算出来的数量建立工作者线程
  214. DWORD nThreadID;
  215. for (int i = 0; i < m_nThreads; i++)
  216. {
  217. THREADPARAMS_WORKER* pThreadParams = new THREADPARAMS_WORKER;
  218. pThreadParams->pIOCPModel = this;
  219. pThreadParams->nThreadNo = i+1;
  220. m_phWorkerThreads[i] = ::CreateThread(0, 0, _WorkerThread, (void *)pThreadParams, 0, &nThreadID);
  221. }
  222. TRACE(" 建立 _WorkerThread %d 个.\n", m_nThreads );
  223. return true;
  224. }
  225. /////////////////////////////////////////////////////////////////
  226. // 初始化Socket
  227. bool CSATTCPServer::_InitializeListenSocket(unsigned int port)
  228. {
  229. // AcceptEx 和 GetAcceptExSockaddrs 的GUID,用于导出函数指针
  230. GUID GuidAcceptEx = WSAID_ACCEPTEX;
  231. GUID GuidGetAcceptExSockAddrs = WSAID_GETACCEPTEXSOCKADDRS;
  232. // 服务器地址信息,用于绑定Socket
  233. struct sockaddr_in ServerAddress;
  234. // 生成用于监听的Socket的信息
  235. m_pListenContext = new PER_SOCKET_CONTEXT;
  236. // 需要使用重叠IO,必须得使用WSASocket来建立Socket,才可以支持重叠IO操作
  237. m_pListenContext->m_Socket = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);
  238. if (INVALID_SOCKET == m_pListenContext->m_Socket)
  239. {
  240. this->_ShowMessage(_T("初始化Socket失败,错误代码: %d.\n"), WSAGetLastError());
  241. return false;
  242. }
  243. else
  244. {
  245. TRACE("WSASocket() 完成.\n");
  246. }
  247. // 将Listen Socket绑定至完成端口中
  248. if( NULL== CreateIoCompletionPort( (HANDLE)m_pListenContext->m_Socket, m_hIOCompletionPort,(DWORD)m_pListenContext, 0))
  249. {
  250. this->_ShowMessage(_T("绑定 Listen Socket至完成端口失败!错误代码: %d/n"), WSAGetLastError());
  251. RELEASE_SOCKET( m_pListenContext->m_Socket );
  252. return false;
  253. }
  254. else
  255. {
  256. TRACE("Listen Socket绑定完成端口 完成.\n");
  257. }
  258. // 填充地址信息
  259. ZeroMemory((char *)&ServerAddress, sizeof(ServerAddress));
  260. ServerAddress.sin_family = AF_INET;
  261. // 这里可以绑定任何可用的IP地址,或者绑定一个指定的IP地址
  262. ServerAddress.sin_addr.s_addr = htonl(INADDR_ANY);
  263. //ServerAddress.sin_addr.s_addr = inet_addr(m_strIP.GetString());
  264. ServerAddress.sin_port = htons(port);
  265. // 绑定地址和端口
  266. if (SOCKET_ERROR == ::bind(m_pListenContext->m_Socket, (struct sockaddr *) &ServerAddress, sizeof(ServerAddress)))
  267. {
  268. this->_ShowMessage(_T("bind()函数执行错误! 错误代码: %d/n"), WSAGetLastError());
  269. return false;
  270. }
  271. else
  272. {
  273. TRACE("bind() 完成.\n");
  274. }
  275. // 开始进行监听
  276. if (SOCKET_ERROR == listen(m_pListenContext->m_Socket,SOMAXCONN))
  277. {
  278. this->_ShowMessage(_T("Listen()函数执行出现错误; 错误代码: %d/n"), WSAGetLastError());
  279. return false;
  280. }
  281. else
  282. {
  283. TRACE("Listen() 完成.\n");
  284. }
  285. // 使用AcceptEx函数,因为这个是属于WinSock2规范之外的微软另外提供的扩展函数
  286. // 所以需要额外获取一下函数的指针,
  287. // 获取AcceptEx函数指针
  288. DWORD dwBytes = 0;
  289. if(SOCKET_ERROR == WSAIoctl(
  290. m_pListenContext->m_Socket,
  291. SIO_GET_EXTENSION_FUNCTION_POINTER,
  292. &GuidAcceptEx,
  293. sizeof(GuidAcceptEx),
  294. &m_lpfnAcceptEx,
  295. sizeof(m_lpfnAcceptEx),
  296. &dwBytes,
  297. NULL,
  298. NULL))
  299. {
  300. this->_ShowMessage(_T("WSAIoctl 未能获取AcceptEx函数指针。错误代码: %d\n"), WSAGetLastError());
  301. this->_DeInitialize();
  302. return false;
  303. }
  304. // 获取GetAcceptExSockAddrs函数指针,也是同理
  305. if(SOCKET_ERROR == WSAIoctl(
  306. m_pListenContext->m_Socket,
  307. SIO_GET_EXTENSION_FUNCTION_POINTER,
  308. &GuidGetAcceptExSockAddrs,
  309. sizeof(GuidGetAcceptExSockAddrs),
  310. &m_lpfnGetAcceptExSockAddrs,
  311. sizeof(m_lpfnGetAcceptExSockAddrs),
  312. &dwBytes,
  313. NULL,
  314. NULL))
  315. {
  316. this->_ShowMessage(_T("WSAIoctl 未能获取GuidGetAcceptExSockAddrs函数指针。错误代码: %d\n"), WSAGetLastError());
  317. this->_DeInitialize();
  318. return false;
  319. }
  320. // 为AcceptEx 准备参数,然后投递AcceptEx I/O请求
  321. for( int i=0;i<MAX_POST_ACCEPT;i++ )
  322. {
  323. // 新建一个IO_CONTEXT
  324. PER_IO_CONTEXT* pAcceptIoContext = m_pListenContext->GetNewIoContext();
  325. if( false==this->_PostAccept( pAcceptIoContext ) )
  326. {
  327. m_pListenContext->RemoveContext(pAcceptIoContext);
  328. return false;
  329. }
  330. }
  331. this->_ShowMessage( _T("投递 %d 个AcceptEx请求完毕"),MAX_POST_ACCEPT );
  332. return true;
  333. }
  334. ////////////////////////////////////////////////////////////
  335. // 最后释放掉所有资源
  336. void CSATTCPServer::_DeInitialize()
  337. {
  338. // 删除客户端列表的互斥量
  339. DeleteCriticalSection(&m_csContextList);
  340. // 关闭系统退出事件句柄
  341. RELEASE_HANDLE(m_hShutdownEvent);
  342. // 释放工作者线程句柄指针
  343. for( int i=0;i<m_nThreads;i++ )
  344. {
  345. RELEASE_HANDLE(m_phWorkerThreads[i]);
  346. }
  347. RELEASE(m_phWorkerThreads);
  348. // 关闭IOCP句柄
  349. RELEASE_HANDLE(m_hIOCompletionPort);
  350. // 关闭监听Socket
  351. RELEASE(m_pListenContext);
  352. this->_ShowMessage(_T("释放资源完毕.\n"));
  353. }
  354. //====================================================================================
  355. //
  356. // 投递完成端口请求
  357. //
  358. //====================================================================================
  359. //////////////////////////////////////////////////////////////////
  360. // 投递Accept请求
  361. bool CSATTCPServer::_PostAccept( PER_IO_CONTEXT* pAcceptIoContext )
  362. {
  363. ASSERT( INVALID_SOCKET!=m_pListenContext->m_Socket );
  364. // 准备参数
  365. DWORD dwBytes = 0;
  366. pAcceptIoContext->m_OpType = ACCEPT_POSTED;
  367. WSABUF *p_wbuf = &pAcceptIoContext->m_wsaBuf;
  368. OVERLAPPED *p_ol = &pAcceptIoContext->m_Overlapped;
  369. // 为以后新连入的客户端先准备好Socket( 这个是与传统accept最大的区别 )
  370. pAcceptIoContext->m_sockAccept = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
  371. if( INVALID_SOCKET==pAcceptIoContext->m_sockAccept )
  372. {
  373. _ShowMessage(_T("创建用于Accept的Socket失败!错误代码: %d"), WSAGetLastError());
  374. return false;
  375. }
  376. // 投递AcceptEx
  377. if(FALSE == m_lpfnAcceptEx( m_pListenContext->m_Socket, pAcceptIoContext->m_sockAccept, p_wbuf->buf, p_wbuf->len - ((sizeof(SOCKADDR_IN)+16)*2),
  378. sizeof(SOCKADDR_IN)+16, sizeof(SOCKADDR_IN)+16, &dwBytes, p_ol))
  379. {
  380. if(WSA_IO_PENDING != WSAGetLastError())
  381. {
  382. _ShowMessage(_T("投递 AcceptEx 请求失败,错误代码: %d"), WSAGetLastError());
  383. return false;
  384. }
  385. }
  386. return true;
  387. }
  388. ////////////////////////////////////////////////////////////
  389. // 在有客户端连入的时候,进行处理
  390. // 流程有点复杂,你要是看不懂的话,就看配套的文档吧....
  391. // 如果能理解这里的话,完成端口的机制你就消化了一大半了
  392. // 总之你要知道,传入的是ListenSocket的Context,我们需要复制一份出来给新连入的Socket用
  393. // 原来的Context还是要在上面继续投递下一个Accept请求
  394. //
  395. bool CSATTCPServer::_DoAccpet( PER_SOCKET_CONTEXT* pSocketContext, PER_IO_CONTEXT* pIoContext )
  396. {
  397. SOCKADDR_IN* ClientAddr = NULL;
  398. SOCKADDR_IN* LocalAddr = NULL;
  399. int remoteLen = sizeof(SOCKADDR_IN), localLen = sizeof(SOCKADDR_IN);
  400. ///////////////////////////////////////////////////////////////////////////
  401. // 1. 首先取得连入客户端的地址信息
  402. // 这个 m_lpfnGetAcceptExSockAddrs 不得了啊~~~~~~
  403. // 不但可以取得客户端和本地端的地址信息,还能顺便取出客户端发来的第一组数据
  404. this->m_lpfnGetAcceptExSockAddrs(pIoContext->m_wsaBuf.buf, pIoContext->m_wsaBuf.len - ((sizeof(SOCKADDR_IN)+16)*2),
  405. sizeof(SOCKADDR_IN)+16, sizeof(SOCKADDR_IN)+16, (LPSOCKADDR*)&LocalAddr, &localLen, (LPSOCKADDR*)&ClientAddr, &remoteLen);
  406. _RecvProcess(pSocketContext, pIoContext);
  407. this->_ShowMessage( _T("客户端 %s:%d 连入."), inet_ntoa(ClientAddr->sin_addr), ntohs(ClientAddr->sin_port) );
  408. //this->_ShowMessage( _T("客户额 %s:%d 信息:%s."),inet_ntoa(ClientAddr->sin_addr), ntohs(ClientAddr->sin_port),pIoContext->m_wsaBuf.buf );
  409. //if ( Global::g_bEnableLog )
  410. //{
  411. // // 解析Json字符串;
  412. // Global::TLog tlog;
  413. // cJSON *pJson = cJSON_Parse(Global::DeCode_URLUNICODE(pIoContext->m_wsaBuf.buf).c_str());
  414. // if ( pJson )
  415. // {
  416. // tlog.report_type = cJSON_GetObjectItem(pJson, _T("ReportType")) ? cJSON_GetObjectItem(pJson, _T("ReportType"))->valuestring : "";
  417. // tlog.report_data = cJSON_GetObjectItem(pJson, _T("prinMsg")) ? cJSON_GetObjectItem(pJson, _T("prinMsg"))->valuestring : "";
  418. // if ( _tcscmp(tlog.report_type.c_str(), _T("printLog")) == 0 )
  419. // {
  420. // Global::WritePythonLog(tlog.report_data.c_str());
  421. // Global::g_time = time(NULL);
  422. // Global::g_lastTime = COleDateTime::GetCurrentTime();
  423. // }
  424. // cJSON_Delete(pJson);
  425. // pJson = NULL;
  426. // }
  427. //}
  428. //////////////////////////////////////////////////////////////////////////////////////////////////////
  429. // 2. 这里需要注意,这里传入的这个是ListenSocket上的Context,这个Context我们还需要用于监听下一个连接
  430. // 所以我还得要将ListenSocket上的Context复制出来一份为新连入的Socket新建一个SocketContext
  431. PER_SOCKET_CONTEXT* pNewSocketContext = new PER_SOCKET_CONTEXT;
  432. pNewSocketContext->m_Socket = pIoContext->m_sockAccept;
  433. memcpy(&(pNewSocketContext->m_ClientAddr), ClientAddr, sizeof(SOCKADDR_IN));
  434. // 参数设置完毕,将这个Socket和完成端口绑定(这也是一个关键步骤)
  435. if( false==this->_AssociateWithIOCP( pNewSocketContext ) )
  436. {
  437. RELEASE( pNewSocketContext );
  438. return false;
  439. }
  440. ///////////////////////////////////////////////////////////////////////////////////////////////////
  441. // 3. 继续,建立其下的IoContext,用于在这个Socket上投递第一个Recv数据请求
  442. PER_IO_CONTEXT* pNewIoContext = pNewSocketContext->GetNewIoContext();
  443. pNewIoContext->m_OpType = RECV_POSTED;
  444. pNewIoContext->m_sockAccept = pNewSocketContext->m_Socket;
  445. // 如果Buffer需要保留,就自己拷贝一份出来
  446. //memcpy( pNewIoContext->m_szBuffer,pIoContext->m_szBuffer,MAX_BUFFER_LEN );
  447. // 绑定完毕之后,就可以开始在这个Socket上投递完成请求了
  448. if( false==this->_PostRecv( pNewIoContext) )
  449. {
  450. pNewSocketContext->RemoveContext( pNewIoContext );
  451. return false;
  452. }
  453. /////////////////////////////////////////////////////////////////////////////////////////////////
  454. // 4. 如果投递成功,那么就把这个有效的客户端信息,加入到ContextList中去(需要统一管理,方便释放资源)
  455. this->_AddToContextList( pNewSocketContext );
  456. ////////////////////////////////////////////////////////////////////////////////////////////////
  457. // 5. 使用完毕之后,把Listen Socket的那个IoContext重置,然后准备投递新的AcceptEx
  458. pIoContext->ResetBuffer();
  459. return this->_PostAccept( pIoContext );
  460. }
  461. ////////////////////////////////////////////////////////////////////
  462. // 投递接收数据请求
  463. bool CSATTCPServer::_PostRecv( PER_IO_CONTEXT* pIoContext )
  464. {
  465. // 初始化变量
  466. DWORD dwFlags = 0;
  467. DWORD dwBytes = 0;
  468. WSABUF *p_wbuf = &pIoContext->m_wsaBuf;
  469. OVERLAPPED *p_ol = &pIoContext->m_Overlapped;
  470. pIoContext->ResetBuffer();
  471. pIoContext->m_OpType = RECV_POSTED;
  472. // 初始化完成后,,投递WSARecv请求
  473. int nBytesRecv = WSARecv( pIoContext->m_sockAccept, p_wbuf, 1, &dwBytes, &dwFlags, p_ol, NULL );
  474. // 如果返回值错误,并且错误的代码并非是Pending的话,那就说明这个重叠请求失败了
  475. if ((SOCKET_ERROR == nBytesRecv) && (WSA_IO_PENDING != WSAGetLastError()))
  476. {
  477. this->_ShowMessage(_T("投递第一个WSARecv失败!"));
  478. return false;
  479. }
  480. return true;
  481. }
  482. /////////////////////////////////////////////////////////////////
  483. // 在有接收的数据到达的时候,进行处理
  484. bool CSATTCPServer::_DoRecv( PER_SOCKET_CONTEXT* pSocketContext, PER_IO_CONTEXT* pIoContext )
  485. {
  486. // 先把上一次的数据显示出现,然后就重置状态,发出下一个Recv请求
  487. SOCKADDR_IN* ClientAddr = &pSocketContext->m_ClientAddr;
  488. //this->_ShowMessage( _T("收到 %s:%d 信息:%s"),inet_ntoa(ClientAddr->sin_addr), ntohs(ClientAddr->sin_port), pIoContext->m_wsaBuf.buf );
  489. _RecvProcess(pSocketContext, pIoContext);
  490. //if ( Global::g_bEnableLog )
  491. //{
  492. // // 解析Json字符串;
  493. // Global::TLog tlog;
  494. // cJSON *pJson = cJSON_Parse(Global::DeCode_URLUNICODE(pIoContext->m_wsaBuf.buf).c_str());
  495. // if ( pJson )
  496. // {
  497. // tlog.report_type = cJSON_GetObjectItem(pJson, _T("ReportType")) ? cJSON_GetObjectItem(pJson, _T("ReportType"))->valuestring : "";
  498. // tlog.report_data = cJSON_GetObjectItem(pJson, _T("prinMsg")) ? cJSON_GetObjectItem(pJson, _T("prinMsg"))->valuestring : "";
  499. // if ( _tcscmp(tlog.report_type.c_str(), _T("printLog")) == 0 )
  500. // {
  501. // Global::WritePythonLog(tlog.report_data.c_str());
  502. // Global::g_time = time(NULL);
  503. // Global::g_lastTime = COleDateTime::GetCurrentTime();
  504. // }
  505. // cJSON_Delete(pJson);
  506. // pJson = NULL;
  507. // }
  508. //
  509. //}
  510. // 然后开始投递下一个WSARecv请求
  511. return _PostRecv( pIoContext );
  512. }
  513. /////////////////////////////////////////////////////
  514. // 将句柄(Socket)绑定到完成端口中
  515. bool CSATTCPServer::_AssociateWithIOCP( PER_SOCKET_CONTEXT *pContext )
  516. {
  517. // 将用于和客户端通信的SOCKET绑定到完成端口中
  518. HANDLE hTemp = CreateIoCompletionPort((HANDLE)pContext->m_Socket, m_hIOCompletionPort, (DWORD)pContext, 0);
  519. if (NULL == hTemp)
  520. {
  521. this->_ShowMessage(_T("执行CreateIoCompletionPort()出现错误.错误代码:%d"),GetLastError());
  522. return false;
  523. }
  524. return true;
  525. }
  526. //====================================================================================
  527. //
  528. // ContextList 相关操作
  529. //
  530. //====================================================================================
  531. //////////////////////////////////////////////////////////////
  532. // 将客户端的相关信息存储到数组中
  533. void CSATTCPServer::_AddToContextList( PER_SOCKET_CONTEXT *pHandleData )
  534. {
  535. EnterCriticalSection(&m_csContextList);
  536. m_arrayClientContext.Add(pHandleData);
  537. LeaveCriticalSection(&m_csContextList);
  538. }
  539. ////////////////////////////////////////////////////////////////
  540. // 移除某个特定的Context
  541. void CSATTCPServer::_RemoveContext( PER_SOCKET_CONTEXT *pSocketContext )
  542. {
  543. EnterCriticalSection(&m_csContextList);
  544. for( int i=0;i<m_arrayClientContext.GetCount();i++ )
  545. {
  546. if( pSocketContext==m_arrayClientContext.GetAt(i) )
  547. {
  548. RELEASE( pSocketContext );
  549. m_arrayClientContext.RemoveAt(i);
  550. break;
  551. }
  552. }
  553. LeaveCriticalSection(&m_csContextList);
  554. }
  555. ////////////////////////////////////////////////////////////////
  556. // 清空客户端信息
  557. void CSATTCPServer::_ClearContextList()
  558. {
  559. EnterCriticalSection(&m_csContextList);
  560. for( int i=0;i<m_arrayClientContext.GetCount();i++ )
  561. {
  562. delete m_arrayClientContext.GetAt(i);
  563. }
  564. m_arrayClientContext.RemoveAll();
  565. LeaveCriticalSection(&m_csContextList);
  566. }
  567. //====================================================================================
  568. //
  569. // 其他辅助函数定义
  570. //
  571. //====================================================================================
  572. ////////////////////////////////////////////////////////////////////
  573. // 获得本机的IP地址
  574. CString CSATTCPServer::GetLocalIP()
  575. {
  576. // 获得本机主机名
  577. char hostname[MAX_PATH] = {0};
  578. gethostname(hostname,MAX_PATH);
  579. struct hostent FAR* lpHostEnt = gethostbyname(hostname);
  580. if(lpHostEnt == NULL)
  581. {
  582. return DEFAULT_IP;
  583. }
  584. // 取得IP地址列表中的第一个为返回的IP(因为一台主机可能会绑定多个IP)
  585. LPSTR lpAddr = lpHostEnt->h_addr_list[0];
  586. // 将IP地址转化成字符串形式
  587. struct in_addr inAddr;
  588. memmove(&inAddr,lpAddr,4);
  589. m_strIP = CString( inet_ntoa(inAddr) );
  590. return m_strIP;
  591. }
  592. ///////////////////////////////////////////////////////////////////
  593. // 获得本机中处理器的数量
  594. int CSATTCPServer::_GetNoOfProcessors()
  595. {
  596. SYSTEM_INFO si;
  597. GetSystemInfo(&si);
  598. return si.dwNumberOfProcessors;
  599. }
  600. /////////////////////////////////////////////////////////////////////
  601. // 在主界面中显示提示信息
  602. void CSATTCPServer::_ShowMessage(const CString szFormat,...) const
  603. {
  604. // 根据传入的参数格式化字符串
  605. CString strMessage;
  606. va_list arglist;
  607. // 处理变长参数
  608. va_start(arglist, szFormat);
  609. strMessage.FormatV(szFormat,arglist);
  610. va_end(arglist);
  611. #if 0
  612. // 在主界面中显示
  613. CMainDlg* pMain = (CMainDlg*)m_pMain;
  614. if( m_pMain!=NULL )
  615. {
  616. pMain->AddInformation(strMessage);
  617. TRACE( strMessage+_T("\n") );
  618. }
  619. #else
  620. // Global::WriteTextLog(strMessage);
  621. #endif
  622. }
  623. #define PAK_LEN sizeof(DataHeader)
  624. void CSATTCPServer::_RecvProcess(PER_SOCKET_CONTEXT* pSocketContext, PER_IO_CONTEXT* pIoContext)
  625. {
  626. // 小于包头;
  627. DataHeader* phead = NULL;
  628. if (pSocketContext->lastData.size() == 0)
  629. {
  630. // 不足包头;
  631. if (PAK_LEN > pIoContext->m_Overlapped.InternalHigh)
  632. {
  633. OutputDebugString("A:不足包头;\n");
  634. pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf, pIoContext->m_Overlapped.InternalHigh);
  635. }
  636. else
  637. {
  638. // 包头充足;
  639. phead = (DataHeader*)pIoContext->m_wsaBuf.buf;
  640. // 完整的包;
  641. if (phead->len == pIoContext->m_Overlapped.InternalHigh)
  642. {
  643. OutputDebugString("A:完整的包;\n");
  644. _TaskProcess(pIoContext, (Package*)pIoContext->m_wsaBuf.buf);
  645. }
  646. // 小包;
  647. else if (phead->len > pIoContext->m_Overlapped.InternalHigh)
  648. {
  649. OutputDebugString("A:小包;\n");
  650. pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf, pIoContext->m_Overlapped.InternalHigh);
  651. }
  652. // 超包;
  653. else if (phead->len < pIoContext->m_Overlapped.InternalHigh)
  654. {
  655. OutputDebugString("A:超包;\n");
  656. pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf + phead->len, pIoContext->m_Overlapped.InternalHigh - phead->len);
  657. _TaskProcess(pIoContext, (Package*)pIoContext->m_wsaBuf.buf);
  658. }
  659. }
  660. }
  661. else
  662. {
  663. int lastlen = pIoContext->m_Overlapped.InternalHigh;
  664. if (pSocketContext->lastData.size() >= PAK_LEN)
  665. {
  666. phead = (DataHeader*)pSocketContext->lastData.data();
  667. if (phead->len <= pSocketContext->lastData.size() + pIoContext->m_Overlapped.InternalHigh)
  668. {
  669. if ( phead->len <= pSocketContext->lastData.size() )
  670. {
  671. OutputDebugString("C:超包;\n");
  672. // 完整包;
  673. _TaskProcess(pIoContext, (Package*)pSocketContext->lastData.substr(0, phead->len).data());
  674. pSocketContext->lastData = pSocketContext->lastData.substr(phead->len);
  675. pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf, pIoContext->m_Overlapped.InternalHigh);
  676. }
  677. else
  678. {
  679. OutputDebugString("D:超包;\n");
  680. lastlen = pSocketContext->lastData.size() + pIoContext->m_Overlapped.InternalHigh - phead->len;
  681. pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf, pIoContext->m_Overlapped.InternalHigh - lastlen);
  682. // 完整包;
  683. _TaskProcess(pIoContext, (Package*)pSocketContext->lastData.data());
  684. // 剩余包;
  685. pSocketContext->lastData.clear();
  686. if (lastlen)
  687. pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf + pIoContext->m_Overlapped.InternalHigh - lastlen, lastlen);
  688. }
  689. }
  690. else
  691. {
  692. OutputDebugString("C:仍不足一个包;\n");
  693. pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf, pIoContext->m_Overlapped.InternalHigh);
  694. }
  695. }
  696. else
  697. {
  698. // 包头剩余长度;
  699. int diflen = PAK_LEN - pSocketContext->lastData.size();
  700. // 仍不足一个包头;
  701. if ( diflen > pIoContext->m_Overlapped.InternalHigh )
  702. {
  703. OutputDebugString("B:仍不足一个包头;\n");
  704. pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf, pIoContext->m_Overlapped.InternalHigh);
  705. }
  706. else
  707. {
  708. // 拼成完整包头;
  709. pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf, diflen);
  710. phead = (DataHeader*)pSocketContext->lastData.data();
  711. // 完整包;
  712. if ( phead->len == PAK_LEN + pIoContext->m_Overlapped.InternalHigh - diflen )
  713. {
  714. OutputDebugString("B:完整包;\n");
  715. pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf + diflen, pIoContext->m_Overlapped.InternalHigh - diflen);
  716. _TaskProcess(pIoContext, (Package*)pSocketContext->lastData.data());
  717. pSocketContext->lastData.clear();
  718. }
  719. // 小包;
  720. else if ( phead->len > PAK_LEN + pIoContext->m_Overlapped.InternalHigh - diflen)
  721. {
  722. OutputDebugString("B:小包;\n");
  723. pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf + diflen, pIoContext->m_Overlapped.InternalHigh - diflen);
  724. }
  725. // 超包;
  726. else if (phead->len < PAK_LEN + pIoContext->m_Overlapped.InternalHigh - diflen)
  727. {
  728. OutputDebugString("B:超包;\n");
  729. // 组完成包;
  730. pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf + diflen, phead->len - PAK_LEN);
  731. _TaskProcess(pIoContext, (Package*)pSocketContext->lastData.data());
  732. pSocketContext->lastData.clear();
  733. int last = pIoContext->m_Overlapped.InternalHigh - diflen - phead->len + PAK_LEN;
  734. if (last)
  735. {
  736. pSocketContext->lastData.append(pIoContext->m_wsaBuf.buf + pIoContext->m_Overlapped.InternalHigh - last, last);
  737. }
  738. }
  739. }
  740. }
  741. }
  742. }
  743. void CSATTCPServer::_TaskProcess(PER_IO_CONTEXT* pIoContext, Package* pak)
  744. {
  745. #ifdef _DEBUG
  746. OutputDebugString("成功获取数据\n");
  747. #endif
  748. DataHeader *pHeader = &pak->header;
  749. if ( !pHeader )
  750. return;
  751. if ( pHeader->protocol == 0xAA ) {
  752. if ( pHeader->cmd == CMD_LOGIN ) {// 登录;
  753. TPLogin *pLogin = (TPLogin*)pak->buf;
  754. bool bRet = CSATExecutor::GetInstance()->Login(pLogin->szUserName, pLogin->szPassword);
  755. // 计算数据包长度;
  756. long len = sizeof(DataHeader)+sizeof(TPResponse);
  757. byte *pbuff = new byte[len];
  758. Package *pSendPak = (Package *)pbuff;
  759. pSendPak->header.protocol = 0xAA;
  760. pSendPak->header.cmd = CMD_LOGIN;
  761. pSendPak->header.len = len;
  762. // 复制内容到指针中;
  763. memcpy(pSendPak->buf, &bRet, sizeof(bool));
  764. int ret = send(pIoContext->m_sockAccept, (const char*)pbuff, len, 0);
  765. }
  766. else if ( pHeader->cmd == CMD_LOGOUT ) {// 登出;
  767. }
  768. else if ( pHeader->cmd == CMD_ADD_DEVICE ) {// 添加设备;
  769. }
  770. else if ( pHeader->cmd == CMD_DEL_DEVICE ) {// 删除设备;
  771. }
  772. else if ( pHeader->cmd == CMD_QUERY_DEVICES ) {// 查询设备;
  773. }
  774. else {// 无效指令;
  775. }
  776. }
  777. else if ( pHeader->protocol == 0xAB )
  778. {
  779. }
  780. }
  781. /////////////////////////////////////////////////////////////////////
  782. // 判断客户端Socket是否已经断开,否则在一个无效的Socket上投递WSARecv操作会出现异常
  783. // 使用的方法是尝试向这个socket发送数据,判断这个socket调用的返回值
  784. // 因为如果客户端网络异常断开(例如客户端崩溃或者拔掉网线等)的时候,服务器端是无法收到客户端断开的通知的
  785. bool CSATTCPServer::_IsSocketAlive(SOCKET s)
  786. {
  787. int nByteSent=send(s,"",0,0);
  788. if (-1 == nByteSent) return false;
  789. return true;
  790. }
  791. ///////////////////////////////////////////////////////////////////
  792. // 显示并处理完成端口上的错误
  793. bool CSATTCPServer::HandleError( PER_SOCKET_CONTEXT *pContext,const DWORD& dwErr )
  794. {
  795. // 如果是超时了,就再继续等吧
  796. if(WAIT_TIMEOUT == dwErr)
  797. {
  798. // 确认客户端是否还活着...
  799. if( !_IsSocketAlive( pContext->m_Socket) )
  800. {
  801. this->_ShowMessage( _T("检测到客户端异常退出!") );
  802. this->_RemoveContext( pContext );
  803. return true;
  804. }
  805. else
  806. {
  807. this->_ShowMessage( _T("网络操作超时!重试中...") );
  808. return true;
  809. }
  810. }
  811. // 可能是客户端异常退出了
  812. else if( ERROR_NETNAME_DELETED==dwErr )
  813. {
  814. this->_ShowMessage( _T("检测到客户端异常退出!") );
  815. this->_RemoveContext( pContext );
  816. return true;
  817. }
  818. else
  819. {
  820. this->_ShowMessage( _T("完成端口操作出现错误,线程退出。错误代码:%d"),dwErr );
  821. return false;
  822. }
  823. }