开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > Internet/网络编程 > P2P编程 > 查看源码
Connector.cpp
资源名称：GGBT.rar　[点击查看]
上传用户：lds876
上传日期：2013-05-25
资源大小：567k
文件大小：10k
源码类别：

P2P编程

开发平台：

Visual C++

Connector.cpp：源码内容
  1. // Connector.cpp: implementation of the CConnector class.
  2. //
  3. //////////////////////////////////////////////////////////////////////
  5. #include "stdafx.h"
  6. #include "testbt.h"
  7. #include "Connector.h"
  10. #include "Connection.h"
  11. #include "Encrypter.h"
  12. #include "Upload.h"
  13. #include "Downloader.h"
  14. #include "Choker.h"
  16. #ifdef _DEBUG
  17. #undef THIS_FILE
  18. static char THIS_FILE[]=__FILE__;
  19. #define new DEBUG_NEW
  20. #endif
  22. long toint(char* pBuf)
  23. {
  24. unsigned char* p = (unsigned char*) pBuf;
  25. return p[0]*0x1000000 + p[1]*0x10000 + p[2]*0x100 + p[3];
  26. }
  28. char* tobinary(long lnum, char* pBuf, long length=4)
  29. {
  30. char* p = (char*)&lnum;
  31. pBuf[0] = p[3];
  32. pBuf[1] = p[2];
  33. pBuf[2] = p[1];
  34. pBuf[3] = p[0];
  35. return pBuf;
  36. }
  38. //////////////////////////////////////////////////////////////////////
  39. // Construction/Destruction
  40. //////////////////////////////////////////////////////////////////////
  42. CConnector::CConnector(long lMaxUploadRate)
  43. {
  44. m_lMaxUploadRate = lMaxUploadRate;
  45. }
  48. CConnector::~CConnector()
  49. {
  50. for (int i=0;  i<m_connections.size(); i++)
  51. {
  52. delete m_connections[i]->m_pDownload;
  53. delete m_connections[i]->m_pUpload;
  54. delete m_connections[i];
  55. }
  56. }
  59. void CConnector::connection_made(CEncryptedConnection *pEConnection)
  60. {
  61. CConnection* pnew = new CConnection(pEConnection, this);
  62. m_connections.push_back(pnew);
  63. pnew->m_pUpload = m_pUploader->make_upload(pnew);
  64. pnew->m_pDownload = m_pDownloader->make_download(pnew);
  65. m_pChoker->connection_made(pnew);
  67. }
  69. void CConnector::Create(CDownloader* pDownloader, CUploader* pUploader, CChoker* pchoker, 
  70.  long lNumPieces, CStorageWrapper* pStorage,  CMeasure* pUpTotalMeasure)
  71. {
  72. m_pDownloader = pDownloader;
  73. m_pUploader = pUploader;
  74. m_lNumPieces = lNumPieces;
  75. m_pStorage = pStorage;
  76. m_pUpTotalMeasure = pUpTotalMeasure;
  77. m_pChoker = pchoker;
  78. m_bEndgame = false;
  79. m_bRateCapped = false;
  80. m_tcap = 0;
  81. m_tTimeUntilUncap = 0;
  83. check_endgame();
  84. }
  87. void CConnector::connection_lost(CEncryptedConnection* pEConnection)
  88. {
  89. bool bfind = false;
  90. for (int i=0;  i<m_connections.size(); i++)
  91. {
  92. if (m_connections[i]->m_pEConnection == pEConnection)
  93. {
  94. m_pChoker->connection_lost(m_connections[i]);
  95. m_connections[i]->m_pDownload->disconnected();
  96. delete m_connections[i]->m_pDownload;
  97. delete m_connections[i]->m_pUpload;
  98. delete m_connections[i];
  99. m_connections.erase(m_connections.begin() + i);
  100. bfind = true;
  101. break;
  102. }
  103. }
  104. assert(bfind);
  105. }
  107. void CConnector::connection_flushed(CEncryptedConnection *pEConnection)
  108. {
  109. bool bfind = false;
  110. for (int i=0;  i<m_connections.size(); i++)
  111. {
  112. if (m_connections[i]->m_pEConnection == pEConnection)
  113. {
  114. m_connections[i]->m_pUpload->flushed();
  115. bfind = true;
  116. break;
  117. }
  118. }
  119. assert(bfind);
  120. }
  122. long CConnector::how_many_connections()
  123. {
  124. return m_connections.size();
  125. }
  126. const CDownloader* const CConnector::GetDownloader() const
  127. {
  128. return m_pDownloader;
  129. };
  131. const CMeasure* const CConnector::GetUpTotalMeasure() const
  132. {
  133. return m_pUpTotalMeasure;
  134. }
  136. void CConnector::got_message(CEncryptedConnection* pEConnection, memstream& memMessage)
  137. {
  138. CConnection* pCon = 0;
  139. for (int i=0;  i<m_connections.size(); i++)
  140. {
  141. if (m_connections[i]->m_pEConnection == pEConnection)
  142. {
  143. pCon = m_connections[i];
  144. break;
  145. }
  146. }
  147. assert(pCon);
  150. char* pBuf = memMessage;
  151. long length = memMessage.size();
  153. char t = (*pBuf);
  155. if (t == BITFIELD && pCon->m_bGotAnything)
  156. {
  157. pEConnection->Close();
  158. return;
  159. }
  160. pCon->m_bGotAnything = true;
  162. if (t == CHOKE || t == UNCHOKE || t == INTERESTED || t == NOT_INTERESTED)
  163. {
  164. if (length != 1)
  165. {
  166. pEConnection->Close();
  167. return;
  168. }
  169. }
  171. switch (t)
  172. {
  173. case CHOKE:
  174. {
  175. pCon->m_pDownload->got_choke();
  176. }
  177. break;
  179. case UNCHOKE:
  180. {
  181. pCon->m_pDownload->got_unchoke();
  182. check_endgame();
  183. }
  184. break;
  186. case INTERESTED:
  187. {
  188. pCon->m_pUpload->got_interested();
  189. }
  190. break;
  192. case NOT_INTERESTED:
  193. {
  194. pCon->m_pUpload->got_not_interested();
  195. }
  196. break;
  198. case HAVE:
  199. {
  200. if (length != 5)
  201. {
  202. pCon->Close();
  203. return;
  204. }
  205. long index = toint(pBuf + 1);
  206. if (index >= m_lNumPieces || index < 0)
  207. {
  208. pCon->Close();
  209. return;
  210. }
  211. pCon->m_pDownload->got_have(index);
  212. check_endgame();
  213. }
  214. break;
  216. case BITFIELD:
  217. {
  218. memstream memBitfield;
  219. memBitfield.write(pBuf+1, length - 1);
  220. vector<bool> vHave;
  222. if (!bitfield_to_booleans(memBitfield, m_lNumPieces, vHave))
  223. {
  224. pCon->Close();
  225. return;
  226. }
  227. pCon->m_pDownload->got_have_bitfield(vHave);
  228. check_endgame();
  229. }
  230. break;
  232. case REQUEST:
  233. {
  234. if (length != 13)
  235. {
  236. pCon->Close();
  237. return;
  238. }
  240. long index = toint(pBuf + 1);
  241. if (index >= m_lNumPieces)
  242. {
  243. pCon->Close();
  244. return;
  245. }
  246. pCon->m_pUpload->got_request(index, toint(pBuf + 5), toint(pBuf+9));
  247. }
  248. break;
  250. case CANCEL:
  251. {
  252. if (length != 13)
  253. {
  254. pCon->Close();
  255. return;
  256. }
  258. long index = toint(pBuf + 1);
  259. if (index >= m_lNumPieces)
  260. {
  261. pCon->Close();
  262. return;
  263. }
  264. pCon->m_pUpload->got_cancel(index, toint(pBuf + 5), toint(pBuf+9));
  266. }
  267. break;
  269. case PIECE:
  270. {
  271. if (length <= 9)
  272. {
  273. pCon->Close();
  274. return;
  275. }
  276. long index = toint(pBuf + 1);
  277. long begin = toint(pBuf+5);
  278. if (index >= m_lNumPieces)
  279. {
  280. pCon->Close();
  281. return;
  282. }
  284. memMessage.TrimLeft(9);
  285. // memstream memPiece;
  286. // memPiece.write(pBuf+9, length - 9);
  287. // if (pCon->m_pDownload->got_piece(index, toint(pBuf+5), memPiece))
  288. if (pCon->m_pDownload->got_piece(index, begin, memMessage))
  289. {
  290. for (int i=0;  i<m_connections.size(); i++)
  291. {
  292. m_connections[i]->send_have(index);
  293. }
  294. }
  295. check_endgame();
  296. }
  297. break;
  298. default:
  299. {
  300. pCon->Close();
  301. return;
  302. }
  303. break;
  304. }
  306. }
  308. void CConnector::check_endgame()
  309. {
  310. if (!m_pDownloader->is_endgame() && m_pStorage->is_everything_pending())
  311. {
  312. m_bEndgame = true;
  313. m_pDownloader->make_endgame();
  314. }
  315. }
  317. void CConnector::_update_upload_rate(long lAmount)
  318. {
  319. m_pUpTotalMeasure->update_rate(lAmount);
  320. if (m_lMaxUploadRate> 0 && m_pUpTotalMeasure->get_rate_noupdate() > m_lMaxUploadRate)
  321. {
  322. m_bRateCapped = true;
  323. time(&m_tcap);
  324. m_tTimeUntilUncap = m_pUpTotalMeasure->time_until_rate(m_lMaxUploadRate);
  326. TRACE("rate capeped() wait time :(%d)rn", m_tTimeUntilUncap);
  327. }
  328. }
  330. void CConnector::uncap()
  331. {
  332. if (!m_bRateCapped)
  333. return;
  335. // check the time to excute?
  336. time_t t;
  337. time(&t);
  338. time_t tspan = t - m_tcap;
  339. assert(tspan >= 0);
  340. if (tspan < 0)
  341. m_tcap = t;
  342. if (tspan < m_tTimeUntilUncap)
  343. return ;
  346. TRACE("Rate uncap()rn");
  348. // flush the smaller rate connection to up the rate, until rate are recapped or all data flushed.
  349. m_bRateCapped = false;
  350. while (!m_bRateCapped)
  351. {
  352. CUpload * pUp = 0;
  353. long lminrate = -1;
  355. for (int i=0; i<m_connections.size(); i++)
  356. {
  357. if (!m_connections[i]->m_pUpload->is_choked() && 
  358. m_connections[i]->m_pUpload->has_queries() &&
  359. m_connections[i]->IsFlush())
  360. {
  361. long lrate = m_connections[i]->m_pUpload->get_rate();
  362. if (lminrate == -1 || lrate < lminrate)
  363. {
  364. pUp = m_connections[i]->m_pUpload;
  365. lminrate = lrate;
  366. }
  367. }
  368. }
  370. if (!pUp) 
  371. break;
  373. TRACE("nRate uncap() excute flush() minrate(%d)rn", lminrate);
  374. pUp->flushed();
  376. if (m_pUpTotalMeasure->get_rate_noupdate() > m_lMaxUploadRate)
  377. break;
  378. }
  380. }
  382. bool CConnector::IsRatecap()
  383. {
  384. return m_bRateCapped;
  385. }
  387. void CConnector::SetMaxUploadRate(long lMaxUploadRate)
  388. {
  389. m_lMaxUploadRate = lMaxUploadRate;
  390. }
  392. void CConnector::OnIdle()
  393. {
  394. bool bSth = false;
  395. vector<CAddrPort> vToPause, vToContinue, vToClose;
  397. BOOL bRet = m_criticalSection.Lock();
  398. if (!bRet) return;
  400. if (!m_vToPause.empty() ||
  401. !m_vToContinue.empty() || 
  402. !m_vToClose.empty())
  403. {
  404. bSth = true;
  405. vToPause = m_vToPause;
  406. vToContinue = m_vToContinue;
  407. vToClose = m_vToClose;
  409. m_vToPause.erase(m_vToPause.begin(), m_vToPause.end());
  410. m_vToContinue.erase(m_vToContinue.begin(), m_vToContinue.end());
  411. m_vToClose.erase(m_vToClose.begin(), m_vToClose.end());
  412. }
  414. bRet = m_criticalSection.Unlock();
  415. if (!bRet) return;
  417. if (bSth)
  418. _OnIdle(vToPause, vToContinue, vToClose);
  419. }
  421. void CConnector::_OnIdle(vector<CAddrPort>& vToPause, vector<CAddrPort>& vToContinue, vector<CAddrPort>& vToClose)
  422. {
  423. for (int i=0; i<vToPause.size(); i++)
  424. {
  425. CAddrPort addrPort = vToPause[i];
  427. for (int j=0; j<m_connections.size(); j++)
  428. {
  429. long lAddr = 0;
  430. short sPort = 0;
  431. m_connections[j]->GetIP(lAddr, sPort);
  433. if (addrPort.m_lAddr == lAddr && addrPort.m_sPort == sPort)
  434. {
  435. TRACE("user pause connectionrn");
  436. m_connections[j]->Pause(true);
  437. break;
  438. }
  439. }
  440. }
  442. for (i=0; i<vToContinue.size(); i++)
  443. {
  444. CAddrPort addrPort = vToContinue[i];
  446. for (int j=0; j<m_connections.size(); j++)
  447. {
  448. long lAddr = 0;
  449. short sPort = 0;
  450. m_connections[j]->GetIP(lAddr, sPort);
  452. if (addrPort.m_lAddr == lAddr && addrPort.m_sPort == sPort)
  453. {
  454. TRACE("user unpause connectionrn");
  455. m_connections[j]->Pause(false);
  456. break;
  457. }
  458. }
  459. }
  461. for (i=0; i<vToClose.size(); i++)
  462. {
  463. CAddrPort addrPort = vToClose[i];
  465. for (int j=0; j<m_connections.size(); j++)
  466. {
  467. long lAddr = 0;
  468. short sPort = 0;
  469. m_connections[j]->GetIP(lAddr, sPort);
  471. if (addrPort.m_lAddr == lAddr && addrPort.m_sPort == sPort)
  472. {
  473. TRACE("user close connectionrn");
  474. m_connections[j]->Close();
  475. break;
  476. }
  477. }
  478. }
  479. }
  481. void CConnector::PausePeer(CAddrPort addrPort, bool bPause)
  482. {
  483. BOOL bRet = m_criticalSection.Lock();
  484. if (!bRet) return;
  486. if (bPause)
  487. m_vToPause.push_back(addrPort);
  488. else
  489. m_vToContinue.push_back(addrPort);
  491. bRet = m_criticalSection.Unlock();
  492. if (!bRet) return;
  493. }
  495. void CConnector::ClosePeer(CAddrPort addrPort)
  496. {
  497. BOOL bRet = m_criticalSection.Lock();
  498. if (!bRet) return;
  500. m_vToClose.push_back(addrPort);
  502. bRet = m_criticalSection.Unlock();
  503. if (!bRet) return;
  504. }