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

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

P2P编程

开发平台：

Visual C++

SPnew.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"
  21. struct Argument
  22. {
  23. int type;
  24. int spchannelcount;
  25. int totalclient;
  26. long long totalupsize;
  27. FILE *f, *xml[MAX_TS];
  28. char *buf;
  29. };
  30. #define SPUPDATE_SLOT 1
  32. #define MAX_CHANNEL 1024 /* max number of Channel */
  34. #define SP4CP_PORT 50001
  36. #define TS4RM_PORT 22169
  37. #define RM2TS_STAT_QUERY 0x20
  38. #define TS2RM_STAT_RESPONSE 0x30
  40. int MAX_CP=512;
  41. int MAX_CS=32;
  43. int MAX_JOB_PER_SESSION=3;
  45. char *Home = "./";
  46. char *PREFIX = "/data/sp/";
  47. char *Database = "user.db";
  48. char *CONFIG = "./asp.cfg";
  49. char *PIDFile = "/var/run/spnew.pid";
  50. char *RULE = "";
  51. char *SERVERIP;
  52. char *DBFILE;
  53. char *MEDIABASE;
  55. char *AUTH_MD5;
  56. int AUTH_USERID;
  58. fd_set osocks;
  59. int NumNewChannel;
  61. struct Message TempMsg;
  63. char *NET_NAME[] = { "edu", "cnc", "uni", "tel" };
  65. long long totalDownBytes=0, totalUpBytes=0;
  66. long long tmpDownBytes=0, tmpUpBytes=0;
  68. char *urlroot;
  69. char *defaultspip;
  70. char *spip[4];
  71. char *tsip[4];
  72. #ifdef TEST
  73. char *WWW_ROOT;
  74. #endif
  75. #ifdef HAVE_MYSQL
  76. char *MYSQL_HOST;
  77. char *MYSQL_USER;
  78. char *MYSQL_PASS;
  79. char *MYSQL_DB;
  80. #endif
  81. char *CAS_ADDR;     // Content Aggregation Server, added by lixingwu, 20070313
  83. int tsSock[MAX_TS];
  84. struct sockaddr_in tsAddr[MAX_TS];
  86. extern struct Channel *ChannelHash[MAX_CHANNEL];
  87. extern struct Channel *ChannelList;
  88. extern int isDelete (struct Channel *pc);
  90. extern int errno;
  92. int JobHighWater = 10000;
  93. int BINDALL=1;
  94. int AuthCS=0;
  95. time_t startTime;
  96. time_t CurrentTime;
  97. int PERIOD;
  98. int SnapShotInterval; //make a snapshot every few second
  99. int spuprecord[MAX_CHANNEL];
  100. struct ServerDesc TRACKER[MAX_TYPE];
  102. char *LOGXML;
  104. struct NamVal ConfigParameters[] = {
  105. {"DBdir", &Home, 's'},
  106. {"AuthCS", &AuthCS, 'd'},
  107. {"MAX_CP",&MAX_CP, 'd'},
  108. {"MAX_CS", &MAX_CS, 'd'},
  109. {"Pidfile", &PIDFile, 's'},
  110. {"Prefix", &PREFIX, 's'},
  111. {"DBfile", &Database, 's'},
  112. {"TrackerIP", &SERVERIP, 's'},
  113. {"Dir", &MEDIABASE, 's'},
  114. {"authid", &AUTH_USERID, 'd'},
  115. {"authmd5", &AUTH_MD5, 's'},
  116. {"periodDump", &PERIOD, 'd'}, //make a periodCheck every periodDump second
  117. {"UrlRoot", &urlroot, 's'},
  118. {"EDUSPIP", &spip[EDUTS], 's'},
  119. {"CNCSPIP", &spip[CNCTS], 's'},
  120. {"UNISPIP", &spip[UNITS], 's'},
  121. {"TELSPIP", &spip[TELTS], 's'},
  122. {"BINDALL", &BINDALL, 'd'},
  123. {"EDUTSIP", &tsip[EDUTS], 's'},
  124. {"CNCTSIP", &tsip[CNCTS], 's'},
  125. {"UNITSIP", &tsip[UNITS], 's'},
  126. {"TELTSIP", &tsip[TELTS], 's'},
  127. {"SnapShotInterval", &SnapShotInterval, 'd'},
  128. {"LogFilePath", &LOGXML, 's'},
  129. #ifdef TEST
  130. {"WWWRoot", &WWW_ROOT, 's'},
  131. #endif
  132. #ifdef HAVE_MYSQL
  133. {"MysqlAddress", &MYSQL_HOST, 's'},
  134. {"User", &MYSQL_USER, 's'},
  135. {"Database", &MYSQL_DB, 's'},
  136. {"Password", &MYSQL_PASS, 's'},
  137. #endif
  138. {"AllowCSList", &RULE, 's'},
  139. {"JobHighWater", &JobHighWater, 'd'},
  140.     {"CAS_ADDR", &CAS_ADDR, 's'}     // Content Aggregation Server, added by lixingwu, 20070313
  141. };
  143. extern int db_end ();
  144. extern int db_init (char *home, char *database);
  145. extern int isAllowed (int id, char *md5, char *cname, float bitrate,
  146.       float *limitedbitrate, int *issave);
  147. extern char *getJobBuffer (struct JobDes *p, int *max);
  148. extern inline void setblockId (struct JobDes *pj, int id);
  149. #ifdef TEST
  150. int buildGTV (struct Channel *pc, int datalen, char *data, int type);
  151. #endif
  152. int send_P2P_SPUPDATE (struct Session *p, struct Channel *pc, char *md5, struct SPUpdate *s);
  153. int send_p2p_err (struct Session *p, unsigned short code, int quit);
  154. int init_sp ();
  155. int handle_new_connection (int sock, int type);
  156. int Clientclosure (int listnum, int type);
  157. void process_child (void);
  158. int init_CP (int listnum);
  159. int process_CP (int listnum);
  160. int closure_CP (int listnum);
  161. int init_CS (int listnum);
  162. int process_CS (int listnum);
  163. int closure_CS (int listnum);
  164. #ifdef HAVE_MYSQL
  165. MYSQL *local_mysql;
  166. #endif
  167. int process_P2P_REQUEST_real (struct Session *p, struct Channel *pc, int id);
  168. void apply_check (struct Channel *p, void *arg);
  169. void apply_idle (struct Session *p, void *arg);
  170. int period_process ();
  171. void process_TS2RM (int type);
  173. extern int locate_mplist_by_id (struct Channel *pc, unsigned int id, char *buf, int max);
  174. extern int check_newplist ();
  175. extern int timer_free ();
  176. extern int timer_process (unsigned int t);
  177. extern void hup_handler (int);
  179. #include "sessions.c"
  181. #define BUILD_NOCH_SPUPDATE(s) do
  182. {
  183. s.minKeySample = -1LL;
  184. s.maxKeySample = -1LL;
  185. s.minBlockID = 0xffffffff;
  186. s.maxBlockID = 0xffffffff;
  187. } while (0)
  189. #define BUILD_ORDER_SPUPDATE(s,pc) do
  190. {
  191. s.minKeySample = -1LL;
  192. s.maxKeySample = -1LL;
  193. s.minBlockID = 0;
  194. s.maxBlockID = (pc->downsize > 0 ? ((pc->downsize - 1) / DEFAULT_BLOCK) : 0)+1;
  195. } while (0)
  197. #define BUILD_CLOSE_SPUPDATE(s) do
  198. {
  199. s.minKeySample = 0;
  200. s.maxKeySample = 0;
  201. s.minBlockID = 0xffffffff;
  202. s.maxBlockID = 0xffffffff;
  203. } while (0)
  205. #define BUILD_MLIST_SPUPDATE(s,pc) do
  206. {
  207. s.minKeySample = -2LL;
  208. s.maxKeySample = pc->pcinfo->mlist->m_totalchannel;
  209. s.minBlockID = 0;
  210. s.maxBlockID = pc->pcinfo->maxID;
  211. } while (0)
  213. #define BUILD_LIVE_SPUPDATE(spupdate,pc) do
  214. {
  215. spupdate.minKeySample = -3LL;
  216. spupdate.maxKeySample = 1;
  217. spupdate.minBlockID = pc->pcinfo->s.minBlockID;
  218. spupdate.maxBlockID = pc->pcinfo->s.maxBlockID;
  219. } while (0)
  221. void send_all_spupdate (struct Channel *pc, struct SPUpdate *s)
  222. {
  223. struct LiveChannelInfo *pcinfo = pc->pcinfo;
  224. struct Edge *pedge;
  225. char buffer[MAX_DATA], *buf;
  226. if (pcinfo == NULL) return;
  228. pcinfo->updated = CurrentTime;
  229. buf = buffer + sizeof (int);
  230. *(unsigned char *) buf = P2P_SPUPDATE;
  231. buf += sizeof (char);
  232. memcpy (buf, pc->channel_md5, MD5_LEN);
  233. buf += MD5_LEN;
  234. if (s == NULL)
  235. memset (buf, 0, sizeof (struct SPUpdate));
  236. else
  237. memcpy (buf, s, sizeof (struct SPUpdate));
  238. buf += sizeof (struct SPUpdate);
  239. *(int *) buffer = buf - buffer;
  240. for (pedge = pc->PeerHead; pedge; pedge = pedge->cnext)
  241. {
  242. if (writeMessage (pedge->me, pc, buffer) < 0)
  243. {
  244. PDEBUG ("send SPUPDATE err.n");
  245. }
  246. }
  247. }
  249. void
  250. process_TS2RM (int type) //assure num of channel 
  251. {
  252. int len;
  253. char buffer[4096];
  254. char *buf;
  255. struct sockaddr_in addr;
  256. int addrlen = sizeof (struct sockaddr);
  257. int i, msgsize, chnlnum;
  258. struct Channel *pChannel;
  260. if ((len =
  261.      recvfrom (tsSock[type], buffer, 4096, 0,
  262.        (struct sockaddr *) &addr, &addrlen)) < 0)
  263. {
  264. perror ("recvfrom ts");
  265. return;
  266. }
  267. PINFO ("got TS msg, len %d. n", len);
  268. buf = buffer;
  269. msgsize = *(int *) buf;
  270. buf += sizeof (int);
  271. if (*(unsigned char *) buf != TS2RM_STAT_RESPONSE)
  272. {
  273. perror ("bad message type from ts");
  274. return;
  275. }
  276. buf += sizeof (unsigned char);
  277. chnlnum = *(int *) buf;
  278. buf += sizeof (int);
  279. if (chnlnum > 100)
  280. {
  281. perror ("too large channel num from ts");
  282. return;
  283. }
  284. for (i = 0; i < chnlnum; ++i)
  285. {
  286. if ((pChannel = findChannel (buf, MD5_LEN)) == NULL)
  287. {
  288. buf += MD5_LEN;
  289. buf += sizeof (int);
  290. continue;
  291. }
  292. buf += MD5_LEN;
  293. pChannel->numofnp[type] = *(int *) buf;
  294. buf += sizeof (int);
  296. PINFO ("%s->%s: client %d.n",
  297. pChannel->pcinfo ? pChannel->channel_name : "",
  298. pChannel->channel_md5, pChannel->numofnp[type]);
  299. }
  300. }
  302. time_t lastCheck, last_snapshot;
  304. int
  305. period_process ()
  306. {
  307. int type;
  308. time_t tmpTime, time_interval;
  309. struct Argument arg;
  310. char buffer[MAX_DATA];
  311. struct tm result;
  312. int *querynum;
  313. static int snapCount = 0;
  315. timer_process (CurrentTime);
  316. check_newplist ();
  317. if (CurrentTime <= lastCheck + PERIOD) //PERIOD =periodDump=60
  318. return 0;
  320. memset (&arg, 0, sizeof (arg));
  321. tmpTime = CurrentTime - startTime;
  322. time_interval = CurrentTime - lastCheck;
  323. #ifdef HAVE_MYSQL
  324. query_mysql (local_mysql, "delete from channel where '1'");
  325. #endif
  326. #ifdef TEST
  327. for (type=0; type<MAX_TS; type++)
  328. {
  329. snprintf (buffer, MAX_DATA, "%s/%s/channel.xml", WWW_ROOT,
  330.    NET_NAME[type]);
  331. if ((arg.xml[type] = fopen (buffer, "w")) == (FILE *) 0)
  332. {
  333. PDEBUG ("Error in open file %sn", buffer);
  334. continue;
  335. }
  336. fprintf (arg.xml[type],
  337.   "<?xml version="1.0" encoding="gbk"?>rn<?xml:stylesheet type="text/xsl" href="simple.xsl" ?>rn<GaoV>rn");
  338. }
  339. #endif
  340. localtime_r (&CurrentTime, &result);
  341. snprintf (buffer, MAX_DATA, "./sp-%d-%d-%d.log", result.tm_year+1900, result.tm_mon+1, result.tm_mday);
  342. if ((arg.f = fopen (buffer, "a")) == NULL)
  343. {
  344. PDEBUG ("Couldn't open sp log file! n");
  345. return -1;
  346. }
  347. fprintf (arg.f, "nn********************Start %d SnapShot of SP, Time: %u/%u %u:%u:%u.n",
  348. snapCount++, result.tm_mon + 1, result.tm_mday, result.tm_hour,
  349.   result.tm_min, result.tm_sec);
  351. fprintf(arg.f, "SP: cur Down %.4f KB. n", ((float)tmpDownBytes)/1024/time_interval);
  352. fprintf(arg.f, "SP: cur Up   %.4f KB. n", ((float)tmpUpBytes)/1024/time_interval);
  353. totalDownBytes += tmpDownBytes;
  354. totalUpBytes += tmpUpBytes;
  355. fprintf(arg.f, "SP: avg Down %.4f KB. n", ((float)totalDownBytes)/1024/tmpTime);
  356. fprintf(arg.f, "SP: avg Up   %.4f KB. n", ((float)totalUpBytes)/1024/tmpTime);
  358. arg.buf = buffer + sizeof (int);
  359. *(unsigned char *) (arg.buf) = RM2TS_STAT_QUERY;//RM2TS=0x20
  360. arg.buf += sizeof (unsigned char);
  361. querynum = (int *) (arg.buf);
  362. arg.buf += sizeof (int);
  364. arg.type = TYPE_CS;
  365. apply_session (TRACKER[TYPE_CS].head, TRACKER[TYPE_CS].maxid+1, apply_idle, &arg);
  366. /*
  367. arg.type = TYPE_CP;
  368. apply_session (TRACKER[TYPE_CP].head, TRACKER[TYPE_CP].maxid+1, apply_idle, &arg);
  369. */
  370. apply_list (ChannelList, apply_check, &arg);
  371. #ifdef TEST
  372. for (type=0; type<MAX_TS; type++)
  373. {
  374. if (arg.xml[type] == NULL)
  375. continue;
  376. fprintf (arg.xml[type], "</GaoV>rn");
  377. fclose (arg.xml[type]);
  378. }
  379. #endif
  380. fprintf (arg.f, "Channel Count : %d.Total client : %d . Total upsize :%lldB . n",
  381.   arg.spchannelcount, arg.totalclient, arg.totalupsize);
  382. fprintf (arg.f,
  383.   "n************************END SnapShot*********************n");
  384. fclose (arg.f);
  386.     // added by lixingwu, 20070313
  387.     // upload channel.xml to server, only edu
  388.     char upload_cmd[MAX_DATA];
  389.     sprintf(upload_cmd, "curl -F filename=@%s/%s/channel.xml -F domain=%s %s",
  390.         WWW_ROOT, NET_NAME[EDUTS], spip[EDUTS], CAS_ADDR);
  391.     printf("%sn", upload_cmd);
  392.     system(upload_cmd);
  394. #ifdef TEST
  395. if (arg.spchannelcount > 0)
  396. {
  397. *querynum = arg.spchannelcount;
  398. *(int *) buffer = arg.buf - buffer;
  399. for (type=0; type<MAX_TS; type++)
  400. {
  401. if (tsSock[type] <= 0)
  402. continue;
  403. if (sendto (tsSock[type], buffer, *(int *) buffer, 0, (struct sockaddr *) &tsAddr[type], sizeof (struct sockaddr_in)) != *(int *) buffer)
  404. {
  405. PDEBUG ("sent to ts %d error. n", type);
  406. return 0;
  407. }
  408. }
  409. }
  410. #endif
  411. logto_xml (time_interval, tmpTime, arg.spchannelcount, arg.totalclient);
  412. lastCheck = CurrentTime;
  413. last_snapshot ++;
  414. tmpDownBytes = tmpUpBytes = 0;
  415. if (last_snapshot > SnapShotInterval)
  416. {
  417. system ("/usr/bin/vmstat -a >> sp.log 2>&1 &");
  418. last_snapshot = 0;
  419. }
  420. return 0;
  421. }
  424. int
  425. main (int argc, char **argv)
  426. {
  427. int i, mode = 1;
  429. if (argc < 2)
  430. {
  431. printf ("usage: %s mode(0 for daemon, 1 for console).n",
  432. argv[0]);
  433. return -1;
  434. }
  435. signal (SIGPIPE, SIG_IGN);
  436. signal (SIGINT, terminate);
  437. signal (SIGHUP, hup_handler);
  439. mode = atoi (argv[1]);
  441. if (mode == 0)
  442. daemon (1, 1);
  444. read_config (CONFIG, ConfigParameters,
  445.      sizeof (ConfigParameters) / sizeof (struct NamVal));
  446. for (i = 0; i < 10 && IN_LOOP > 0; i++)
  447. {
  448. /*
  449.    pid_t pid;
  450.    if ((pid = fork ()) == 0)
  451.    {
  452.  */
  453. FD_ZERO (&osocks);
  454. if (init_sp () < 0) // || initLOG () < 0)
  455. {
  456. PDEBUG ("init_sp error, exit...n");
  457. exit (-1);
  458. }
  459. process_child ();
  460. /*
  461.    } else if (pid < 0)
  462.    {
  463.    perror ("fork");
  464.    exit (pid);
  465.    } else
  466.    {
  467.    waitpid (pid, NULL, 0);
  468.    }
  469.  */
  470. }
  471. return 0;
  472. }
  474. int
  475. process_P2P_HELLO (struct Session *p, struct Message *m)
  476. {
  477. struct SPUpdate spupdate;
  478. struct Edge *pedge;
  479. struct Channel *pc;
  480. int listnum;
  481. PDEBUG ("CP %d.%d.%d.%d:%d join channel %.32sn",
  482. IPADDR (p->host), p->port, m->buffer);
  483. listnum = p - TRACKER[TYPE_CP].head;
  484. if ((pc = findChannel (m->buffer, MD5_LEN)) == NULL)
  485. {
  486. if ((pc = findOrder (m->buffer, MD5_LEN)) == NULL &&
  487. (pc = newOrder (m->buffer)) == NULL)
  488. {
  489. BUILD_NOCH_SPUPDATE(spupdate);
  490. send_P2P_SPUPDATE (p, pc, m->buffer, &spupdate);
  491. PDEBUG ("Cannot find channel %.32sn",
  492. m->buffer);
  493. //Clientclosure (listnum, TYPE_CP);
  494. } else
  495. {
  496. BUILD_ORDER_SPUPDATE(spupdate,pc);
  497. send_P2P_SPUPDATE (p, pc, m->buffer, &spupdate);
  498. PDEBUG ("channel %.32s spupdate %d~%d. n",
  499. m->buffer, spupdate.minBlockID,
  500. spupdate.maxBlockID);
  501. }
  502. } else
  503. {
  504. for (pedge = p->header; pedge && pedge->head == pc;
  505.      pedge = pedge->enext);
  506. if (!pedge)
  507. {
  508. pedge = newEdge (pc, p);
  509. pc->numclient++;
  510. }
  511. if (pc->pcinfo)
  512. {
  513. if (pc->pcinfo->status > 0)
  514. {
  515. // this channel has been closed
  516. BUILD_CLOSE_SPUPDATE(spupdate);
  517. send_P2P_SPUPDATE (p, pc, m->buffer, &spupdate);
  518. PDEBUG ( "channel %.32s has been closed. n",
  519. m->buffer);
  520. //Clientclosure (listnum, TYPE_CP);
  521. } else if (pc->pcinfo->mlist != NULL)
  522. {
  523. BUILD_MLIST_SPUPDATE (spupdate, pc);
  524. send_P2P_SPUPDATE (p, pc, m->buffer, &spupdate);
  525. sendMedia (p, pc);
  526. PDEBUG ("Channel %.32s is a playlist.n", m->buffer);
  527. } else
  528. {
  529. BUILD_LIVE_SPUPDATE(spupdate,pc);
  530. send_P2P_SPUPDATE (p, pc, m->buffer, &spupdate);
  531. sendMedia (p, pc);
  532. }
  533. } else
  534. {
  535. BUILD_ORDER_SPUPDATE(spupdate,pc);
  536. send_P2P_SPUPDATE (p, pc, m->buffer, &spupdate);
  537. PDEBUG ("channel %.32s spupdate %d~%d. n",
  538. m->buffer, spupdate.minBlockID,
  539. spupdate.maxBlockID);
  540. }
  541. }
  542. return 0;
  543. }
  545. int
  546. process_P2P_PUSHLIST (struct Session *p, struct Message *m)
  547. {
  548. struct Edge *pedge;
  549. struct Channel *pc;
  550. char *buf;
  551. int i,size, type;
  553. if ((pc = findChannel (m->buffer, MD5_LEN)) == NULL)
  554. pc = newOrder (m->buffer);
  555. else if (pc->pcinfo != NULL)
  556. {
  557. for (pedge = p->header; pedge && pedge->head != pc;
  558.      pedge = pedge->enext);
  559. if (pedge == NULL)
  560. {
  561. pedge = newEdge (pc, p);
  562. pc->numclient++;
  563. }
  564. }
  565. if (pc == NULL)
  566. return -1;
  568. if (p->numjob >= MAX_JOB_PER_SESSION)
  569. return -2;
  571. buf = m->buffer + MD5_LEN;
  572. type = *(unsigned char *) buf;
  573. buf += sizeof (char);
  574. size = *(unsigned char *) buf;
  575. buf += sizeof (char);
  576. if (type)
  577. {
  578. deleteChannel (p, pc);
  579. for (i=0; i<size; i++)
  580. if (process_P2P_REQUEST_real (p, pc, ((unsigned int *)buf)[i]) < 0) 
  581. return -1;
  582. } else
  583. {
  584. for (i=0; i<size; i++)
  585. if (process_P2P_REQUEST_real (p, pc, ((unsigned int *)buf)[i]) < 0)
  586. return -1;
  587. buf += size * sizeof (int);
  588. size = *(unsigned char *) buf;
  589. buf += sizeof (char);
  590. deleteJob (p, pc, (unsigned int *) buf, size);
  591. }
  592. return 0;
  593. }
  595. int
  596. process_P2P_REQUEST_real (struct Session *p, struct Channel *pc, int id)
  597. {
  598. struct JobDes *pj = newJob ();
  599. char *buf, *buffer;
  600. int listnum, size=0, max;
  601. struct SPUpdate spupdate;
  603. if (pj == NULL) return -1;
  604. buffer = getJobBuffer (pj, &max);
  605. listnum = p - TRACKER[TYPE_CP].head;
  606. buf = buffer + sizeof (int);
  607. *(unsigned char *) buf = P2P_RESPONSE;
  608. buf += sizeof (char);
  609. memcpy (buf, pc->channel_md5, MD5_LEN);
  610. buf += MD5_LEN;
  612. if (pc->pcinfo == NULL)
  613. {
  614. *(int *) buf = id;
  615. buf += sizeof (int);
  616. if ((size =
  617.      locate_order_by_id (pc, id, buf + sizeof (int),
  618.  max)) < 0 && size != -2)
  619. {
  620. // BLOCK NOT FOUND
  621. spupdate.minKeySample = -1LL;
  622. spupdate.maxKeySample = -1LL;
  623. spupdate.minBlockID = 0xffffffff;
  624. spupdate.maxBlockID = 0xffffffff;
  625. send_P2P_SPUPDATE (p, pc, pc->channel_md5, &spupdate);
  626. size = 0;
  627. *(int *) buf = 0;
  628. buf += sizeof (int);
  630. } else if (size == -2)
  631. {
  632. PDEBUG ("Leave blocks %d to next round.n", id);
  633. return -1;
  634. } else
  635. {
  636. // block found
  637. *(int *) buf = size;
  638. buf += sizeof (int) + size;
  639. p->last_transferblock = CurrentTime;
  640. }
  641. //              Clientclosure (listnum, TYPE_CP);
  642. } else if (id >= 0 && pc->pcinfo->mlist != NULL
  643.    && (size = locate_mplist_by_id (pc, id, buf, max - 32)) > 0)
  644. {
  645. p->last_transferblock = CurrentTime;
  646. buf += 2 * sizeof (int) + size;
  648. } else if (id >= 0 && pc->pcinfo->mlist == NULL
  649.    && (size = locate_by_id (pc, id, buf, max - 32)) > 0)
  650. {
  651. p->last_transferblock = CurrentTime;
  652. buf += 2 * sizeof (int) + size;
  653. } else if (size == -2)
  654. {
  655. assert (0);
  656. PDEBUG ("Leave blocks %d to next round.n", id);
  657. return -1;
  658. } else
  659. {
  660. *(int *) buf = id;
  661. buf += sizeof (int);
  662. size = 0;
  663. *(int *) buf = 0;
  664. buf += sizeof (int);
  665. PDEBUG ("Cannot find block id %d required by client %d.%d.%d.%d.n",
  666. id, IPADDR (p->host));
  667. }
  668. *(int *) buffer = buf - buffer;
  669. setblockId (pj, id);
  670. writeDATAMessage (p, pc, pj);
  671. // PDEBUG ("Write block %d to %d.%d.%d.%dn", id,
  672. // IPADDR (p->host));
  673. return 0;
  674. }
  677. int
  678. init_CP (int listnum)
  679. {
  680. return 0;
  681. }
  683. int
  684. process_CP (int listnum)
  685. {
  686. int ret;
  687. struct Session *p = &(TRACKER[TYPE_CP].head[listnum]);
  688. struct Message *m = (struct Message *) (p->buf + p->start);
  690. tmpDownBytes += m->len;
  692. switch (m->type)
  693. {
  694. case P2P_HELLO:
  695. ret=  process_P2P_HELLO (p, m);
  696. break;
  697. case P2P_PUSHLIST:
  698. ret = process_P2P_PUSHLIST (p, m);
  699. break;
  700. case P2P_MSG:
  701. break;
  702. default:
  703. ret = -1;
  704. break;
  705. }
  706. switch (ret)
  707. {
  708. case -1:
  709. PDEBUG ("Message processing error from client %d.%d.%d.%dn",
  710. IPADDR (p->host));
  711. Clientclosure (listnum, TYPE_CP);
  712. return -1;
  713. case -2:
  714. return -2;
  715. default:
  716. return 0;
  717. }
  718. }
  720. int
  721. closure_CP (int listnum)
  722. {
  723. struct Session *p = &(TRACKER[TYPE_CP].head[listnum]);
  724. //      struct Channel *pc = p->pc;
  725. struct Edge *pedge, *nextedge;
  726. PDEBUG ("CP disconnected from %d.%d.%d.%d:%dn",
  727. IPADDR (p->host), p->port);
  728. for (pedge = p->header; pedge; pedge = nextedge)
  729. {
  730. nextedge = pedge->enext;
  731. if (pedge->head)
  732. pedge->head->numclient--;
  733. delEdge (pedge);
  734. }
  735. FD_CLR (p->socket, &osocks);
  736. close (p->socket);
  737. FREE (p->buf);
  738. deleteAll (p);
  739. memset (p, 0, sizeof (struct Session));
  740. return 0;
  741. }
  744. int
  745. init_CS (int listnum)
  746. {
  747. return 0;
  748. }
  751. int
  752. process_CS2SP_REGISTER (int listnum, char *msg)
  753. {
  754. int i, errmsg;
  755. char cname[MAX_LINE], cmd5[MD5_LEN + 1], buffer[MAX_DATA];
  756. //char escape_buf[MAX_LINE];
  757. char md5[MD5_LEN + 1];
  758. struct Session *p = &(TRACKER[TYPE_CS].head[listnum]), *source;
  759. struct Channel *pc;
  760. float bitrate, limitedbitrate=10000.0;
  761. int id, startblock, size, maxblocksize, datalen, issave=0;
  763. size = *(unsigned char *) msg;
  764. msg += sizeof (char);
  765. if (size > sizeof (cname) || size <= 0) //wrong Message
  766. {
  767. Clientclosure (listnum, TYPE_CS);
  768. return -1;
  769. } else
  770. memcpy (cname, msg, size);
  771. cname[size] = 0;
  772. msg += size;
  773. //      memcpy (cmd5, msg, MD5_LEN);
  774. //      msg += MD5_LEN;
  775. id = *(int *) msg;
  776. msg += sizeof (int);
  778. sprintf (buffer, "%d@%s_%s", id, defaultspip, cname);
  779. md5_calc ((unsigned char *) md5, (unsigned char *) buffer,
  780.   strlen (buffer));
  781. for (i = 0; i < MD5_LEN; i += 2)
  782. sprintf (cmd5 + i, "%02x", (unsigned char) md5[i / 2]);
  783. cmd5[MD5_LEN] = 0;
  785. memcpy (md5, msg, MD5_LEN);
  786. msg += MD5_LEN;
  787. maxblocksize = *(int *) msg;
  788. msg += sizeof (int);
  789. bitrate = *(float *) msg;
  790. msg += sizeof (float);
  791. datalen = *(unsigned short *) msg;
  792. if (datalen > MAX_LINE)
  793. {
  794. Clientclosure (listnum, TYPE_CS);
  795. return -1;
  796. }
  797. msg += sizeof (short);
  798. if (AuthCS && (errmsg =
  799.      isAllowed (id, md5, cname, bitrate, &limitedbitrate, &issave)) < 0)
  800. {
  801. PDEBUG ("User %d is not allowed to newchannel %s.n",
  802. id, cname);
  803. send_p2p_err (p, -errmsg, 1);
  804. Clientclosure (listnum, TYPE_CS);
  805. return -1;
  806. }
  808. startblock = 0;
  809. if ((pc = findChannel (cmd5, MD5_LEN)) != NULL)
  810. {
  811. if (pc->pcinfo == NULL || pc->pcinfo->mlist != NULL)
  812. {
  813. PDEBUG ("The channel %s is a playlist.n", cname);
  814. send_p2p_err (p, ERR_INTERNAL, 1);
  815. Clientclosure (listnum, TYPE_CS);
  816. return -1;
  817. }
  818. if ((source = pc->pcinfo->dataSource) != NULL)
  819. {
  820. Clientclosure (source - TRACKER[TYPE_CS].head,
  821.        TYPE_CS);
  822. }
  823. }
  824. if ((pc = newLiveChannel (cname, p, cmd5, bitrate,
  825.     maxblocksize)) != (struct Channel *) 0)
  826. {
  827. p->pc = pc;
  828. pc->pcinfo->userid = id;
  829. pc->pcinfo->limitedBitRate = limitedbitrate;
  830. pc->pcinfo->isSave = issave;
  831. pc->pcinfo->dataSource = &(TRACKER[TYPE_CS].head[listnum]);
  832. pc->pcinfo->startid = pc->pcinfo->maxID = (CurrentTime - FIX_MAGIC) * 16;
  833. } else
  834. {
  835. PDEBUG ("newLiveChannel failed.n");
  836. send_p2p_err (p, ERR_INTERNAL, 1);
  837. Clientclosure (listnum, TYPE_CS);
  838. return -1;
  839. }
  840. *(int *) buffer = 9;
  841. *(char *) (buffer + sizeof (int)) = SP2CS_WELCOME;
  842. *(int *) (buffer + sizeof (int) + sizeof (char)) = startblock;
  843. if (writeMessage (p, pc, buffer) < 0)
  844. {
  845. send_p2p_err (p, ERR_INTERNAL, 1);
  846. Clientclosure (listnum, TYPE_CS);
  847. return -1;
  848. }
  849. pc->pcinfo->cur_channel = pc->pcinfo->max_channel = 1;
  850. pc->pcinfo->media = calloc (1, sizeof (struct MediaData));
  851. pc->pcinfo->media[0].start = 0;
  852. pc->pcinfo->media[0].len = -1;
  853. pc->pcinfo->media[0].dlen = datalen;
  854. pc->pcinfo->media[0].data = calloc (1, datalen);
  855. memcpy (pc->pcinfo->media[0].data, msg, datalen);
  856. #ifdef TEST
  857. /*
  858. for (type = 0; type < MAX_TS; ++type)
  859. buildGTV (pc, datalen, msg, type);
  860. */
  861. for (i=0; i<MAX_TS; i++) {
  862. if (buildGTV (pc, datalen, msg, i) < 0)
  863. continue;
  864. }
  865. #endif
  866. #ifdef HAVE_MYSQL
  867. //      query_mysql (local_mysql, "delete from channel where ChannelMD5 = "%s"", cmd5);
  868. //      mysql_escape_string (escape_buf, msg, datalen);
  869. //      query_mysql (local_mysql, "insert into channel (ChannelName, ChannelBitrate, ChannelAttachData, ChannelMD5, ChannelOwnerID) values ("%s","%f", "%s", "%s", "%d")", cname, bitrate, escape_buf, cmd5, id);
  870. #endif
  871. return 0;
  872. }
  874. int
  875. process_CS2SP_UPDATE (int listnum, float rate) //only update bitrate
  876. {
  877. struct Session *p = &(TRACKER[TYPE_CS].head[listnum]);
  878. struct Channel *pc = p->pc;
  879. struct LiveChannelInfo *pcinfo = pc->pcinfo;
  881. pcinfo->bitrate = rate;
  882. if (rate > pcinfo->limitedBitRate)
  883. {
  884. send_p2p_err (p, ERR_EXCEED_BITRATE, 1);
  885. Clientclosure (listnum, TYPE_CS);
  886. return -1;
  887. }
  888. #ifdef HAVE_MYSQL
  889. //      query_mysql (local_mysql, "update channel set ChannelBitrate = "%f" where ChannelMD5 = "%s"", rate, pc->channel_md5);
  890. #endif
  892. return 0;
  893. }
  895. int
  896. process_CS2SP_BLOCK (int listnum, char *msg)
  897. {
  898. char *buf, buffer[MAX_DATA];
  899. struct Edge *pedge;
  900. int size; // max=TRACKER[TYPE_CP].maxid+1;
  901. struct Session *p = &(TRACKER[TYPE_CS].head[listnum]);
  902. //      struct Session *q=TRACKER[TYPE_CP].head;
  903. struct Channel *pc = p->pc;
  904. struct LiveChannelInfo *pcinfo;
  906. if (p->pc == NULL || (pcinfo = p->pc->pcinfo) == NULL || pcinfo->mlist != NULL)
  907. {
  908. PDEBUG ("Unmatched channeln");
  909. Clientclosure (listnum, TYPE_CS);
  910. return -1;
  911. }
  913. pcinfo = p->pc->pcinfo;
  914. if ((size = saveBlock (pc, msg, p)) > 0)
  915. {
  916. // directly send this block to connected CPs
  917. buf = buffer + sizeof (int);
  918. *(unsigned char *) buf = P2P_RESPONSE;
  919. buf += sizeof (char);
  920. memcpy (buf, pc->channel_md5, MD5_LEN);
  921. buf += MD5_LEN;
  922. p->last_transferblock = CurrentTime;
  923. memcpy (buf, msg, size + 2 * sizeof (int));
  924. buf += size + 2 * sizeof (int);
  925. *(int *) buffer = buf - buffer;
  926. for (pedge = pc->PeerHead; pedge; pedge = pedge->cnext)
  927. {
  928. if (pedge->me->numjob >= MAX_JOB_PER_SESSION)
  929. continue;
  930. pedge->me->last_transferblock = CurrentTime;
  931. // PDEBUG ("write %u ",
  932. // ((unsigned int *) msg)[0]);
  933. if (-1 == writeMessage (pedge->me, pc, buffer))
  934. PDEBUG ("buffer is fulln");
  935. // else
  936. // PDEBUG ("OKn");
  937. }
  938. if (pcinfo->updated <= CurrentTime + SPUPDATE_SLOT) //pcinfo is livechannel
  939. send_all_spupdate (pc, &(pcinfo->s));
  940. } else
  941. {
  942. PDEBUG ("save Block Failed ! size %d, %dn", size,
  943. listnum);
  944. Clientclosure (listnum, TYPE_CS);
  945. }
  946. return 0;
  947. }
  949. int
  950. process_CS (int listnum)
  951. {
  952. struct Session *p = &(TRACKER[TYPE_CS].head[listnum]);
  953. struct Message *m = (struct Message *) (p->buf + p->start);
  955. tmpDownBytes += m->len;
  957. switch (m->type)
  958. {
  959. case CS2SP_REGISTER:
  960. process_CS2SP_REGISTER (listnum, m->buffer);
  961. break;
  962. case CS2SP_UPDATE:
  963. process_CS2SP_UPDATE (listnum,
  964.       *(float *) (m->buffer));
  965. break;
  966. case CS2SP_BLOCK:
  967. process_CS2SP_BLOCK (listnum, m->buffer);
  968. break;
  969. default:
  970. Clientclosure (listnum, TYPE_CS);
  971. return -1;
  972. }
  973. return 0;
  974. }
  976. int
  977. closure_CS (int listnum)
  978. {
  979. struct Session *p = &(TRACKER[TYPE_CS].head[listnum]);
  980. struct Channel *pc = p->pc;
  981. PDEBUG ("CS disconnected from %d.%d.%d.%d:%dn",
  982. IPADDR (p->host), p->port);
  984. if (pc)
  985. {
  986. if (pc->pcinfo)
  987. {
  988. pc->pcinfo->dataSource = NULL;
  989. pc->pcinfo->status = 1;
  990. }
  991. freeLiveChannel (pc, NULL);
  992. }
  993. FD_CLR (TRACKER[TYPE_CS].head[listnum].socket, &osocks);
  994. close (TRACKER[TYPE_CS].head[listnum].socket);
  995. FREE (p->buf);
  996. deleteAll (p);
  997. memset (&(TRACKER[TYPE_CS].head[listnum]), 0,
  998. sizeof (struct Session));
  999. return 0;
  1000. }
  1003. int
  1004. init_sp ()
  1005. {
  1006. FILE *pidf;
  1007. struct rlimit rl;
  1008. char *index;
  1009. int type;
  1010. char buffer[MAX_DATA];
  1012. rl.rlim_cur = rl.rlim_max = 1000000;
  1013. if (setrlimit (RLIMIT_NOFILE, &rl) != 0)
  1014. {
  1015. perror ("setrlimit");
  1016. }
  1017. for (type=0; type<MAX_TS; type++)
  1018. {
  1019. if (spip[type] != NULL && strlen (spip[type]) >= MIN_IPADDR_LEN)
  1020. {
  1021. defaultspip = spip[type];
  1022. break;
  1023. }
  1024. }
  1025. if (defaultspip == NULL)
  1026. defaultspip = "127.0.0.1";
  1028. OPENLOG;
  1029. #ifdef DEBUG
  1030. system ("ulimit -a");
  1031. if (getrlimit (RLIMIT_CORE, &rl) != 0)
  1032. {
  1033. perror ("getrlimit");
  1034. }
  1035. PDEBUG ("Get core limit %d:%dn", (int)rl.rlim_cur, (int)rl.rlim_max);
  1036. rl.rlim_cur = rl.rlim_max = (rlim_t )10240000;
  1037. if (setrlimit (RLIMIT_CORE, &rl) != 0)
  1038. {
  1039. perror ("setrlimit");
  1040. }
  1041. if (getrlimit (RLIMIT_CORE, &rl) != 0)
  1042. {
  1043. perror ("getrlimit");
  1044. }
  1045. PDEBUG ("Set core limit to %d:%dn", (int)rl.rlim_cur, (int)rl.rlim_max);
  1046. system ("ulimit -a");
  1047. #endif
  1048. #ifdef HAVE_MYSQL
  1049. if ((local_mysql =
  1050.      init_mysql (MYSQL_HOST, MYSQL_USER, MYSQL_PASS, MYSQL_DB,
  1051.  "/var/run/mysqld/mysqld.sock")) == 0)
  1052. {
  1053. PDEBUG ("Error in init_mysql.n");
  1054. exit (1);
  1055. }
  1056. #endif
  1057. TRACKER[TYPE_CP].type = TYPE_CP; //allocate CP ServerDesc  TYPE_CP = 0
  1058. TRACKER[TYPE_CP].port = SP4CP_PORT; //SP4CP_PORT = 50001
  1059. TRACKER[TYPE_CP].cur = 0; //current client connection 
  1060. TRACKER[TYPE_CP].max = MAX_CP; //MAX_CP = 2048
  1061. TRACKER[TYPE_CP].init = init_CP; //the function pointer of init_CP,return a debug message
  1062. TRACKER[TYPE_CP].process = process_CP; //the function pointer of process_CP, switch TYPE
  1063. TRACKER[TYPE_CP].closure = closure_CP; //the funciont pointer of closure_CP,return debug msg and freejob
  1064. TRACKER[TYPE_CP].head = calloc (sizeof (struct Session), TRACKER[TYPE_CP].max); //allocate session memory
  1065. switch (BINDALL)
  1066. {
  1067. case 0:
  1068. if ((TRACKER[TYPE_CP].sock = init_server (spip[0], SP4CP_PORT)) < 0) //PORT = 50001
  1069. return -1;
  1070. break;
  1071. default:
  1072. if ((TRACKER[TYPE_CP].sock = init_server (NULL, SP4CP_PORT)) < 0) //PORT = 50001
  1073. return -1;
  1074. break;
  1075. }
  1076. FD_SET (TRACKER[TYPE_CP].sock, &osocks);
  1077. TRACKER[TYPE_CS].type = TYPE_CS; //allocate CP ServerDese,TYPE_CS = 1
  1078. TRACKER[TYPE_CS].port = SP4CS_PORT;
  1079. TRACKER[TYPE_CS].cur = 0;
  1080. TRACKER[TYPE_CS].max = MAX_CS; //MAX_CS = 512
  1081. TRACKER[TYPE_CS].init = init_CS; //return a debug message        
  1082. TRACKER[TYPE_CS].process = process_CS;
  1083. TRACKER[TYPE_CS].closure = closure_CS;
  1084. TRACKER[TYPE_CS].head =
  1085. calloc (sizeof (struct Session), TRACKER[TYPE_CS].max);
  1086. switch (BINDALL)
  1087. {
  1088. case 0:
  1089. if ((TRACKER[TYPE_CS].sock = init_server (spip[0], SP4CS_PORT)) < 0) //port = 
  1090. return -1;
  1091. break;
  1092. default:
  1093. if ((TRACKER[TYPE_CS].sock = init_server (NULL, SP4CS_PORT)) < 0) //port = 
  1094. return -1;
  1095. break;
  1096. }
  1097. FD_SET (TRACKER[TYPE_CS].sock, &osocks);
  1098. if (db_init (Home, Database) < 0)
  1099. return -1;
  1101. memset (tsAddr, 0, sizeof (struct sockaddr_in) * MAX_TS); //sockaddr_in type
  1102. for (type = 0; type < MAX_TS; ++type)
  1103. {
  1104. #ifdef TEST
  1105. sprintf (buffer, "rm -f %s/%s/*.gtv", WWW_ROOT,
  1106.  NET_NAME[type]);
  1107. system (buffer);
  1108. sprintf (buffer, "rm -f %s/%s/*.mediadata", WWW_ROOT,
  1109.  NET_NAME[type]);
  1110. system (buffer);
  1111. sprintf (buffer, "rm -f %s/%s/channel.xml", WWW_ROOT,
  1112.  NET_NAME[type]);
  1113. system (buffer);
  1114. #endif
  1115. sprintf (buffer, "rm -fr %s/%s/*", PREFIX, LIVE_PREFIX);
  1116. system (buffer);
  1118. if (tsip[type] == NULL) continue;
  1119. tsAddr[type].sin_family = AF_INET;
  1120. tsAddr[type].sin_port = htons (TS4RM_PORT);
  1121. index = strchr (tsip[type], ':');
  1122. if (index == NULL)
  1123. inet_aton (tsip[type], &tsAddr[type].sin_addr);
  1124. else
  1125. {
  1126. *index = 0;
  1127. inet_aton (tsip[type], &tsAddr[type].sin_addr);
  1128. *index = ':';
  1129. }
  1130. tsSock[type] = socket (PF_INET, SOCK_DGRAM, 0); //upd connection
  1131. if (tsSock[type] < 0)
  1132. return -1;
  1133. }
  1134. mkdir (PREFIX, 0777);
  1135. sprintf (buffer, "%s/%s", PREFIX, LIVE_PREFIX);
  1136. mkdir (buffer, 0777);
  1137. sprintf (buffer, "%s/%s", PREFIX, ORDER_PREFIX);
  1138. mkdir (buffer, 0777);
  1139. sprintf (buffer, "%s/%s", PREFIX, PLIST_PREFIX);
  1140. mkdir (buffer, 0777);
  1141. sprintf (buffer, "%s/%s", PREFIX, PROG_PREFIX);
  1142. mkdir (buffer, 0777);
  1143. if ((pidf = fopen (PIDFile, "w")) == NULL)
  1144. {
  1145. PDEBUG ("Cannot open pidfile.n");
  1146. return -1;
  1147. }
  1148. fprintf (pidf, "%dn", getpid ());
  1149. fclose (pidf);
  1150. return 0;
  1151. }
  1153. #ifdef TEST
  1154. /*
  1155. int
  1156. buildMediaData (struct Channel *pc, int datalen, char *data, int type)
  1157. {
  1158. int i;
  1159. char olddata[MAX_DATA];
  1160. char buffer[MAX_DATA];
  1161. struct stat stbuf;
  1162. FILE *f;
  1163. assert (pc->pcinfo);
  1164. snprintf (buffer, MAX_LINE, "%s/%s/%s.mediadata", WWW_ROOT, NET_NAME[type], pc->channel_name);
  1165. if (stat (buffer, &stbuf) == 0)
  1166. {
  1167. if (stbuf.st_size != datalen)
  1168. {
  1169. PDEBUG ("old media data size %d not match new %dn", (int)(stbuf.st_size), datalen);
  1170. return -1;
  1171. }
  1172. if ((f = fopen (buffer, "r")) == NULL)
  1173. {
  1174. PDEBUG ("cannot open mediadata file %sn", buffer);
  1175. perror ("fopen");
  1176. return -1;
  1177. }
  1178. if (fread (olddata, datalen, 1, f) == 1)
  1179. {
  1180. for (i=0; i<datalen; i++)
  1181. {
  1182. if (data[i] != olddata[i])
  1183. {
  1184. PDEBUG ("media data not match %s, %dn", buffer, i);
  1185. fclose (f);
  1186. return -1;
  1187. }
  1188. }
  1189. } else
  1190. {
  1191. PDEBUG ("Error in read old mediadata %s.n", buffer);
  1192. fclose (f);
  1193. return -1;
  1194. }
  1195. fclose (f);
  1196. return 0;
  1197. } else if ((f = fopen (buffer, "w")) == NULL)
  1198. {
  1199. PDEBUG ("cannot open gtv file %sn", buffer);
  1200. perror ("fopen");
  1201. return -1;
  1202. }
  1203. fwrite (data, datalen, 1, f);
  1204. fclose (f);
  1205. return 0;
  1206. }
  1207. */
  1209. int
  1210. buildGTV (struct Channel *pc, int datalen, char *data, int type)
  1211. {
  1212. char buffer[MAX_DATA];
  1213. FILE *f;
  1214. assert (pc->pcinfo);
  1215. snprintf (buffer, MAX_LINE, "%s/%s/%s.gtv", WWW_ROOT, NET_NAME[type], pc->channel_name);
  1216. // snprintf (buffer, MAX_LINE, "%s/%s.gtv", pc->fname, pc->channel_name);
  1217. if ((f = fopen (buffer, "w")) == NULL)
  1218. {
  1219. PDEBUG ("cannot open gtv file %sn", buffer);
  1220. perror ("fopen");
  1221. return -1;
  1222. }
  1223. if (pc->pcinfo)
  1224. {
  1225. if (pc->pcinfo->mlist != NULL)
  1226. sprintf (buffer,
  1227.  "CSUserID=%drnBlockSize=%drnBitRate=%frnChannelName=%srnPlaylist=truernResourceHash=%srnTrackServer=%srnSuperPeer=%s:50001rnDataLength=%drnData=",
  1228.  pc->pcinfo->userid, pc->maxblocksize, pc->pcinfo->bitrate,
  1229.  pc->channel_name, pc->channel_md5, tsip[type], spip[type],
  1230.  datalen);
  1231. else
  1232. sprintf (buffer,
  1233.  "CSUserID=%drnBlockSize=%drnBitRate=%frnChannelName=%srnResourceHash=%srnTrackServer=%srnSuperPeer=%s:50001rnDataLength=%drnData=",
  1234.  pc->pcinfo->userid, pc->maxblocksize, pc->pcinfo->bitrate,
  1235.  pc->channel_name, pc->channel_md5, tsip[type], spip[type],
  1236.  datalen);
  1237. } else
  1238. sprintf (buffer,
  1239.  "GTVHome=%s/rnBlockSize=%drnResourceHash=%srnTrackServer=%srnSuperPeer=%s:50001rnDataLength=%drnData=",
  1240.  urlroot, pc->maxblocksize, pc->channel_md5,
  1241.  tsip[type], spip[type], datalen);
  1242. fwrite (buffer, strlen (buffer), 1, f);
  1243. fwrite (data, datalen, 1, f);
  1244. fclose (f);
  1245. /* snprintf (buffer, MAX_LINE, "%s/%s/%s.mediadata", WWW_ROOT, NET_NAME[type], pc->channel_name);
  1246. if (stat (buffer, &stbuf) == 0)
  1247. {
  1248. if (stbuf.st_size != datalen)
  1249. {
  1250. PDEBUG ("old media data size %d not match new %dn", (int)(stbuf.st_size), datalen);
  1251. return -1;
  1252. }
  1253. if ((f = fopen (buffer, "r")) == NULL)
  1254. {
  1255. PDEBUG ("cannot open mediadata file %sn", buffer);
  1256. perror ("fopen");
  1257. return -1;
  1258. }
  1259. if (fread (olddata, datalen, 1, f) == 1)
  1260. {
  1261. for (i=0; i<datalen; i++)
  1262. {
  1263. if (data[i] != olddata[i])
  1264. {
  1265. PDEBUG ("media data not match %s, %dn", buffer, i);
  1266. fclose (f);
  1267. return -1;
  1268. }
  1269. }
  1270. } else
  1271. {
  1272. PDEBUG ("Error in read old mediadata %s.n", buffer);
  1273. fclose (f);
  1274. return -1;
  1275. }
  1276. fclose (f);
  1277. return 0;
  1278. } else if ((f = fopen (buffer, "w")) == NULL)
  1279. {
  1280. PDEBUG ("cannot open gtv file %sn", buffer);
  1281. perror ("fopen");
  1282. return -1;
  1283. }
  1284. fwrite (data, datalen, 1, f);
  1285. fclose (f);
  1286. */
  1287.     // added by lixingwu, 20070313
  1288.     // upload gtv files to server
  1289.     char upload_cmd[MAX_DATA];
  1290.     sprintf(upload_cmd, "curl -F filename=@%s/%s/%s.gtv -F domain=%s %s",
  1291.         WWW_ROOT, NET_NAME[type], pc->channel_name, spip[type], CAS_ADDR);
  1292.     printf("%sn", upload_cmd);
  1293.     system(upload_cmd);
  1294.     
  1295.     return 0;
  1296. }
  1297. #endif
  1299. int
  1300. send_P2P_SPUPDATE (struct Session *p, struct Channel *pc, char *md5, struct SPUpdate *s)
  1301. {
  1302. char buffer1[MAX_DATA];
  1303. char *buf;
  1304. buf = buffer1 + sizeof (int);
  1305. *(unsigned char *) buf = P2P_SPUPDATE; //P2P_SPUPDATE=1 in ProTocol file
  1306. buf += sizeof (char);
  1307. memcpy (buf, md5, MD5_LEN);
  1308. buf += MD5_LEN;
  1309. memcpy (buf, s, sizeof (struct SPUpdate));
  1310. buf += sizeof (struct SPUpdate);
  1311. *(int *) buffer1 = buf - buffer1;
  1312. if (writeMessage (p, pc, buffer1) < 0)
  1313. return -1;
  1314. return 0;
  1315. }
  1317. int
  1318. send_p2p_err (struct Session *p, unsigned short code, int quit)
  1319. {
  1320. char buffer1[MAX_DATA];
  1321. char *buf;
  1322. buf = buffer1 + sizeof (int);
  1323. *(unsigned char *) buf = P2P_MSG;
  1324. buf += sizeof (char);
  1325. (*(unsigned short *) buf) = code;
  1326. buf += sizeof (short);
  1327. *(int *) buf = quit;
  1328. buf += sizeof (int);
  1329. *(int *) buffer1 = buf - buffer1;
  1330. PDEBUG ("Send error msg type %hd to %pn", code, p);
  1331. if (writeMessage (p, NULL, buffer1) < 0)
  1332. return -1;
  1333. return 0;
  1334. }
  1336. void apply_idle (struct Session *p, void *arg)
  1337. {
  1338. struct Argument *parg = (struct Argument *)arg;
  1340. if (CurrentTime - p->last_transferblock >= MAX_TRANSFER_IDLE)
  1341. {
  1342. fprintf (parg->f, "%s Session timeout! %ld n", parg->type == TYPE_CS? "CS":"CP", CurrentTime - p->last_transferblock);
  1343. Clientclosure (p - TRACKER[parg->type].head, parg->type);
  1344. }
  1345. }
  1347. void apply_check (struct Channel *p, void *arg)
  1348. {
  1349. struct Argument *parg = (struct Argument *)arg;
  1350. int type;
  1351. struct LiveChannelInfo *pc = p->pcinfo;
  1352. if (pc && pc->status <= 0)
  1353. {
  1354. #ifdef HAVE_MYSQL
  1355. query_mysql (local_mysql,
  1356.      "insert into channel(ChannelName, ChannelBitrate, ChannelMD5, ChannelElapsed, ChannelRange) values ('%s','%f','%d','%d')",
  1357.      pc->channel_name, pc->bitrate,
  1358.      p->channel_md5,
  1359.      time (NULL) - p->ctime,
  1360.      pc->s.maxKeySample -
  1361.      pc->s.minKeySample);
  1362. #endif
  1363. #ifdef TEST
  1364. for (type=0; type<MAX_TS; type++)
  1365. {
  1366. if (parg->xml[type] == NULL)
  1367. continue;
  1368. fprintf (parg->xml[type], "<Channel Name="%s" Desc="%s" File="%s.gtv" NumClient="%d" BitRate="%d" Start="%ld" End="-1" Elapsed="%ld"/>rn",
  1369.  p->channel_name, "gtv", //pc->userid,
  1370.  p->channel_name, p->numofnp[type],
  1371.  (int) (pc->bitrate * 8),
  1372.  (long) p->ctime, (long) CurrentTime);
  1373. }
  1374. #endif
  1375. PINFO ("query %s->%s n",
  1376. p->channel_name, p->channel_md5);
  1377. memcpy (parg->buf, p->channel_md5, MD5_LEN);
  1378. parg->buf += MD5_LEN;
  1379. fprintf (parg->f, "Channel %s have %d client. upsize %lldB, avg speed %f; downsize %lldB, avg speed %f, reported %f, real/reported is %f%%.n",
  1380.  p->channel_md5, p->numclient,
  1381.  p->upsize, ((float)(p->upsize)) / (CurrentTime - lastCheck), p->downsize, ((float)(p->downsize)) / (CurrentTime - lastCheck), p->pcinfo !=NULL ? p->pcinfo->bitrate:0, p->pcinfo != NULL && p->pcinfo->bitrate != 0 ? (((float)(p->downsize)) / (CurrentTime - lastCheck)*100/p->pcinfo->bitrate): 0);
  1382. fprintf (parg->f, "Live SPUpdate : SampleMin %lld SampleMax %lld BlockMin %d BlockMax %d n",
  1383.  pc->s.minKeySample,
  1384.  pc->s.maxKeySample,
  1385.  pc->s.minBlockID,
  1386.  pc->s.maxBlockID);
  1387. parg->spchannelcount ++;
  1388. parg->totalclient += p->numclient;
  1389. parg->totalupsize += p->upsize;
  1390. } else if (pc && pc->isSave == 0)
  1391. freeLiveChannel (p, NULL);
  1392. }