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

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

P2P编程

开发平台：

Visual C++

CPnew.c：源码内容
  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.  
  20. #include "echo.h"
  22. #define TYPE_NP 0
  23. #define TYPE_CP 1
  24. #define TYPE_SCP 2
  25. #define TYPE_ECP 3
  26. #define TYPE_GCP 4
  28. #define MEDIATYPE_FIRST 0.149
  30. #define MAX_IDLE 90 /* destruct a connection to sp/np after 90s */
  31. //#define P2PS_PORT 50002
  32. #define TS4CP_PORT 22168
  34. #define REQUEST_AHEAD 15 /* Request how many blocks if a block doesn't exist in cache */
  36. int MAX_P2PS=500;
  37. int MAX_P2PC=512;
  38. int MAX_JOB_PER_SESSION=3;
  40. //char *CONFIG = "./acp.cfg";
  41. char *PREFIX="/data/cp/";
  42. char *SCP_CHANNEL;
  43. char *ECP_REGION;
  44. int isGCP;
  45. char *SERVERIP;
  46. char *BINDIP;
  47. char *PIDFile="/var/run/cpnew.pid";
  48. int MAX_BANDWIDTH;
  50. char *AUTH_MD5;
  51. int  AUTH_USERID;
  53. fd_set osocks;
  54. int BINDALL;
  55. #ifdef HAVE_TS
  56. int TSSOCK = 0;
  57. //#define RANDOM_PORT 3947
  58. #endif
  59. time_t CurrentTime;
  60. time_t startTime;
  61. int SnapShotInterval;
  62. int NearPeerInterval = 30;
  64. int isSet = 0; /* whether TS has returned WELCOME message */
  65. struct TSMessage UDPMsg;
  67. extern struct Channel *ChannelHash[MAX_CHANNEL];
  68. extern struct Channel *ChannelList;
  70. struct sockaddr_in TSADDR;
  71. socklen_t addrlen = sizeof (struct sockaddr_in);
  73. // calculate avg & cur speed
  74. long long totalDownBytes=0, totalUpBytes=0;
  75. long long tmpDownBytes=0, tmpUpBytes=0;
  77. struct ServerDesc TRACKER[MAX_TYPE];
  79. char *LOGXML;
  81. extern int errno;
  83. int JobHighWater = 10000;
  84. int MaxNPPerChannel = 300;
  85. int cfgP2PS_PORT = 23;
  86. int cfgCP2TS_PORT = CP2TS_PORT;
  88. struct NamVal ConfigParameters[]
  89. = 
  90. {
  91. {"Prefix", &PREFIX, 's'},
  92. {"MAX_NP", &MAX_P2PS, 'd'},
  93. {"MAX_SP", &MAX_P2PC, 'd'},
  94. {"Pidfile", &PIDFile, 's'},
  95. {"TrackerIP", &SERVERIP, 's'},
  96. {"SCP", &SCP_CHANNEL, 's'},
  97. {"GCP", &isGCP, 'd'},
  98. {"authid", &AUTH_USERID, 'd'},
  99. {"authmd5", &AUTH_MD5, 's'},
  100. {"ECP", &ECP_REGION, 's'},
  101. {"BandWidth", &MAX_BANDWIDTH, 'd'},
  102. {"BindIP", &BINDIP, 's'},
  103. {"SnapShotInterval",&SnapShotInterval,'d'},
  104. {"CP4NP_PORT", &cfgP2PS_PORT, 'd'},
  105. {"CP2TS_PORT", &cfgCP2TS_PORT, 'd'},
  106. {"LogFilePath", &LOGXML, 's'},
  107. {"NearPeerInterval", &NearPeerInterval, 'd'},
  108. {"JobHighWater", &JobHighWater, 'd'},
  109. {"MaxNPPerChannel", &MaxNPPerChannel, 'd'},
  110. {"BINDALL", &BINDALL, 'd'}
  111. };
  113. int register_cp ();
  114. int init_cp ();
  115. int handle_new_connection(int sock, int type);
  116. int Clientclosure (int listnum, int type);
  117. void process_child (void);
  118. int init_P2PS (int listnum);
  119. int process_P2PS (int listnum);
  120. int closure_P2PS (int listnum);
  121. int init_P2PC (int listnum);
  122. int process_P2PC (int listnum);
  123. int closure_P2PC (int listnum);
  124. int sendMessage (int sock, char *ptr, struct sockaddr_in *dest);
  125. int process_TS2CP_PEERS (char *buf);
  126. void process_type (int type, fd_set *socks, fd_set *wsocks, fd_set *esocks);
  127. int reconnect (struct Session *p);
  128. char *parseECP (char *str, char *buf);
  129. int closure_TS ();
  130. int periodCheck (float KBPSused);
  131. void makeSnapShot(int count, int time_interval);
  132. int send_nearpeers (struct Channel *pc, struct Edge *pme);
  133. int send_nearpeers_toall (struct Channel *pc);
  134. #ifdef HAVE_TS
  135. int process_TS();
  136. #endif
  137. int send_P2P_PUSHLIST (struct Channel *pc, int id);
  138. int period_process (void);
  139. extern char *getJobBuffer (struct JobDes *p, int *max);
  140. extern inline void setblockId (struct JobDes *pj, int id);
  142. #include "sessions.c"
  143. #define INIT_MAXQ(pc,s,maxq) do
  144. {
  145. if (s->maxBlockID == 0) return -1;
  146. if (s->maxBlockID >= MAX_QUEUE) maxq = MAX_QUEUE;
  147. else maxq = s->maxBlockID;
  148. if (pc->pcinfo->indisk == NULL)
  149. {
  150. pc->pcinfo->max_queue = maxq;
  151. pc->pcinfo->indisk = calloc (maxq, 1);
  152. pc->pcinfo->bitflag = calloc ((maxq+7)/8, 1);
  153. if (pc->pcinfo->indisk == NULL || pc->pcinfo->bitflag == NULL)
  154. return -1;
  155. } else if (maxq < pc->pcinfo->max_queue)
  156. {
  157. pc->pcinfo->max_queue = maxq;
  158. pc->pcinfo->indisk = realloc (pc->pcinfo->indisk, maxq);
  159. pc->pcinfo->bitflag = realloc (pc->pcinfo->bitflag, (maxq+7)/8);
  160. if (pc->pcinfo->indisk == NULL || pc->pcinfo->bitflag == NULL)
  161. return -1;
  162. }
  163. } while (0)
  165. int period_process (void)
  166. {
  167. static time_t last_snapshot;
  168. static int snapCount = 0;
  170. if (CurrentTime - last_snapshot > SnapShotInterval)
  171. {
  172. makeSnapShot(snapCount++, CurrentTime-last_snapshot);
  173. system("/usr/bin/vmstat -a >> cp.log 2>&1 &");
  174. last_snapshot = CurrentTime;
  175. }
  176. return 0;
  177. }
  179. int main(int argc, char **argv)
  180. {
  181. int i, mode = 1, tmp = 0;
  183. if (argc < 2)
  184. {
  185. printf("usage: %s mode(0 for daemon, 1 for console).n", argv[0]);
  186. return -1;
  187. }
  188. signal (SIGPIPE, SIG_IGN);
  189. signal (SIGINT, terminate);
  191. mode = atoi (argv[1]);
  193. if (mode == 0)
  194. daemon(1,1);
  196. read_config (CONFIG, ConfigParameters, sizeof(ConfigParameters)/sizeof(struct NamVal));
  198. for(i = 0 ; i < argc; ++i)
  199. {
  200. if(strncmp(argv[i], "tcp=", 4) == 0)
  201. {
  202. tmp = atoi(argv[i]+4);
  203. if(tmp > 0 && tmp < 65535)
  204. cfgP2PS_PORT = tmp;
  205. }
  206. else if(strncmp(argv[i], "udp=", 4) == 0)
  207. {
  208. tmp = atoi (argv[i]+4);
  209. if(tmp > 0 && tmp < 65535)
  210. cfgCP2TS_PORT = tmp;
  212. }
  213. }
  215. for (i=0; i<10 && IN_LOOP > 0; i++)
  216. {
  217. /*
  218. pid_t pid;
  219. if ((pid = fork ()) == 0)
  220. {
  221. */
  222. FD_ZERO(&osocks);
  223. if (init_cp () < 0)// || initLOG () < 0)
  224. {
  225. PDEBUG ("init_cp error, exit...n");
  226. exit (-1);
  227. }
  228. process_child ();
  229. /*
  230. } else if (pid < 0)
  231. {
  232. perror ("fork");
  233. exit (pid);
  234. } else
  235. {
  236. waitpid (pid, NULL, 0);
  237. }
  238. */
  239. }
  240. return 0;
  241. }
  243. int init_P2PS (int listnum)
  244. {
  245. return 0;
  246. }
  248. int process_P2P_HELLO (struct Session *p, struct Message *m)
  249. {
  250. struct Edge *pedge = NULL;
  251. float version = *(float *)(m->buffer);
  252. char *buf = m->buffer + sizeof (float);
  253. struct Channel *pc;
  254. struct LiveChannelInfo *pcinfo;
  255. int listnum;
  256. // char *buf, buffer[MAX_DATA];
  258. PINFO ("RECV P2P_HELLO. n");
  259. p->version = version;
  260. listnum = p - TRACKER[TYPE_P2PS].head;
  261. if ((pc = findChannel (buf, MD5_LEN)) == NULL)
  262. {
  263. if ((pc = newLiveChannel (buf, NULL, buf, 0, 0)) != (struct Channel *)0)
  264. {
  265. pedge=newEdge (pc, p);
  266. pc->numclient ++;
  267. } else
  268. {
  269. Clientclosure (listnum, TYPE_P2PS);
  270. return -1;
  271. }
  272. p->pc = pc;
  273. } else
  274. {
  275. if (pc->numclient > MaxNPPerChannel)
  276. {
  277. Clientclosure (listnum, TYPE_P2PS);
  278. return -1;
  279. }
  280. for (pedge=p->header; pedge && pedge->head != pc; pedge=pedge->enext);
  281. if (pedge == NULL)
  282. {
  283. pedge=newEdge (pc, p);
  284. pc->numclient ++;
  285. }
  286. p->pc = pc;
  287. if (pc->pcinfo->dataSource) 
  288. {
  289. /*
  290. // connect exist, send p2p_hello to check channel state.
  291. buf = buffer+sizeof (int);
  292. *(unsigned char *)buf = P2P_HELLO;
  293. buf += sizeof (char);
  294. memcpy (buf, pc->channel_md5, MD5_LEN);
  295. buf += MD5_LEN;
  296. *(unsigned char *)buf = 0;
  297. buf += sizeof (char);
  298. *(int *)buffer = buf - buffer;
  299. if (writeMessage (pc->pcinfo->dataSource, buffer) < 0)
  300. {
  301. Clientclosure (listnum, TYPE_P2PC);
  302. return -1;
  303. }
  304. PDEBUG("sent P2P_HELLO to SP. n");
  305. */
  306. buf += MD5_LEN + sizeof (char);
  307. memcpy (&(p->addr), buf, sizeof (p->addr));
  308. if (pedge) send_nearpeers (pc, pedge);
  309. return 0;
  310. }
  311. PDEBUG("NO connection to SP. try connect.n");
  312. }
  313. buf += MD5_LEN + sizeof (char);
  314. memcpy (&(p->addr), buf, sizeof (p->addr));
  315. buf += sizeof (p->addr);
  317. pcinfo = pc->pcinfo;
  318. pcinfo->numofsp = (unsigned char)*buf;
  319. buf += sizeof (char);
  320. if (pcinfo->numofsp == 0 || pcinfo->numofsp > MAX_REPSP)
  321. {
  322. Clientclosure (listnum, TYPE_P2PS);
  323. return -1;
  324. }
  325. memcpy (&(pcinfo->SPLIST), buf, pcinfo->numofsp*sizeof(struct NormalAddress));
  326. buf += pcinfo->numofsp*sizeof (struct NormalAddress);
  327. if (buf - m->buffer + NORMAL_HEADER > m->len)
  328. {
  329. PDEBUG ("Invalid message %d, length %d not enoughn", m->type, m->len);
  330. Clientclosure (listnum, TYPE_P2PS);
  331. return -1;
  332. }
  333. if (isGCP || SCP_CHANNEL)
  334. {
  335. if ((listnum = newChannel (pc, pcinfo->SPLIST, pcinfo->numofsp, 0)) < 0)
  336. {
  337. PDEBUG("Unable connect to SP.n");
  338. Clientclosure (p-TRACKER[TYPE_P2PS].head, TYPE_P2PS);
  339. return -1;
  340. }
  341. } else /* ECP, need peers */
  342. {
  343. #ifdef HAVE_TS
  344. UDPMsg.type = CP2TS_NEED_PEERS;
  345. UDPMsg.len = 12+MD5_LEN;
  346. memcpy (UDPMsg.buffer, pc->channel_md5, MD5_LEN);
  347. /* isSet whether TS has returned WELCOME message */
  348. if (isSet && sendMessage (TSSOCK, (char *)&UDPMsg, &TSADDR) < 0)
  349. {
  350. PDEBUG ("exit...n");
  351. exit (1);
  352. }
  353. #endif
  354. }
  355. if (pedge) send_nearpeers (pc, pedge);
  356. return 0;
  357. }
  359. int process_P2P_MEDIATYPE (int listnum, struct Message *m)
  360. {
  361. struct Channel *pc;
  362. char *buf, *media, *name, *channel_name;
  363. int start, length, size, proglen, chnllen;
  365. buf = m->buffer;
  366. if ((pc = findChannel (buf, MD5_LEN)) == NULL)
  367. return -1;
  368. buf += MD5_LEN;
  369. start = *(int *)buf;
  370. buf += sizeof (int);
  371. length = *(int *)buf;
  372. buf += sizeof (int);
  373. size = *(int *)buf;
  374. buf += sizeof (int);
  375. media = buf;
  376. buf += size;
  377. proglen = *(unsigned char *)buf;
  378. buf += sizeof (char);
  379. name = buf;
  380. buf += proglen;
  381. buf += sizeof (int);
  382. chnllen = *(unsigned char *)buf;
  383. buf += sizeof (char);
  384. channel_name = buf;
  385. buf += chnllen;
  386. if (buf - (char *)m > m->len)
  387. return -1;
  388. addMedia (pc, start, length, size, media, name, channel_name);
  389. return 0;
  390. }
  392. int process_P2P_PUSHLIST (struct Session *p, struct Message *m)
  393. {
  394. struct Channel *pc;
  395. struct Edge *pedge;
  396. char *buf;
  397. int i, type, listnum, size;
  399. listnum = p - TRACKER[TYPE_P2PS].head;
  400. buf = m->buffer;
  401. if (p->npcp == TYPE_CP)
  402. {
  403. if ((pc = findChannel (buf, MD5_LEN)) == NULL)
  404. {
  405. if ((pc = newLiveChannel (m->buffer, NULL, m->buffer, 0, 0)) != (struct Channel *)0)
  406. {
  407. pedge=newEdge (pc, p);
  408. pc->numclient ++;
  409. }
  410. } else
  411. {
  412. for (pedge=p->header; pedge && pedge->head != pc; pedge=pedge->enext);
  413. if (pedge == NULL)
  414. {
  415. pedge=newEdge (pc, p);
  416. pc->numclient ++;
  417. }
  418. }
  419. buf += MD5_LEN;
  420. } else
  421. {
  422. pc = p->pc;
  423. }
  425. if (pc == NULL && p->npcp != TYPE_CP)
  426. {
  427. Clientclosure (listnum, TYPE_P2PS);
  428. return -1;
  429. }
  430. if (p->numjob >= MAX_JOB_PER_SESSION)
  431. return -2;
  433. type = *(unsigned char *)buf;
  434. buf += sizeof (char);
  435. size = *(unsigned char *)buf;
  436. buf += sizeof (char);
  437. if (type)
  438. {
  439. deleteChannel (p, pc);
  440. for (i=0; i<size; i++)
  441. if (process_P2P_REQUEST_real (p, pc, ((unsigned int *)buf)[i]) < 0)
  442. return -1;
  443. } else
  444. {
  445. for (i=0; i<size; i++)
  446. if (process_P2P_REQUEST_real (p, pc, ((unsigned int *)buf)[i]) < 0)
  447. return -1;
  448. buf += size*sizeof(int);
  449. size = *(unsigned char *)buf;
  450. buf += sizeof (char);
  451. deleteJob (p, pc, (unsigned int *)buf, size);
  452. }
  453. if (buf - m->buffer + NORMAL_HEADER > m->len)
  454. {
  455. PDEBUG ("Invalid message %d, length %d not enoughn", m->type, m->len);
  456. Clientclosure (listnum, TYPE_P2PS);
  457. return -1;
  458. }
  460. return 0;
  461. }
  463. int process_P2P_REQUEST_real (struct Session *p, struct Channel *pc, int id)
  464. {
  465. struct JobDes *pj = newJob ();
  466. char *buf, *buffer;
  467. struct LiveChannelInfo *pcinfo;
  468. int size, max, i;
  470. buffer = getJobBuffer (pj, &max);
  471. buf = buffer + sizeof (int);
  472. *(unsigned char *)buf = P2P_RESPONSE;
  473. buf += sizeof (char);
  474. if (p->npcp == TYPE_CP)
  475. {
  476. memcpy (buf, pc->channel_md5, MD5_LEN);
  477. buf += MD5_LEN;
  478. }
  479. pcinfo = pc->pcinfo;
  480. if (p->first == 0)
  481. {
  482. p->first ++;
  483. if (p->version >= MEDIATYPE_FIRST)
  484. sendIdMedia (p, pc, id, 0);
  485. }
  486. if (pcinfo->dataSource == NULL)
  487. {
  488. *(int *)buf = id;
  489. buf += sizeof (int);
  490. *(int *)buf = 0;
  491. buf += sizeof (int);
  492. if (isGCP || SCP_CHANNEL)
  493. {
  494. if ((i = newChannel (pc, pcinfo->SPLIST, pcinfo->numofsp, 0)) < 0)
  495. freeLiveChannel (pc, NULL);
  496. } else
  497. {
  498. #ifdef HAVE_TS
  499. UDPMsg.type = CP2TS_NEED_PEERS;
  500. UDPMsg.len = 12+MD5_LEN;
  501. memcpy (UDPMsg.buffer, pc->channel_md5, MD5_LEN);
  502. /* isSet whether TS has returned WELCOME message */
  503. if (isSet && sendMessage (TSSOCK, (char *)&UDPMsg, &TSADDR) < 0)
  504. {
  505. PDEBUG ("exit...n");
  506. exit (1);
  507. }
  508. #endif
  509. }
  510. } else if ((size=locate_by_id (pc, id, buf, max-32)) > 0)
  511. {
  512. buf += 2*sizeof (int) + size;
  513. if (p->version >= MEDIATYPE_FIRST && (i=isHit (pc, id)) >= 0)
  514. sendHitMedia (p, pc, i, id, 0);
  515. p->last_transferblock = CurrentTime;
  516. } else if (size == -2)
  517. {
  518. assert (0);
  519. PDEBUG ("Leave block %d to next round.n", id);
  520. return -1;
  521. } else if (id >= 0)
  522. {
  523. *(int *)buf = id;
  524. buf += sizeof (int);
  525. *(int *)buf = 0;
  526. buf += sizeof (int);
  527. PINFO ("no block %dn", id);
  528. send_P2P_PUSHLIST (pc, id);
  529. }
  530. *(int *)buffer = buf - buffer;
  531. setblockId(pj, id);
  532. writeDATAMessage(p,pc, pj);
  533. // PDEBUG ("Write block %dn", id);
  534. return 0;
  535. }
  537. int process_P2PS (int listnum)
  538. {
  539. struct Session *p = &(TRACKER[TYPE_P2PS].head[listnum]);
  540. struct Message *m = (struct Message *)(p->buf+p->start);
  542. tmpDownBytes += m->len;
  544. switch (m->type)
  545. {
  546. case P2P_HELLO:
  547. if (process_P2P_HELLO (p, m) == -2)
  548. return -2;
  549. break;
  550. case P2P_PUSHLIST:
  551. if (process_P2P_PUSHLIST (p, m) == -2)
  552. return -2;
  553. break;
  554. case P2P_REPORT: /* At present no action */
  555. break;
  556. case P2P_MSG:
  557. break;
  558. case P2P_SPUPDATE:
  559. break;
  560. case P2P_RESPONSE:
  561. break;
  562. case P2P_NEAR_PEERS:
  563. break;
  564. case P2P_REQMEDIA:
  565. sendIdMedia (p, p->pc, *(int *)(m->buffer), 0);
  566. break;
  567. default:
  568. PDEBUG ("Unknown message format from clientn");
  569. Clientclosure (listnum, TYPE_P2PS);
  570. return -1;
  571. }
  572. return 0;
  573. }
  575. int closure_P2PS (int listnum)
  576. {
  577. struct Session *p = &(TRACKER[TYPE_P2PS].head[listnum]);
  578. // struct Channel *pc = p->pc;
  579. struct Edge *pedge, *prevedge;
  580. PDEBUG ("CP disconnected from %d.%d.%d.%d:%d.n", IPADDR (p->host), p->port);
  581. for (pedge=p->header; pedge; pedge=prevedge)
  582. {
  583. pedge->head->numclient --;
  584. prevedge=pedge->enext;
  585. delEdge (pedge);
  586. }
  587. FD_CLR(p->socket, &osocks);
  588. close (p->socket);
  589. FREE (p->buf);
  590. deleteAll (p);
  591. memset (p, 0, sizeof (struct Session));
  592. return 0;
  593. }
  596. int init_P2PC (int listnum)
  597. {
  598. return 0;
  599. }
  602. int newChannel (struct Channel *pc, struct NormalAddress *client, int n, int flag)
  603. {
  604. struct Session *p;
  605. int listnum, newconn = -1, max, sock_flag;
  606. struct Session *head;
  607. struct sockaddr_in addr;
  608. char *buf, buffer[MAX_DATA];
  610. if (pc == NULL) return -1;
  611. head = TRACKER[TYPE_P2PC].head;
  612. max = TRACKER[TYPE_P2PC].max;
  613. for (listnum = 0; listnum < max; listnum ++)
  614. {
  615. if(head[listnum].socket == 0)
  616. {
  617. if ((newconn = socket (PF_INET, SOCK_STREAM, 0)) < 0)
  618. {
  619. perror ("socket||gethostbyname");
  620. return -1;
  621. }
  622. memset (&addr, 0, sizeof (addr));
  623. addr.sin_port = client[0].sin_port;
  624. addr.sin_addr = client[0].sin_addr;
  625. addr.sin_family = AF_INET;
  626. if ((sock_flag = connect_nonb(newconn, &addr, sizeof (addr))) == -1)
  627. {
  628. close (newconn);
  629. return -1;
  630. }
  631. head[listnum].socket = newconn;
  632. head[listnum].type = TYPE_P2PC;
  633. head[listnum].flag = 1;
  634. head[listnum].sock_flag = sock_flag;
  635. head[listnum].buf = NEW();
  636. head[listnum].pc = pc;
  637. head[listnum].time_sec = CurrentTime;
  638. head[listnum].totalup = 0;
  639. head[listnum].last_transferblock = CurrentTime;
  640. FD_SET(newconn, &osocks);
  641. if (listnum > TRACKER[TYPE_P2PC].maxid)
  642. TRACKER[TYPE_P2PC].maxid = listnum;
  643. break;
  644. }
  645. }
  646. if (listnum >= max)
  647. {
  648. PDEBUG ("no space left for new incoming client.");
  649. close (newconn);
  650. return -1;
  651. }
  652. TRACKER[TYPE_P2PC].cur ++;
  653. (*(TRACKER[TYPE_P2PC].init)) (listnum);
  654. p = &(TRACKER[TYPE_P2PC].head[listnum]);
  655. pc->pcinfo->dataSource = p;
  656. pc->upsize = 0;
  657. pc->downsize = 0;
  659. PDEBUG("Connect to %s:%d. and send P2P_HELLO.n", inet_ntoa(addr.sin_addr), ntohs(addr.sin_port));
  662. buf = buffer+sizeof (int);
  663. *(unsigned char *)buf = P2P_HELLO;
  664. buf += sizeof (char);
  665. memcpy (buf, pc->channel_md5, MD5_LEN);
  666. buf += MD5_LEN;
  667. *(unsigned char *)buf = 0;
  668. buf += sizeof (char);
  669. if (flag == TYPE_CP)
  670. {
  671. *(unsigned char *)buf = pc->pcinfo->numofsp;
  672. buf += sizeof (char);
  673. if (pc->pcinfo->numofsp)
  674. {
  675. memcpy (buf, &(pc->pcinfo->SPLIST), pc->pcinfo->numofsp*sizeof (struct NormalAddress));
  676. buf += pc->pcinfo->numofsp*sizeof (struct NormalAddress);
  677. }
  678. }
  679. *(int *)buffer = buf - buffer;
  680. if (writeMessage (p, pc, buffer) < 0)
  681. {
  682. perror ("CP: write SP");
  683. Clientclosure (listnum, TYPE_P2PC);
  684. return -1;
  685. }
  686. return listnum;
  687. }
  690. int process_P2P_SPUPDATE (int listnum, struct Message *m)
  691. {
  692. int maxq;
  693. struct Edge *pedge;
  694. char buffer[MAX_DATA];
  695. struct Channel *pc;
  696. struct SPUpdate *s = (struct SPUpdate *)(m->buffer+MD5_LEN);
  697. int bShouldCloseSP = 0;
  699. if (m->len < sizeof (struct SPUpdate) + MD5_LEN + 5)
  700. {
  701. PDEBUG ("Invalid message %d, length %d not enoughn", m->type, m->len);
  702. Clientclosure (listnum, TYPE_P2PC);
  703. return -1;
  704. }
  705. *(int *)buffer = sizeof (struct SPUpdate) + sizeof(int) + sizeof (char);
  706. *(unsigned char *)(buffer + sizeof (int)) = P2P_SPUPDATE;
  707. PINFO ("Recv SPUPDATE(%lld,%lld,%u,%u).n", s->minKeySample, s->maxKeySample, s->minBlockID, s->maxBlockID);
  708. memcpy (buffer+sizeof(int)+sizeof(char), s, sizeof(struct SPUpdate));
  710. if ((pc=findChannel (m->buffer, MD5_LEN)) == NULL) return -1;
  712. if (s->minBlockID == 0xffffffff && s->maxBlockID == 0xffffffff)
  713. {
  714. pc->pcinfo->dataSource = NULL; /* indication of end */
  715. if(s->minKeySample == -1 && s->maxKeySample == -1)
  716. PDEBUG("NO SUCH RESOURCE!n"); // no such resource
  717. else if(s->minKeySample == 0 && s->maxKeySample == 0)
  718. PDEBUG("CHANNEL HAS BEEN CLOSED!n");// channel has been closed
  719. else
  720. PDEBUG("UNKNOWN MESSAGE! 1n"); // unknown message
  721. bShouldCloseSP = 1; // should close connection
  722. } else
  723. {
  724. if(s->minBlockID == 0 && s->maxBlockID == 0 && s->minKeySample == 0 && s->maxKeySample == 0)
  725. {
  726. PDEBUG("END OF CHANNEL!n"); // end of channel
  727. bShouldCloseSP = 1;
  728. } else if (s->minKeySample == -1ULL && s->maxKeySample == -1ULL)
  729. {
  730. INIT_MAXQ(pc,s,maxq);
  731. return 0;
  732. } else if (s->minKeySample == -2LL)
  733. {
  734. INIT_MAXQ(pc,s,maxq);
  735. pc->type = T_PLIST;
  736. pc->pcinfo->max_channel = (int)(s->maxKeySample);
  737. if (pc->pcinfo->max_channel <= 0 || pc->pcinfo->max_channel >= MAX_FILEINPUT)
  738. return -1;
  739. if (pc->pcinfo->media != NULL)
  740. freeMedia (pc);
  741. pc->pcinfo->media = calloc (sizeof (struct MediaData), pc->pcinfo->max_channel);
  742. return 0;
  743. } else if (s->minKeySample == -3LL)
  744. {
  745. pc->pcinfo->max_channel = 1;
  746. if (pc->pcinfo->media != NULL)
  747. freeMedia (pc);
  748. pc->pcinfo->media = calloc (sizeof (struct MediaData), 1);
  749. } else
  750. {
  751. memcpy (&(pc->pcinfo->s), s, sizeof (struct SPUpdate));
  752. // request block after spupdate, not wait!
  753. // now, block will be sent automaticlly by SP
  754. // send_P2P_PUSHLIST (pc, s->maxBlockID);
  755. }
  756. }
  757. {
  758. int i = 0;
  759. unsigned char vcode = 0;
  760. for (i = 0; i < sizeof (struct SPUpdate); ++i) {
  761. vcode += ((unsigned char*)s)[i];
  762. }
  763. buffer[*(int*)buffer] = vcode;
  764. ++*(int*)buffer;
  765. }
  766. for (pedge=pc->PeerHead; pedge; pedge=pedge->cnext)
  767. {
  768. if (pedge->me->npcp == TYPE_CP)
  769. {
  770. // PDEBUG ("Send SPUPDATE to CP %d.n", pedge->me-TRACKER[TYPE_P2PC].head);
  771. writeMessage (pedge->me, pc, (char *)m);
  772. } else
  773. {
  774. // PDEBUG ("Send SPUPDATE to NP %d.n", pedge->me-TRACKER[TYPE_P2PS].head);
  775. writeMessage (pedge->me, pc, buffer);
  776. }
  777. }
  778. if(bShouldCloseSP != 0 || pedge == pc->PeerHead/* no NP*/)
  779. return -1; // Close Connection to SP
  780. return 0;
  781. }
  783. int process_P2P_RESPONSE (int listnum, struct Message *m)
  784. {
  785. int size;
  786. struct Session *p = &(TRACKER[TYPE_P2PC].head[listnum]);
  787. struct Channel *pc;
  788. char *msg = m->buffer+MD5_LEN;
  791. if ((pc=findChannel (m->buffer, MD5_LEN)) == NULL)
  792. return -1;
  794.   if ((size = saveBlock (pc, msg, p)) <= 0)
  795. {
  796.     PDEBUG ("save block error, size %d, %dn", size, listnum);
  797. return -1;
  798. // Clientclosure (listnum, TYPE_P2PC);
  799. }
  800. p->last_transferblock = CurrentTime;
  801. return 0;
  802. }
  804. int process_P2PC (int listnum)
  805. {
  806. struct Session *p = &(TRACKER[TYPE_P2PC].head[listnum]);
  807. struct Message *m = (struct Message *)(p->buf+p->start);
  809. tmpDownBytes += m->len;
  811. switch (m->type)
  812. {
  813. case P2P_SPUPDATE:
  814. if(process_P2P_SPUPDATE (listnum, m) < 0)
  815. {
  816. Clientclosure(listnum, TYPE_P2PC);
  817. }
  818. break;
  819. case P2P_RESPONSE:
  820. process_P2P_RESPONSE (listnum, m);
  821. break;
  822. case P2P_MSG:
  823. break;
  824. case P2P_MEDIATYPE:
  825. process_P2P_MEDIATYPE (listnum, m);
  826. break;
  827. default:
  828. PDEBUG("Err msg type from SP.n");
  829. Clientclosure (listnum, TYPE_P2PC);
  830. return -1;
  831. }
  832. return 0;
  833. }
  835. int closure_P2PC (int listnum)
  836. {
  837. // struct Edge *pedge, *prevedge;
  838. // char buffer[MAX_DATA], *buf;
  839. struct Session *p=&(TRACKER[TYPE_P2PC].head[listnum]);
  840. struct Channel *pc = p->pc;
  842. if (pc)
  843. {
  844. /*
  845. buf = buffer + sizeof (int);
  846. *(unsigned char *)buf = P2P_SPUPDATE;
  847. buf += sizeof (char);
  848. memcpy (buf, pc->channel_md5, MD5_LEN);
  849. buf += MD5_LEN;
  850. memset (buf, 0, sizeof (struct SPUpdate));
  851. buf += sizeof (struct SPUpdate);
  852. *(int *)buffer = buf - buffer;
  853. for (pedge=pc->PeerHead; pedge; pedge=prevedge)
  854. {
  855. pc->numclient --;
  856. writeMessage (pedge->me, buffer);
  857. prevedge = pedge->cnext;
  858. delEdge (pedge);
  859. }
  860. */
  861. pc->pcinfo->dataSource = NULL;
  862. }
  863. PDEBUG ("SP disconnected from %d.%d.%d.%d:%d.n", IPADDR (p->host), p->port);
  864. FD_CLR(p->socket, &osocks);
  865. close (p->socket);
  866. FREE (p->buf);
  867. deleteAll (p);
  868. memset (p, 0, sizeof (struct Session));
  869. return 0;
  870. }
  872. #ifdef HAVE_TS
  873. int register_cp () //send UDP msg
  874. {
  875. const int max_times = 10;
  876. int i;
  877. char *buf;
  878. char buffer[MAX_DATA];
  880. isSet = 0;
  882. if (TSSOCK == 0)
  883. {
  884. for (i=0; i<max_times; i++)
  885. {
  886. if (BINDALL == 0)
  887. TSSOCK = init_udp (BINDIP, cfgCP2TS_PORT);
  888. else
  889. TSSOCK = init_udp (NULL, cfgCP2TS_PORT);
  890. if (TSSOCK > 0) break;
  891. PDEBUG("Sleep 1000. cause init UDP port %d failed.", cfgCP2TS_PORT);
  892. sleep (1000);
  893. }
  894. if (TSSOCK <= 0)
  895. {
  896. PDEBUG ("exit...n");
  897. exit (1); //the max times try init_udp failure
  898. }
  899. }
  900. buf = buffer + sizeof (int);
  901. *(unsigned char *)buf = CP2TS_REGISTER;
  902. buf += sizeof (char);
  903. *(int *)buf = AUTH_USERID;
  904. buf += sizeof (int);
  905. strncpy (buf, AUTH_MD5, MD5_LEN);
  906. buf += MD5_LEN;
  907. *(unsigned short *)buf = htons (cfgP2PS_PORT);
  908. buf += sizeof (short);
  909. if (SCP_CHANNEL) //Now only GCP is available
  910. {
  911. *buf = CT_SPECIFIED_RES;
  912. buf += sizeof (char);
  913. memcpy (buf, SCP_CHANNEL, strlen(SCP_CHANNEL)+1);
  914. buf += strlen (SCP_CHANNEL)+1;
  915. } else if (ECP_REGION)
  916. {
  917. *buf = CT_EDGE;
  918. buf += sizeof (char);
  919. buf = parseECP (ECP_REGION, buf);
  920. } else
  921. {
  922. *buf = CT_GENERAL;
  923. buf += sizeof (char);
  924. }
  925. *(int *)buffer = buf - buffer; //the size of buffer;
  926. TSADDR.sin_port = htons (TS4CP_PORT);
  927. TSADDR.sin_addr.s_addr = inet_addr (SERVERIP);
  928. TSADDR.sin_family = AF_INET;
  929. if (sendMessage (TSSOCK, buffer, &TSADDR) < 0) //send register msg
  930. {
  931. PDEBUG ("Cannot write to servern");
  932. return -1;
  933. }
  934. return 0;
  935. }
  936. #endif
  938. int init_cp ()
  939. {
  940. FILE *pidf;
  941. struct rlimit rl;
  942. char buffer[MAX_DATA];
  943. rl.rlim_cur = rl.rlim_max = 1000000;
  944. if (setrlimit (RLIMIT_NOFILE, &rl) != 0)
  945. {
  946. perror ("getrlimit");
  947. }
  948. OPENLOG;
  949. #ifdef DEBUG
  950. system ("ulimit -a");
  951. if (getrlimit (RLIMIT_CORE, &rl) != 0)
  952. {
  953. perror ("getrlimit");
  954. }
  955. fprintf (stderr, "Get core limit %d:%dn", (int)rl.rlim_cur, (int)rl.rlim_max);
  956. rl.rlim_cur = rl.rlim_max = (rlim_t )102400;
  957. if (setrlimit (RLIMIT_CORE, &rl) != 0)
  958. {
  959. perror ("getrlimit");
  960. }
  961. if (getrlimit (RLIMIT_CORE, &rl) != 0)
  962. {
  963. perror ("getrlimit");
  964. }
  965. fprintf (stderr, "Set core limit to %d:%dn", (int)rl.rlim_cur, (int)rl.rlim_max);
  966. system ("ulimit -a");
  967. #endif
  968. #ifdef HAVE_TS
  969. if (register_cp () < 0)
  970. {
  971. PDEBUG ("Cannot init TS connectionn");
  972. return -1;
  973. }
  974. #endif
  975. TRACKER[TYPE_P2PS].flag = FLAG_SERVER;
  976. TRACKER[TYPE_P2PS].type = TYPE_P2PS;
  977. TRACKER[TYPE_P2PS].port = cfgP2PS_PORT;
  978. TRACKER[TYPE_P2PS].cur = 0;
  979. TRACKER[TYPE_P2PS].max = MAX_P2PS;
  980. TRACKER[TYPE_P2PS].init = init_P2PS;
  981. TRACKER[TYPE_P2PS].process = process_P2PS;
  982. TRACKER[TYPE_P2PS].closure = closure_P2PS;
  983. TRACKER[TYPE_P2PS].head = calloc (sizeof (struct Session), TRACKER[TYPE_P2PS].max);
  984. switch (BINDALL)
  985. {
  986. case 0:
  987. if ((TRACKER[TYPE_P2PS].sock = init_server (BINDIP, cfgP2PS_PORT)) < 0)
  988. return -1;
  989. break;
  990. default:
  991. if ((TRACKER[TYPE_P2PS].sock = init_server (NULL, cfgP2PS_PORT)) < 0)
  992. return -1;
  993. break;
  994. }
  996. FD_SET(TRACKER[TYPE_P2PS].sock, &osocks);
  997. TRACKER[TYPE_P2PC].flag = FLAG_CLIENT;
  998. TRACKER[TYPE_P2PC].type = TYPE_P2PC;
  999. TRACKER[TYPE_P2PC].port = 0;
  1000. TRACKER[TYPE_P2PC].cur = 0;
  1001. TRACKER[TYPE_P2PC].max = MAX_P2PC;
  1002. TRACKER[TYPE_P2PC].init = init_P2PC;
  1003. TRACKER[TYPE_P2PC].process = process_P2PC;
  1004. TRACKER[TYPE_P2PC].closure = closure_P2PC;
  1005. TRACKER[TYPE_P2PC].head = calloc (sizeof (struct Session), TRACKER[TYPE_P2PC].max);
  1007. mkdir (PREFIX, 0777);
  1008. snprintf (buffer, MAX_DATA, "%s/%s", PREFIX, LIVE_PREFIX);
  1009. mkdir (buffer, 0777);
  1010. snprintf (buffer, MAX_DATA, "rm -fr %s/%s/*", PREFIX, LIVE_PREFIX);
  1011. system (buffer);
  1012. if ((pidf = fopen (PIDFile, "w")) == NULL)
  1013. {
  1014. PDEBUG ("Cannot open pidfile.n");
  1015. return -1;
  1016. }
  1017. fprintf (pidf, "%dn", getpid ());
  1018. fclose (pidf);
  1019. return 0;
  1020. }
  1022. #ifdef HAVE_TS
  1023. int process_TS()
  1024. {
  1025. struct sockaddr_in dest;
  1026. int addr_len = sizeof (dest);
  1027. struct Message m;
  1028. int i;
  1030. if ((i = recvfrom (TSSOCK, &m, sizeof (struct Message), 0, (struct sockaddr *)&dest, &addr_len)) <= 0)
  1031. {
  1032. PDEBUG ("Error in recving ts message.n");
  1033. register_cp ();
  1034. return 0;
  1035. }
  1036. switch (m.type)
  1037. {
  1038. case TS2CP_WELCOME:
  1039. memcpy (&UDPMsg, &m, 12);
  1040. PDEBUG("recv WELCOME from TS.n");
  1041. /* isSet whether TS has returned WELCOME message */
  1042. isSet = 1;
  1043. break;
  1044. case TS2CP_PEERS:
  1045. process_TS2CP_PEERS (m.buffer);
  1046. break;
  1047. case TS2CP_MSG:
  1048. if (*(char *)(m.buffer+sizeof(short)))
  1049. {
  1050. PDEBUG ("Error in TS2CP_MSG. n");
  1051. register_cp ();
  1052. }
  1053. break;
  1054. default:
  1055. PDEBUG ("Error in trackerserver message formatn");
  1056. register_cp ();
  1057. return -1;
  1058. }
  1059. return 0;
  1060. }
  1062. int closure_TS ()
  1063. {
  1064. struct TSMessage *m = &UDPMsg;
  1066. m->type = CP2TS_LOGOUT;
  1067. m->len = 12;
  1068. sendMessage (TSSOCK, (char *)m, &TSADDR);
  1069. return 0;
  1070. }
  1072. int process_TS2CP_PEERS (char *buf)
  1073. {
  1074. struct Channel *pc;
  1075. int listnum;
  1076. char *channel_md5;
  1077. unsigned char cpsize;
  1078. struct NormalAddress *CPlist;
  1079. unsigned char peersize;
  1080. struct PeerInfoWithAddr *pinfo;
  1083. channel_md5 = buf;
  1084. buf += MD5_LEN;
  1085. cpsize = *(unsigned char *)buf;
  1086. buf += sizeof (char);
  1087. CPlist = (struct NormalAddress *)buf;
  1088. buf += sizeof(struct NormalAddress)*cpsize;
  1089. peersize = *(unsigned char *)buf;
  1090. buf += sizeof (char);
  1091. pinfo = (struct PeerInfoWithAddr *)buf;
  1092. // now find the channel
  1093. if ((pc = findChannel (channel_md5, MD5_LEN)) == NULL) return -1;
  1094. if ((listnum = newChannel (pc, CPlist, cpsize, TYPE_CP)) < 0)
  1095. return -1;
  1096. return 0;
  1097. }
  1098. #endif
  1101. char *parseECP (char *str, char *buf)
  1102. {
  1103. char  *buffer = buf;
  1104. int flag = -1;
  1105. unsigned char c;
  1106. unsigned char part;
  1107. buf += sizeof (int);
  1108. for (part=0; *str ;str++)
  1109. {
  1110. c = *str;
  1111. switch (c)
  1112. {
  1113. case ':':
  1114. flag = 0;
  1115. break;
  1116. case '.':
  1117. if (flag < 2)
  1118. {
  1119. *(unsigned char *)buf = part;
  1120. buf += sizeof (char);
  1121. }
  1122. part = 0;
  1123. flag ++;
  1124. break;
  1125. default:
  1126. if (c >= '0' && c <= '9')
  1127. part = c;
  1128. break;
  1129. }
  1130. }
  1131. *(int *)buffer = (buf - buffer - sizeof (int))/sizeof(short);
  1132. return buf;
  1133. }
  1135. int periodCheck (float KBPSused)
  1136. {
  1137. struct Session *head;
  1138. int max, listnum;
  1139. struct statistics
  1140. {
  1141. unsigned int resnum;
  1142. unsigned short connnum;
  1143. float bandwidth;
  1144. } stat;
  1145. stat.bandwidth = KBPSused/(MAX_BANDWIDTH*1024)/8;
  1147. /* isSet whether TS has returned WELCOME message */
  1148. if (isSet == 0)
  1149. {
  1150. register_cp ();
  1151. return 0;
  1152. }
  1153. max = TRACKER[TYPE_P2PC].maxid + 1;
  1154. head = TRACKER[TYPE_P2PC].head;
  1155. memset (&stat, 0, sizeof(stat));
  1156. for (listnum = 0; listnum < max; listnum++)
  1157. {
  1158. if (head[listnum].socket > 0)
  1159. {
  1160. stat.resnum ++;
  1161. stat.connnum ++;
  1162. }
  1163. }
  1164. max = TRACKER[TYPE_P2PS].maxid + 1;
  1165. head = TRACKER[TYPE_P2PS].head;
  1166. for (listnum = 0; listnum < max; listnum++)
  1167. {
  1168. if (head[listnum].socket > 0)
  1169. {
  1170. if (head[listnum].pc == NULL &&
  1171. head[listnum].header == NULL &&
  1172. CurrentTime - head[listnum].last_transferblock > MAX_TRANSFER_IDLE)
  1173. {
  1174. PDEBUG ("timeout %d from NP %d to %d.n", listnum, head[listnum].last_transferblock, (int)CurrentTime);
  1175. Clientclosure (listnum, TYPE_P2PS);
  1176. }
  1177. else
  1178. stat.connnum ++;
  1179. }
  1180. }
  1182. #ifdef HAVE_TS
  1183. *(unsigned int *)(UDPMsg.buffer) = stat.resnum;
  1184. *(unsigned short *)(UDPMsg.buffer+sizeof(int)) = stat.connnum;
  1185. *(float *)(UDPMsg.buffer+sizeof(int)+sizeof(short)) = stat.bandwidth;
  1186. if(stat.connnum <= MAX_P2PS)
  1187. *(unsigned char *)(UDPMsg.buffer+sizeof(int)+sizeof(short)+sizeof(float)) = 0;
  1188. else
  1189. *(unsigned char *)(UDPMsg.buffer+sizeof(int)+sizeof(short)+sizeof(float)) = 1;
  1190. UDPMsg.len = 23;
  1191. UDPMsg.type = CP2TS_UPDATE;
  1192. if (sendMessage (TSSOCK, (char *)&UDPMsg, &TSADDR) < 0)
  1193. {
  1194. PDEBUG ("exit...n");
  1195. exit (1);
  1196. }
  1197. #endif
  1198. PDEBUG("Res Num: %d. Connection Num: %d. BandWidth Usage: %.4f. n", stat.resnum, stat.connnum, stat.bandwidth);
  1199. return 0;
  1200. }
  1202. void makeSnapShot(int count, int time_interval)
  1203. {
  1204. time_t tmpTime;
  1205. struct tm result;
  1206. struct Channel *pc, *nextpc;
  1207. // struct Session *ps;
  1208. // struct Edge *pe;
  1209. int cpchannelcount = 0;
  1210. int totalclient = 0;
  1211. long long totalupsize = 0, totaldownsize = 0;
  1212. FILE *f;
  1213. char buffer[MAX_DATA];
  1215. if (time_interval <= 0)
  1216. return;
  1217. localtime_r(&CurrentTime, &result);
  1218. sprintf (buffer, "./cp-%d-%d-%d.log", result.tm_year+1900, result.tm_mon+1, result.tm_mday);
  1219. if ((f = fopen(buffer,"a")) == NULL)
  1220. {
  1221. PDEBUG("Couldn't open cp.log file! n");
  1222. return;
  1223. }
  1224. fseeko(f, 0, SEEK_END);
  1226. // 1. start CP SnapShot 
  1227. fprintf(f, "nn**************Start %d SnapShot of CP, Time : %u/%u %u:%u:%u.*********  n",count,result.tm_mon+1, result.tm_mday, result.tm_hour, result.tm_min, result.tm_sec);
  1229. // 2. log speed
  1230. fprintf(f, "CP: cur Down %.4f KB. n", ((float)tmpDownBytes)/1024/time_interval);
  1231. fprintf(f, "CP: cur Up   %.4f KB. n", ((float)tmpUpBytes)/1024/time_interval);
  1233. periodCheck(((float)tmpDownBytes+tmpUpBytes)/1024/time_interval);
  1235. totalDownBytes += tmpDownBytes;
  1236. totalUpBytes += tmpUpBytes;
  1237. tmpTime = CurrentTime - startTime; 
  1238. fprintf(f, "CP: avg Down %.4f KB. n", ((float)totalDownBytes)/1024/tmpTime);
  1239. fprintf(f, "CP: avg Up   %.4f KB. n", ((float)totalUpBytes)/1024/tmpTime);
  1241. // 3. log channel state
  1242. for (pc=ChannelList; pc; pc=nextpc)
  1243. {
  1244. nextpc = pc->lnext;
  1245. ++cpchannelcount;
  1246. totalclient += pc->numclient;
  1247. totalupsize += pc->upsize;
  1248. totaldownsize += pc->downsize;
  1249. fprintf(f,"Channel %s have %d client. Down size %lld, avg speed %f. Up Size %lld, avg speed %f. n",pc->fname,pc->numclient,pc->downsize, ((float)(pc->downsize)) / time_interval, pc->upsize, ((float)(pc->upsize))/ time_interval);
  1250. /*
  1251. for (pe=pc->PeerHead; pe; pe = pe->cnext)
  1252. {
  1253. // if bitrate < 300kb/s ,then kill it
  1254. if (pe->me->totalup/(CurrentTime - pe->me->time_sec)/1024 < 300)
  1255. fprintf(f,"Session bitrate:%lld .Too slow ! n",pe->me->totalup/(CurrentTime - pe->me->time_sec));
  1256. }
  1257. */
  1258. if (pc->numclient == 0)
  1259. freeLiveChannel (pc, NULL);
  1260. else if (CurrentTime - pc->last_nearpeer > NearPeerInterval)
  1261. {
  1262. send_nearpeers_toall (pc);
  1263. pc->last_nearpeer = CurrentTime;
  1264. }
  1266. }
  1267. fprintf(f,"Channel Count : %d. Total client : %d. Total dowsize: %lld. Total upsize %lld n",cpchannelcount,totalclient,totaldownsize,totalupsize);
  1269. fprintf(f,"n*********************End of SnapShot************************n");
  1270. fclose(f);
  1271. logto_xml (time_interval, tmpTime, cpchannelcount, totalclient);
  1272. tmpDownBytes = tmpUpBytes = 0;
  1273. }
  1275. int reconnect (struct Session *p)
  1276. {
  1277. struct NormalAddress *client;
  1278. struct sockaddr_in addr;
  1279. if (p->pc == NULL || p->pc->pcinfo->numofsp <= p->flag) return -1;
  1281. close (p->socket);
  1282. FD_CLR(p->socket, &osocks);
  1283. if ((p->socket = socket (PF_INET, SOCK_STREAM, 0)) < 0)
  1284. {
  1285. perror ("socket||gethostbyname");
  1286. p->socket = 0;
  1287. return -1;
  1288. }
  1289. client = &(p->pc->pcinfo->SPLIST[p->flag]);
  1290. memset (&addr, 0, sizeof (addr));
  1291. addr.sin_port = client->sin_port;
  1292. addr.sin_addr = client->sin_addr;
  1293. addr.sin_family = AF_INET;
  1294. p->flag ++;
  1295. if ((p->sock_flag = connect_nonb(p->socket, &addr, sizeof (addr))) == -1)
  1296. {
  1297. close (p->socket);
  1298. return -1;
  1299. }
  1301. FD_SET(p->socket, &osocks);
  1302. p->time_sec = CurrentTime;
  1303. p->totalup = 0;
  1304. return p->flag;
  1305. }
  1307. int send_P2P_PUSHLIST (struct Channel *pc, int id)
  1308. {
  1309. unsigned char *ptr;
  1310. char buffer[MAX_DATA], *buf;
  1311. struct LiveChannelInfo *pcinfo = pc->pcinfo;
  1312. int i, j;
  1314. buf = buffer+sizeof (int);
  1315. *(unsigned char *)buf = P2P_PUSHLIST;
  1316. buf += sizeof (char);
  1317. memcpy (buf, pc->channel_md5, MD5_LEN);
  1318. buf += MD5_LEN;
  1319. *(unsigned char *)buf = 0;
  1320. buf += sizeof (char);
  1321. ptr = buf;
  1322. buf += sizeof (char);
  1323. *ptr = 0;
  1324. for (i=id; (i >= pcinfo->s.minBlockID && i <= pcinfo->s.maxBlockID) && i<id+REQUEST_AHEAD; i++)
  1325. {
  1326. j = i % pcinfo->max_queue;
  1327. if (pcinfo->indisk[j] == (i/pcinfo->max_queue+1) || isSet (pcinfo->bitflag, j))
  1328. continue;
  1329. *(int *)buf = i;
  1330. buf += sizeof (int);
  1331. (*ptr) ++;
  1332. setBit (pcinfo->bitflag, j);
  1333. }
  1334. if (*ptr > 0)
  1335. {
  1336. *(unsigned char *)buf = 0;
  1337. buf += sizeof (char);
  1338. *(int *)buffer = buf - buffer;
  1339. writeMessage (pcinfo->dataSource, pc, buffer);
  1340. }
  1341. return 0;
  1342. }
  1344. int send_nearpeers (struct Channel *pc, struct Edge *pme)
  1345. {
  1346. char buffer[MAX_DATA], *buf, *ptr;
  1347. struct Edge *pedge;
  1348. int i;
  1349. buf = buffer+sizeof (int);
  1350. *(unsigned char *)buf = P2P_NEAR_PEERS;
  1351. buf += sizeof (char);
  1352. ptr = buf;
  1353. buf += sizeof (char);
  1354. for (i=0, pedge=pme->cnext; i<MAX_NEARPEER; pedge=pedge->cnext)
  1355. {
  1356. if (pedge == NULL)
  1357. pedge = pc->PeerHead;
  1358. if (pedge == pme)
  1359. break;
  1360. memcpy (buf, &(pedge->me->addr), sizeof (struct PeerInfoWithAddr));
  1361. buf += sizeof (struct PeerInfoWithAddr);
  1362. i++;
  1363. }
  1364. if (i > 0)
  1365. {
  1366. *(unsigned char *)ptr = i;
  1367. *(int *)buffer = buf - buffer;
  1368. writeMessage (pme->me, pc, buffer);
  1369. }
  1370. return 0;
  1371. }
  1373. int send_nearpeers_toall (struct Channel *pc)
  1374. {
  1375. struct Edge *pedge;
  1376. for (pedge=pc->PeerHead; pedge; pedge=pedge->cnext)
  1377. send_nearpeers (pc, pedge);
  1378. return 0;
  1379. }