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

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

P2P编程

开发平台：

Visual C++

livechannel.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. int NumNewChannel;
  22. struct Channel *ChannelHash[MAX_CHANNEL];
  23. struct Channel *ChannelList;
  24. extern char *PREFIX;
  25. extern struct ServerDesc TRACKER[MAX_TYPE];
  26. extern int Clientclosure (int listnum, int type);
  28. inline int hash_str (unsigned char *str, int len)
  29. {
  30. int hash;
  31. for (hash=0; len; len--, str++)
  32. hash += (hash << 5) - hash + (*str);
  33. return hash & (MAX_CHANNEL - 1);
  34. }
  36. struct Channel *getChannel (struct Channel **phash, char *name, int len)
  37. {
  38. int id = hash_str (name, len);
  39. struct Channel *p;
  40. for (p=phash[id]; p; p=p->next)
  41. {
  42. if (strncmp (name, p->channel_md5, len) == 0)
  43. return p;
  44. }
  45. PDEBUG("Cannot findChannel hash %.32s.n", name);
  46. return NULL;
  47. }
  49. void apply_hash (struct Channel **phash, void apply(struct Channel *, void *), void *p)
  50. {
  51. int i;
  52. struct Channel *pc, *nextpc;
  53. for (i=0; i<MAX_CHANNEL; i++)
  54. {
  55. for (pc=phash[i]; pc; pc=nextpc)
  56. {
  57. nextpc = pc->next;
  58. apply (pc, p);
  59. }
  60. }
  61. }
  63. void apply_list (struct Channel *plist, void apply(struct Channel *, void *), void *p)
  64. {
  65. struct Channel *pc, *nextpc;
  66. for (pc=plist; pc; pc=nextpc)
  67. {
  68. nextpc = pc->lnext;
  69. apply (pc, p);
  70. }
  71. }
  73. int freeChannel (struct Channel **phash, struct Channel **plist, int *count, struct Channel *p)
  74. {
  75. int id = hash_str (p->channel_md5, MD5_LEN);
  76. struct Edge *nextedge, *pedge;
  77. struct Channel *pchannel;
  78. if (phash[id] == p)
  79. {
  80. phash[id] = p->next;
  81. } else
  82. {
  83. for (pchannel=phash[id]; pchannel; pchannel=pchannel->next)
  84. {
  85. if (pchannel->next == p)
  86. {
  87. pchannel->next = p->next;
  88. break;
  89. }
  90. }
  91. if (!pchannel) return -1;
  92. }
  93. if ((*plist) == p)
  94. {
  95. *plist = p->lnext;
  96. } else
  97. {
  98. for (pchannel=*plist; pchannel; pchannel=pchannel->lnext)
  99. {
  100. if (pchannel->lnext == p)
  101. {
  102. pchannel->lnext = p->lnext;
  103. break;
  104. }
  105. }
  106. if (!pchannel) return -1;
  107. }
  108. (*count) --;
  109. #ifdef __CP_SOURCE
  110. if (p->db != NULL) fclose (p->db);
  111. #endif
  112. if (p->pcinfo)
  113. {
  114. free_livechannel (p);
  115. free (p->pcinfo);
  116. }
  117. for (pedge=p->PeerHead; pedge; pedge=nextedge)
  118. {
  119. p->numclient --;
  120. nextedge = pedge->cnext;
  121. delEdge (pedge);
  122. }
  123. free (p);
  124. return 0;
  125. }
  127. inline struct Channel *findChannel (char *name, int len)
  128. {
  129. return getChannel (ChannelHash, name, len);
  130. }
  132. void freeLiveChannel (struct Channel *pc, void *p)
  133. {
  134. freeChannel (ChannelHash, &ChannelList, &NumNewChannel, pc);
  135. }
  137. void freeAllChannel ()
  138. {
  139. apply_hash (ChannelHash, freeLiveChannel, NULL);
  140. }
  142. inline void buildLivePath (char *buf, int len, char *md5)
  143. {
  144. snprintf (buf, len, "%s/%s%.2s/", PREFIX, LIVE_PREFIX, md5);
  145. mkdir (buf, 0777);
  146. strcat (buf, md5);
  147. }
  150. #ifdef __CP_SOURCE
  151. int locate_by_id (struct Channel *pc, unsigned int id, char *buf, int max)
  152. {
  153. int i, *msg;
  154. struct LiveChannelInfo *c = pc->pcinfo;
  155. if (c == NULL)
  156. return -1;
  157. if (c->indisk == NULL)
  158. {
  159. c->max_queue = MAX_QUEUE;
  160. c->indisk = calloc (c->max_queue, 1);
  161. c->bitflag = calloc ((c->max_queue+7)/8, 1);
  162. // if (c->indisk == NULL || c->bitflag == NULL)
  163. PDEBUG ("allocate memory,%p,%pn",c->indisk, c->bitflag);
  164. return -1;
  165. }
  166. i = id % c->max_queue;
  167. if (c->indisk[i] == 0 || (pc->type != T_PLIST && c->indisk[i] != (id/c->max_queue + 1)))
  168. {
  169. PINFO ("empty flag %d.n", c->indisk[i]);
  170. return -1;
  171. }
  173. if (pc->maxblocksize + 2*sizeof(int) > max)
  174. {
  175. PDEBUG ("too small buffer %d for %d", max, pc->maxblocksize);
  176. return -2;
  177. }
  179. if (fseeko (pc->db, ((off_t)(i)) * (pc->maxblocksize + 2*sizeof (int)), SEEK_SET) != 0)
  180. {
  181. PDEBUG ("fseek failed.n");
  182. return -1;
  183. }
  184. if ((i=fread (buf, 1, pc->maxblocksize+2*sizeof (int), pc->db)) <= 2*sizeof (int) || i < ((int *)buf)[1]+2*sizeof(int))
  185. {
  186. PDEBUG ("Only %d read [%d]n", i, ((int *)buf)[1]);
  187. return -1;
  188. }
  189. msg = (int *)buf;
  190. if (pc->type == T_PLIST)
  191. msg[0] = id;
  192. if (msg[0] != id || msg[1] > pc->maxblocksize || msg[1] <= 0)
  193. {
  194. PDEBUG ("Message read is [%d %d]n", msg[0], msg[1]);
  195. return -1;
  196. }
  197. pc->upsize += msg[1];
  198. return msg[1];
  199. }
  201. int saveBlock (struct Channel *c, char *buf, struct Session *p)
  202. {
  203. unsigned int pos, id, size;
  204. struct LiveChannelInfo *pcinfo;
  206. assert (buf);
  207. if ((!c) || (pcinfo = c->pcinfo) == NULL)
  208. {
  209. PDEBUG ("saveBlock c is null.n");
  210. return -1;
  211. }
  212. if (pcinfo->indisk == NULL)
  213. {
  214. pcinfo->max_queue = MAX_QUEUE;
  215. pcinfo->indisk = calloc (pcinfo->max_queue, 1);
  216. pcinfo->bitflag = calloc ((pcinfo->max_queue+7)/8, 1);
  217. if (pcinfo->indisk == NULL || pcinfo->bitflag == NULL)
  218. return -1;
  219. }
  220. id = ((unsigned int *)buf)[0];
  221. size = ((unsigned int *)buf)[1];
  222. pos = id % pcinfo->max_queue;
  223. if (id > 0) clrBit (pcinfo->bitflag, pos);
  224. if (size > MAX_BLOCK_SIZE || size <= MIN_BLOCK_SIZE || id < 0)
  225. {
  226. PDEBUG ("saveBlock:size is %d and id is %d.n", size, id);
  227. return 0;
  228. }
  230. if (c->maxblocksize == 0)
  231. c->maxblocksize = size;
  232. else if (size > c->maxblocksize)
  233. {
  234. PDEBUG ("saveBlock:maxblocksize is %d, size is %d and id is %d.n", c->maxblocksize, size, id);
  235. return 0;
  236. }
  237. // PDEBUG ("Recv %d(%d) Save2 %d, dataSource %p and now is %pn", (int)id, (int)size, (int)pos, pcinfo->dataSource, p);
  239. if (fseeko (c->db, ((off_t)(pos)) * (c->maxblocksize + 2*sizeof (int)), SEEK_SET) != 0)
  240. {
  241. PDEBUG ("Error in fsseko.n");
  242. return -1;
  243. }
  245. if (fwrite (buf, size+2*sizeof(int), 1, c->db) != 1)
  246. {
  247. PDEBUG ("fwrite error in saveBlock:%.32s:%s", c->channel_md5, c->fname);
  248. return -1;
  249. }
  251. if (pcinfo->dataSource != p)
  252. {
  253. PDEBUG ("dataSource %p is not equal p %p.n", pcinfo->dataSource, p);
  254. pcinfo->dataSource = p;
  255. }
  257. fflush (c->db);
  258. pcinfo->total ++;
  259. c->downsize += size;
  260. if (id > pcinfo->maxID) pcinfo->maxID = id;
  261. pcinfo->indisk[pos] = (id/pcinfo->max_queue) + 1;
  262. if (pcinfo->indisk[pos] == 0)
  263. PDEBUG("Too large id %ds", id);
  264. return size;
  265. }
  267. int init_livechannel (struct Channel *p)
  268. {
  269. if ((p->db = fopen (p->fname, "w+")) == (FILE *)0)
  270. return -1;
  271. p->pcinfo->isSave = 0;
  272. return 0;
  273. }
  275. int free_livechannel (struct Channel *p)
  276. {
  277. struct LiveChannelInfo *pcinfo = p->pcinfo;
  278. free (pcinfo->indisk);
  279. free (pcinfo->bitflag);
  280. unlink (p->fname);
  281. if (pcinfo->dataSource != NULL)
  282. {
  283. pcinfo->dataSource->pc = NULL;
  284. PDEBUG ("close CS source for channel %.32s.n", p->channel_md5);
  285. Clientclosure (pcinfo->dataSource-TRACKER[TYPE_P2PC].head, TYPE_P2PC);
  286. pcinfo->dataSource = NULL;
  287. }
  288. freeMedia (p);
  289. return 0;
  290. }
  291. #endif
  293. #ifdef __SP_SOURCE
  294. extern time_t CurrentTime;
  295. extern char *NET_NAME[];
  296. extern char *WWW_ROOT;
  297. extern char *defaultspip;
  298. int Changed = 1;
  299. extern int timer_add (unsigned int t, TimerFunc process, void *entity, void *data);
  300. extern int writeMessage (struct Session *p, struct Channel *pc, char *ptr);
  301. extern int buildGTV (struct Channel *pc, int datalen, char *data, int type);
  302. extern inline void freeProgram (struct Channel *, void *);
  303. extern void send_all_spupdate (struct Channel *pc, struct SPUpdate *s);
  305. void hup_handler (int sig)
  306. {
  307. Changed ++;
  308. }
  309. inline void buildPListPath (char *buf, int len, char *md5)
  310. {
  311. snprintf (buf, len, "%s/%s/", PREFIX, PLIST_PREFIX);
  312. strcat (buf, md5);
  313. }
  315. FILE * open_keyfile (struct Channel *p)
  316. {
  317. char buffer[MAX_DATA];
  318. struct stat stbuf;
  319. if (p == NULL) return NULL;
  320. snprintf (buffer, MAX_LINE, "%s/keysample", p->fname);
  321. if (stat (buffer, &stbuf) == 0)
  322. {
  323. if (!S_ISREG (stbuf.st_mode))
  324. {
  325. PDEBUG ("File %s exist and not a regular file", buffer);
  326. return NULL;
  327. }
  328. }
  329. return fopen (buffer, "r");
  330. }
  332. int send_mplist_spupdate (struct Channel *pc, void * data)
  333. {
  334. int iddiff, prev;
  335. char *buf, buffer[MAX_DATA];
  336. struct Edge *pedge;
  337. struct LiveChannelInfo *pcinfo;
  338. struct SPUpdate s;
  339. struct logrec lrec;
  340. time_t slot;
  342. if (pc == NULL || pc->pcinfo == NULL || pc->pcinfo->mlist == NULL
  343. || pc->pcinfo->maxID == 0)
  344. {
  345. PDEBUG ("Wrong playlist data.n");
  346. return -1;
  347. }
  348. pcinfo = pc->pcinfo;
  349. if (pcinfo->s.maxBlockID == 0)
  350. {
  351. fseek (pcinfo->keyfile, 0, SEEK_SET);
  352. if (fread (&lrec, 1, sizeof(struct logrec), pcinfo->keyfile) != sizeof (struct logrec))
  353. {
  354. PDEBUG ("Cannot read keysamplen");
  355. return -1;
  356. }
  357. pcinfo->s.minBlockID = (((CurrentTime - FIX_MAGIC ) * 16 + pcinfo->maxID - 1) / pcinfo->maxID) * pcinfo->maxID;
  358. pcinfo->s.maxBlockID = lrec.id;
  359. if (pcinfo->s.maxKeySample == 0)
  360. pcinfo->s.maxKeySample = lrec.keysample;
  361. if (pcinfo->s.minKeySample == 0)
  362. pcinfo->s.minKeySample = CurrentTime;
  363. // pcinfo->s.minKeySample = lrec.keysample;
  364. }
  366. s = pcinfo->s;
  367. s.maxBlockID += s.minBlockID;
  368. s.maxKeySample = ((long long)CurrentTime) * 10000000;
  369. s.minKeySample = ((long long)(s.minKeySample)) * 10000000;
  371. if (pcinfo->updated < CurrentTime && pc->numclient > 0)
  372. {
  373. pcinfo->updated = CurrentTime;
  374. buf = buffer + sizeof (int);
  375. *(unsigned char *) buf = P2P_SPUPDATE;
  376. buf += sizeof (char);
  377. memcpy (buf, pc->channel_md5, MD5_LEN);
  378. buf += MD5_LEN;
  379. memcpy (buf, &s, sizeof (struct SPUpdate));
  380. buf += sizeof (struct SPUpdate);
  381. *(int *) buffer = buf - buffer;
  382. for (pedge = pc->PeerHead; pedge; pedge = pedge->cnext)
  383. {
  384. if (writeMessage (pedge->me, pc, buffer) < 0)
  385. {
  386. PDEBUG ("send SPUPDATE err.n");
  387. }
  388. }
  389. PINFO ("Send spupdate in %d, %d.n", (int)(pcinfo->s.maxKeySample), (int)(pcinfo->s.minKeySample));
  390. }
  392. if (fread (&lrec, 1, sizeof (struct logrec), pcinfo->keyfile) != sizeof (struct logrec))
  393. {
  394. fclose (pcinfo->keyfile);
  395. prev = pcinfo->mlist->m_cursampleid;
  396. pcinfo->mlist->m_cursampleid ++;
  397. if (pcinfo->mlist->m_cursampleid >= pcinfo->mlist->m_totalchannel)
  398. pcinfo->mlist->m_cursampleid = 0;
  399. if ((pcinfo->keyfile = open_keyfile (pcinfo->mlist->m_lists[pcinfo->mlist->m_cursampleid])) == NULL)
  400. {
  401. PDEBUG ("Error in new keysample filen");
  402. return -1;
  403. }
  404. fseek (pcinfo->keyfile, SEEK_SET, 0);
  405. if (fread (&lrec, 1, sizeof (struct logrec), pcinfo->keyfile) != sizeof (struct logrec))
  406. {
  407. PDEBUG ("Cannot read keysamplen");
  408. return -1;
  409. }
  410. iddiff = pcinfo->mlist->m_lists[prev]->pcinfo->maxID + lrec.id - pcinfo->mlist->m_lastmaxid;
  411. slot = lrec.keysample;
  412. pcinfo->s.maxKeySample = lrec.keysample;
  413. } else
  414. {
  415. if ((slot = lrec.keysample - pcinfo->s.maxKeySample) < 0)
  416. slot = 0;
  417. iddiff = lrec.id - pcinfo->mlist->m_lastmaxid;
  418. pcinfo->s.maxKeySample += slot;
  419. }
  420. pcinfo->mlist->m_lastmaxid = lrec.id;
  421. pcinfo->s.maxBlockID += iddiff;
  422. timer_add (CurrentTime+slot, (TimerFunc)send_mplist_spupdate, pc, NULL);
  423. return 0;
  424. }
  426. int init_mplistchannel (struct Channel *p)
  427. {
  428. struct LiveChannelInfo *pcinfo = p->pcinfo;
  429. if (p->maxblocksize == 0) p->maxblocksize = DEFAULT_BLOCK;
  430. pcinfo->max_queue = BLOCK_PER_FILE;
  431. if ((pcinfo->keyfile = open_keyfile (pcinfo->mlist->m_lists[0])) == NULL)
  432. {
  433. PDEBUG ("Error in new keysample filen");
  434. return -1;
  435. }
  436. pcinfo->total = 0;
  437. pcinfo->isSave = 0;
  438. timer_add (CurrentTime, (TimerFunc)send_mplist_spupdate, p, NULL);
  439. return 0;
  440. }
  443. struct Channel *newMPListChannel (char *name, char *cmd5, float bitrate, int maxblocksize, int nchannel, struct Channel **pchannel)
  444. {
  445. int i, id, startID=0;
  446. struct Channel *p;
  447. struct LiveChannelInfo *pcinfo;
  448. if (NumNewChannel >= MAX_CHANNEL) return (struct Channel *)0;
  449. p = (struct Channel *)calloc (sizeof (struct Channel), 1);
  450. memcpy (p->channel_md5, cmd5, MD5_LEN);
  451. if (name) strncpy (p->channel_name, name, sizeof (p->channel_name));
  452. p->channel_md5[MD5_LEN] = 0;
  453. p->upsize = 0;
  454. p->downsize = 0;
  455. p->maxblocksize = maxblocksize;
  456. p->ctime = time (NULL);
  458. p->pcinfo = (struct LiveChannelInfo *)calloc (sizeof (struct LiveChannelInfo), 1);
  459. pcinfo = p->pcinfo;
  460. pcinfo->bitrate = bitrate;
  461. pcinfo->mlist = (struct MList *)calloc (sizeof (struct MList), 1);
  462. pcinfo->mlist->m_totalchannel = nchannel;
  463. pcinfo->media = calloc (nchannel, sizeof (struct MediaData));
  465. pcinfo->max_channel = nchannel;
  466. for (i=0; i<nchannel; i++)
  467. {
  468. pchannel[i]->ref ++;
  469. pcinfo->mlist->m_lists[i] = pchannel[i];
  470. pcinfo->mlist->m_startID[i] = startID;
  471. pcinfo->maxID += pchannel[i]->pcinfo->maxID;
  472. addMedia (p, startID, pchannel[i]->pcinfo->maxID, pchannel[i]->pcinfo->media[0].dlen, pchannel[i]->pcinfo->media[0].data, pchannel[i]->channel_name);
  473. startID += pchannel[i]->pcinfo->maxID;
  474. }
  475. if (init_mplistchannel (p) < 0)
  476. {
  477. PDEBUG ("newPlistChannel error for %p.", p);
  478. free_livechannel (p);
  479. free (pcinfo);
  480. free (p);
  481. return (struct Channel *)0;
  482. }
  484. id = hash_str (p->channel_md5, MD5_LEN);
  485. PDEBUG("newMPlistChannel hash %.32s(fname=%s) to %d.n", p->channel_md5, p->fname, id);
  486. p->next = ChannelHash[id];
  487. ChannelHash[id] = p;
  488. p->lnext = ChannelList;
  489. ChannelList = p;
  490. NumNewChannel ++;
  491. return p;
  492. }
  494. struct Channel *add_mplist_channel (char *buffer, char *md5)
  495. {
  496. int i=0;
  497. FILE *in;
  498. struct stat stbuf;
  499. struct Channel *pc=NULL, *pchannel[MAX_FILEINPUT];
  500. char *data=NULL, cname[MAX_DATA], buf[MAX_DATA];
  501. float bitrate=0.0;
  502. int bsize=16384, dlen=0;
  503. if (stat (buffer, &stbuf) < 0) return NULL;
  504. if (S_ISREG (stbuf.st_mode))
  505. {
  506. if ((in=fopen (buffer, "r")) == NULL)
  507. {
  508. PDEBUG ("Error in open file %s.n", buffer);
  509. return NULL;
  510. }
  511. while (fgets (buf, MAX_DATA, in))
  512. {
  513. switch (i)
  514. {
  515. case 0:
  516. bitrate = atof (buf);
  517. i++;
  518. break;
  519. case 1:
  520. if (buf[strlen(buf)-1] == 'r' || buf[strlen(buf)-1] == 'n')
  521. buf[strlen(buf)-1] = 0;
  522. strncpy (cname, buf, sizeof(cname));
  523. i++;
  524. break;
  525. default:
  526. buf[MD5_LEN] = 0;
  527. if ((pc = getProgrambymd5 (buf, MD5_LEN)) != NULL)
  528. {
  529. pchannel[i-2] = pc;
  530. i++;
  531. }
  532. break;
  533. }
  534. }
  535. fclose (in);
  536. if (i <= 2) return NULL;
  537. dlen = pchannel[0]->pcinfo->media[0].dlen;
  538. data = pchannel[0]->pcinfo->media[0].data;
  539. // snprintf (chname, MAX_DATA, "%s/%s/%s.mediadata", WWW_ROOT, NET_NAME[0], pchannel[0]->channel_name);
  540. // data = read_file (chname, &dlen);
  541. if ((pc=newMPListChannel (cname, md5, bitrate, bsize, i-2, pchannel)) == NULL)
  542. {
  543. PDEBUG ("Error in newMPListChannel %s.n", md5);
  544. // free (data);
  545. return NULL;
  546. }
  547. if (pc != NULL)
  548. strncpy (pc->fname, buffer, CHNLURL_LEN);
  549. #ifdef TEST
  550. for (i=0; i<MAX_TS; i++) {
  551. if (dlen <= 0 || data == NULL || buildGTV (pc, dlen, data, i) < 0)
  552. continue;
  553. }
  554. #endif
  555. // free (data);
  556. }
  557. return pc;
  558. }
  560. void apply_update (struct Channel *p, void *arg)
  561. {
  562. struct stat stbuf;
  563. // char buffer[MAX_DATA];
  564. struct LiveChannelInfo *pc = p->pcinfo;
  565. if (pc && pc->mlist != NULL)
  566. {
  567. if (stat (p->fname, &stbuf) != 0)
  568. pc->status = 1;
  569. }
  570. }
  573. int check_newplist ()
  574. {
  575. int i;
  576. struct Channel *pc;
  577. DIR *entry;
  578. struct dirent *pd;
  579. char buffer[MAX_DATA];
  580. char tmp0[MD5_LEN+1], tmp[MD5_LEN+1];
  581. // if (Changed == 0) return 0;
  582. snprintf (buffer, MAX_DATA, "%s/%s/", PREFIX, PLIST_PREFIX);
  583. if ((entry = opendir (buffer)) == NULL)
  584. {
  585. return -1;
  586. }
  587. while ((pd = readdir (entry)) != NULL)
  588. {
  589. if (strcmp (pd->d_name, ".") == 0 || strcmp (pd->d_name, "..") == 0 || strlen (pd->d_name) != MD5_LEN)
  590. continue;
  591. snprintf (buffer, MAX_DATA, "%s_%s", defaultspip, pd->d_name);
  592. md5_calc ((unsigned char *) tmp0,
  593. (unsigned char *) buffer, strlen (buffer));
  594. for (i = 0; i < MD5_LEN; i += 2)
  595. sprintf (tmp + i, "%02x", (unsigned char) tmp0[i / 2]);
  596. tmp[MD5_LEN] = 0;
  597. if ((pc = findChannel (tmp, strlen (tmp))) != NULL)
  598. {
  599. if (pc->pcinfo != NULL) pc->pcinfo->status = 0;
  600. continue;
  601. }
  602. snprintf (buffer, MAX_DATA, "%s/%s/%s", PREFIX, PLIST_PREFIX, pd->d_name);
  603. add_mplist_channel (buffer, tmp);
  604. }
  605. closedir (entry);
  606. apply_list (ChannelList, apply_update, NULL);
  607. Changed = 0;
  608. return 0;
  609. }
  611. int locate_mplist_by_id (struct Channel *pc, unsigned int id, char *buf, int max)
  612. {
  613. int result;
  614. int i, j;
  615. struct LiveChannelInfo *c = pc->pcinfo;
  616. if (c == NULL)
  617. {
  618. PDEBUG ("c is null.n");
  619. return -1;
  620. }
  621. i = id % c->maxID;
  622. for (j=0; j<pc->pcinfo->mlist->m_totalchannel; j++)
  623. if (pc->pcinfo->mlist->m_startID[j] > i) break;
  624. j--;
  625. if (j < 0)
  626. {
  627. PDEBUG ("Internal error in locate id %d.n", id);
  628. return -1;
  629. }
  630. if ((result = locateprog_by_id (pc->pcinfo->mlist->m_lists[j], i-pc->pcinfo->mlist->m_startID[j], buf, max)) > 0)
  631. ((int *)buf)[0] = id;
  632. return result;
  633. }
  635. // Return 1 to indicate write available, return 0 to indicate now writable.
  636. inline int newChannelFile (struct Channel *p)
  637. {
  638. int result;
  639. struct stat stbuf;
  640. char buffer[MAX_LINE];
  641. struct LiveChannelInfo *pcinfo = p->pcinfo;
  642. if (pcinfo->numinput >= MAX_FILEINPUT)
  643. {
  644. PDEBUG ("Max file input has been reached, %s:%d.n", p->fname,pcinfo->numinput);
  645. return -1;
  646. }
  647. snprintf (buffer, MAX_LINE, "%s/%d", p->fname, pcinfo->numinput);
  648. if (stat (buffer, &stbuf) == 0)
  649. {
  650. if (stbuf.st_size / (p->maxblocksize+2*sizeof (int)) < pcinfo->max_queue)
  651. {
  652. if ((pcinfo->input[pcinfo->numinput] = fopen (buffer, "a+")) == NULL)
  653. {
  654. PDEBUG ("Cannot open file %s.n", buffer);
  655. return -1;
  656. }
  657. result = 1;
  658. } else
  659. {
  660. if ((pcinfo->input[pcinfo->numinput] = fopen (buffer, "r+")) == NULL)
  661. {
  662. PDEBUG ("Cannot open file %s.n", buffer);
  663. return -1;
  664. }
  665. result = 0;
  666. }
  667. pcinfo->numblocks = stbuf.st_size /(p->maxblocksize+2*sizeof (int));
  668. pcinfo->maxID += stbuf.st_size / (p->maxblocksize+2*sizeof(int));
  669. if (stbuf.st_size % (p->maxblocksize + 2*sizeof(int))) return -1;
  670. } else
  671. {
  672. if ((pcinfo->input[pcinfo->numinput] = fopen (buffer, "w+")) == NULL)
  673. {
  674. PDEBUG ("Cannot open file %s.n", buffer);
  675. return -1;
  676. }
  677. pcinfo->numblocks = 0;
  678. result = 1;
  679. }
  680. pcinfo->numinput ++;
  681. return result;
  682. }
  684. int init_livechannel (struct Channel *p)
  685. {
  686. int i;
  687. struct stat stbuf;
  688. char buffer[MAX_LINE];
  689. struct LiveChannelInfo *pcinfo = p->pcinfo;
  690. if (stat (p->fname, &stbuf) == 0 && (!S_ISDIR (stbuf.st_mode)))
  691. {
  692. PDEBUG ("directory %s not exist or not a dir.n", p->fname);
  693. return -1;
  694. }
  695. mkdir (p->fname, 0777);
  696. if (p->maxblocksize == 0) p->maxblocksize = DEFAULT_BLOCK;
  697. pcinfo->max_queue = BLOCK_PER_FILE;
  698. while ((i = newChannelFile (p)) == 0);
  699. if (i < 0) return -1;
  700. p->db = pcinfo->input[pcinfo->numinput-1];
  701. snprintf (buffer, MAX_LINE, "%s/keysample", p->fname);
  702. if (stat (buffer, &stbuf) == 0)
  703. {
  704. if (!S_ISREG (stbuf.st_mode))
  705. {
  706. PDEBUG ("File %s exist and not a regular file", buffer);
  707. return -1;
  708. }
  709. }
  710. pcinfo->keyfile = fopen (buffer, "a+");
  711. if (pcinfo->keyfile == NULL)
  712. {
  713. PDEBUG ("File %s can not be opened.n", buffer);
  714. return -1;
  715. }
  716. pcinfo->total = 0;
  717. return 0;
  718. }
  721. int locate_by_id (struct Channel *pc, unsigned int id, char *buf, int max)
  722. {
  723. int i, pos, *msg;
  724. struct LiveChannelInfo *c = pc->pcinfo;
  725. if (c == NULL || id > c->maxID)
  726. {
  727. PDEBUG ("c is %p and id is (%d,%d).n", c, c->maxID, id);
  728. return -1;
  729. }
  730. if (pc->maxblocksize + 2*sizeof(int) > max)
  731. {
  732. PDEBUG ("too small buffer %d for %d", max, pc->maxblocksize);
  733. return -2;
  734. }
  735. i = (id - c->startid) % c->max_queue;
  736. pos = (id - c->startid) / c->max_queue;
  737. if (pos >= c->numinput || c->input[pos] == NULL)
  738. {
  739. PDEBUG ("file %d does not exist. (%d,%p)n", pos, c->numinput, c->input[pos]);
  740. return -1;
  741. }
  743. if (fseeko (c->input[pos], ((off_t)(i)) * (pc->maxblocksize + 2*sizeof (int)), SEEK_SET) != 0)
  744. {
  745. PDEBUG ("Fssek failed. (%d, %d, %d)n", pos, i, pc->maxblocksize);
  746. return -1;
  747. }
  748. if ((i=fread (buf, 1, pc->maxblocksize+2*sizeof (int), c->input[pos])) <= 2*sizeof (int) || i < ((int *)buf)[1]+2*sizeof(int))
  749. {
  750. PDEBUG ("Fread failed. (%d, %d, %d)n", pos, i, pc->maxblocksize);
  751. return -1;
  752. }
  753. msg = (int *)buf;
  754. if (msg[1] > pc->maxblocksize || msg[1] <= 0)
  755. {
  756. PDEBUG ("msg format error. (%d, %d, %d)n", msg[0], msg[1], pc->maxblocksize);
  757. return -1;
  758. }
  759. PINFO ("Found block. (%d, %d, %d, %d)n", id, msg[0], msg[1], pc->maxblocksize);
  760. // msg[0] = id;
  761. pc->upsize += msg[1];
  762. return msg[1];
  763. }
  765. int saveBlock (struct Channel *c, char *buf, struct Session *p)
  766. {
  767. struct logrec lrec;
  768. int id, size;
  769. int j=0;
  770. unsigned long long keysample;
  771. struct SPUpdate *s;
  772. struct LiveChannelInfo *pcinfo;
  774. if ((!c) || (pcinfo = c->pcinfo) == NULL)
  775. {
  776. PDEBUG ("saveBlock c is null.n");
  777. return -1;
  778. }
  779. assert (buf);
  780. size = ((int *)buf)[1];
  781. if (size > MAX_BLOCK_SIZE || size <= MIN_BLOCK_SIZE)
  782. {
  783. PDEBUG ("saveBlock size is %d and id is %d.n", size, id);
  784. return 0;
  785. }
  787. if (c->maxblocksize == 0)
  788. {
  789. c->maxblocksize = size;
  790. pcinfo->max_queue = BLOCK_PER_FILE;
  791. }
  792. else if (size > c->maxblocksize)
  793. {
  794. PDEBUG ("saveBlock maxblocksize is %d, size is %d and id is %d.n", c->maxblocksize, size, id);
  795. return 0;
  796. }
  797. if (pcinfo->max_queue == 0)
  798. {
  799. PDEBUG ("c->max_queue is 0n");
  800. return -1;
  801. }
  802. // PDEBUG ("Recv %d(%d) Save2 (%d,%d,%d), dataSource %p and now is %pn", ((int*)buf)[0], (int)size, pcinfo->numinput, pcinfo->numblocks, pcinfo->maxID, pcinfo->dataSource, p);
  804. id = pcinfo->maxID;
  805. ((int *)buf)[0] = id;
  806. if (fseeko (c->db, ((off_t)(pcinfo->numblocks)) * (c->maxblocksize + 2*sizeof (int)), SEEK_SET) != 0)
  807. {
  808. PDEBUG ("Error in fsseko.n");
  809. return -1;
  810. }
  812. if (fwrite (buf, c->maxblocksize+2*sizeof(int), 1, c->db) != 1)
  813. {
  814. PDEBUG ("fwrite error in saveBlock:%.32s:%s", c->channel_md5, c->fname);
  815. return -1;
  816. }
  818. j = ((int *)buf)[2];
  819. if (j <= 0 || j >= size-sizeof(keysample)) keysample = 0;
  820. else keysample = *(unsigned long long *) (buf+2*sizeof(int)+j);
  821. s = &(pcinfo->s);
  822. if (keysample > 0)
  823. {
  824. if(s->maxKeySample < keysample)
  825. {
  826. //the increasement should not be larger than 1000 seconds!
  827. if(s->maxKeySample == 0 || (keysample-s->maxKeySample)/10000000 < 1000)
  828. {
  829. s->maxKeySample = keysample;
  830. } else
  831. PDEBUG("Error Keysample at block %d,(%lld,%lld). n", id, keysample, s->maxKeySample);
  832. }
  833. if (s->minKeySample == 0 || keysample < s->minKeySample)
  834. {
  835. s->minKeySample = keysample;
  836. }
  837. lrec.id = id;
  838. lrec.keysample = keysample/10000000;
  839. fwrite (&lrec, sizeof (lrec), 1, pcinfo->keyfile);
  840. fflush (pcinfo->keyfile);
  841. }
  842. if (id < s->minBlockID || s->maxBlockID == 0)
  843. {
  844. s->minBlockID = id;
  845. }
  846. if (id > s->maxBlockID)
  847. {
  848. s->maxBlockID = id;
  849. }
  851. if (pcinfo->dataSource != p)
  852. {
  853. PDEBUG ("dataSource %p is not equal p %p.n", pcinfo->dataSource, p);
  854. pcinfo->dataSource = p;
  855. }
  858. fflush (c->db);
  859. pcinfo->total ++;
  860. pcinfo->maxID ++;
  861. pcinfo->numblocks ++;
  862. c->downsize += size;
  863. if (pcinfo->numblocks >= pcinfo->max_queue)
  864. {
  865. while ((j = newChannelFile (c)) == 0);
  866. if (j < 0) return -1;
  867. c->db = pcinfo->input[pcinfo->numinput-1];
  868. }
  869. return size;
  870. }
  872. int free_livechannel (struct Channel *p)
  873. {
  874. char buffer[MAX_LINE];
  875. struct LiveChannelInfo *pcinfo = p->pcinfo;
  876. int i;
  877. send_all_spupdate (p, NULL);
  878. for (i=0; i<pcinfo->numinput; i++)
  879. {
  880. if (pcinfo->input[i]) fclose (pcinfo->input[i]);
  881. if (!pcinfo->isSave)
  882. {
  883. snprintf (buffer, MAX_LINE, "%s/%d", p->fname, i);
  884. unlink (buffer);
  885. }
  886. }
  887. p->db = NULL;
  888. if (pcinfo->keyfile != NULL) fclose (pcinfo->keyfile);
  889. if (!pcinfo->isSave)
  890. {
  891. snprintf (buffer, MAX_LINE, "%s/keysample", p->fname);
  892. unlink (buffer);
  893. rmdir (p->fname);
  894. }
  895. if (pcinfo->dataSource != NULL)
  896. {
  897. pcinfo->dataSource->pc = NULL;
  898. Clientclosure (pcinfo->dataSource - TRACKER[TYPE_CS].head, TYPE_CS);
  899. pcinfo->dataSource = NULL;
  900. }
  901. if (pcinfo->mlist)
  902. {
  903. timer_remove (p, NULL);
  904. for (i=0; i<pcinfo->max_channel; i++)
  905. {
  906. if (pcinfo->mlist->m_lists[i] != NULL)
  907. {
  908. pcinfo->mlist->m_lists[i]->ref --;
  909. if (pcinfo->mlist->m_lists[i]->ref <= 0)
  910. freeProgram (pcinfo->mlist->m_lists[i], NULL);
  911. }
  912. }
  913. free (pcinfo->mlist);
  914. }
  915. freeMedia (p);
  916. #ifdef TEST
  917. for (i=0; i<MAX_TS; i++)
  918. {
  919. sprintf (buffer, "%s/%s/%s.gtv", WWW_ROOT, NET_NAME[i], p->channel_name);
  920. remove (buffer);
  921. }
  922. #endif
  923. return 0;
  924. }
  925. #endif
  927. struct Channel *newLiveChannel (char *name, struct Session *source, char *cmd5, float bitrate, int maxblocksize)
  928. {
  929. int id;
  930. struct Channel *p;
  931. struct LiveChannelInfo *pcinfo;
  932. if (NumNewChannel >= MAX_CHANNEL) return (struct Channel *)0;
  933. p = (struct Channel *)calloc (sizeof (struct Channel), 1);
  934. memcpy (p->channel_md5, cmd5, MD5_LEN);
  935. if (name) strncpy (p->channel_name, name, sizeof (p->channel_name));
  936. p->channel_md5[MD5_LEN] = 0;
  937. p->upsize = 0;
  938. p->downsize = 0;
  939. p->maxblocksize = maxblocksize;
  940. p->ctime = time (NULL);
  942. p->pcinfo = (struct LiveChannelInfo *)calloc (sizeof (struct LiveChannelInfo), 1);
  943. pcinfo = p->pcinfo;
  944. pcinfo->dataSource = source;
  945. pcinfo->bitrate = bitrate;
  947. buildLivePath (p->fname, CHNLURL_LEN, cmd5);
  948. if (init_livechannel (p) < 0)
  949. {
  950. PDEBUG ("newLiveChannel error for %p.", p);
  951. free_livechannel (p);
  952. free (pcinfo);
  953. free (p);
  954. return (struct Channel *)0;
  955. }
  957. id = hash_str (p->channel_md5, MD5_LEN);
  958. PDEBUG("newLiveChannel hash %.32s(fname=%s) to %d.n", p->channel_md5, p->fname, id);
  959. p->next = ChannelHash[id];
  960. ChannelHash[id] = p;
  961. p->lnext = ChannelList;
  962. ChannelList = p;
  963. NumNewChannel ++;
  964. return p;
  965. }