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

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

P2P编程

开发平台：

Visual C++

sessions.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. #define FENCE_DATA 1024
  21. int IN_LOOP = 1;
  22. struct cachetype *BufferCacheHead;
  23. inline char *NEW ()
  24. {
  25. char *result;
  26. if (BufferCacheHead != NULL)
  27. {
  28. result = (char *) BufferCacheHead;
  29. BufferCacheHead = BufferCacheHead->next;
  30. } else
  31. {
  32. result = calloc (1, MAX_DATA+FENCE_DATA);
  33. }
  34. assert (result != NULL);
  35. return result;
  36. }
  38. inline void FREE (void *p)
  39. {
  40. if (p == NULL) return;
  41. ((struct cachetype *)p)->next = BufferCacheHead;
  42. BufferCacheHead = p;
  43. }
  45. inline void closure_cache ()
  46. {
  47. struct cachetype *p, *nextp;
  48. for (p=BufferCacheHead; p; p=nextp)
  49. {
  50. nextp = p->next;
  51. free (p);
  52. }
  53. BufferCacheHead = NULL;
  54. }
  56. void terminate (int sig)
  57. {
  58. IN_LOOP --;
  59. }
  61. int handle_new_connection(int sock, int type)
  62. {
  63. struct sockaddr_in addr;
  64. socklen_t addrlen = sizeof (struct sockaddr_in);
  65. int listnum, max, flags;
  66. struct Session *head;
  67. #ifdef SO_LINGER
  68. struct linger ling;
  69. #endif
  70. int newconn = accept(sock, (struct sockaddr *)(&addr), &addrlen);
  71. int keepalive = 1;
  73. if (newconn < 0)
  74. {
  75. perror("accept");
  76. return -1;
  77. }
  78. #ifdef SO_LINGER
  79. ling.l_onoff = 1;
  80. ling.l_linger = 0;
  81. setsockopt (newconn, SOL_SOCKET, SO_LINGER, &ling, sizeof (struct linger));
  82. #endif
  83. setsockopt (newconn, SOL_SOCKET, SO_KEEPALIVE, &keepalive, sizeof (keepalive));
  85. flags = fcntl(newconn, F_GETFL, 0);
  86. flags |= O_NONBLOCK;
  87. fcntl(newconn, F_SETFL, flags);
  89. if (TRACKER[type].cur == TRACKER[type].max)
  90. {
  91. char *buf, buffer[MAX_DATA];
  92. buf = buffer+sizeof (int);
  93. *(unsigned char *)buf = P2P_MSG;
  94. buf += sizeof (char);
  95. *(unsigned short *)buf = ERR_CONNECTION_FULL;
  96. buf += sizeof (short);
  97. *(unsigned char *)buf = 1;
  98. buf += sizeof (char);
  99. *(unsigned int *)buffer = buf - buffer;
  100. write (newconn, buffer, *(unsigned int *)buffer);
  101. close (newconn);
  102. return -1;
  103. }
  105. head = TRACKER[type].head;
  106. max = TRACKER[type].max;
  107. for (listnum = 0; listnum < max; listnum ++)
  108. {
  109. if(head[listnum].socket == 0)
  110. {
  111. head[listnum].socket = newconn;
  112. head[listnum].type = type;
  113. getpeername (newconn, (struct sockaddr *)(&addr), &addrlen);
  114. head[listnum].host = ntohl(addr.sin_addr.s_addr);
  115. head[listnum].port = ntohs(addr.sin_port);
  116. head[listnum].buf = NEW ();
  117. #ifdef __CP_SOURCE
  118. head[listnum].totalup = 0;
  119. #endif
  120. head[listnum].time_sec = CurrentTime;
  121. head[listnum].last_transferblock = CurrentTime;
  122. FD_SET(newconn, &osocks);
  123. if (listnum > TRACKER[type].maxid)
  124. TRACKER[type].maxid = listnum;
  125. break;
  126. }
  127. }
  128. if (listnum >= max)
  129. {
  130. PDEBUG ("no space left for incoming client type %d.", type);
  131. close (newconn);
  132. return -1;
  133. }
  134. TRACKER[type].cur ++;
  135. return (*(TRACKER[type].init)) (listnum);
  136. }
  138. int Clientclosure (int listnum, int type)
  139. {
  140. (*(TRACKER[type].closure)) (listnum);
  141. while (TRACKER[type].maxid == listnum && TRACKER[type].head[listnum].socket == 0 && listnum > 0)
  142. {
  143. listnum --;
  144. TRACKER[type].maxid --;
  145. }
  146. return (--TRACKER[type].cur);
  147. }
  149. void my_exit() __attribute__((noreturn, destructor));
  150. #ifdef __SP_SOURCE
  151. extern struct Channel *ProgramHash[MAX_CHANNEL];
  152. #endif
  153. extern void freeJobCache ();
  155. void my_exit (int err)
  156. {
  157. int max, listnum, type;
  158. struct Session *head;
  159. int (*closure) (int);
  160. for (type=MAX_TYPE-1; type >=0; type --)
  161. {
  162. max = TRACKER[type].maxid+1;
  163. closure = TRACKER[type].closure;
  164. if ((head = TRACKER[type].head) != NULL)
  165. {
  166. for (listnum=0; listnum<max; listnum++)
  167. {
  168. if (head[listnum].socket > 0)
  169. (*closure) (listnum);
  170. }
  171. free (head);
  172. }
  173. #ifdef __CP_SOURCE
  174. if (TRACKER[type].flag == FLAG_SERVER) close (TRACKER[type].sock);
  175. #endif
  176. #ifdef __SP_SOURCE
  177. close (TRACKER[type].sock);
  178. #endif
  179. }
  180. #ifdef __CP_SOURCE
  181. #ifdef HAVE_TS
  182. closure_TS ();
  183. close (TSSOCK);
  184. #endif
  185. #endif
  186. PDEBUG ("exit...n");
  187. freeAllChannel ();
  188. #ifdef __SP_SOURCE
  189. db_end ();
  190. timer_free ();
  191. freeAllProgram ();
  192. freeAllOrder ();
  193. #endif
  194. free_config (ConfigParameters, sizeof(ConfigParameters)/sizeof(struct NamVal));
  195. closure_cache ();
  196. freeJobCache ();
  197. unlink (PIDFile);
  198. CLOSELOG;
  199. exit (err);
  200. }
  201. /* check the message in p->buf, wheter is complete */
  202. inline int checkComplete (struct Session *p)
  203. {
  204. int len;
  205. if (p->socket == 0 || p->off < sizeof (int)+sizeof(char)) return 0;
  206. len = *(unsigned int *)(p->buf+p->start);
  207. if (len >= MAX_BLOCK_SIZE) return -1;
  208. return (p->off >= len?len:0);
  209. }
  211. inline void my_memmove (char *dst, char *src, int len)
  212. {
  213. int i;
  214. if (len <= 0 || dst == src) return;
  215. for (i=0; i<len; i++)
  216. *dst++ = *src++;
  217. }
  218. /* after process a message, update the buf position in p */
  219. inline int updateBuf (struct Session *p, int len)
  220. {
  221. if (p->socket == 0) return 0;
  222. p->start += len;
  223. p->off -= len;
  224. return 0;
  225. }
  228. void process_type (int type, fd_set *socks, fd_set *wsocks, fd_set *esocks)
  229. {
  230. struct ServerDesc *ser = &(TRACKER[type]);
  231. struct Session *p;
  232. int (*process)(int listnum);
  233. int listnum, max, this_read, this_write;
  235. max = ser->maxid + 1;
  236. process = ser->process; //the pointer of function process_P2PC P2PS
  237. for (listnum = 0; listnum < max; listnum++)
  238. {
  239. p = &(ser->head[listnum]);
  240. #ifdef __CP_SOURCE
  241. if (p->flag > 0 && (CurrentTime - p->time_sec) > MAX_CONN)
  242. {
  243. if (reconnect (p) < 0)
  244. {
  245. PDEBUG("Reconnect Error.n");
  246. Clientclosure (listnum, type);
  247. }
  248. continue;
  249. }
  250. #endif
  251. if (p->socket != 0 && FD_ISSET (p->socket, wsocks))
  252. {
  253. #ifdef __CP_SOURCE
  254. struct sockaddr_in addr;
  255. socklen_t addrlen = sizeof (struct sockaddr_in);
  256. if (p->flag != 0)
  257. {
  258. p->flag = 0;
  259. getpeername (p->socket, (struct sockaddr *)(&addr), &addrlen);
  260. p->port = ntohs(addr.sin_port);
  261. /* restore file status flags */
  262. fcntl(p->socket, F_SETFL, p->sock_flag);
  264. }
  265. #endif
  266. if ((this_write = processJobs (p)) < 0)
  267. {
  268. perror ("PP: Write");
  269. Clientclosure (listnum, type);
  270. } else
  271. {
  272. // PDEBUG ("Send %d to %dn", this_write, listnum);
  273. tmpUpBytes += this_write;
  274. #ifdef __CP_SOURCE
  275. p->totalup += this_write;
  276. #endif
  277. }
  278. // continue;
  279. }
  280. if (p->socket != 0 && FD_ISSET (p->socket, socks))
  281. {
  282. if (p->start + p->off  >= MAX_DATA-1)
  283. {
  284. my_memmove (p->buf, p->buf+p->start, p->off);
  285. p->start = 0;
  286. } else if (p->off == 0)
  287. p->start = 0;
  288. if ((this_read = read(p->socket, p->buf+p->start+p->off, MAX_DATA -p->start- p->off)) <= 0)
  289. {
  290. perror("PP: read err");
  291. PDEBUG("socket: %d, p->start: %d. p->off: %d. n", p->socket, p->start, p->off);
  292. Clientclosure (listnum, type);
  293. } else
  294. {
  295. p->off += this_read;
  296. }
  297. }
  298. while (p->socket != 0 && (this_read = checkComplete (p)) > 0)
  299. {
  300. if ((*process) (listnum) == -2) break;
  301. updateBuf (p, this_read);
  302. }
  303. if (p->socket != 0)
  304. {
  305. if (this_read < 0 || this_read > MAX_DATA)
  306. {
  307. PDEBUG ("Too long message, cut off, length is %d, and p->off is %d, p->start is %d.n", this_read, p->off, p->start);
  308. Clientclosure (listnum, type);
  309. }
  310. }
  311. }
  312. }
  315. void process_child (void)
  316. {
  317. int readsocks; //?
  318. struct Session *head;
  319. int highsock;
  320. fd_set socks, wsocks, esocks;
  321. int type, listnum, max;
  322. struct timeval tm;
  323. #ifdef __CP_SOURCE
  324. time_t last_handle_conn = 0;
  325. #endif
  327. startTime = time(NULL);
  329. while (IN_LOOP > 0)
  330. {
  331. FD_ZERO(&socks);
  332. FD_ZERO(&esocks);
  333. FD_ZERO(&wsocks);
  334. #ifdef __CP_SOURCE
  335. #ifdef HAVE_TS
  336. highsock = TSSOCK;
  337. FD_SET(TSSOCK, &socks);
  338. #else
  339. highsock = 0;
  340. #endif
  341. #endif
  342. CurrentTime = time (NULL);
  343. #ifdef __SP_SOURCE
  344. highsock = 0;
  345. for(type=0; type<MAX_TS; type++)
  346. {
  347. if (tsSock[type] <= 0) continue;
  348. if (tsSock[type] > highsock)
  349. highsock = tsSock[type];
  350. FD_SET(tsSock[type], &socks);
  351. }
  352. #endif
  353. for (type=0; type<MAX_TYPE; type++)//type is P2PC and P2PS 
  354. {
  355. #ifdef __CP_SOURCE
  356. if (TRACKER[type].flag == FLAG_SERVER && highsock < TRACKER[type].sock)
  357. highsock = TRACKER[type].sock;
  358. if(CurrentTime - last_handle_conn >= 1) 
  359. {
  360. last_handle_conn = CurrentTime;
  361. }
  362. if (TRACKER[type].flag == FLAG_SERVER) FD_SET(TRACKER[type].sock, &socks);
  363. #endif
  364. #ifdef __SP_SOURCE
  365. if (highsock < TRACKER[type].sock)
  366. highsock = TRACKER[type].sock;
  367. FD_SET(TRACKER[type].sock, &socks);
  368. #endif
  371. max = TRACKER[type].maxid + 1;
  372. head = TRACKER[type].head;
  373. for (listnum = 0; listnum < max; listnum++)
  374. {
  375. if (head[listnum].socket != 0)
  376. {
  377. if (head[listnum].head)
  378. FD_SET(head[listnum].socket, &wsocks);
  379. else
  380. FD_SET(head[listnum].socket, &socks);
  381. if (head[listnum].socket > highsock)
  382. highsock = head[listnum].socket;
  383. }
  384. }
  385. }
  387. tm.tv_sec = 0;
  388. tm.tv_usec = 10000;
  389. readsocks = select(highsock+1, &socks, &wsocks, &esocks, &tm);
  390. if (readsocks <= 0)
  391. goto NEXTROUND;
  392. #ifdef __CP_SOURCE
  393. #ifdef HAVE_TS
  394. if (FD_ISSET(TSSOCK, &socks))
  395. process_TS ();
  396. if (FD_ISSET(TSSOCK, &esocks))
  397. {
  398. PDEBUG ("exit...n");
  399. exit (errno);
  400. }
  401. #endif
  402. #endif
  403. #ifdef __SP_SOURCE
  404. for(type=0; type < MAX_TS; ++type)
  405. {
  406. if(tsSock[type] > 0 && FD_ISSET(tsSock[type], &socks))
  407. process_TS2RM(type);    //
  409. }
  410. #endif
  411. for (type=0; type<MAX_TYPE; type++)
  412. {
  413. #ifdef __CP_SOURCE
  414. if (TRACKER[type].flag == FLAG_SERVER && FD_ISSET(TRACKER[type].sock, &socks))
  415. #endif
  416. #ifdef __SP_SOURCE
  417. if (FD_ISSET(TRACKER[type].sock, &socks))
  418. #endif
  419. handle_new_connection (TRACKER[type].sock, type);
  420. #ifdef __CP_SOURCE
  421. if (TRACKER[type].flag == FLAG_SERVER && FD_ISSET(TRACKER[type].sock, &esocks))
  422. #endif
  423. #ifdef __SP_SOURCE
  424. if (FD_ISSET(TRACKER[type].sock, &esocks))
  425. #endif
  426. {
  427. PDEBUG ("exit...n");
  428. exit (errno);
  429. }
  431. process_type (type, &socks, &wsocks, &esocks);
  432. }
  433. NEXTROUND:
  434. period_process ();
  435. }
  436. }
  438. /* assume the message has been transfered into p->buf */
  439. inline void writeDATAMessage (struct Session *p, struct Channel *pc, struct JobDes *ptr)
  440. {
  441. ptr->len += (*(int *)(ptr->buffer));
  442. addJob (p, pc, ptr);
  443. }
  445. int writeMessage (struct Session *p, struct Channel *pc, char *ptr)
  446. {
  447. struct JobDes *pj;
  448. char *buffer;
  449. int max = 0;
  450. int new = 0;
  451. int len = *(int *)ptr;
  452. if ((pj = findEnoughBuffer (p, pc, len)) == NULL)
  453. {
  454. new = 1;
  455. if ((pj = newJob ()) == NULL)
  456. return -1;
  457. }
  458. buffer = getJobBuffer (pj, &max);
  459. memcpy (buffer, ptr, len);
  460. pj->len += len;
  461. if (new)
  462. addJob (p, pc, pj);
  463. return 0;
  464. }
  466. struct Edge *newEdge (struct Channel *head, struct Session *me)
  467. {
  468. struct Edge *result = malloc (sizeof (struct Edge));
  469. result->head = head;
  470. result->me = me;
  471. result->cnext = head->PeerHead;
  472. head->PeerHead = result;
  473. result->enext = me->header;
  474. me->header = result;
  475. return result;
  476. }
  478. int delEdge (struct Edge *e)
  479. {
  480. struct Channel *pchannel=e->head;
  481. struct Session *psession=e->me;
  482. struct Edge *pedge;
  484. if (pchannel)
  485. {
  486. if (pchannel->PeerHead == e) pchannel->PeerHead = e->cnext;
  487. else
  488. {
  489. for (pedge=pchannel->PeerHead; pedge && pedge->cnext && pedge->cnext != e; pedge = pedge->cnext);
  490. if (pedge && pedge->cnext)
  491. pedge->cnext = e->cnext;
  492. }
  493. }
  494. if (psession)
  495. {
  496. if (psession->header == e) psession->header = e->enext;
  497. else
  498. {
  499. for (pedge=psession->header; pedge && pedge->enext && pedge->enext != e; pedge = pedge->enext);
  500. if (pedge && pedge->enext)
  501. pedge->enext = e->enext;
  502. }
  503. }
  504. if (psession->pc == pchannel) psession->pc = NULL;
  505. free (e);
  506. return 0;
  507. }
  509. void apply_session (struct Session *head, int size, void apply(struct Session *, void *), void *p)
  510. {
  511. int i;
  512. for (i = 0; i < size; i++)
  513. {
  514. if (head[i].socket > 0) apply (&(head[i]), p);
  515. }
  516. }
  518. int logto_xml (unsigned int time_interval, unsigned int tmpTime, unsigned int channelcount, unsigned int totalclient)
  519. {
  520. FILE *logf;
  521. if (LOGXML == NULL || LOGXML[0] == 0 || (logf = fopen(LOGXML,"w")) == NULL)
  522. {
  523. PDEBUG("Couldn't open xml log %s!.n", LOGXML);
  524. return -1;
  525. }
  526. fprintf(logf, "<?xml version="1.0" encoding="iso-8859-1"?>n");
  527. #ifdef __SP_SOURCE
  528. fprintf(logf, "<SP>n");
  529. #endif
  530. #ifdef __CP_SOURCE
  531. fprintf(logf, "<CP>n");
  532. #endif
  533. fprintf(logf, "<ElapsedTime>%ld</ElapsedTime>n", CurrentTime-startTime);
  534. fprintf(logf, "<CurrentIncoming>%.4f</CurrentIncoming>n", ((float)tmpDownBytes)/1024/time_interval);
  535. fprintf(logf, "<CurrentOutgoing>%.4f</CurrentOutgoing>n", ((float)tmpUpBytes)/1024/time_interval);
  536. fprintf(logf, "<AverageIncoming>%.4f</AverageIncoming>n", ((float)totalDownBytes)/1024/tmpTime);
  537. fprintf(logf, "<AverageOutgoing>%.4f</AverageOutgoing>n", ((float)totalUpBytes)/1024/tmpTime);
  538. fprintf(logf, "<ActiveChannel>%d</ActiveChannel>n", channelcount);
  539. fprintf(logf, "<OnlineUser>%d</OnlineUser>n", totalclient);
  540. fprintf(logf, "<TotalIncoming>%lld</TotalIncoming>n", totalDownBytes);
  541. fprintf(logf, "<TotalOutgoing>%lld</TotalOutgoing>n", totalUpBytes);
  542. #ifdef __SP_SOURCE
  543. fprintf(logf, "</SP>n");
  544. #endif
  545. #ifdef __SP_SOURCE
  546. fprintf(logf, "</CP>n");
  547. #endif
  548.         fclose(logf);
  549. return 0;
  550. }