SATTCPServer.cpp 31 KB

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