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

源码开发语言/平台
当前位置：首页 > 源码/资料 > Internet/网络编程 > P2P编程 > 查看源码
P2PClient.cpp
资源名称：p2p_vod.rar　[点击查看]
上传用户：liguizhu
上传日期：2015-11-01
资源大小：2422k
文件大小：39k
源码类别：

P2P编程

开发平台：

Visual C++

P2PClient.cpp：源码内容
  1. /*
  2.  *  Openmysee
  3.  *
  4.  *  This program is free software; you can redistribute it and/or modify
  5.  *  it under the terms of the GNU General Public License as published by
  6.  *  the Free Software Foundation; either version 2 of the License, or
  7.  *  (at your option) any later version.
  8.  *
  9.  *  This program is distributed in the hope that it will be useful,
  10.  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  11.  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12.  *  GNU General Public License for more details.
  13.  *
  14.  *  You should have received a copy of the GNU General Public License
  15.  *  along with this program; if not, write to the Free Software
  16.  *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  17.  *
  18.  */
  19. // P2PClient.cpp: implementation of the P2PClient class.
  20. //
  21. //////////////////////////////////////////////////////////////////////
  23. #include "stdafx.h"
  24. #include "P2PClient.h"
  25. #include "Communicator.h"
  27. namespace NPLayer1 {
  29. //////////////////////////////////////////////////////////////////////
  30. // Construction/Destruction
  31. //////////////////////////////////////////////////////////////////////
  33. P2PClient::P2PClient() 
  34. : recvPointer(0)
  35. , sendPointer(0) 
  36. , isSameLan(false)
  37. {
  38. valid = FALSE;
  39. errStr[0] = 0;
  40. }
  42. P2PClient::~P2PClient() {
  43. if(valid)
  44. SetInvalid();
  45. }
  47. BOOL P2PClient::SetValid(Communicator* c, const ConnectingPeer& peer) {
  48. comm = c;
  49. recvOff = 0;
  50. valid = TRUE;
  52. m_Socket = peer.sock;
  53. isIncoming= peer.isIncoming;
  54. isForFree = peer.isForFree;
  55. isPassive = peer.isPassive;
  56. isSameRes = false;
  57. isSameLan = peer.isSameLan;
  59. // 到这里为止,我们完全不知道对方有哪些块
  60. remoteInterval.Clear();
  62. remotePush.Clear();
  63. localPush.Clear();
  64. sentMediaArray.Clear();
  66. connectionBeginTime = lastRecvDataTime = lastSendDataTime = lastRecvDataHdTime = timeGetTime();
  67. bGotFirstMsg = FALSE;
  69. memcpy((PeerInfoWithAddr*)&remotePeer, (PeerInfoWithAddr*)&peer, sizeof(PeerInfoWithAddr));
  70. comm->logFile.StatusOut("Connected on %s.", comm->p2pMgr.FormatIPAddress(remotePeer));
  72. if(GetIsCachePeer()) {
  73. remoteVersion = comm->cfgData.COMMUNICATOR_VERSION;
  74. remotePeer.layer = 0;
  75. }
  76. else {
  77. remoteVersion = 0.0f;
  78. remotePeer.layer = 0xff;
  79. }
  81. // clear transfer calculator members
  82. ClearTransferInfo();
  83. lastRequestStartTime = 0;
  84. lastRequestBytes = 0;
  85. m_reqStartTime = 0;
  86. m_transUsedTime = 0;
  88. if(!SendHello())
  89. return FALSE;
  90. CorePeerInfo thisPeer;
  91. comm->p2pMgr.GetSelfInfo(thisPeer);
  92. if(!SendReport(thisPeer, true))
  93. return FALSE;
  95. if(GetIsCachePeer()) {
  96. // 请求重新分配所有push list
  97. comm->p2pMgr.RedistributeAllBlocks(this);
  98. }
  99. return TRUE;
  100. }
  102. void P2PClient::SetInvalid() {
  103. if(valid) {
  104. valid = FALSE;
  106. remotePush.Clear();
  108. comm->logFile.StatusOut("Clear TCPPacket Send list...");
  109. while (!m_sendList.empty()) {
  110. comm->p2pMgr.ReleasePacket(m_sendList.front());
  111. m_sendList.pop_front();
  112. }
  114. comm->logFile.StatusOut("Disconnected from %s.", comm->p2pMgr.FormatIPAddress(remotePeer));
  116. comm->p2pMgr.SafeCloseSocket(m_Socket); // no force disconnect
  118. // 此处没有排除低版本客户端的情况，所以统计会有少许偏差
  119. comm->p2pMgr.ConnectionClosed(isIncoming, timeGetTime()-connectionBeginTime);
  120. }
  121. }
  123. BOOL P2PClient::BaseRecv() {
  124. int ret = recv(m_Socket, recvBuf+recvOff, P2P_BUF_SIZE-recvOff, 0);
  125. if(ret < 0) {
  126. DWORD lastError = ::WSAGetLastError();
  127. if (WSAEWOULDBLOCK != lastError) {
  128. comm->logFile.StatusErr("Receiving data on TCP", lastError);
  129. return FALSE;
  130. }
  131. else
  132. return TRUE;
  133. }
  134. else if(0 == ret) {
  135. comm->logFile.StatusOut("Connection has been disconnected gracefully.");
  136. return FALSE;
  137. }
  139. recvOff += ret;
  140. AddIncomingBytes(ret);
  141. comm->p2pMgr.AddIncomingBytes(ret);
  143. BOOL retVal = FALSE;
  145. for(;;) {
  146. // because multiple msgs can be received at once.
  147. // keep call parseMsg() till !MSG_COMPLETE
  148. MSG_STATE ms = ParseMsg(); 
  149. bool bBadAddr = false;
  151. switch(ms) {
  152. case MSG_COMPLETE:
  153. continue;
  154. case MSG_UNCOMPLETE:
  155. retVal = TRUE;
  156. break;
  157. case MSG_ERR_SIZE:
  158. sprintf(errStr, "错误的消息大小!");
  159. break;
  160. case MSG_ERR_TYPE: 
  161. sprintf(errStr, "错误的消息类型!");
  162. break;
  163. case MSG_DIFF_CHNL:
  164. sprintf(errStr, "属于不同的频道!");
  165. bBadAddr = true;
  166. break;
  167. case MSG_NOMORE_CONS:
  168. sprintf(errStr, "已经存在一个连接!");
  169. break;
  170. case MSG_ERR_LIST_SIZE:
  171. sprintf(errStr, "错误的列表大小!");
  172. break;
  173. case MSG_SEND_ERR:
  174. sprintf(errStr, "发送消息错误!");
  175. break;
  176. case MSG_SAVEDATA_ERR:
  177. sprintf(errStr, "无法保存数据!");
  178. break;
  179. case MSG_UNMATCH_BLOCKID:
  180. sprintf(errStr, "返回了并未请求的块!");
  181. break;
  182. case MSG_NOSUCH_RES_HERE:
  183. sprintf(errStr, "本机没有这个资源!");
  184. bBadAddr = true;
  185. break;
  186. case MSG_REMOTE_ERR:
  187. // 错误信息已经在errStr中了
  188. break;
  189. case MSG_CHNL_CLOSED:
  190. sprintf(errStr, "频道关闭!!");
  191. // 发送没有频道关闭的消息给外界
  192. comm->PostErrMessage(PNT_CHNL_CLOSED, 0, true);
  193. break;
  194. case MSG_NOSUCH_RES_SP:
  195. sprintf(errStr, "SP没有这个资源!!");
  196. // 发送没有资源的消息给外界
  197. comm->PostErrMessage(PNT_NO_SUCH_RES, 0, true);
  198. break;
  199. case MSG_CHNL_END:
  200. sprintf(errStr, "频道结束了!");
  201. // 发送频道结束的消息给外界
  202. comm->PostErrMessage(PNT_CHNL_ENDED, 0, true);
  203. break;
  204. case MSG_LOW_VERSION:
  205. sprintf(errStr, "对方客户端版本过低！");
  206. bBadAddr = true;
  208. comm->p2pMgr.AddLowVersionConCount();
  209. break;
  210. default:
  211. sprintf(errStr, "未知错误类型!");
  212. bBadAddr = true;
  213. break;
  214. }
  215. if(strlen(errStr) > 0) {
  216. comm->logFile.StatusOut("来自%s的错误: %s", comm->p2pMgr.FormatIPAddress(remotePeer), errStr);
  217. errStr[0] = 0;
  219. // reject this client
  220. comm->p2pMgr.AddBadAddr(remotePeer);
  221. }
  222. break;
  223. }
  224. return retVal;
  225. }
  227. MSG_STATE P2PClient::ParseMsg() {
  228. // 如果过小，则不是正常的包
  229. if(recvOff < sizeof(int)+sizeof(BYTE))
  230. return MSG_UNCOMPLETE;
  232. // 把移动指针放到数据的起始地址
  233. recvPointer = recvBuf;
  235. // 读取消息大小
  236. CSClient::CopyMoveSrc(&msgSize, recvPointer, sizeof(msgSize));
  238. // 读取消息类型
  239. UINT8 msgType;
  240. CSClient::CopyMoveSrc(&msgType, recvPointer, sizeof(msgType));
  242. // msgSize是否正常
  243. if(msgSize > P2P_BUF_SIZE || msgSize < sizeof(int)+sizeof(BYTE))
  244. return MSG_ERR_SIZE;
  246. // 因为P2P_RESPONSE包含数据，可能传输比较慢
  247. // 在这里至少我们知道发送的请求的到回应了，要根据这个判断请求是否超时
  248. if(msgType == P2P_RESPONSE) {
  249. lastRecvDataHdTime = timeGetTime();
  250. }
  252. // 是否包含完成的消息
  253. if(recvOff < msgSize)
  254. return MSG_UNCOMPLETE;
  256. MSG_STATE ret = MSG_COMPLETE;
  257. switch(msgType) {
  258. case P2P_HELLO:
  259. ret = OnHello();
  260. break;
  261. case P2P_SPUPDATE:
  262. ret = OnSPUpdate();
  263. break;
  264. case P2P_REPORT:
  265. ret = OnReport();
  266. break;
  267. case P2P_NEAR_PEERS:
  268. ret = OnNearPeers();
  269. break;
  270. case P2P_PUSHLIST:
  271. ret = OnPushList();
  272. break;
  273. case P2P_RESPONSE:
  274. ret = OnResponse();
  275. break;
  276. case P2P_MSG:
  277. ret = OnMsg();
  278. break;
  279. case P2P_REQMEDIA:
  280. ret = OnReqMedia();
  281. break;
  282. case P2P_MEDIATYPE:
  283. ret = OnMediaType();
  284. break;
  285. default:
  286. ret = MSG_ERR_TYPE;
  287. break;
  288. }
  290. // copy left data to start of recvBuf
  291. if(recvOff >= msgSize) {
  292. memcpy(recvBuf, recvBuf+msgSize, recvOff-msgSize);
  293. recvOff -= msgSize;
  294. }
  296. return ret;
  297. }
  299. MSG_STATE P2PClient::OnHello() {
  300. assert(!GetIsCachePeer());
  302. // NP version
  303. CSClient::CopyMoveSrc(&remoteVersion, recvPointer, sizeof(remoteVersion));
  304. if(remoteVersion < comm->cfgData.ACCEPT_VERSION) {
  305. comm->logFile.StatusOut("Reject low version client %s %.5f.", comm->p2pMgr.FormatIPAddress(remotePeer), remoteVersion);
  306. return MSG_LOW_VERSION;
  307. }
  309. // 对方需要的资源Hash码
  310. char resHashCode[MD5_LEN+1];
  311. resHashCode[MD5_LEN] = 0;
  312. CSClient::CopyMoveSrc(resHashCode, recvPointer, MD5_LEN);
  314. // 是否被动连接,因为对方可能是因为收到TS2NP_CONNECTO，被动连接过来的
  315. // 那本连接就算做outgoing，对方会将本连接算做incoming
  316. bool passiveConnect = false;
  317. CSClient::CopyMoveSrc(&passiveConnect, recvPointer, sizeof(passiveConnect));
  318. if(passiveConnect)
  319. isIncoming = false;
  321. // 对方Peer信息
  322. CSClient::CopyMoveSrc(&remotePeer, recvPointer, sizeof(remotePeer));
  324. comm->logFile.StatusOut("Got P2P_HELLO from to %s.", comm->p2pMgr.FormatIPAddress(remotePeer));
  326. // 如果对方资源和本机当前资源相同，则是非Free的连接
  327. if(comm->currRes && memcmp(resHashCode, comm->currRes->GetHashCode().data(), MD5_LEN) == 0) {
  328. isForFree = false;
  329. isSameRes = true;
  330. }
  331. else {
  332. return MSG_NOSUCH_RES_HERE;
  333. }
  335. if(isIncoming || passiveConnect) {
  336. // 此时isIncoming为true，说明是accept进来的连接,要在这里判断是否重复连接
  337. // 前面的passiveConnect可能改变了isIncoming的值, 
  338. // 而passiveConnect必定是incoming的连接，所以一并在这里判断
  339. if(!comm->p2pMgr.CheckN4One(remotePeer, isForFree, true, MAX_CONNECTION_PER_NP))
  340. return MSG_NOMORE_CONS;
  341. }
  343. // 如果对方不是CachePeer, 则第一个收到的消息应该在这里
  344. bGotFirstMsg = TRUE;
  346. if (remotePeer.outerIP.sin_addr.s_addr == comm->localAddress.outerIP.sin_addr.s_addr) {
  347. isSameLan = true;
  348. }
  350. comm->p2pMgr.ConnectionEstablished(isIncoming);
  352. return MSG_COMPLETE;
  353. }
  355. MSG_STATE P2PClient::OnSPUpdate() {
  356. if(!comm->currRes) {
  357. assert(0);
  358. return MSG_ERR_TYPE;
  359. }
  361. MSG_STATE state = MSG_COMPLETE;
  362. UINT oldMaxBlockID = UINT_MAX;
  364. // 记录旧的最大块ID
  365. oldMaxBlockID = comm->currRes->GetSPUpdate().maxBlockID;
  367. // 收到的SPUpdate
  368. SPUpdate tmpUpdate;
  369. CSClient::CopyMoveSrc(&tmpUpdate, recvPointer, sizeof(tmpUpdate));
  371. comm->logFile.StatusOut("Recv SPUpdate %d->%d from %s", tmpUpdate.minBlockID, tmpUpdate.maxBlockID, 
  372. comm->p2pMgr.FormatIPAddress(remotePeer));
  374. // 计算收到的SPUpdate的校验码
  375. PBYTE temp = reinterpret_cast<PBYTE>(recvPointer-sizeof(tmpUpdate));
  376. BYTE calsum = 0;
  377. for(int i = 0; i < sizeof(tmpUpdate); ++i) {
  378. calsum += *temp;
  379. temp++;
  380. }
  382. // 如果是不带校验的SPUpdate，则不接受
  383. if(msgSize != 5+sizeof(SPUpdate)+1) {
  384. comm->logFile.StatusOut("Old SPUpdate!");
  385. return MSG_COMPLETE;
  386. }
  388. // 读取对方发送的SPUpdate校验码
  389. BYTE sum = 0;
  390. CSClient::CopyMoveSrc(&sum, recvPointer, sizeof(sum));
  392. // 比较两个SPUpdate的校验码，必须符合
  393. if(calsum != sum) {
  394. comm->logFile.StatusOut("Bad SPUpdate, err sum!");
  395. return MSG_COMPLETE;
  396. }
  398. if(tmpUpdate.minBlockID == UINT_MAX && tmpUpdate.maxBlockID == UINT_MAX) {
  399. if(tmpUpdate.minKeySample == 0xffffffffffffffff && tmpUpdate.maxKeySample == 0xffffffffffffffff)
  400. state = MSG_NOSUCH_RES_SP; // SP 上没有这个资源
  401. else if(tmpUpdate.minKeySample == 0 && tmpUpdate.maxKeySample == 0)
  402. state = MSG_CHNL_CLOSED; // 这个频道已经关闭
  403. else 
  404. assert(0); // 错误的消息
  405. }
  406. if(tmpUpdate.minBlockID == 0 && tmpUpdate.maxBlockID == 0 && 
  407. tmpUpdate.minKeySample == 0 && tmpUpdate.maxKeySample == 0)
  408. state = MSG_CHNL_END;
  410. if(state == MSG_COMPLETE) {
  411. if(GetIsCachePeer()) {
  412. remoteInterval.Clear();
  413. remoteInterval.AddInterval(tmpUpdate.minBlockID, tmpUpdate.maxBlockID-tmpUpdate.minBlockID);
  414. }
  416. // 如果收到SPUpdate的maxBlockID比本机SPUpdate的maxBlockID更大
  417. // 则更新本机SPUpdate，并向比本机层数更高的连接广播
  418. if(tmpUpdate.maxBlockID > oldMaxBlockID) {
  419. comm->currRes->SetSPUpdate(tmpUpdate, sum);
  421. // 检查PushList状况，如果只剩0块了，就要求重新分配所有连接的块
  422. // 直播系统中连上CP的连接，不会接收到P2P_REPORT，所以只能根据spUpdate决定是否开始请求数据
  423. if(remotePush.GetValidSize() == 0 && GetIsCachePeer())
  424. comm->p2pMgr.RedistributeAllBlocks(this);
  426. // 广播最新的SPUpdate
  427. comm->p2pMgr.BroadCastSPUpdate(tmpUpdate, sum);
  428. }
  429. }
  430. else {
  431. // 广播频道的非正常状态
  432. comm->p2pMgr.BroadCastSPUpdate(tmpUpdate, sum);
  433. }
  435. return state;
  436. }
  438. MSG_STATE P2PClient::OnReport() {
  439. // 复制对方的信息
  440. CSClient::CopyMoveSrc((CorePeerInfo*)&remotePeer, recvPointer, sizeof(CorePeerInfo));
  442. // 是否更新全部区间列表
  443. bool bRefresh;
  444. CSClient::CopyMoveSrc(&bRefresh, recvPointer, sizeof(bRefresh));
  445. if(bRefresh)
  446. remoteInterval.Clear();
  448. // 如果REFRESH，那么只有一组区间；如果不是，那么有两组区间，先后是增加和删除
  449. for(UINT8 j = 0; j < (bRefresh?1:2); ++j) {
  450. // read interval count
  451. UINT8 intervalNum = 0;
  452. CSClient::CopyMoveSrc(&intervalNum, recvPointer, sizeof(intervalNum));
  453. assert(recvPointer - recvBuf < 1000);
  455. // read intervals
  456. for(UINT8 i = 0; i < intervalNum; ++i) {
  457. BlockInterval temp;
  458. CSClient::CopyMoveSrc(&temp, recvPointer, sizeof(temp));
  459. if(j == 0)
  460. remoteInterval.AddInterval(temp.start, temp.size);
  461. else
  462. remoteInterval.DelInterval(temp.start, temp.size);
  463. }
  465. // 如果对方新增了Block
  466. if(j == 0 && intervalNum > 0) {
  467. // 检查PushList状况，如果只剩0块了，就要求重新分配所有连接的块
  468. if(remotePush.GetValidSize() == 0 && comm->currRes)
  469. comm->p2pMgr.RedistributeAllBlocks(this);
  470. }
  471. }
  473. //comm->logFile.StatusOut("Recv P2P_REPORT from %s", comm->p2pMgr.FormatIPAddress(remotePeer));
  475. return MSG_COMPLETE;
  476. }
  478. MSG_STATE P2PClient::OnNearPeers() {
  479. // get size of peer list and move the pointer
  480. UINT8 listSize;
  481. CSClient::CopyMoveSrc(&listSize, recvPointer, sizeof(listSize));
  483. // 检查listSize是否正常
  484. if(static_cast<int>(listSize*sizeof(PeerInfoWithAddr)) > recvBuf+P2P_BUF_SIZE-recvPointer)
  485. return MSG_ERR_LIST_SIZE;
  487. // read peer one by one, and add to known peer list
  488. for(UINT8 i = 0; i < listSize; ++i) {
  489. PeerInfoWithAddr peer;
  490. CSClient::CopyMoveSrc(&peer, recvPointer, sizeof(peer));
  491. peer.isCachePeer = false;
  492. comm->p2pMgr.AddPeerInfo(peer);
  493. }
  494. comm->logFile.StatusOut("Recv P2P_NEAR_PEERS(%d NP) from %s.", listSize, comm->p2pMgr.FormatIPAddress(remotePeer));
  495. return MSG_COMPLETE;
  496. }
  498. MSG_STATE P2PClient::OnPushList() {
  499. // 当没有当前资源时，不给对方发送数据
  500. if(!comm->currRes)
  501. return MSG_SEND_ERR;
  504. // 是否更新整个PushList
  505. bool bRefresh;
  506. CSClient::CopyMoveSrc(&bRefresh, recvPointer, sizeof(bRefresh));
  507. if(bRefresh) {
  508. // 清空所有准备发送的块（已经开始发送的块，不用清空）
  509. while(DeleteSendingBlock(UINT_MAX));
  510. localPush.Clear();
  511. }
  513. UINT blockID = UINT_MAX;
  514. if(bRefresh) {
  515. UINT8 count;
  516. CSClient::CopyMoveSrc(&count, recvPointer, sizeof(count));
  517. assert(count < 30);
  518. for(UINT8 i = 0; i < count; ++i) {
  519. CSClient::CopyMoveSrc(&blockID, recvPointer, sizeof(blockID));
  520. // 查找本机有没有这个块，如果没有，立即告知对方；
  521. if(!comm->currRes->FindBlock(blockID)) {
  522. if(!SendResponse(blockID, false))
  523. return MSG_SEND_ERR;
  524. }
  525. else 
  526. localPush.AddBlock(blockID);
  527. }
  528. if(count)
  529. comm->logFile.StatusOut("Got request refresh(%d...) from %s.", blockID, comm->p2pMgr.FormatIPAddress(remotePeer));
  530. }
  531. else {
  532. UINT8 count;
  533. CSClient::CopyMoveSrc(&count, recvPointer, sizeof(count));
  534. assert(count < 2);
  535. for(UINT8 i = 0; i < count; ++i) {
  536. CSClient::CopyMoveSrc(&blockID, recvPointer, sizeof(blockID));
  537. // 查找本机有没有这个块，如果没有，立即告知对方；
  538. if(!comm->currRes->FindBlock(blockID)) {
  539. if(!SendResponse(blockID, false))
  540. return MSG_SEND_ERR;
  541. }
  542. else 
  543. localPush.AddBlock(blockID);
  544. }
  545. if(count)
  546. comm->logFile.StatusOut("Got request add(%d) from %s.", blockID, comm->p2pMgr.FormatIPAddress(remotePeer));
  547. CSClient::CopyMoveSrc(&count, recvPointer, sizeof(count));
  548. assert(count < 2);
  549. for(UINT8 i = 0; i < count; ++i) {
  550. CSClient::CopyMoveSrc(&blockID, recvPointer, sizeof(blockID));
  551. // 如果在localPush中查找并删除此块
  552. if(!localPush.DelBlock(blockID)) {
  553. // 如果在localPush中没有找到，则说明此块已经在发送列表中了，甚至可能已经被发送了
  554. // 尝试从发送列表中查找并删除此块
  555. if(!DeleteSendingBlock(blockID)) {
  556. comm->logFile.StatusOut("Ahhhhh..., the block has already been sent!");
  557. }
  558. }
  559. }
  560. if(count)
  561. comm->logFile.StatusOut("Got request del(%d) from %s.", blockID, comm->p2pMgr.FormatIPAddress(remotePeer));
  562. }
  564. // 发送push列表中的第一个块
  565. if(!SendFirstPushBlock())
  566. return MSG_SEND_ERR;
  567. return MSG_COMPLETE;
  568. }
  570. MSG_STATE P2PClient::OnResponse() {
  571. // 如果对方是CachePeer, 则第一个收到的消息应该在这里
  572. if(GetIsCachePeer())
  573. bGotFirstMsg = TRUE;
  575. MSG_STATE ret = MSG_COMPLETE;
  577. // 回应的BlockID和大小
  578. UINT blockID;
  579. CSClient::CopyMoveSrc(&blockID, recvPointer, sizeof(blockID));
  580. UINT blockSize;
  581. CSClient::CopyMoveSrc(&blockSize, recvPointer, sizeof(blockSize));
  583. // 检查Block大小是否正常
  584. if(static_cast<int>(blockSize) > recvBuf+P2P_BUF_SIZE-recvPointer) {
  585. ret = MSG_ERR_SIZE;
  586. assert(0);
  587. }
  589. // 检查对方返回的Block是不是在Push List之中，如果在，则删除之；
  590. // 如果不在，则检查是否在其他连接的Push List之中，如果在，则删除之
  591. DWORD reqTime;
  592. if(!remotePush.DelBlock(blockID, &reqTime))
  593. comm->p2pMgr.ReclaimBlocks(this, blockID);
  595. if(ret == MSG_COMPLETE) {
  596. if(blockSize > 0) {
  597. // 记录此次得到数据的时间
  598. lastRecvDataTime = timeGetTime();
  600. // 记录本次传输所用的时间
  601. m_transUsedTime += timeGetTime()-m_reqStartTime;
  602. //assert(lastRecvDataTime-m_reqStartTime < 25000);
  604. // 确实已经下载了，虽然可能发生存储错误
  605. comm->p2pMgr.AddBlockDataDown(blockSize);
  606. if(GetIsCachePeer())
  607. comm->p2pMgr.AddCPData(blockSize);
  609. // 存储此块
  610. P2P_RETURN_TYPE bSuccess = comm->currRes->PutBlock(blockID, blockSize, reinterpret_cast<PBYTE>(recvPointer));
  611. recvPointer += blockSize;
  613. if(bSuccess < PRT_OK) {
  614. if(bSuccess == PRT_BUFFER_FULL) {
  615. comm->logFile.StatusOut("缓冲区已满");
  616. Sleep(300);
  617. }
  618. else if(bSuccess == PRT_BAD_BLOCK) {
  619. if(!remotePeer.isCachePeer) {
  620. // 这个块是错误的,不再从这个NP下载此块
  621. remoteInterval.DelInterval(blockID, 1);
  622. }
  623. }
  624. else {
  625. comm->logFile.StatusErr("写入磁盘错误", GetLastError());
  626. // 发送写入磁盘错误消息
  627. comm->PostErrMessage(PNT_DISK_ERR, 0, true);
  628. ret = MSG_SAVEDATA_ERR;
  629. assert(0);
  630. }
  631. }
  633. char* temp = "";
  634. #ifdef _DEBUG
  635. MD5 md5(reinterpret_cast<BYTE*>(recvPointer-blockSize), blockSize);
  636. temp = md5.hex_digest();
  637. #endif
  638. comm->logFile.StatusOut("Got %sblock %d(%d)(%s)(used %dms) from %s.", 
  639. (bSuccess==PRT_BAD_BLOCK?"bad ":""), blockID, blockSize, temp, lastRecvDataTime-m_reqStartTime, 
  640. comm->p2pMgr.FormatIPAddress(remotePeer));
  641. #ifdef _DEBUG
  642. delete [] temp;
  644. // 记录下次传输开始的时间
  645. m_reqStartTime = timeGetTime();
  646. #endif
  647. }
  648. else {
  649. comm->logFile.StatusOut("Got block %d(%d) from %s.", 
  650. blockID, blockSize, comm->p2pMgr.FormatIPAddress(remotePeer));
  651. }
  652. // 既然得到了数据，登录不上TS的问题就算了：）
  653. comm->csClient.ResetLoginFail();
  654. }
  656. if(ret == MSG_COMPLETE) {
  657. if(blockSize == 0) {
  658. // 对方没有这个块，从区间表中删除，并将此块交给其他连接
  659. remoteInterval.DelInterval(blockID, 1);
  660. }
  661. // 检查PushList状况，如果只剩0块了，就要求重新分配所有连接的块
  662. if(remotePush.GetValidSize() == 0)
  663. comm->p2pMgr.RedistributeAllBlocks(this);
  664. }
  666. return ret;
  667. }
  669. MSG_STATE P2PClient::OnMsg() {
  670. //assert(GetIsCachePeer());
  671. // 错误代码
  672. UINT16 errCode;
  673. CSClient::CopyMoveSrc(&errCode, recvPointer, sizeof(errCode));
  675. // 是否需要断开连接
  676. bool shouldDisconnect;
  677. CSClient::CopyMoveSrc(&shouldDisconnect, recvPointer, sizeof(shouldDisconnect));
  679. // 根据错误代码处理
  680. switch(errCode) {
  681. case ERR_PROTOCOL_FORMAT:
  682. sprintf(errStr, "协议错误");
  683. break;
  684. case ERR_AUTHORIZATION:
  685. sprintf(errStr, "验证错误");
  686. break;
  687. case ERR_INTERNAL:
  688. sprintf(errStr, "未知错误");
  689. break;
  690. case ERR_CONNECTION_FULL:
  691. sprintf(errStr, "对方连接已满");
  692. break;
  693. default:
  694. shouldDisconnect = true;
  695. }
  696. if(shouldDisconnect) {
  697. // 停止连接这个地址
  698. comm->p2pMgr.AddBadAddr(remotePeer);
  699. return MSG_REMOTE_ERR;
  700. }
  702. return MSG_COMPLETE;
  703. }
  705. MSG_STATE P2PClient::OnReqMedia() {
  706. UINT blockID = 0;
  707. CSClient::CopyMoveSrc(&blockID, recvPointer, sizeof(blockID));
  709. if(!SendMediaType(blockID))
  710. return MSG_SEND_ERR;
  711. return MSG_COMPLETE;
  712. }
  714. MSG_STATE P2PClient::OnMediaType() {
  715. assert(comm->currRes);
  716. if(!comm->currRes)
  717. return MSG_DIFF_CHNL;
  719. // 一些较早版本的客户端会读取gtv文件中的媒体类型，然后发送很大的mediainterval，这种消息不予接受
  720. if(remoteVersion < comm->cfgData.ADD_PROGNAME_VERSION) {
  721. return MSG_COMPLETE;
  722. }
  724. MediaInterval mInterval;
  725. // 区间
  726. CSClient::CopyMoveSrc(&mInterval.start, recvPointer, sizeof(mInterval.start));
  727. CSClient::CopyMoveSrc(&mInterval.size, recvPointer, sizeof(mInterval.size));
  729. UINT len = 0;
  730. CSClient::CopyMoveSrc(&len, recvPointer, sizeof(len));
  731. if(len > 1024) {
  732. assert(0);
  733. return MSG_ERR_LIST_SIZE;
  734. }
  736. // 首先读取视频编码格式
  737. CSClient::CopyMoveSrc(&mInterval.videoType, recvPointer, sizeof(mInterval.videoType));
  738. if(mInterval.videoType.cbFormat > 1024) {
  739. assert(0);
  740. return MSG_ERR_LIST_SIZE;
  741. }
  742. mInterval.videoData = new BYTE[mInterval.videoType.cbFormat];
  743. CSClient::CopyMoveSrc(mInterval.videoData, recvPointer, mInterval.videoType.cbFormat);
  745. // 然后读取音频编码格式
  746. CSClient::CopyMoveSrc(&mInterval.audioType, recvPointer, sizeof(mInterval.audioType));
  747. if(mInterval.audioType.cbFormat > 1024) {
  748. assert(0);
  749. return MSG_ERR_LIST_SIZE;
  750. }
  751. mInterval.audioData = new BYTE[mInterval.audioType.cbFormat];
  752. CSClient::CopyMoveSrc(mInterval.audioData, recvPointer, mInterval.audioType.cbFormat);
  754. assert(len == mInterval.videoType.cbFormat+sizeof(mInterval.videoType)+mInterval.audioType.cbFormat+sizeof(mInterval.audioType));
  756. // 超过ADD_PROGNAME_VERSION的版本将发送节目的名字
  757. if(remoteVersion > comm->cfgData.ADD_PROGNAME_VERSION) {
  758. CSClient::CopyMoveSrc(&mInterval.pnamesize, recvPointer, sizeof(mInterval.pnamesize));
  759. /// 直播频道没有轮播节目名
  760. if(mInterval.pnamesize && mInterval.pnamesize + recvPointer-recvBuf <= msgSize) {
  761. mInterval.pname = new char[mInterval.pnamesize+1];
  762. CSClient::CopyMoveSrc(mInterval.pname, recvPointer, mInterval.pnamesize);
  763. mInterval.pname[mInterval.pnamesize] = 0;
  764. }
  765. else
  766. mInterval.pnamesize = 0; // 既然不能复制, 一定要置空！
  768. // 超过ADD_PROGNAME_VERSION的版本将发送节目的时间长度和频道的名字
  769. if(remoteVersion > comm->cfgData.ADD_PROGTIME_VERSION) {
  770. CSClient::CopyMoveSrc(&mInterval.progtime, recvPointer, sizeof(mInterval.progtime));
  771. CSClient::CopyMoveSrc(&mInterval.cnamesize, recvPointer, sizeof(mInterval.cnamesize));
  772. if(mInterval.cnamesize && mInterval.cnamesize + recvPointer-recvBuf <= msgSize) {
  773. mInterval.cname = new char[mInterval.cnamesize+1];
  774. CSClient::CopyMoveSrc(mInterval.cname, recvPointer, mInterval.cnamesize);
  775. mInterval.cname[mInterval.cnamesize] = 0;
  776. }
  777. else
  778. mInterval.cnamesize = 0; // 既然不能复制, 一定要置空！
  779. }
  780. }
  782. // add to interval array
  783. comm->currRes->AddMediaInterval(mInterval);
  785. return MSG_COMPLETE;
  786. }
  788. BOOL P2PClient::SendHello() {
  789. // 不可能没有currRes
  790. if(!comm->currRes)
  791. return FALSE;
  793. TCPPacket* packet;
  794. if(!SendBegin(packet, P2P_HELLO))
  795. return FALSE;
  797. // NP version
  798. CSClient::CopyMoveDes(sendPointer, &comm->cfgData.COMMUNICATOR_VERSION, sizeof(float));
  800. // 当前资源的Hash码
  801. CSClient::CopyMoveDes(sendPointer, comm->currRes->GetHashCode().data(), MD5_LEN);
  803. // 是否被动连接
  804. CSClient::CopyMoveDes(sendPointer, &isPassive, sizeof(isPassive));
  806. // 本机信息
  807. PeerInfoWithAddr thisPeer;
  808. memcpy(static_cast<P2PAddress*>(&thisPeer), &comm->localAddress, sizeof(comm->localAddress));
  809. comm->p2pMgr.GetSelfInfo(thisPeer);
  810. CSClient::CopyMoveDes(sendPointer, &thisPeer, sizeof(thisPeer));
  812. // 如果对方是CachePeer，还要发送SP的地址列表
  813. if(GetIsCachePeer()) {
  814. NormalAddress* spIPList = comm->currRes->GetSPList();
  815. if(spIPList) {
  816. UINT8 spIPListSize = comm->currRes->GetSPListSize();
  817. CSClient::CopyMoveDes(sendPointer, &spIPListSize, sizeof(spIPListSize));
  818. for(UINT8 i = 0; i < spIPListSize; ++i)
  819. CSClient::CopyMoveDes(sendPointer, &spIPList[i], sizeof(spIPList[i]));
  820. }
  821. }
  823. comm->logFile.StatusOut("Sending P2P_HELLO to %s.", comm->p2pMgr.FormatIPAddress(remotePeer));
  825. SendEnd(packet);
  827. return TRUE;
  828. }
  830. BOOL P2PClient::SendReport(const CorePeerInfo& peer, bool bRefresh) {
  831. TCPPacket* packet;
  832. if(!SendBegin(packet, P2P_REPORT))
  833. return FALSE;
  835. // PeerInfoWithAddr of local peer
  836. CSClient::CopyMoveDes(sendPointer, &peer, sizeof(peer));
  838. // 如果没有当前资源，则发送bRefresh=true和intervalNum=0
  839. if(!comm->currRes)
  840. bRefresh = true;
  842. // 是否更新全部区间列表
  843. CSClient::CopyMoveDes(sendPointer, &bRefresh, sizeof(bRefresh));
  845. // send block intervals
  846. // TODO: 这里存在重复代码，有待改进
  847. UINT8 intervalNum = comm->currRes?0xff:0;
  848. if(bRefresh) {
  849. if(comm->currRes) {
  850. // 如果是更新全部区间，则取得当前资源全部区间的列表，并发送
  851. comm->currRes->GetAllIntervals((BlockInterval*)(sendPointer+sizeof(intervalNum)), intervalNum);
  852. }
  853. CSClient::CopyMoveDes(sendPointer, &intervalNum, sizeof(intervalNum));
  854. sendPointer += intervalNum*sizeof(BlockInterval);
  855. }
  856. else {
  857. // 如果是更新变化的区间，则先后写入增加的区间和减少的区间
  858. if(comm->currRes)
  859. comm->currRes->GetDiffIntervals((BlockInterval*)(sendPointer+sizeof(intervalNum)), intervalNum, false, true);
  860. CSClient::CopyMoveDes(sendPointer, &intervalNum, sizeof(intervalNum));
  861. sendPointer += intervalNum*sizeof(BlockInterval);
  863. if(comm->currRes)
  864. comm->currRes->GetDiffIntervals((BlockInterval*)(sendPointer+sizeof(intervalNum)), intervalNum, false, false);
  865. CSClient::CopyMoveDes(sendPointer, &intervalNum, sizeof(intervalNum));
  866. sendPointer += intervalNum*sizeof(BlockInterval);
  867. }
  869. SendEnd(packet);
  871. //comm->logFile.StatusOut("Sending Report To %s", comm->p2pMgr.FormatIPAddress(remotePeer));
  872. return TRUE;
  873. }
  875. BOOL P2PClient::SendSPUpdate(const SPUpdate& spUpdate, UINT8 selfLayer, BYTE sum) {
  876. // 如果对方比自己离CP更近，那就不用发了
  877. if(selfLayer > remotePeer.layer)
  878. return TRUE;
  879. if(!isSameRes)
  880. return TRUE;
  882. TCPPacket* packet;
  883. if(!SendBegin(packet, P2P_SPUPDATE))
  884. return FALSE;
  886. // SPUpdate
  887. CSClient::CopyMoveDes(sendPointer, &spUpdate, sizeof(spUpdate));
  889. // SPUpdate sum
  890. CSClient::CopyMoveDes(sendPointer, &sum, sizeof(sum));
  892. SendEnd(packet);
  893. return TRUE;
  894. }
  896. BOOL P2PClient::SendNearPeers() {
  897. // 对方资源与本机相同时，才能发送有用的PeerInfo给它
  898. if(!isSameRes)
  899. return TRUE;
  901. TCPPacket* packet;
  902. if(!SendBegin(packet, P2P_NEAR_PEERS))
  903. return FALSE;
  905. list<PeerInfoWithAddr> peerList;
  906. comm->p2pMgr.GetPeersForNeighbour(peerList, 20, remotePeer);
  907. // 如果PeerList中元素个数大于0xff，则删除更多的元素
  908. UINT8 listSize = min(0xff, peerList.size());
  909. peerList.resize(listSize);
  911. if(listSize == 0) {
  912. // 释放这个Packet
  913. comm->p2pMgr.ReleasePacket(packet);
  914. return TRUE;
  915. }
  917. // 列表大小
  918. CSClient::CopyMoveDes(sendPointer, &listSize, sizeof(listSize));
  920. // write peerInfo one by one
  921. for(PeerAddrIt it = peerList.begin(); it != peerList.end(); ++it) {
  922. CSClient::CopyMoveDes(sendPointer, &*it, sizeof(PeerInfoWithAddr));
  923. }
  925. comm->logFile.StatusOut("Sending P2P_NEAR_PEERS to %s.", comm->p2pMgr.FormatIPAddress(remotePeer));
  927. SendEnd(packet);
  928. return TRUE;
  929. }
  931. BOOL P2PClient::SendPushList(UINT blockID, bool bAdd) {
  932. // 是否更新全部
  933. bool bRefresh = (blockID == UINT_MAX);
  934. // 如果大小为零就不必更新
  935. if(bRefresh && remotePush.GetValidSize() == 0)
  936. return TRUE;
  938. TCPPacket* packet;
  939. if(!SendBegin(packet, P2P_PUSHLIST))
  940. return FALSE;
  942. CSClient::CopyMoveDes(sendPointer, &bRefresh, sizeof(bRefresh));
  944. if(bRefresh) {
  945. // 更新整个PushList
  946. UINT8 count = 0xff;
  947. remotePush.CopyPushList(reinterpret_cast<UINT*>(sendPointer+sizeof(count)), count);
  948. CSClient::CopyMoveDes(sendPointer, &count, sizeof(count));
  949. sendPointer += count*sizeof(UINT);
  951. comm->logFile.StatusOut("Sending P2P_PUSHLIST %s block(%d) ... to %s.", 
  952. bRefresh?"Refresh":(bAdd?"Add":"Del"), count, comm->p2pMgr.FormatIPAddress(remotePeer));
  953. }
  954. else {
  955. UINT8 count = bAdd?1:0;
  956. CSClient::CopyMoveDes(sendPointer, &count, sizeof(count));
  957. if(bAdd)
  958. CSClient::CopyMoveDes(sendPointer, &blockID, sizeof(blockID));
  959. count = bAdd?0:1;
  960. CSClient::CopyMoveDes(sendPointer, &count, sizeof(count));
  961. if(!bAdd)
  962. CSClient::CopyMoveDes(sendPointer, &blockID, sizeof(blockID));
  964. comm->logFile.StatusOut("Sending P2P_PUSHLIST %s block %d to %s.", 
  965. bAdd?"Add":"Del",blockID, comm->p2pMgr.FormatIPAddress(remotePeer));
  966. }
  968. SendEnd(packet);
  969. return TRUE;
  970. } 
  972. BOOL P2PClient::SendResponse(UINT blockID, bool tryGetData) {
  973. // 此块所在的区间，其媒体类型是否已经发送过了
  974. if(!sentMediaArray.FindBlock(blockID)) {
  975. // 没有发送过，则发送此块所在区间的媒体类型
  976. if(!SendMediaType(blockID))
  977. return FALSE;
  978. }
  980. TCPPacket* packet;
  981. if(!SendBegin(packet, P2P_RESPONSE))
  982. return FALSE;
  984. // 请求的BlockID
  985. CSClient::CopyMoveDes(sendPointer, &blockID, sizeof(blockID));
  987. // 检查是否存在这个Block，取出他
  988. UINT blockSize = 0;
  989. // 如果需要获取数据，则尝试获取数据；如果只是为了告诉对方此块不存在，则不作尝试；
  990. if(comm->currRes && tryGetData) {
  991. P2P_RETURN_TYPE ret = comm->currRes->GetBlock(blockID, blockSize, sendPointer+sizeof(blockSize));
  992. if(ret < PRT_OK) {
  993. // 一定要释放这个packet
  994. comm->p2pMgr.ReleasePacket(packet);
  995. return ret;
  996. }
  997. }
  999. // write block size (bSize==0 means no data)
  1000. CSClient::CopyMoveDes(sendPointer, &blockSize, sizeof(blockSize));
  1002. // 将指针移动到数据的末尾
  1003. sendPointer += blockSize;
  1005. comm->logFile.StatusOut("Sending P2P_RESPONSE block(%d)(%d) to %s.", 
  1006. blockID, blockSize, comm->p2pMgr.FormatIPAddress(remotePeer));
  1008. if(blockSize > 0) {
  1009. // 记录此次发送数据的时间
  1010. lastSendDataTime = timeGetTime();
  1011. }
  1013. SendEnd(packet);
  1015. if(tryGetData && blockSize == 0) {
  1016. // 如果想要发送这个块，结果发现本机没有，则尝试发送下一个块
  1017. return SendFirstPushBlock();
  1018. }
  1019. return TRUE;
  1020. }
  1022. BOOL P2PClient::SendReqMedia(UINT blockID) {
  1023. TCPPacket* packet;
  1024. if(!SendBegin(packet, P2P_REQMEDIA))
  1025. return FALSE;
  1027. CSClient::CopyMoveDes(sendPointer, &blockID, sizeof(blockID));
  1029. SendEnd(packet);
  1030. return TRUE;
  1031. }
  1033. BOOL P2PClient::SendMediaType(UINT blockID) {
  1034. if(!comm->currRes)
  1035. return FALSE;
  1036. // 查找媒体类型
  1037. MediaInterval mInterval;
  1038. if(!comm->currRes->GetMediaInterval(blockID, mInterval))
  1039. return TRUE;
  1041. TCPPacket* packet;
  1042. if(!SendBegin(packet, P2P_MEDIATYPE))
  1043. return FALSE;
  1045. // 写入区间
  1046. CSClient::CopyMoveDes(sendPointer, &mInterval.start, sizeof(mInterval.start));
  1047. CSClient::CopyMoveDes(sendPointer, &mInterval.size, sizeof(mInterval.size));
  1049. // 写入媒体数据长度
  1050. UINT len = sizeof(mInterval.videoType)+mInterval.videoType.cbFormat+sizeof(mInterval.audioType)+mInterval.audioType.cbFormat;
  1051. CSClient::CopyMoveDes(sendPointer, &len, sizeof(len));
  1053. // 写入媒体数据
  1054. CSClient::CopyMoveDes(sendPointer, &mInterval.videoType, sizeof(mInterval.videoType));
  1055. CSClient::CopyMoveDes(sendPointer, mInterval.videoData, mInterval.videoType.cbFormat);
  1056. CSClient::CopyMoveDes(sendPointer, &mInterval.audioType, sizeof(mInterval.audioType));
  1057. CSClient::CopyMoveDes(sendPointer, mInterval.audioData, mInterval.audioType.cbFormat);
  1059. // 超过ADD_PROGNAME_VERSION的版本将发送节目的名字
  1060. if(remoteVersion > comm->cfgData.ADD_PROGNAME_VERSION) {
  1061. // 写入节目的名字
  1062. CSClient::CopyMoveDes(sendPointer, &mInterval.pnamesize, sizeof(mInterval.pnamesize));
  1063. CSClient::CopyMoveDes(sendPointer, mInterval.pname, mInterval.pnamesize);
  1064. // 超过ADD_PROGTIME_VERSION的版本将发送节目的时间和频道名
  1065. if(remoteVersion > comm->cfgData.ADD_PROGTIME_VERSION) {
  1066. // 写入节目的时间长度
  1067. CSClient::CopyMoveDes(sendPointer, &mInterval.progtime, sizeof(mInterval.progtime));
  1068. // 写入频道的名字
  1069. CSClient::CopyMoveDes(sendPointer, &mInterval.cnamesize, sizeof(mInterval.cnamesize));
  1070. CSClient::CopyMoveDes(sendPointer, mInterval.cname, mInterval.cnamesize);
  1071. }
  1072. }
  1074. SendEnd(packet);
  1076. sentMediaArray.AddInterval(mInterval);
  1077. return TRUE;
  1078. }
  1080. BOOL P2PClient::SendBegin(TCPPacket*& packet, UINT8 msgType) {
  1081. packet= comm->p2pMgr.AllocatePacket();
  1082. if(!packet)
  1083. return FALSE;
  1085. // 先留着消息大小不写，到最后再写
  1086. sendPointer = packet->buf+sizeof(UINT);
  1087. CSClient::CopyMoveDes(sendPointer, &msgType, sizeof(msgType));
  1089. return TRUE;
  1090. }
  1092. void P2PClient::SendEnd(TCPPacket*& packet) {
  1093. // 消息的大小就是移动的指针减去初始的指针
  1094. packet->size = sendPointer-packet->buf;
  1095. packet->sent = 0;
  1096. memcpy(packet->buf, &packet->size, sizeof(packet->size));
  1098. m_sendList.push_back(packet);
  1099. }
  1101. // 添加一个要Push的块
  1102. bool P2PClient::AddPushBlock(UINT blockID, bool bRefreshing) {
  1103. if(!remotePush.AddBlock(blockID, timeGetTime()))
  1104. return false;
  1105. if(remotePush.GetValidSize() == 1) {
  1106. // pushlist大小从0到1，说明新的下载即将开始
  1107. lastRecvDataHdTime = timeGetTime();
  1109. // pushlist大小从0到1，记录此时的时间，用于统计实际传输速度
  1110. m_reqStartTime = timeGetTime();
  1111. //comm->logFile.StatusOut("Request Start Time %d", m_reqStartTime);
  1112. }
  1113. if(!bRefreshing) {
  1114. // 如果不是正在更新整个PushList，就立即发送消息
  1115. if(!SendPushList(blockID, true)) {
  1116. if(!remotePush.DelBlock(blockID))
  1117. assert(0);
  1118. return false;
  1119. }
  1120. }
  1121. return true;
  1122. }
  1124. // 查找一个PushList中的块
  1125. bool P2PClient::FindPushBlock(const UINT blockID) const {
  1126. return remotePush.FindBlock(blockID);
  1127. }
  1129. // 删除一个要Push的块
  1130. bool P2PClient::DelPushBlock(UINT blockID) {
  1131. // 从PushList中查找并删除此块
  1132. if(!remotePush.DelBlock(blockID))
  1133. return false;
  1134. // 发送删除一个块的消息
  1135. return SendPushList(blockID, false) == TRUE;
  1136. }
  1138. // Push List是否已满
  1139. bool P2PClient::IsPushListFull() {
  1140. return (remotePush.GetValidSize() == remotePush.GetTotalSize());
  1141. }
  1143. // 取得Pushlist大小
  1144. UINT8 P2PClient::GetTotalPushSize() {
  1145. return remotePush.GetTotalSize();
  1146. }
  1148. // 清空PushList
  1149. void P2PClient::ClearPush() {
  1150. remotePush.Clear();
  1151. }
  1153. // 发送整个PushList，因为之前重新分配了所有连接的块
  1154. void P2PClient::SendPush() {
  1155. if(!SendPushList(UINT_MAX, false))
  1156. assert(0);
  1157. }
  1159. // 重新加载PushList，只有在普通的下载时才会被调用
  1160. void P2PClient::ReloadPushList() {
  1161. if(remotePush.GetValidSize() > 0)
  1162. return;
  1164. UINT array[P2P_HIGH_PUSH];
  1165. std::fill(array, array+P2P_HIGH_PUSH, UINT_MAX);
  1166. UINT count = 0;
  1168. // 计算此连接一次push的大小
  1169. UINT pushNum = P2P_MID_PUSH; // default
  1170. // 注意两者单位不同,前者是BPS, 后者是KBPS
  1171. if(avgDownSpeed/1024/comm->currRes->GetBitRate() < 0.1667f)
  1172. pushNum = P2P_LOW_PUSH;
  1173. else if(avgDownSpeed/1024/comm->currRes->GetBitRate() > 0.5f)
  1174. pushNum = P2P_HIGH_PUSH;
  1175. //else 
  1176. // the default value
  1178. // 1. 获取可以下载的区间列表
  1179. IntervalArray downloadable;
  1180. comm->currRes->GetDownloadableArray(remoteInterval, downloadable);
  1181. if(downloadable.IsEmpty()) {
  1182. return;
  1183. }
  1185. // 2. 遍历此区间列表，根据条件排除不可以下载的块，直到遍历完毕或者找到pushNum个Block
  1186. BlockInterval temp;
  1187. while(!downloadable.IsEmpty() && count < pushNum) {
  1188. downloadable.PopFront(temp);
  1189. if(temp.size == 0)
  1190. break;
  1191. for(UINT i = temp.start; i < temp.start+temp.size && count < pushNum; ++i) {
  1192. // 如果此块有别人正在下载，则去除此块
  1193. if(comm->p2pMgr.FindInAllPushList(i, this))
  1194. continue;
  1195. // 如果有其他NP有此块，那么就不要从CP上下载了, 前提是缓冲进度100%而且有此块的NP下载速度大于5.0KBPS
  1196. if(GetIsCachePeer() && 
  1197. comm->currRes->GetBufferPercent() == 100 && 
  1198. comm->p2pMgr.FindInAllRemoteInterval(i, this, false, 5.0f) > 0)
  1199. continue;
  1200. // 如果是外网节点，且内网节点上有，则不从这下载
  1201. if (GetIsSameLan() &&
  1202. comm->p2pMgr.FindInAllLanRemoteInterval(i, this, 5.0f) > 0)
  1203. continue;
  1205. // 记录一个可以下载的块
  1206. array[count++] = i;
  1207. }
  1208. }
  1210. if(count) {
  1211. // 根据缓冲情况决定是否要随机下载
  1212. if(comm->currRes->GetBufferPercent() == 100) {
  1213. random_shuffle(array, array+count);
  1214. comm->logFile.StatusOut("Do by random %d.", pushNum);
  1215. }
  1216. else
  1217. comm->logFile.StatusOut("Do by order %d.", pushNum);
  1219. // 添加到pushlist中，并发送
  1220. UINT addNum = 0;
  1221. for(UINT i = 0; i < count && addNum < pushNum; ++i) {
  1222. if(array[i] != UINT_MAX) {
  1223. if(!AddPushBlock(array[i], true))
  1224. break;
  1225. addNum++;
  1226. }
  1227. }
  1228. if(addNum)
  1229. SendPush();
  1230. }
  1231. }
  1233. // 发送push列表中的第一个块
  1234. BOOL P2PClient::SendFirstPushBlock() {
  1235. // 是否有需要发送的块，如果有，取出第一个（也是最小的一个）
  1236. UINT blockID;
  1237. localPush.GetFirstBlock(blockID);
  1238. if(blockID == UINT_MAX)
  1239. return TRUE;
  1241. // 检查当前发送列表中是否已经有本连接的块，
  1242. // 1. 如果没有，那么发送；
  1243. // 2. 如果有，则暂时不发送
  1244. if(!CheckSendingBlock())
  1245. return SendResponse(blockID, true);
  1247. localPush.AddBlock(blockID);
  1248. return TRUE;
  1249. }
  1252. BOOL P2PClient::IsIdleTooLong() {
  1253. DWORD tNow = timeGetTime();
  1255. // 对于Listener, 是上次发送数据的时间；对于Connector，是上次收到数据的时间
  1256. //int transferIdle = isIncoming?tNow-lastSendDataTime:tNow-lastRecvDataTime;
  1257. int transferIdle = min(tNow-lastSendDataTime, tNow-lastRecvDataTime);
  1258. // 明明push list非空，但是却没有收到数据头部的时间
  1259. int responseIdle = tNow-lastRecvDataHdTime;
  1261. GenerateTransferInfo(TRUE);
  1262. TransferInfo tiTemp;
  1263. GetTransferInfo(tiTemp);
  1264. comm->logFile.StatusOut("%s: transfer idle %d. response idle %d. rn
  1265. tcurr down/up:%.2f/%.2f, avg down/up:%.2f/%.2f, total down/up:%.2f/%.2fMB. layer: %d. bandwidth: %.4fKBPS %s.", 
  1266. comm->p2pMgr.FormatIPAddress(remotePeer), 
  1267. transferIdle, responseIdle, 
  1268. tiTemp.currDownSpeed/1024, tiTemp.currUpSpeed/1024, 
  1269. tiTemp.avgDownSpeed/1024, tiTemp.avgUpSpeed/1024, 
  1270. tiTemp.totalDownBytes/1024.f/1024.f, tiTemp.totalUpBytes/1024.f/1024.f, 
  1271. GetLayer(), GetBandWidth(), isIncoming?"in":"out");
  1273. BOOL kill = FALSE;
  1274. if(bGotFirstMsg) {
  1275. if(isIncoming)
  1276. kill = (transferIdle > MAX_INCOMING_SENDDATA_IDLE);
  1277. else
  1278. kill = (transferIdle > MAX_OUTGOING_RECVDATA_IDLE);
  1280. // TOOD: 如果pushlist非空，但是有一段时间没有数据的头部了，那么就杀掉此连接
  1281. //       注意此timeout时间比transfer timeout要短一些
  1282. kill = kill || (remotePush.GetValidSize() != 0 && responseIdle > MAX_RESPONSE_IDLE);
  1283. }
  1284. else {
  1285. int startIdle = transferIdle;
  1286. if(GetIsCachePeer()) {
  1287. kill = (startIdle > MAX_CP_FIRST_MSG_IDLE);
  1288. }
  1289. else
  1290. kill = (startIdle > MAX_NP_FIRST_MSG_IDLE);
  1291. }
  1293. if(kill) {
  1294. comm->logFile.StatusOut("Kill Idle %s.", comm->p2pMgr.FormatIPAddress(remotePeer));
  1295. }
  1296. return kill;
  1297. }
  1299. // 检查当前发送列表中是否已经有某连接的块，
  1300. // 1. 如果有，返回TRUE
  1301. // 2. 如果没有，那么返回FALSE；
  1302. BOOL P2PClient::CheckSendingBlock() {
  1303. TCPPacket* packet = NULL;
  1304. UINT temp;
  1305. for(TCPPackIt it = m_sendList.begin(); it != m_sendList.end(); ++it) {
  1306. packet = *it;
  1308. // 如果消息类型不是P2P_RESPONSE，接着找下一个
  1309. if(packet->buf[4] != P2P_RESPONSE)
  1310. continue;
  1312. // 如果blockSize为0，接着找下一个
  1313. memcpy(&temp, packet->GetBlockSize(), sizeof(temp));
  1314. if(temp == 0)
  1315. continue;
  1317. // 已经有准备发送的块
  1318. return TRUE;
  1319. }
  1320. // 没有找到,说明可以发送blockID
  1321. return FALSE;
  1322. }
  1324. // 从发送列表中查找一个块，如果此块准备发送（packet->send == 0），则删除并返回TRUE；
  1325. // 如果此块已经开始发送，则直接返回TRUE。如未找到，则返回FALSE
  1326. // 如果blockID是UINT_MAX，则删除任意一个准备发送的块
  1327. BOOL P2PClient::DeleteSendingBlock(UINT blockID) {
  1328. TCPPacket* packet = NULL;
  1329. UINT temp;
  1330. UINT index = 0;
  1331. for(TCPPackIt it = m_sendList.begin(); it != m_sendList.end(); ++it, index++) {
  1332. packet = *it;
  1334. // 如果消息类型不是P2P_RESPONSE，接着找下一个
  1335. if(packet->GetMsgType() != P2P_RESPONSE)
  1336. continue;
  1338. // 如果不是目标块，则接着找下一个
  1339. // 如果blockID是UINT_MAX，则删除所有准备发送的块
  1340. memcpy(&temp, packet->buf+5, sizeof(temp));
  1341. if(temp != blockID && blockID != UINT_MAX)
  1342. continue;
  1344. // 如果已经开始发送，就不能打断，返回TRUE
  1345. if(packet->sent > 0)
  1346. return TRUE;
  1348. // 如果是尚未开始发送的块，则删除，返回TRUE
  1349. comm->p2pMgr.ReleasePacket(packet);
  1350. m_sendList.erase(it);
  1351. return TRUE;
  1352. }
  1353. // 没有找到
  1354. return FALSE;
  1355. }
  1357. }