工具项目

TcpAgent.cpp 34KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541
  1. /*
  2. * Copyright: JessMA Open Source (ldcsaa@gmail.com)
  3. *
  4. * Author : Bruce Liang
  5. * Website : http://www.jessma.org
  6. * Project : https://github.com/ldcsaa
  7. * Blog : http://www.cnblogs.com/ldcsaa
  8. * Wiki : http://www.oschina.net/p/hp-socket
  9. * QQ Group : 75375912, 44636872
  10. *
  11. * Licensed under the Apache License, Version 2.0 (the "License");
  12. * you may not use this file except in compliance with the License.
  13. * You may obtain a copy of the License at
  14. *
  15. * http://www.apache.org/licenses/LICENSE-2.0
  16. *
  17. * Unless required by applicable law or agreed to in writing, software
  18. * distributed under the License is distributed on an "AS IS" BASIS,
  19. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  20. * See the License for the specific language governing permissions and
  21. * limitations under the License.
  22. */
  23. #include "stdafx.h"
  24. #include "TcpAgent.h"
  25. #include "../Common/Src/WaitFor.h"
  26. #include <malloc.h>
  27. #include <process.h>
  28. const CInitSocket CTcpAgent::sm_wsSocket;
  29. EnHandleResult CTcpAgent::TriggerFireConnect(TSocketObj* pSocketObj)
  30. {
  31. CReentrantSpinLock locallock(pSocketObj->csRecv);
  32. return TRIGGER(FireConnect(pSocketObj));
  33. }
  34. EnHandleResult CTcpAgent::TriggerFireReceive(TSocketObj* pSocketObj, TBufferObj* pBufferObj)
  35. {
  36. EnHandleResult rs = (EnHandleResult)HR_CLOSED;
  37. if(TSocketObj::IsValid(pSocketObj))
  38. {
  39. CReentrantSpinLock locallock(pSocketObj->csRecv);
  40. if(TSocketObj::IsValid(pSocketObj))
  41. {
  42. rs = TRIGGER(FireReceive(pSocketObj, (BYTE*)pBufferObj->buff.buf, pBufferObj->buff.len));
  43. }
  44. }
  45. return rs;
  46. }
  47. EnHandleResult CTcpAgent::TriggerFireSend(TSocketObj* pSocketObj, TBufferObj* pBufferObj)
  48. {
  49. EnHandleResult rs = TRIGGER(FireSend(pSocketObj, (BYTE*)pBufferObj->buff.buf, pBufferObj->buff.len));
  50. if(rs == HR_ERROR)
  51. {
  52. TRACE("<S-CNNID: %Iu> OnSend() event should not return 'HR_ERROR' !!\n", pSocketObj->connID);
  53. ASSERT(FALSE);
  54. }
  55. if(pBufferObj->ReleaseSendCounter() == 0)
  56. AddFreeBufferObj(pBufferObj);
  57. return rs;
  58. }
  59. EnHandleResult CTcpAgent::TriggerFireClose(TSocketObj* pSocketObj, EnSocketOperation enOperation, int iErrorCode)
  60. {
  61. CReentrantSpinLock locallock(pSocketObj->csRecv);
  62. return FireClose(pSocketObj, enOperation, iErrorCode);
  63. }
  64. void CTcpAgent::SetLastError(EnSocketError code, LPCSTR func, int ec)
  65. {
  66. m_enLastError = code;
  67. ::SetLastError(ec);
  68. TRACE("%s --> Error: %d, EC: %d\n", func, code, ec);
  69. }
  70. BOOL CTcpAgent::Start(LPCTSTR lpszBindAddress, BOOL bAsyncConnect)
  71. {
  72. if(!CheckParams() || !CheckStarting())
  73. return FALSE;
  74. PrepareStart();
  75. if(ParseBindAddress(lpszBindAddress))
  76. if(CreateCompletePort())
  77. if(CreateWorkerThreads())
  78. {
  79. m_bAsyncConnect = bAsyncConnect;
  80. m_enState = SS_STARTED;
  81. return TRUE;
  82. }
  83. EXECUTE_RESTORE_ERROR(Stop());
  84. return FALSE;
  85. }
  86. BOOL CTcpAgent::CheckParams()
  87. {
  88. if ((m_enSendPolicy >= SP_PACK && m_enSendPolicy <= SP_DIRECT) &&
  89. ((int)m_dwMaxConnectionCount > 0) &&
  90. ((int)m_dwWorkerThreadCount > 0 && m_dwWorkerThreadCount <= MAX_WORKER_THREAD_COUNT) &&
  91. ((int)m_dwSocketBufferSize >= MIN_SOCKET_BUFFER_SIZE) &&
  92. ((int)m_dwFreeSocketObjLockTime >= 0) &&
  93. ((int)m_dwFreeSocketObjPool >= 0) &&
  94. ((int)m_dwFreeBufferObjPool >= 0) &&
  95. ((int)m_dwFreeSocketObjHold >= m_dwFreeSocketObjPool) &&
  96. ((int)m_dwFreeBufferObjHold >= m_dwFreeBufferObjPool) &&
  97. ((int)m_dwKeepAliveTime >= 1000 || m_dwKeepAliveTime == 0) &&
  98. ((int)m_dwKeepAliveInterval >= 1000 || m_dwKeepAliveInterval == 0) )
  99. return TRUE;
  100. SetLastError(SE_INVALID_PARAM, __FUNCTION__, ERROR_INVALID_PARAMETER);
  101. return FALSE;
  102. }
  103. void CTcpAgent::PrepareStart()
  104. {
  105. m_bfActiveSockets.Reset(m_dwMaxConnectionCount);
  106. m_lsFreeSocket.Reset(m_dwFreeSocketObjHold);
  107. m_bfObjPool.SetItemCapacity((int)m_dwSocketBufferSize);
  108. m_bfObjPool.SetPoolSize((int)m_dwFreeBufferObjPool);
  109. m_bfObjPool.SetPoolHold((int)m_dwFreeBufferObjHold);
  110. m_bfObjPool.Prepare();
  111. }
  112. BOOL CTcpAgent::CheckStarting()
  113. {
  114. CSpinLock locallock(m_csState);
  115. if(m_enState == SS_STOPPED)
  116. m_enState = SS_STARTING;
  117. else
  118. {
  119. SetLastError(SE_ILLEGAL_STATE, __FUNCTION__, ERROR_INVALID_OPERATION);
  120. return FALSE;
  121. }
  122. return TRUE;
  123. }
  124. BOOL CTcpAgent::CheckStoping()
  125. {
  126. if(m_enState != SS_STOPPED)
  127. {
  128. CSpinLock locallock(m_csState);
  129. if(HasStarted())
  130. {
  131. m_enState = SS_STOPPING;
  132. return TRUE;
  133. }
  134. while(m_enState != SS_STOPPED)
  135. ::Sleep(30);
  136. }
  137. SetLastError(SE_ILLEGAL_STATE, __FUNCTION__, ERROR_INVALID_OPERATION);
  138. return FALSE;
  139. }
  140. BOOL CTcpAgent::ParseBindAddress(LPCTSTR lpszBindAddress)
  141. {
  142. BOOL isOK = FALSE;
  143. BOOL bBind = (lpszBindAddress && lpszBindAddress[0] != 0);
  144. if(!bBind) lpszBindAddress = DEFAULT_IPV4_BIND_ADDRESS;
  145. HP_SOCKADDR addr;
  146. if(::sockaddr_A_2_IN(lpszBindAddress, 0, addr))
  147. {
  148. SOCKET sock = socket(addr.family, SOCK_STREAM, IPPROTO_TCP);
  149. if(sock != INVALID_SOCKET)
  150. {
  151. if(::bind(sock, addr.Addr(), addr.AddrSize()) != SOCKET_ERROR)
  152. {
  153. m_pfnConnectEx = ::Get_ConnectEx_FuncPtr(sock);
  154. m_pfnDisconnectEx = ::Get_DisconnectEx_FuncPtr(sock);
  155. ASSERT(m_pfnConnectEx);
  156. ASSERT(m_pfnDisconnectEx);
  157. if(bBind) addr.Copy(m_soAddr);
  158. isOK = TRUE;
  159. }
  160. else
  161. SetLastError(SE_SOCKET_BIND, __FUNCTION__, ::WSAGetLastError());
  162. ::ManualCloseSocket(sock);
  163. }
  164. else
  165. SetLastError(SE_SOCKET_CREATE, __FUNCTION__, ::WSAGetLastError());
  166. }
  167. else
  168. SetLastError(SE_SOCKET_CREATE, __FUNCTION__, ::WSAGetLastError());
  169. return isOK;
  170. }
  171. BOOL CTcpAgent::CreateCompletePort()
  172. {
  173. m_hCompletePort = ::CreateIoCompletionPort(INVALID_HANDLE_VALUE, nullptr, 0, 0);
  174. if(m_hCompletePort == nullptr)
  175. SetLastError(SE_CP_CREATE, __FUNCTION__, ::GetLastError());
  176. return (m_hCompletePort != nullptr);
  177. }
  178. BOOL CTcpAgent::CreateWorkerThreads()
  179. {
  180. BOOL isOK = TRUE;
  181. for(DWORD i = 0; i < m_dwWorkerThreadCount; i++)
  182. {
  183. HANDLE hThread = (HANDLE)_beginthreadex(nullptr, 0, WorkerThreadProc, (LPVOID)this, 0, nullptr);
  184. if(hThread)
  185. m_vtWorkerThreads.push_back(hThread);
  186. else
  187. {
  188. SetLastError(SE_WORKER_THREAD_CREATE, __FUNCTION__, ::GetLastError());
  189. isOK = FALSE;
  190. break;
  191. }
  192. }
  193. return isOK;
  194. }
  195. BOOL CTcpAgent::Stop()
  196. {
  197. if(!CheckStoping())
  198. return FALSE;
  199. DisconnectClientSocket();
  200. WaitForClientSocketClose();
  201. WaitForWorkerThreadEnd();
  202. ReleaseClientSocket();
  203. FireShutdown();
  204. ReleaseFreeSocket();
  205. ReleaseFreeBuffer();
  206. CloseCompletePort();
  207. Reset();
  208. return TRUE;
  209. }
  210. void CTcpAgent::Reset()
  211. {
  212. m_phSocket.Reset();
  213. m_soAddr.Reset();
  214. m_pfnConnectEx = nullptr;
  215. m_pfnDisconnectEx = nullptr;
  216. m_enState = SS_STOPPED;
  217. }
  218. void CTcpAgent::DisconnectClientSocket()
  219. {
  220. ::WaitWithMessageLoop(150);
  221. DWORD size = 0;
  222. unique_ptr<CONNID[]> ids = m_bfActiveSockets.GetAllElementIndexes(size);
  223. for(DWORD i = 0; i < size; i++)
  224. Disconnect(ids[i]);
  225. }
  226. void CTcpAgent::ReleaseClientSocket()
  227. {
  228. VERIFY(m_bfActiveSockets.IsEmpty());
  229. m_bfActiveSockets.Reset();
  230. }
  231. TSocketObj* CTcpAgent::GetFreeSocketObj(CONNID dwConnID, SOCKET soClient)
  232. {
  233. DWORD dwIndex;
  234. TSocketObj* pSocketObj = nullptr;
  235. if(m_lsFreeSocket.TryLock(&pSocketObj, dwIndex))
  236. {
  237. if(::GetTimeGap32(pSocketObj->freeTime) >= m_dwFreeSocketObjLockTime)
  238. VERIFY(m_lsFreeSocket.ReleaseLock(nullptr, dwIndex));
  239. else
  240. {
  241. VERIFY(m_lsFreeSocket.ReleaseLock(pSocketObj, dwIndex));
  242. pSocketObj = nullptr;
  243. }
  244. }
  245. if(!pSocketObj) pSocketObj = CreateSocketObj();
  246. pSocketObj->Reset(dwConnID, soClient);
  247. return pSocketObj;
  248. }
  249. void CTcpAgent::AddFreeSocketObj(TSocketObj* pSocketObj, EnSocketCloseFlag enFlag, EnSocketOperation enOperation, int iErrorCode)
  250. {
  251. if(!InvalidSocketObj(pSocketObj))
  252. return;
  253. CloseClientSocketObj(pSocketObj, enFlag, enOperation, iErrorCode);
  254. m_bfActiveSockets.Remove(pSocketObj->connID);
  255. TSocketObj::Release(pSocketObj);
  256. if(!m_lsFreeSocket.TryPut(pSocketObj))
  257. {
  258. m_lsGCSocket.PushBack(pSocketObj);
  259. if(m_lsGCSocket.Size() > m_dwFreeSocketObjPool)
  260. ReleaseGCSocketObj();
  261. }
  262. }
  263. void CTcpAgent::ReleaseGCSocketObj(BOOL bForce)
  264. {
  265. TSocketObj* pSocketObj = nullptr;
  266. DWORD now = ::TimeGetTime();
  267. while(m_lsGCSocket.PopFront(&pSocketObj))
  268. {
  269. if(bForce || (int)(now - pSocketObj->freeTime) >= (int)m_dwFreeSocketObjLockTime)
  270. DeleteSocketObj(pSocketObj);
  271. else
  272. {
  273. m_lsGCSocket.PushBack(pSocketObj);
  274. break;
  275. }
  276. }
  277. }
  278. BOOL CTcpAgent::InvalidSocketObj(TSocketObj* pSocketObj)
  279. {
  280. BOOL bDone = FALSE;
  281. if(TSocketObj::IsValid(pSocketObj))
  282. {
  283. CReentrantSpinLock locallock(pSocketObj->csRecv);
  284. CCriSecLock locallock2(pSocketObj->csSend);
  285. if(TSocketObj::IsValid(pSocketObj))
  286. {
  287. TSocketObj::Invalid(pSocketObj);
  288. bDone = TRUE;
  289. }
  290. }
  291. return bDone;
  292. }
  293. void CTcpAgent::AddClientSocketObj(CONNID dwConnID, TSocketObj* pSocketObj)
  294. {
  295. ASSERT(FindSocketObj(dwConnID) == nullptr);
  296. pSocketObj->connTime = ::TimeGetTime();
  297. pSocketObj->activeTime = pSocketObj->connTime;
  298. VERIFY(m_bfActiveSockets.ReleaseLock(dwConnID, pSocketObj));
  299. }
  300. void CTcpAgent::ReleaseFreeSocket()
  301. {
  302. TSocketObj* pSocketObj = nullptr;
  303. while(m_lsFreeSocket.TryGet(&pSocketObj))
  304. DeleteSocketObj(pSocketObj);
  305. VERIFY(m_lsFreeSocket.IsEmpty());
  306. m_lsFreeSocket.Reset();
  307. ReleaseGCSocketObj(TRUE);
  308. VERIFY(m_lsGCSocket.IsEmpty());
  309. }
  310. TSocketObj* CTcpAgent::CreateSocketObj()
  311. {
  312. TSocketObj* pSocketObj = (TSocketObj*)m_phSocket.Alloc(sizeof(TSocketObj));
  313. ASSERT(pSocketObj);
  314. pSocketObj->TSocketObj::TSocketObj(m_bfObjPool);
  315. return pSocketObj;
  316. }
  317. void CTcpAgent::DeleteSocketObj(TSocketObj* pSocketObj)
  318. {
  319. ASSERT(pSocketObj);
  320. pSocketObj->TSocketObj::~TSocketObj();
  321. m_phSocket.Free(pSocketObj);
  322. }
  323. TBufferObj* CTcpAgent::GetFreeBufferObj(int iLen)
  324. {
  325. ASSERT(iLen >= -1 && iLen <= (int)m_dwSocketBufferSize);
  326. TBufferObj* pBufferObj = m_bfObjPool.PickFreeItem();
  327. if(iLen < 0) iLen = m_dwSocketBufferSize;
  328. pBufferObj->buff.len = iLen;
  329. return pBufferObj;
  330. }
  331. void CTcpAgent::AddFreeBufferObj(TBufferObj* pBufferObj)
  332. {
  333. m_bfObjPool.PutFreeItem(pBufferObj);
  334. }
  335. void CTcpAgent::ReleaseFreeBuffer()
  336. {
  337. m_bfObjPool.Clear();
  338. }
  339. TSocketObj* CTcpAgent::FindSocketObj(CONNID dwConnID)
  340. {
  341. TSocketObj* pSocketObj = nullptr;
  342. if(m_bfActiveSockets.Get(dwConnID, &pSocketObj) != TSocketObjPtrPool::GR_VALID)
  343. pSocketObj = nullptr;
  344. return pSocketObj;
  345. }
  346. void CTcpAgent::CloseClientSocketObj(TSocketObj* pSocketObj, EnSocketCloseFlag enFlag, EnSocketOperation enOperation, int iErrorCode, int iShutdownFlag)
  347. {
  348. ASSERT(TSocketObj::IsExist(pSocketObj));
  349. if(enFlag == SCF_CLOSE)
  350. TriggerFireClose(pSocketObj, SO_CLOSE, SE_OK);
  351. else if(enFlag == SCF_ERROR)
  352. TriggerFireClose(pSocketObj, enOperation, iErrorCode);
  353. SOCKET socket = pSocketObj->socket;
  354. pSocketObj->socket = INVALID_SOCKET;
  355. ::ManualCloseSocket(socket, iShutdownFlag);
  356. }
  357. BOOL CTcpAgent::GetLocalAddress(CONNID dwConnID, TCHAR lpszAddress[], int& iAddressLen, USHORT& usPort)
  358. {
  359. ASSERT(lpszAddress != nullptr && iAddressLen > 0);
  360. TSocketObj* pSocketObj = FindSocketObj(dwConnID);
  361. if(TSocketObj::IsValid(pSocketObj))
  362. return ::GetSocketLocalAddress(pSocketObj->socket, lpszAddress, iAddressLen, usPort);
  363. return FALSE;
  364. }
  365. BOOL CTcpAgent::GetRemoteAddress(CONNID dwConnID, TCHAR lpszAddress[], int& iAddressLen, USHORT& usPort)
  366. {
  367. ASSERT(lpszAddress != nullptr && iAddressLen > 0);
  368. TSocketObj* pSocketObj = FindSocketObj(dwConnID);
  369. if(TSocketObj::IsExist(pSocketObj))
  370. {
  371. ADDRESS_FAMILY usFamily;
  372. return ::sockaddr_IN_2_A(pSocketObj->remoteAddr, usFamily, lpszAddress, iAddressLen, usPort);
  373. }
  374. return FALSE;
  375. }
  376. BOOL CTcpAgent::GetRemoteHost(CONNID dwConnID, TCHAR lpszHost[], int& iHostLen, USHORT& usPort)
  377. {
  378. ASSERT(lpszHost != nullptr && iHostLen > 0);
  379. BOOL isOK = FALSE;
  380. TSocketObj* pSocketObj = FindSocketObj(dwConnID);
  381. if(TSocketObj::IsExist(pSocketObj))
  382. {
  383. int iLen = pSocketObj->host.GetLength() + 1;
  384. if(iHostLen >= iLen)
  385. {
  386. memcpy(lpszHost, CA2CT(pSocketObj->host), iLen * sizeof(TCHAR));
  387. usPort = pSocketObj->remoteAddr.Port();
  388. isOK = TRUE;
  389. }
  390. iHostLen = iLen;
  391. }
  392. return isOK;
  393. }
  394. BOOL CTcpAgent::GetRemoteHost(CONNID dwConnID, LPCSTR* lpszHost, USHORT* pusPort)
  395. {
  396. *lpszHost = nullptr;
  397. TSocketObj* pSocketObj = FindSocketObj(dwConnID);
  398. if(TSocketObj::IsExist(pSocketObj))
  399. {
  400. *lpszHost = pSocketObj->host;
  401. if(pusPort)
  402. *pusPort = pSocketObj->remoteAddr.Port();
  403. }
  404. return (*lpszHost != nullptr && (*lpszHost)[0] != 0);
  405. }
  406. BOOL CTcpAgent::SetConnectionExtra(CONNID dwConnID, PVOID pExtra)
  407. {
  408. TSocketObj* pSocketObj = FindSocketObj(dwConnID);
  409. return SetConnectionExtra(pSocketObj, pExtra);
  410. }
  411. BOOL CTcpAgent::SetConnectionExtra(TSocketObj* pSocketObj, PVOID pExtra)
  412. {
  413. if(TSocketObj::IsExist(pSocketObj))
  414. {
  415. pSocketObj->extra = pExtra;
  416. return TRUE;
  417. }
  418. return FALSE;
  419. }
  420. BOOL CTcpAgent::GetConnectionExtra(CONNID dwConnID, PVOID* ppExtra)
  421. {
  422. TSocketObj* pSocketObj = FindSocketObj(dwConnID);
  423. return GetConnectionExtra(pSocketObj, ppExtra);
  424. }
  425. BOOL CTcpAgent::GetConnectionExtra(TSocketObj* pSocketObj, PVOID* ppExtra)
  426. {
  427. ASSERT(ppExtra != nullptr);
  428. if(TSocketObj::IsExist(pSocketObj))
  429. {
  430. *ppExtra = pSocketObj->extra;
  431. return TRUE;
  432. }
  433. return FALSE;
  434. }
  435. BOOL CTcpAgent::SetConnectionReserved(CONNID dwConnID, PVOID pReserved)
  436. {
  437. TSocketObj* pSocketObj = FindSocketObj(dwConnID);
  438. return SetConnectionReserved(pSocketObj, pReserved);
  439. }
  440. BOOL CTcpAgent::SetConnectionReserved(TSocketObj* pSocketObj, PVOID pReserved)
  441. {
  442. if(TSocketObj::IsExist(pSocketObj))
  443. {
  444. pSocketObj->reserved = pReserved;
  445. return TRUE;
  446. }
  447. return FALSE;
  448. }
  449. BOOL CTcpAgent::GetConnectionReserved(CONNID dwConnID, PVOID* ppReserved)
  450. {
  451. TSocketObj* pSocketObj = FindSocketObj(dwConnID);
  452. return GetConnectionReserved(pSocketObj, ppReserved);
  453. }
  454. BOOL CTcpAgent::GetConnectionReserved(TSocketObj* pSocketObj, PVOID* ppReserved)
  455. {
  456. ASSERT(ppReserved != nullptr);
  457. if(TSocketObj::IsExist(pSocketObj))
  458. {
  459. *ppReserved = pSocketObj->reserved;
  460. return TRUE;
  461. }
  462. return FALSE;
  463. }
  464. BOOL CTcpAgent::SetConnectionReserved2(CONNID dwConnID, PVOID pReserved2)
  465. {
  466. TSocketObj* pSocketObj = FindSocketObj(dwConnID);
  467. return SetConnectionReserved2(pSocketObj, pReserved2);
  468. }
  469. BOOL CTcpAgent::SetConnectionReserved2(TSocketObj* pSocketObj, PVOID pReserved2)
  470. {
  471. if(TSocketObj::IsExist(pSocketObj))
  472. {
  473. pSocketObj->reserved2 = pReserved2;
  474. return TRUE;
  475. }
  476. return FALSE;
  477. }
  478. BOOL CTcpAgent::GetConnectionReserved2(CONNID dwConnID, PVOID* ppReserved2)
  479. {
  480. TSocketObj* pSocketObj = FindSocketObj(dwConnID);
  481. return GetConnectionReserved2(pSocketObj, ppReserved2);
  482. }
  483. BOOL CTcpAgent::GetConnectionReserved2(TSocketObj* pSocketObj, PVOID* ppReserved2)
  484. {
  485. ASSERT(ppReserved2 != nullptr);
  486. if(TSocketObj::IsExist(pSocketObj))
  487. {
  488. *ppReserved2 = pSocketObj->reserved2;
  489. return TRUE;
  490. }
  491. return FALSE;
  492. }
  493. BOOL CTcpAgent::IsPauseReceive(CONNID dwConnID, BOOL& bPaused)
  494. {
  495. TSocketObj* pSocketObj = FindSocketObj(dwConnID);
  496. if(TSocketObj::IsValid(pSocketObj))
  497. {
  498. bPaused = pSocketObj->paused;
  499. return TRUE;
  500. }
  501. return FALSE;
  502. }
  503. BOOL CTcpAgent::GetPendingDataLength(CONNID dwConnID, int& iPending)
  504. {
  505. TSocketObj* pSocketObj = FindSocketObj(dwConnID);
  506. if(TSocketObj::IsValid(pSocketObj))
  507. {
  508. iPending = pSocketObj->Pending();
  509. return TRUE;
  510. }
  511. return FALSE;
  512. }
  513. DWORD CTcpAgent::GetConnectionCount()
  514. {
  515. return m_bfActiveSockets.Elements();
  516. }
  517. BOOL CTcpAgent::GetAllConnectionIDs(CONNID pIDs[], DWORD& dwCount)
  518. {
  519. return m_bfActiveSockets.GetAllElementIndexes(pIDs, dwCount);
  520. }
  521. BOOL CTcpAgent::GetConnectPeriod(CONNID dwConnID, DWORD& dwPeriod)
  522. {
  523. BOOL isOK = TRUE;
  524. TSocketObj* pSocketObj = FindSocketObj(dwConnID);
  525. if(TSocketObj::IsValid(pSocketObj))
  526. dwPeriod = ::GetTimeGap32(pSocketObj->connTime);
  527. else
  528. isOK = FALSE;
  529. return isOK;
  530. }
  531. BOOL CTcpAgent::GetSilencePeriod(CONNID dwConnID, DWORD& dwPeriod)
  532. {
  533. if(!m_bMarkSilence)
  534. return FALSE;
  535. BOOL isOK = TRUE;
  536. TSocketObj* pSocketObj = FindSocketObj(dwConnID);
  537. if(TSocketObj::IsValid(pSocketObj))
  538. dwPeriod = ::GetTimeGap32(pSocketObj->activeTime);
  539. else
  540. isOK = FALSE;
  541. return isOK;
  542. }
  543. BOOL CTcpAgent::Disconnect(CONNID dwConnID, BOOL bForce)
  544. {
  545. BOOL isOK = FALSE;
  546. TSocketObj* pSocketObj = FindSocketObj(dwConnID);
  547. if(TSocketObj::IsValid(pSocketObj))
  548. {
  549. if(bForce)
  550. isOK = ::PostIocpDisconnect(m_hCompletePort, dwConnID);
  551. else
  552. isOK = m_pfnDisconnectEx(pSocketObj->socket, nullptr, 0, 0);
  553. }
  554. return isOK;
  555. }
  556. BOOL CTcpAgent::DisconnectLongConnections(DWORD dwPeriod, BOOL bForce)
  557. {
  558. if(dwPeriod > MAX_CONNECTION_PERIOD)
  559. return FALSE;
  560. DWORD size = 0;
  561. unique_ptr<CONNID[]> ids = m_bfActiveSockets.GetAllElementIndexes(size);
  562. DWORD now = ::TimeGetTime();
  563. for(DWORD i = 0; i < size; i++)
  564. {
  565. CONNID connID = ids[i];
  566. TSocketObj* pSocketObj = FindSocketObj(connID);
  567. if(TSocketObj::IsValid(pSocketObj) && (int)(now - pSocketObj->connTime) >= (int)dwPeriod)
  568. Disconnect(connID, bForce);
  569. }
  570. return TRUE;
  571. }
  572. BOOL CTcpAgent::DisconnectSilenceConnections(DWORD dwPeriod, BOOL bForce)
  573. {
  574. if(!m_bMarkSilence)
  575. return FALSE;
  576. if(dwPeriod > MAX_CONNECTION_PERIOD)
  577. return FALSE;
  578. DWORD size = 0;
  579. unique_ptr<CONNID[]> ids = m_bfActiveSockets.GetAllElementIndexes(size);
  580. DWORD now = ::TimeGetTime();
  581. for(DWORD i = 0; i < size; i++)
  582. {
  583. CONNID connID = ids[i];
  584. TSocketObj* pSocketObj = FindSocketObj(connID);
  585. if(TSocketObj::IsValid(pSocketObj) && (int)(now - pSocketObj->activeTime) >= (int)dwPeriod)
  586. Disconnect(connID, bForce);
  587. }
  588. return TRUE;
  589. }
  590. void CTcpAgent::WaitForClientSocketClose()
  591. {
  592. while(m_bfActiveSockets.Elements() > 0)
  593. ::WaitWithMessageLoop(100);
  594. }
  595. void CTcpAgent::WaitForWorkerThreadEnd()
  596. {
  597. int count = (int)m_vtWorkerThreads.size();
  598. for(int i = 0; i < count; i++)
  599. ::PostIocpExit(m_hCompletePort);
  600. int remain = count;
  601. int index = 0;
  602. while(remain > 0)
  603. {
  604. int wait = min(remain, MAXIMUM_WAIT_OBJECTS);
  605. HANDLE* pHandles = (HANDLE*)_alloca(sizeof(HANDLE) * wait);
  606. for(int i = 0; i < wait; i++)
  607. pHandles[i] = m_vtWorkerThreads[i + index];
  608. VERIFY(::WaitForMultipleObjects((DWORD)wait, pHandles, TRUE, INFINITE) == WAIT_OBJECT_0);
  609. for(int i = 0; i < wait; i++)
  610. ::CloseHandle(pHandles[i]);
  611. remain -= wait;
  612. index += wait;
  613. }
  614. m_vtWorkerThreads.clear();
  615. }
  616. void CTcpAgent::CloseCompletePort()
  617. {
  618. if(m_hCompletePort != nullptr)
  619. {
  620. ::CloseHandle(m_hCompletePort);
  621. m_hCompletePort = nullptr;
  622. }
  623. }
  624. UINT WINAPI CTcpAgent::WorkerThreadProc(LPVOID pv)
  625. {
  626. CTcpAgent* pAgent = (CTcpAgent*)pv;
  627. while(TRUE)
  628. {
  629. DWORD dwErrorCode = NO_ERROR;
  630. DWORD dwBytes;
  631. OVERLAPPED* pOverlapped;
  632. TSocketObj* pSocketObj;
  633. BOOL result = ::GetQueuedCompletionStatus
  634. (
  635. pAgent->m_hCompletePort,
  636. &dwBytes,
  637. (PULONG_PTR)&pSocketObj,
  638. &pOverlapped,
  639. INFINITE
  640. );
  641. if(pOverlapped == nullptr)
  642. {
  643. EnIocpAction action = pAgent->CheckIocpCommand(pOverlapped, dwBytes, (ULONG_PTR)pSocketObj);
  644. if(action == IOCP_ACT_CONTINUE)
  645. continue;
  646. else if(action == IOCP_ACT_BREAK)
  647. break;
  648. }
  649. TBufferObj* pBufferObj = CONTAINING_RECORD(pOverlapped, TBufferObj, ov);
  650. CONNID dwConnID = pSocketObj->connID;
  651. if (!result)
  652. {
  653. DWORD dwFlag = 0;
  654. DWORD dwSysCode = ::GetLastError();
  655. if(pAgent->HasStarted())
  656. {
  657. SOCKET sock = pBufferObj->client;
  658. result = ::WSAGetOverlappedResult(sock, &pBufferObj->ov, &dwBytes, FALSE, &dwFlag);
  659. if (!result)
  660. {
  661. dwErrorCode = ::WSAGetLastError();
  662. TRACE("GetQueuedCompletionStatus error (<A-CNNID: %Iu> SYS: %d, SOCK: %d, FLAG: %d)\n", dwConnID, dwSysCode, dwErrorCode, dwFlag);
  663. }
  664. }
  665. else
  666. dwErrorCode = dwSysCode;
  667. ASSERT(dwSysCode != 0 && dwErrorCode != 0);
  668. }
  669. pAgent->HandleIo(dwConnID, pSocketObj, pBufferObj, dwBytes, dwErrorCode);
  670. }
  671. pAgent->OnWorkerThreadEnd(::GetCurrentThreadId());
  672. return 0;
  673. }
  674. EnIocpAction CTcpAgent::CheckIocpCommand(OVERLAPPED* pOverlapped, DWORD dwBytes, ULONG_PTR ulCompKey)
  675. {
  676. ASSERT(pOverlapped == nullptr);
  677. EnIocpAction action = IOCP_ACT_CONTINUE;
  678. CONNID dwConnID = (CONNID)ulCompKey;
  679. switch(dwBytes)
  680. {
  681. case IOCP_CMD_SEND : DoSend(dwConnID) ; break;
  682. case IOCP_CMD_DISCONNECT: ForceDisconnect(dwConnID) ; break;
  683. case IOCP_CMD_EXIT : action = IOCP_ACT_BREAK ; break;
  684. default : CheckError(FindSocketObj(dwConnID), SO_CLOSE, (int)dwBytes);
  685. }
  686. return action;
  687. }
  688. void CTcpAgent::ForceDisconnect(CONNID dwConnID)
  689. {
  690. AddFreeSocketObj(FindSocketObj(dwConnID), SCF_CLOSE);
  691. }
  692. void CTcpAgent::HandleIo(CONNID dwConnID, TSocketObj* pSocketObj, TBufferObj* pBufferObj, DWORD dwBytes, DWORD dwErrorCode)
  693. {
  694. ASSERT(pBufferObj != nullptr);
  695. ASSERT(pSocketObj != nullptr);
  696. if(dwErrorCode != NO_ERROR)
  697. {
  698. HandleError(dwConnID, pSocketObj, pBufferObj, dwErrorCode);
  699. return;
  700. }
  701. if(dwBytes == 0 && pBufferObj->operation != SO_CONNECT)
  702. {
  703. AddFreeSocketObj(pSocketObj, SCF_CLOSE);
  704. AddFreeBufferObj(pBufferObj);
  705. return;
  706. }
  707. pBufferObj->buff.len = dwBytes;
  708. switch(pBufferObj->operation)
  709. {
  710. case SO_CONNECT:
  711. HandleConnect(dwConnID, pSocketObj, pBufferObj);
  712. break;
  713. case SO_SEND:
  714. HandleSend(dwConnID, pSocketObj, pBufferObj);
  715. break;
  716. case SO_RECEIVE:
  717. HandleReceive(dwConnID, pSocketObj, pBufferObj);
  718. break;
  719. default:
  720. ASSERT(FALSE);
  721. }
  722. }
  723. void CTcpAgent::HandleError(CONNID dwConnID, TSocketObj* pSocketObj, TBufferObj* pBufferObj, DWORD dwErrorCode)
  724. {
  725. CheckError(pSocketObj, pBufferObj->operation, dwErrorCode);
  726. AddFreeBufferObj(pBufferObj);
  727. }
  728. void CTcpAgent::HandleConnect(CONNID dwConnID, TSocketObj* pSocketObj, TBufferObj* pBufferObj)
  729. {
  730. ::SSO_UpdateConnectContext(pSocketObj->socket, 0);
  731. pSocketObj->SetConnected();
  732. if(TriggerFireConnect(pSocketObj) != HR_ERROR)
  733. DoReceive(pSocketObj, pBufferObj);
  734. else
  735. {
  736. AddFreeSocketObj(pSocketObj, SCF_NONE);
  737. AddFreeBufferObj(pBufferObj);
  738. }
  739. }
  740. void CTcpAgent::HandleSend(CONNID dwConnID, TSocketObj* pSocketObj, TBufferObj* pBufferObj)
  741. {
  742. int iLength = -(long)(pBufferObj->buff.len);
  743. switch(m_enSendPolicy)
  744. {
  745. case SP_PACK:
  746. {
  747. ::InterlockedExchangeAdd(&pSocketObj->sndCount, iLength);
  748. TriggerFireSend(pSocketObj, pBufferObj);
  749. DoSendPack(pSocketObj);
  750. }
  751. break;
  752. case SP_SAFE:
  753. {
  754. ::InterlockedExchangeAdd(&pSocketObj->sndCount, iLength);
  755. TriggerFireSend(pSocketObj, pBufferObj);
  756. DoSendSafe(pSocketObj);
  757. }
  758. break;
  759. case SP_DIRECT:
  760. {
  761. TriggerFireSend(pSocketObj, pBufferObj);
  762. }
  763. break;
  764. default:
  765. ASSERT(FALSE);
  766. }
  767. }
  768. void CTcpAgent::HandleReceive(CONNID dwConnID, TSocketObj* pSocketObj, TBufferObj* pBufferObj)
  769. {
  770. pSocketObj->recving = FALSE;
  771. if(m_bMarkSilence) pSocketObj->activeTime = ::TimeGetTime();
  772. EnHandleResult hr = TriggerFireReceive(pSocketObj, pBufferObj);
  773. if(hr == HR_OK || hr == HR_IGNORE)
  774. {
  775. if(ContinueReceive(pSocketObj, pBufferObj, hr))
  776. DoReceive(pSocketObj, pBufferObj);
  777. }
  778. else if(hr == HR_CLOSED)
  779. {
  780. AddFreeBufferObj(pBufferObj);
  781. }
  782. else
  783. {
  784. TRACE("<A-CNNID: %Iu> OnReceive() event return 'HR_ERROR', connection will be closed !\n", dwConnID);
  785. AddFreeSocketObj(pSocketObj, SCF_ERROR, SO_RECEIVE, ENSURE_ERROR_CANCELLED);
  786. AddFreeBufferObj(pBufferObj);
  787. }
  788. }
  789. BOOL CTcpAgent::ContinueReceive(TSocketObj* pSocketObj, TBufferObj* pBufferObj, EnHandleResult& hr)
  790. {
  791. int rs = NO_ERROR;
  792. for(int i = 0; i < MAX_IOCP_CONTINUE_RECEIVE || MAX_IOCP_CONTINUE_RECEIVE < 0; i++)
  793. {
  794. if(hr != HR_OK && hr != HR_IGNORE)
  795. break;
  796. if(pBufferObj->buff.len != m_dwSocketBufferSize)
  797. break;
  798. pBufferObj->buff.len = m_dwSocketBufferSize;
  799. rs =::NoBlockReceiveNotCheck(pBufferObj);
  800. if(rs != NO_ERROR)
  801. break;
  802. hr = TriggerFireReceive(pSocketObj, pBufferObj);
  803. }
  804. if(rs != NO_ERROR && rs != WSAEWOULDBLOCK)
  805. {
  806. if(rs == WSAEDISCON)
  807. AddFreeSocketObj(pSocketObj, SCF_CLOSE);
  808. else
  809. CheckError(pSocketObj, SO_RECEIVE, rs);
  810. AddFreeBufferObj(pBufferObj);
  811. return FALSE;
  812. }
  813. return TRUE;
  814. }
  815. int CTcpAgent::DoReceive(TSocketObj* pSocketObj, TBufferObj* pBufferObj)
  816. {
  817. int result = NO_ERROR;
  818. BOOL bNeedFree = FALSE;
  819. if(pSocketObj->paused)
  820. bNeedFree = TRUE;
  821. else
  822. {
  823. CReentrantSpinLock locallock(pSocketObj->csRecv);
  824. if(pSocketObj->paused || pSocketObj->recving)
  825. bNeedFree = TRUE;
  826. else
  827. {
  828. pBufferObj->buff.len = m_dwSocketBufferSize;
  829. result = ::PostReceive(pSocketObj, pBufferObj);
  830. }
  831. }
  832. if(result != NO_ERROR)
  833. {
  834. CheckError(pSocketObj, SO_RECEIVE, result);
  835. bNeedFree = TRUE;
  836. }
  837. if(bNeedFree) AddFreeBufferObj(pBufferObj);
  838. return result;
  839. }
  840. BOOL CTcpAgent::PauseReceive(CONNID dwConnID, BOOL bPause)
  841. {
  842. TSocketObj* pSocketObj = FindSocketObj(dwConnID);
  843. if(!TSocketObj::IsValid(pSocketObj))
  844. {
  845. ::SetLastError(ERROR_OBJECT_NOT_FOUND);
  846. return FALSE;
  847. }
  848. if(!pSocketObj->HasConnected())
  849. {
  850. ::SetLastError(ERROR_INVALID_STATE);
  851. return FALSE;
  852. }
  853. if(pSocketObj->paused == bPause)
  854. return TRUE;
  855. pSocketObj->paused = bPause;
  856. if(!bPause)
  857. return (DoReceive(pSocketObj, GetFreeBufferObj()) == NO_ERROR);
  858. return TRUE;
  859. }
  860. BOOL CTcpAgent::Connect(LPCTSTR lpszRemoteAddress, USHORT usPort, CONNID* pdwConnID, PVOID pExtra)
  861. {
  862. ASSERT(lpszRemoteAddress && usPort != 0);
  863. DWORD result = NO_ERROR;
  864. SOCKET soClient = INVALID_SOCKET;
  865. if(!pdwConnID)
  866. pdwConnID = CreateLocalObject(CONNID);
  867. *pdwConnID = 0;
  868. HP_SOCKADDR addr;
  869. if(!HasStarted())
  870. result = ERROR_INVALID_STATE;
  871. else
  872. {
  873. result = CreateClientSocket(lpszRemoteAddress, usPort, soClient, addr);
  874. if(result == NO_ERROR)
  875. {
  876. result = PrepareConnect(*pdwConnID, soClient);
  877. if(result == NO_ERROR)
  878. result = ConnectToServer(*pdwConnID, lpszRemoteAddress, usPort, soClient, addr, pExtra);
  879. }
  880. }
  881. if(result != NO_ERROR)
  882. {
  883. if(soClient != INVALID_SOCKET)
  884. ::ManualCloseSocket(soClient);
  885. ::SetLastError(result);
  886. }
  887. return (result == NO_ERROR);
  888. }
  889. DWORD CTcpAgent::CreateClientSocket(LPCTSTR lpszRemoteAddress, USHORT usPort, SOCKET& soClient, HP_SOCKADDR& addr)
  890. {
  891. if(!::GetSockAddrByHostName(lpszRemoteAddress, usPort, addr))
  892. return WSAEADDRNOTAVAIL;
  893. BOOL bBind = m_soAddr.IsSpecified();
  894. if(bBind && m_soAddr.family != addr.family)
  895. return WSAEAFNOSUPPORT;
  896. DWORD result = NO_ERROR;
  897. soClient = socket(addr.family, SOCK_STREAM, IPPROTO_TCP);
  898. if(soClient == INVALID_SOCKET)
  899. result = ::WSAGetLastError();
  900. else
  901. {
  902. VERIFY(::SSO_ReuseAddress(soClient, m_bReuseAddress) != SOCKET_ERROR);
  903. BOOL bOnOff = (m_dwKeepAliveTime > 0 && m_dwKeepAliveInterval > 0);
  904. ::SSO_KeepAliveVals(soClient, bOnOff, m_dwKeepAliveTime, m_dwKeepAliveInterval);
  905. const HP_SOCKADDR& bindAddr = bBind ? m_soAddr : HP_SOCKADDR::AnyAddr(addr.family);
  906. if(::bind(soClient, bindAddr.Addr(), bindAddr.AddrSize()) == SOCKET_ERROR)
  907. result = ::WSAGetLastError();
  908. }
  909. return result;
  910. }
  911. DWORD CTcpAgent::PrepareConnect(CONNID& dwConnID, SOCKET soClient)
  912. {
  913. if(!m_bfActiveSockets.AcquireLock(dwConnID))
  914. return ERROR_CONNECTION_COUNT_LIMIT;
  915. if(TRIGGER(FirePrepareConnect(dwConnID, soClient)) == HR_ERROR)
  916. {
  917. VERIFY(m_bfActiveSockets.ReleaseLock(dwConnID, nullptr));
  918. return ENSURE_ERROR_CANCELLED;
  919. }
  920. return NO_ERROR;
  921. }
  922. DWORD CTcpAgent::ConnectToServer(CONNID dwConnID, LPCTSTR lpszRemoteAddress, USHORT usPort, SOCKET soClient, const HP_SOCKADDR& addr, PVOID pExtra)
  923. {
  924. TBufferObj* pBufferObj = GetFreeBufferObj();
  925. TSocketObj* pSocketObj = GetFreeSocketObj(dwConnID, soClient);
  926. addr.Copy(pSocketObj->remoteAddr);
  927. pSocketObj->extra = pExtra;
  928. pSocketObj->host = lpszRemoteAddress;
  929. AddClientSocketObj(dwConnID, pSocketObj);
  930. DWORD result = NO_ERROR;
  931. BOOL bNeedFree = TRUE;
  932. if(m_bAsyncConnect)
  933. {
  934. VERIFY(::SSO_NoBlock(pSocketObj->socket) == NO_ERROR);
  935. if(::CreateIoCompletionPort((HANDLE)soClient, m_hCompletePort, (ULONG_PTR)pSocketObj, 0))
  936. result = ::PostConnect(m_pfnConnectEx, pSocketObj->socket, pSocketObj->remoteAddr, pBufferObj);
  937. else
  938. result = ::GetLastError();
  939. }
  940. else
  941. {
  942. if(::connect(soClient, addr.Addr(), addr.AddrSize()) != SOCKET_ERROR)
  943. {
  944. VERIFY(::SSO_NoBlock(pSocketObj->socket) == NO_ERROR);
  945. if(::CreateIoCompletionPort((HANDLE)soClient, m_hCompletePort, (ULONG_PTR)pSocketObj, 0))
  946. {
  947. pSocketObj->SetConnected();
  948. if(TriggerFireConnect(pSocketObj) != HR_ERROR)
  949. {
  950. result = DoReceive(pSocketObj, pBufferObj);
  951. bNeedFree = FALSE;
  952. }
  953. else
  954. result = ENSURE_ERROR_CANCELLED;
  955. }
  956. else
  957. result = ::GetLastError();
  958. }
  959. else
  960. result = ::WSAGetLastError();
  961. }
  962. if(result != NO_ERROR)
  963. {
  964. if(bNeedFree)
  965. {
  966. AddFreeSocketObj(pSocketObj, SCF_NONE);
  967. AddFreeBufferObj(pBufferObj);
  968. }
  969. }
  970. return result;
  971. }
  972. BOOL CTcpAgent::Send(CONNID dwConnID, const BYTE* pBuffer, int iLength, int iOffset)
  973. {
  974. ASSERT(pBuffer && iLength > 0);
  975. if(iOffset != 0) pBuffer += iOffset;
  976. WSABUF buffer;
  977. buffer.len = iLength;
  978. buffer.buf = (char*)pBuffer;
  979. return SendPackets(dwConnID, &buffer, 1);
  980. }
  981. BOOL CTcpAgent::DoSendPackets(CONNID dwConnID, const WSABUF pBuffers[], int iCount)
  982. {
  983. ASSERT(pBuffers && iCount > 0);
  984. TSocketObj* pSocketObj = FindSocketObj(dwConnID);
  985. if(!TSocketObj::IsValid(pSocketObj))
  986. {
  987. ::SetLastError(ERROR_OBJECT_NOT_FOUND);
  988. return FALSE;
  989. }
  990. return DoSendPackets(pSocketObj, pBuffers, iCount);
  991. }
  992. BOOL CTcpAgent::DoSendPackets(TSocketObj* pSocketObj, const WSABUF pBuffers[], int iCount)
  993. {
  994. ASSERT(pSocketObj && pBuffers && iCount > 0);
  995. if(!pSocketObj->HasConnected())
  996. {
  997. ::SetLastError(ERROR_INVALID_STATE);
  998. return FALSE;
  999. }
  1000. int result = NO_ERROR;
  1001. if(pBuffers && iCount > 0)
  1002. {
  1003. CCriSecLock locallock(pSocketObj->csSend);
  1004. if(TSocketObj::IsValid(pSocketObj))
  1005. result = SendInternal(pSocketObj, pBuffers, iCount);
  1006. else
  1007. result = ERROR_OBJECT_NOT_FOUND;
  1008. }
  1009. else
  1010. result = ERROR_INVALID_PARAMETER;
  1011. if(result != NO_ERROR)
  1012. {
  1013. if(m_enSendPolicy == SP_DIRECT && TSocketObj::IsValid(pSocketObj))
  1014. ::PostIocpClose(m_hCompletePort, pSocketObj->connID, result);
  1015. ::SetLastError(result);
  1016. }
  1017. return (result == NO_ERROR);
  1018. }
  1019. int CTcpAgent::SendInternal(TSocketObj* pSocketObj, const WSABUF pBuffers[], int iCount)
  1020. {
  1021. int result = NO_ERROR;
  1022. for(int i = 0; i < iCount; i++)
  1023. {
  1024. int iBufLen = pBuffers[i].len;
  1025. if(iBufLen > 0)
  1026. {
  1027. BYTE* pBuffer = (BYTE*)pBuffers[i].buf;
  1028. ASSERT(pBuffer);
  1029. switch(m_enSendPolicy)
  1030. {
  1031. case SP_PACK: result = SendPack(pSocketObj, pBuffer, iBufLen); break;
  1032. case SP_SAFE: result = SendSafe(pSocketObj, pBuffer, iBufLen); break;
  1033. case SP_DIRECT: result = SendDirect(pSocketObj, pBuffer, iBufLen); break;
  1034. default: ASSERT(FALSE); result = ERROR_INVALID_INDEX; break;
  1035. }
  1036. if(result != NO_ERROR)
  1037. break;
  1038. }
  1039. }
  1040. return result;
  1041. }
  1042. int CTcpAgent::SendPack(TSocketObj* pSocketObj, const BYTE* pBuffer, int iLength)
  1043. {
  1044. return CatAndPost(pSocketObj, pBuffer, iLength);
  1045. }
  1046. int CTcpAgent::SendSafe(TSocketObj* pSocketObj, const BYTE* pBuffer, int iLength)
  1047. {
  1048. return CatAndPost(pSocketObj, pBuffer, iLength);
  1049. }
  1050. int CTcpAgent::SendDirect(TSocketObj* pSocketObj, const BYTE* pBuffer, int iLength)
  1051. {
  1052. int result = NO_ERROR;
  1053. int iRemain = iLength;
  1054. while(iRemain > 0)
  1055. {
  1056. int iBufferSize = min(iRemain, (int)m_dwSocketBufferSize);
  1057. TBufferObj* pBufferObj = GetFreeBufferObj(iBufferSize);
  1058. memcpy(pBufferObj->buff.buf, pBuffer, iBufferSize);
  1059. result = ::PostSend(pSocketObj, pBufferObj);
  1060. LONG sndCounter = pBufferObj->ReleaseSendCounter();
  1061. if(sndCounter == 0 || result != NO_ERROR)
  1062. {
  1063. AddFreeBufferObj(pBufferObj);
  1064. if(result != NO_ERROR)
  1065. break;
  1066. }
  1067. iRemain -= iBufferSize;
  1068. pBuffer += iBufferSize;
  1069. }
  1070. return result;
  1071. }
  1072. int CTcpAgent::CatAndPost(TSocketObj* pSocketObj, const BYTE* pBuffer, int iLength)
  1073. {
  1074. int result = NO_ERROR;
  1075. pSocketObj->sndBuff.Cat(pBuffer, iLength);
  1076. pSocketObj->pending += iLength;
  1077. if(pSocketObj->IsCanSend() && pSocketObj->IsSmooth() && !::PostIocpSend(m_hCompletePort, pSocketObj->connID))
  1078. result = ::GetLastError();
  1079. return result;
  1080. }
  1081. int CTcpAgent::DoSend(CONNID dwConnID)
  1082. {
  1083. TSocketObj* pSocketObj = FindSocketObj(dwConnID);
  1084. if(TSocketObj::IsValid(pSocketObj))
  1085. return DoSend(pSocketObj);
  1086. return ERROR_OBJECT_NOT_FOUND;
  1087. }
  1088. int CTcpAgent::DoSend(TSocketObj* pSocketObj)
  1089. {
  1090. switch(m_enSendPolicy)
  1091. {
  1092. case SP_PACK: return DoSendPack(pSocketObj);
  1093. case SP_SAFE: return DoSendSafe(pSocketObj);
  1094. default: ASSERT(FALSE); return ERROR_INVALID_INDEX;
  1095. }
  1096. }
  1097. int CTcpAgent::DoSendPack(TSocketObj* pSocketObj)
  1098. {
  1099. if(!pSocketObj->IsCanSend())
  1100. return NO_ERROR;
  1101. int result = NO_ERROR;
  1102. if(pSocketObj->IsPending() && pSocketObj->TurnOffSmooth())
  1103. {
  1104. {
  1105. CCriSecLock locallock(pSocketObj->csSend);
  1106. if(!TSocketObj::IsValid(pSocketObj))
  1107. return ERROR_OBJECT_NOT_FOUND;
  1108. if(pSocketObj->IsPending())
  1109. result = SendItem(pSocketObj);
  1110. pSocketObj->TurnOnSmooth();
  1111. }
  1112. if(result == WSA_IO_PENDING && pSocketObj->IsSmooth())
  1113. ::PostIocpSend(m_hCompletePort, pSocketObj->connID);
  1114. }
  1115. if(!IOCP_SUCCESS(result))
  1116. CheckError(pSocketObj, SO_SEND, result);
  1117. return result;
  1118. }
  1119. int CTcpAgent::DoSendSafe(TSocketObj* pSocketObj)
  1120. {
  1121. if(pSocketObj->sndCount == 0 && !pSocketObj->IsSmooth())
  1122. {
  1123. CCriSecLock locallock(pSocketObj->csSend);
  1124. if(!TSocketObj::IsValid(pSocketObj))
  1125. return ERROR_OBJECT_NOT_FOUND;
  1126. if(pSocketObj->sndCount == 0)
  1127. pSocketObj->smooth = TRUE;
  1128. }
  1129. if(!pSocketObj->IsCanSend())
  1130. return NO_ERROR;
  1131. int result = NO_ERROR;
  1132. if(pSocketObj->IsPending() && pSocketObj->IsSmooth())
  1133. {
  1134. CCriSecLock locallock(pSocketObj->csSend);
  1135. if(!TSocketObj::IsValid(pSocketObj))
  1136. return ERROR_OBJECT_NOT_FOUND;
  1137. if(pSocketObj->IsPending() && pSocketObj->IsSmooth())
  1138. {
  1139. pSocketObj->smooth = FALSE;
  1140. result = SendItem(pSocketObj);
  1141. if(result == NO_ERROR)
  1142. pSocketObj->smooth = TRUE;
  1143. }
  1144. }
  1145. if(!IOCP_SUCCESS(result))
  1146. CheckError(pSocketObj, SO_SEND, result);
  1147. return result;
  1148. }
  1149. int CTcpAgent::SendItem(TSocketObj* pSocketObj)
  1150. {
  1151. int result = NO_ERROR;
  1152. while(pSocketObj->sndBuff.Size() > 0)
  1153. {
  1154. TBufferObj* pBufferObj = pSocketObj->sndBuff.PopFront();
  1155. int iBufferSize = pBufferObj->buff.len;
  1156. ASSERT(iBufferSize > 0 && iBufferSize <= (int)m_dwSocketBufferSize);
  1157. pSocketObj->pending -= iBufferSize;
  1158. ::InterlockedExchangeAdd(&pSocketObj->sndCount, iBufferSize);
  1159. result = ::PostSendNotCheck(pSocketObj, pBufferObj);
  1160. LONG sndCounter = pBufferObj->ReleaseSendCounter();
  1161. if(sndCounter == 0 || !IOCP_SUCCESS(result))
  1162. AddFreeBufferObj(pBufferObj);
  1163. if(result != NO_ERROR)
  1164. break;
  1165. }
  1166. return result;
  1167. }
  1168. BOOL CTcpAgent::SendSmallFile(CONNID dwConnID, LPCTSTR lpszFileName, const LPWSABUF pHead, const LPWSABUF pTail)
  1169. {
  1170. CAtlFile file;
  1171. CAtlFileMapping<> fmap;
  1172. WSABUF szBuf[3];
  1173. HRESULT hr = ::MakeSmallFilePackage(lpszFileName, file, fmap, szBuf, pHead, pTail);
  1174. if(FAILED(hr))
  1175. {
  1176. ::SetLastError(HRESULT_CODE(hr));
  1177. return FALSE;
  1178. }
  1179. return SendPackets(dwConnID, szBuf, 3);
  1180. }
  1181. void CTcpAgent::CheckError(TSocketObj* pSocketObj, EnSocketOperation enOperation, int iErrorCode)
  1182. {
  1183. if(iErrorCode != WSAENOTSOCK && iErrorCode != ERROR_OPERATION_ABORTED)
  1184. AddFreeSocketObj(pSocketObj, SCF_ERROR, enOperation, iErrorCode);
  1185. }