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

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

P2P编程

开发平台：

Visual C++

TSnew.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 <stdio.h>
  21. #include <stdlib.h>
  22. #include <stdarg.h>
  23. #include <ctype.h>
  24. #include <sys/types.h>
  25. #include <sys/time.h>
  26. #include <sys/stat.h>
  27. #include <sys/wait.h>
  28. #include <sys/resource.h>
  29. #include <sys/socket.h>
  30. #include <netinet/in.h>
  31. #include <arpa/inet.h>
  32. #include <netdb.h>
  33. #include <unistd.h>
  34. #include <fcntl.h>
  35. #include <errno.h>
  36. #include <assert.h>
  37. #include <syslog.h>
  38. #include <string.h>
  39. #include <time.h>
  40. #include <assert.h>
  42. #include "ProtocolDefine.h"
  43. #include "StructDefine.h"
  44. #include "ErrorDefine.h"
  45. #include "util.h"
  46. #include "findcp.h"
  48. #define MAX_BIND 10
  49. #define AUTH_HEADER             12
  50. #define NORMAL_HEADER           5
  51. #define MAX_NET_NUM 5
  53. #define MAX_CLIENT 3072
  55. #define MAX_PEER 0x10 /* max peers returned by NEED_PEERS */
  57. #define IPADDR_LEN 16 /*****IP地址最大长度*****/
  58. #define MD5_LEN 32
  59. #define SERVICE_LEN 128
  61. #define TYPE_NP 0
  62. #define TYPE_CP 1
  65. #define MAX_CHANNEL 0x400 //1024
  66. #define MAX_NP 0x40000 //262144
  67. #define MAX_CP 0x10000 //65536
  68. #define MAX_RM 0x4 //4
  70. //#define TS4CP_PORT 22168
  71. //#define TS4RM_PORT 22169
  73. #define MAX_DATA 20000
  74. #define MAX_LINE 1024
  75. #define MAX_IDLE 120
  76. #define SILENCE_TIME 2
  77. #define BUILD_SOCKADDR(host,port,client)
  78. {
  79. memset ((char *)&client, 0, sizeof (client));
  80. client.sin_port = htons (pnp->port);
  81. client.sin_family = AF_INET;
  82. client.sin_addr.s_addr = htonl (pnp->host);
  83. }
  85. #define FREE_TRACKER(tracker)
  86. {
  87. int i;
  88. max = tracker.max;
  89. closure = tracker.closure;
  90. if ((head = tracker.head) != NULL)
  91. {
  92. for (listnum=0; listnum<max; listnum++)
  93. {
  94. if (head[listnum].socket > 0)
  95. (*closure) (head+listnum);
  96. }
  97. free (head);
  98. }
  99. if (tracker.hash) free (NPTRACKER.hash);
  100. for (i=0; tracker.sock[i] != 0 && i<MAX_BIND; i++)
  101. close (tracker.sock[i]);
  102. }
  103. #define MAX_INTERVAL 20
  104. #define OBSERVE_LAYER 10
  105. #define BASEDIR "/home/channel/"
  107. int current_log_count = 0;
  108. #define MAX_LOG_COUNT 100 // call fflush() when current_log_count reaches MAX_LOG_COUNT
  110. // 用来表示NP上面所有的快区间, 目前块区间由一个开始字段和一个长度字段来标识. 
  111. struct Interval
  112. {
  113. unsigned int start;
  114. unsigned int len;
  115. };
  116. // Generic format of messages between TS and other peers
  117. // Most often used in login process
  118. struct Message
  119. {
  120. unsigned int len;
  121. unsigned char type;
  122. char buffer[MAX_DATA];
  123. };
  124. // format of messages between TS and other peers
  125. // with authcodes specified
  126. struct TSMessage
  127. {
  128. unsigned int len;
  129. unsigned char type;
  130. unsigned int authcode1:24;
  131. unsigned int authcode2;
  132. char buffer[MAX_DATA];
  133. } UDPMsg;
  135. struct ChannelInfo
  136. {
  137. char md5[MD5_LEN+1];
  138. unsigned char numinter;
  139. struct Interval inter[MAX_INTERVAL];
  140. };
  143. struct Channel
  144. {
  145. char name[MD5_LEN+1];
  146. unsigned int numclient;
  147. unsigned int accumclient;
  148. unsigned int latest_time;
  149. #ifndef SORT_NET
  150. struct Edge *PeerHead;
  151. #else
  152. unsigned int nclient_net[MAX_NET_NUM];
  153. struct Edge *PeerHead[MAX_NET_NUM];
  154. #endif
  155. struct Session *SCPhead;
  156. struct Channel *next;
  157. };
  159. struct Edge
  160. {
  161. struct Channel *head;
  162. struct Session *me;
  163. struct Edge *cnext;
  164. struct Edge *enext;
  165. unsigned char numinter;
  166. unsigned int current;
  167. struct Interval inter[MAX_INTERVAL];
  168. };
  170. struct NPInfo
  171. {
  172. struct CorePeerInfo p;
  173. struct TransferInfo t;
  174. struct StatInfo  s;
  175. int    startBlock; // starting position of NP
  176. int numchannel;
  177. struct Edge *cur;
  178. struct Edge *header;
  179. };
  181. struct CPInfo
  182. {
  183. char type;
  184. unsigned char numHeads;
  185. unsigned short connnum;
  186. unsigned char maxConn;
  187. char parameter[42];
  188. int userid;
  189. int resnum;
  190. float band;
  191. char servicetype[128];
  192. };
  194. struct Session
  195. {
  196. int type;
  197. int socket;
  198. unsigned int host;
  199. unsigned int intra;
  200. unsigned short port;
  201. unsigned short npport;
  202. #ifdef SORT_NET
  203. unsigned int net;
  204. struct Session *cachepeer[MAX_NET_NUM];
  205. #else
  206. struct Session *cachepeer;
  207. #endif
  208. unsigned int auth;
  209. unsigned int time_sec;
  210. unsigned int last_access;
  211. union
  212. {
  213. struct NPInfo p;
  214. struct CPInfo cp;
  215. } u;
  216. float clientVer;
  217. struct Session *hnext;
  218. };
  220. FILE *statlog = NULL; // statistics log file
  222. extern int errno;
  223. int OUTPUT_STAT;
  224. time_t CurTimeSec;
  225. time_t startTime;
  226. char *CONFIG="./ats.cfg";
  228. char *LOCALHOST;
  229. #ifdef HAVE_MYSQL
  230. char *MYSQL_HOST="localhost";
  231. char *MYSQL_USER="root";
  232. char *MYSQL_PASS="gtv";
  233. char *MYSQL_DB="gtv";
  234. #endif
  236. int CurrentSock;
  238. fd_set osocks;
  239. int highsock;
  241. char *LOGXML;
  243. struct TransferInfo Transfer;
  244. struct sockaddr_in UDPCLIENT;
  246. extern int init();
  247. extern int readconfig(char * filename);
  248. extern const char* find_ip_from_list(unsigned long ip);
  249. extern int findcppeers(unsigned long ip, void *p);
  250. extern int findnettype(const char *servicetype, void* p);
  251. extern void add_cp_to_list(void *p);
  252. extern void remove_cp_from_list(void *p);
  253. extern const char* find_cp_service_type(unsigned long ip);
  255. #define MAX_POLLUTE 1000
  256. int BINDALL;
  257. int Polluted; /* should we call periodLOG? if polluted>MAX_POLLUTE call */
  258. struct Array cfgTS4NP_PORT;
  259. struct Array cfgTS4CP_PORT;
  260. struct Array cfgTS4RM_PORT;
  261. int SnapShotInterval; // make a snap shot every few seconds
  262. float MIN_CLIENT_VERSION; // client version should be bigger than this one.
  264. long long np2tsLoginCount = 0;
  265. long long np2tsResListCount = 0;
  266. long long np2tsReqResCount = 0;
  267. long long np2tsDelResCount = 0;
  268. long long np2tsReportCount = 0;
  269. long long np2tsNeedPeerCount = 0;
  270. long long np2tsLogoutCount = 0;
  272. long long ts2npWelcomeCount = 0;
  273. long long ts2npPeersCount = 0;
  274. long long ts2npConnectToCount = 0;
  275. long long ts2npMsgCount = 0;
  277. /*************************************************** 
  278.  * SessionCluster是有关每种服务的信息.             *
  279.  * 其中有一个指向该服务相关的各个Session的指针.    *
  280.  ***************************************************/ 
  281. struct SessionCluster
  282. {
  283. unsigned int port[MAX_BIND];
  284. unsigned int maxbuf;
  285. int sock[MAX_BIND]; // 描述所使用的socket
  286. int cur; // current number of sessions 当前客户端的个数
  287. int max; // maximum number of sessions 最多容纳客户端的个数
  288. int maxid; // maxid: maximum session index currently in the list. for optimization of search
  289. struct Session *head; /* pointer to the session pool 
  290.          head[0]为第一个Session,head[max-1]为最后一个session */
  291. struct Session **hash; // session hash table
  292. int (*process) (struct Session *); //这一服务中消息的处理函数
  293. int (*closure) (struct Session *); //这一服务中需要的析构函数
  294. };
  296. struct Session *GCPCHOICE;
  297. struct Channel *ChannelHash[MAX_CHANNEL];
  298. struct SessionCluster NPTRACKER, CPTRACKER;
  299. #ifdef HAVE_RM
  300. struct SessionCluster RMTRACKER;
  301. #endif
  302. #ifdef SORT_NET
  303. char *NETFN;
  304. #endif
  306. struct NamVal ConfigParameters[]
  307. = 
  308. {
  309. #ifdef HAVE_MYSQL
  310. {"MysqlAddress", &MYSQL_HOST, 's'},
  311. {"User", &MYSQL_USER, 's'},
  312. {"Password", &MYSQL_PASS, 's'},
  313. {"Database", &MYSQL_DB, 's'},
  314. #endif
  315. #ifdef SORT_NET
  316. {"Netfile", &NETFN, 's'},
  317. #endif
  318. {"Bind", &LOCALHOST, 's'},
  319. {"TS4NP_PORT", &cfgTS4NP_PORT, 'a'},
  320. {"TS4CP_PORT", &cfgTS4CP_PORT, 'a'},
  321. {"TS4RM_PORT", &cfgTS4RM_PORT, 'a'},
  322. {"SnapShotInterval", &SnapShotInterval, 'd'},
  323. {"ClientVersion", &MIN_CLIENT_VERSION, 'f'},
  324. {"LogFilePath", &LOGXML, 's'},
  325. {"BINDALL", &BINDALL, 'd'}
  326. };
  328. #ifdef DEBUG
  329. #define PDEBUG(fmt, args...) fprintf(stderr, "TS: (%s,%d)" fmt, __FILE__, __LINE__, ## args)
  330. #else
  331. #define PDEBUG(fmt, args...) do {} while (0)
  332. #endif
  334. #ifdef SORT_NET
  335. #define MAX_NET 2048
  336. struct networks
  337. {
  338. unsigned int host;
  339. unsigned int mask;
  340. int net;
  341. } NETBLOCKS[MAX_NET];
  343. unsigned int MASKS[33]={0x0,0x80000000,0xc0000000,0xe0000000,0xf0000000,0xf8000000,0xfc000000,0xfe000000,0xff000000,0xff800000,0xffc00000,0xffe00000,0xfff00000,0xfff80000,0xfffc0000,0xfffe0000,0xffff0000,0xffff8000,0xffffc000,0xffffe000,0xfffff000,0xfffff800,0xfffffc00,0xfffffe00,0xffffff00,0xffffff80,0xffffffc0,0xffffffe0,0xfffffff0,0xfffffff8,0xfffffffc,0xfffffffe,0xffffffff};
  345. // globals
  346. int maxNet; // max id
  349. int readNETBLOCK (char *fname); //用来读入network配置文件
  350. struct networks *getnetwork (unsigned int host, struct networks *head, int n);
  351. int compareNet (const void *a, const void *b);
  352. #endif
  354. struct Message ErrMSG;
  356. #define SEND_NPMSG(sock,msg,code,quit,client)
  357. {
  358. ErrMSG.len = 8;
  359. ErrMSG.type = msg;
  360. ++ts2npMsgCount;
  361. *(unsigned short *)(ErrMSG.buffer) = code;
  362. *(unsigned char *)(ErrMSG.buffer+sizeof(short)) = quit;
  363. sendMessage(sock,(char *)&ErrMSG,client);
  364. }
  366. int init_ts ();
  367. void my_exit () __attribute__((noreturn, destructor));
  368. void process_child (void);
  370. int init_NP (struct Session *);
  371. int process_NP (int idsock);
  372. int closure_NP (struct Session *);
  373. int process_NP2TS_LOGIN (struct Message *);
  374. int process_NP2TS_REPORT (struct Session *, struct TSMessage *);
  375. int process_NP2TS_REPORT2 (struct Session *, struct TSMessage *);
  376. int process_NP2TS_RES_LIST (struct Session *, struct TSMessage *);
  377. int process_NP2TS_REQ_RES (struct Session *, struct TSMessage *);
  378. int process_NP2TS_DEL_RES (struct Session *, struct TSMessage *);
  379. int process_NP2TS_NEED_PEERS (struct Session *, struct TSMessage *m);
  380. int process_NP2TS_QUERY_RES (struct Session *p, struct TSMessage *m);
  381. int process_NEED_PEERS_real (struct Session *p, char *md5, int needcp, unsigned int cur, unsigned char layer);
  383. int findNPPeers (struct Channel *pc, struct Session *me, int playing, int num, char **buffer, char *buffer1);
  384. int findCPPeers (unsigned long host, char *md5, char **buffer);
  386. int init_CP (struct Session *);
  387. int process_CP (int idsock);
  388. int closure_CP (struct Session *);
  389. int process_CP2TS_UPDATE (struct Session *, struct TSMessage *m);
  390. int process_CP2TS_REGISTER (struct Message *m);
  391. int process_CP2TS_NEED_PEERS (struct Session *p, struct TSMessage *m);
  393. void periodLOG (int s);
  394. void makeSnapShot(int count);
  395. //int memlog(char *pwd,char *cmd);
  396. void freeChannel (struct Channel *p);
  397. int check_valid (struct Edge *e, int play);
  398. int merge (struct Interval *head, int total, struct Interval *_new, int num);
  399. int delete_interval (struct Interval *head, int total, struct Interval *_new, int num);
  401. #ifdef HAVE_RM
  402. int init_RM (struct Session *p);
  403. int process_RM (int idsock);
  404. int closure_RM (struct Session *p);
  405. #endif
  406. int logto_xml (long channelcount, unsigned int totalclient, long long totalstay);
  408. inline int hash_np (int h, int p)
  409. {
  410. int id = (((h & (0xffffffff-MAX_NP+1)) >> 14) ^ (h & (MAX_NP-1)) ^ (p & 0xffff)) & (MAX_NP -1);
  411. return id?id:(MAX_NP-1);
  412. }
  414. inline int hash_cp (int h, int p)
  415. {
  416. int id = (((h & (0xffffffff-MAX_CP+1)) >> 12) ^ (h & (MAX_CP-1)) ^ (p & 0xffff)) & (MAX_CP -1);
  417. return id?id:(MAX_CP-1);
  418. }
  420. inline int hash_str (unsigned char *str, int len)
  421. {
  422. int hash;
  423. for (hash=0; len; len--, str++)
  424. hash += (hash << 5) - hash + (*str);
  425. return hash & (MAX_CHANNEL-1);
  426. }
  428. // process_child: 主要函数, 这一函数主要用来设置socks和wsocks.
  429. void process_child (void)
  430. {
  431. time_t last_update=0, last_snapshot=0;
  432. int snapCount = 0;
  433. int i, readsocks;
  434. struct timeval tm;
  435. fd_set socks, esocks;
  437. startTime = time(NULL);
  438. for (i=0; ; i++) 
  439. {
  440. CurTimeSec = time (NULL);
  441. if (CurTimeSec-last_update > MAX_IDLE)
  442. {
  443. periodLOG (1); // close timeout peers
  444. last_update = CurTimeSec;
  445. Polluted = 0; // clear dirty flag
  446. }
  447. if(CurTimeSec - last_snapshot > SnapShotInterval)
  448. {
  449. makeSnapShot(snapCount++); // read it!
  450. system("/usr/bin/vmstat >> ts.log 2>&1 &");
  451. last_snapshot = CurTimeSec;
  452. }
  454. socks = osocks;
  455. esocks = osocks;
  457. tm.tv_sec = 1;
  458. tm.tv_usec = 0;
  459. readsocks = select(highsock+1, &socks, (fd_set *) 0, &esocks, &tm);
  460. if (readsocks <= 0)
  461. continue;
  462. for (i=0; i<MAX_BIND && NPTRACKER.sock[i] != 0; i++)
  463. {
  464. if (FD_ISSET (NPTRACKER.sock[i], &socks))
  465. {
  466. CurrentSock = i;
  467. process_NP (i);
  468. }
  469. // should have a 'break' here to cut off unnecessary check
  470. }
  471. for (i=0; i<MAX_BIND && CPTRACKER.sock[i] != 0; i++)
  472. {
  473. if (FD_ISSET (CPTRACKER.sock[i], &socks))
  474. {
  475. CurrentSock = i;
  476. process_CP (i);
  477. }
  478. }
  479. #ifdef HAVE_RM
  480. for (i=0; i<MAX_BIND && RMTRACKER.sock[i] != 0; i++)
  481. {
  482. if (FD_ISSET(RMTRACKER.sock[i], &socks))
  483. {
  484. CurrentSock = i;
  485. process_RM (i);
  486. }
  487. }
  488. #endif
  489. }
  490. }
  492. void my_exit ()
  493. {
  494. int max, listnum;
  495. struct Session *head;
  496. int (*closure) (struct Session *);
  498. #ifdef HAVE_RM
  499. FREE_TRACKER(RMTRACKER);
  500. #endif
  502. FREE_TRACKER(NPTRACKER);
  503. FREE_TRACKER(CPTRACKER);
  505. // free configuration file.
  506. // 参数为和read_config相同的struct NamVal *以及项数.
  507. free_config (ConfigParameters, sizeof(ConfigParameters)/sizeof(struct NamVal));
  508. #ifdef HAVE_MYSQL
  509. mysql_close (local_mysql);
  510. #endif
  511. PDEBUG ("exit...n");
  512. exit (0);
  513. }
  515. inline void my_memmove (char *dst, char *src, int len)
  516. {
  517. int i;
  518. if (len <= 0 || dst == src) return;
  519. for (i=0; i<len; i++)
  520. *dst++ = *src++;
  521. }
  523. struct Channel *findChannel (char *name, int len)
  524. {
  525. int id = hash_str ((unsigned char*)name, len);
  526. struct Channel *p;
  527. for (p=ChannelHash[id]; p; p=p->next)
  528. {
  529. if (strncmp (name, p->name, len) == 0)
  530. return p;
  531. }
  532. return NULL;
  533. }
  535. struct Edge *newEdge (struct Channel *head, struct Session *me)
  536. {
  537. struct Edge *result = (struct Edge *)malloc (sizeof (struct Edge));
  538. result->head = head;
  539. result->me = me;
  540. #ifndef SORT_NET
  541. result->cnext = head->PeerHead;
  542. head->PeerHead = result;
  543. #else
  544. result->cnext = head->PeerHead[me->net];
  545. head->PeerHead[me->net] = result;
  546. head->nclient_net[me->net] ++;
  547. #endif
  548. result->enext = me->u.p.header;
  549. me->u.p.header = result;
  550. head->numclient ++;
  551. head->accumclient ++;
  552. return result;
  553. }
  555. int delEdge (struct Edge *e)
  556. {
  557. struct Channel *pchannel=e->head;
  558. struct Session *psession=e->me;
  559. struct Edge *pedge;
  561. if (pchannel)
  562. {
  563. #ifndef SORT_NET
  564. if (pchannel->PeerHead == e) pchannel->PeerHead = e->cnext;
  565. #else
  566. pchannel->nclient_net[psession->net] --;
  567. if (pchannel->PeerHead[psession->net] == e) pchannel->PeerHead[psession->net] = e->cnext;
  568. #endif
  569. else
  570. {
  571. #ifndef SORT_NET
  572. for (pedge=pchannel->PeerHead; pedge && pedge->cnext && pedge->cnext != e; pedge = pedge->cnext);
  573. #else
  574. for (pedge=pchannel->PeerHead[psession->net]; pedge && pedge->cnext && pedge->cnext != e; pedge = pedge->cnext);
  575. #endif
  576. if (pedge && pedge->cnext) pedge->cnext = e->cnext;
  577. }
  578. pchannel->numclient --;
  579. if (pchannel->numclient == 0)
  580. freeChannel (pchannel);
  581. }
  582. if (psession)
  583. {
  584. if (psession->u.p.header == e) psession->u.p.header = e->enext;
  585. else
  586. {
  587. for (pedge=psession->u.p.header; pedge && pedge->enext && pedge->enext != e; pedge = pedge->enext);
  588. if (pedge && pedge->enext) pedge->enext = e->enext;
  589. }
  590. }
  591. if (psession->u.p.cur == e) psession->u.p.cur = NULL;
  592. free (e);
  593. return 0;
  594. }
  597. struct Channel *newChannel (char *name, int len)
  598. {
  599. int id = hash_str ((unsigned char*)name, len);
  600. struct Channel *result = (struct Channel *)calloc (1, sizeof (struct Channel));
  601. if (!result) return NULL;
  602. if (len != MD5_LEN)
  603. {
  604. free (result);
  605. return NULL;
  606. }
  607. memcpy (result->name, name, len);
  608. result->name[len] = 0;
  609. result->latest_time = CurTimeSec;
  610. result->next = ChannelHash[id];
  611. ChannelHash[id] = result;
  612. return result;
  613. }
  615. void freeChannel (struct Channel *p)
  616. {
  617. int id = hash_str ((unsigned char*)(p->name), MD5_LEN);
  618. struct Channel *q;
  619. if (ChannelHash[id] == p)
  620. ChannelHash[id] = p->next;
  621. else
  622. {
  623. for (q=ChannelHash[id]; q && q->next != p; q=q->next);
  624. if (q) q->next = p->next;
  625. }
  626. free (p);
  627. }
  629. // join in a channel for session *p
  630. // Add a session-channel mapping (aka. an edge)
  631. int addSession (struct Session *p, struct ChannelInfo *c, int isdefault, unsigned int cur)
  632. {
  633. struct Channel *pc;
  634. struct Edge *pedge = NULL;
  635. if ((pc=findChannel (c->md5, MD5_LEN)) == NULL) /* Since the channel do not exist, the edge does not exist. 
  636.    So there is no need to search the list. */
  637. {
  638. if ((pc = newChannel (c->md5, MD5_LEN)) == NULL)
  639. return -1;
  640. } else{
  641. for (pedge=p->u.p.header; pedge && pedge->head != pc; pedge=pedge->enext);
  642. }
  643. if (pedge == NULL)
  644. {
  645. pedge = newEdge (pc, p);
  646. }
  647. pedge->numinter = c->numinter;
  648. memcpy (pedge->inter, c->inter, c->numinter*sizeof(struct Interval));
  649. pedge->current = cur;
  650. if (isdefault) p->u.p.cur = pedge;
  651. return 0;
  652. }
  654. // delete corresponding session in the session-channel map
  655. int delSession (struct Session *p)
  656. {
  657. struct Edge *pedge, *prevedge;
  658. for (pedge=p->u.p.header; pedge; pedge=prevedge)
  659. {
  660. prevedge = pedge->enext;
  661. delEdge (pedge);
  662. }
  663. return 0;
  664. }
  666. void periodLOG (int s)
  667. {
  668. int i;
  669. struct Session *p;
  670. struct tm result;
  672. localtime_r (&CurTimeSec, &result);
  673. if (s)
  674. {
  675. for (i=0,p=NPTRACKER.head; i<=NPTRACKER.maxid; i++, p++)
  676. {
  677. if (p->socket > 0 && CurTimeSec - p->last_access > MAX_IDLE)
  678. {
  679. closure_NP (p);
  680. }
  681. }
  682. for (i=0,p=CPTRACKER.head; i<=CPTRACKER.maxid; i++, p++)
  683. {
  684. if (p->socket > 0 && CurTimeSec - p->last_access > MAX_IDLE)
  685. {
  686. closure_CP (p);
  687. }
  688. }
  689. }
  691. PDEBUG ("STAT %u,%u/%u %u:%u:%u.n", result.tm_year+1900, result.tm_mon+1, result.tm_mday, result.tm_hour, result.tm_min, result.tm_sec);
  692. PDEBUG ("STAT: "%ld小时%ld分钟%ld秒"n", CurTimeSec/3600, (CurTimeSec%3600)/60, CurTimeSec%60);
  693. return;
  694. }
  696. void makeSnapShot(int count)
  697. {
  698. FILE *f = fopen ("./ts.log", "a");
  700. if (!f)
  701. {
  702. PDEBUG("Couldn't open log file!.n");
  703. return;
  704. }
  705. // seek to the end for write
  706. fseek(f, 0, SEEK_END);
  708. // 1. start SnapShot 
  709. struct tm result;
  710. localtime_r (&CurTimeSec, &result);
  711. fprintf (f, "nn********************Start %d SnapShot, Time: %u/%u %u:%u:%u.n", count, result.tm_mon+1, result.tm_mday, result.tm_hour, result.tm_min, result.tm_sec);
  713. // 2. log cpu & mem state
  714. //memlog(BASEDIR,"./log.sh");
  716. // 3. log message count
  717. fprintf(f, "Login: %lld. ResList: %lld. ReqRes: %lld. DelRes: %lld. Report: %lld. NeedPeers: %lld. Logout: %lld. n", np2tsLoginCount, np2tsResListCount, np2tsReqResCount, np2tsDelResCount, np2tsReportCount, np2tsNeedPeerCount, np2tsLogoutCount);
  718. fprintf(f, "Welcome: %lld. Peers: %lld. ConnectTo: %lld. Msg: %lld.n", ts2npWelcomeCount, ts2npPeersCount, ts2npConnectToCount, ts2npMsgCount);
  719. fflush(f);
  721. // 4. log channel state
  722. struct Channel *pChannel;
  723. int i, total=0, j, watchingCount, noIdleCount;
  724. struct Edge* pedge;
  725. int layerarray[OBSERVE_LAYER], noconnectioncount, higherlayercount;
  726. long long totalup, totaldown;
  727. long long totalcurrblock;
  728. long ts_ChannelCount = 0;
  729. long long ts_stay, totalstay = 0;
  730. for (i=0; i<MAX_CHANNEL; i++)
  731. {
  732. for (pChannel=ChannelHash[i]; pChannel; pChannel=pChannel->next)
  733. {
  734. ++ts_ChannelCount;
  735. total += pChannel->numclient;
  736. watchingCount = 0;
  737. noIdleCount = 0;
  738. ts_stay = 0;
  739. totalup = totaldown = totalcurrblock = 0;
  740. memset(layerarray, 0, OBSERVE_LAYER*sizeof(int));
  741. noconnectioncount = higherlayercount = 0;
  742. #ifndef SORT_NET
  743. for (j=0, pedge=pChannel->PeerHead; pedge && j < pChannel->numclient; pedge=pedge->cnext)
  744. #else
  745. {
  746. int mnetnum;
  747. for (mnetnum = 0; mnetnum < MAX_NET_NUM; mnetnum ++)
  748. {
  749. for (j=0, pedge=pChannel->PeerHead[mnetnum]; pedge && j < pChannel->nclient_net[mnetnum]; pedge=pedge->cnext)
  750. #endif
  751. {
  752. if(pedge->me->socket <= 0)
  753. continue;
  754. if(pedge->me->u.p.p.layer == 0xff)
  755. noconnectioncount++;
  756. else if(pedge->me->u.p.p.layer < OBSERVE_LAYER)
  757. layerarray[pedge->me->u.p.p.layer]++;
  758. else
  759. higherlayercount++;
  760. ts_stay += CurTimeSec - pedge->me->time_sec;
  761. totalup += pedge->me->u.p.t.totalDownBytes;
  762. totaldown += pedge->me->u.p.t.totalUpBytes;
  763. if(pedge->current != 0xffffffff)
  764. {
  765. totalcurrblock += pedge->current;
  766. watchingCount++;
  767. if(CurTimeSec < pedge->me->last_access + 40)
  768. noIdleCount++;
  769. }
  770. }
  771. #ifdef SORT_NET
  772. }
  773. }
  774. #endif
  775. totalstay += ts_stay;
  776. fprintf(f, "%s: accumulated %d clients, current %d clients and  %d/%d is watchings. Average Stay Time : %fn", pChannel->name, pChannel->accumclient, pChannel->numclient, watchingCount,noIdleCount,pChannel->numclient==0?0:(ts_stay/(double)pChannel->numclient));
  777. fprintf(f, "total down: %llu, total up: %llu, down-up: %llu. Avg currBlock: %llu n", totaldown, totalup, (totaldown-totalup), watchingCount==0?0:(totalcurrblock/watchingCount));
  778. fprintf(f, "No Connect: %d. Higher: %d. n Detail: ", noconnectioncount, higherlayercount);
  779. for(j = 0; j < OBSERVE_LAYER; j++)
  780. {
  781. if(layerarray[j] != 0)
  782. fprintf(f, "%d: %d(%.1f), ", j, layerarray[j ], pChannel->numclient==0?0:(layerarray[j]/(double)pChannel->numclient));
  783. }
  784. fprintf(f, "n");
  785. }
  786. }
  787. fprintf(f,"ChannelCount: %ld n",ts_ChannelCount);
  788. fprintf(f, "n ********************END SnapShot.nn");
  789. fclose(f);
  790. logto_xml (ts_ChannelCount, total, totalstay);
  791. }
  792. #if 0
  793. int memlog(char *pwd, char *cmd)
  794. {
  795. int pid = fork();
  796. if (pid == 0)
  797. {
  798. char buffer[10];
  799. daemon(0,1);
  800. snprintf(buffer,10,"%s",cmd);
  801. chdir(pwd);
  802. execlp("/bin/sh", "sh" ,"-c",buffer,NULL);
  803. }
  804. else if(pid < 0)
  805. {
  806. perror("fork error!!!");
  807. return -1;
  808. }
  809. return 0;
  810. }
  811. #endif
  813. #ifdef SORT_NET
  814. int compareNet (const void *a, const void *b)
  815. {
  816. struct networks *p = (struct networks *)a;
  817. struct networks *q = (struct networks *)b;
  818. if (p->host > q->host) return 1;
  819. else if (p->host < q->host) return -1;
  820. return 0;
  821. }
  823. // This is implemented with recursion
  824. struct networks *getnetwork (unsigned int host, struct networks *head, int n)
  825. {
  826. int i = n/2-1;
  827. unsigned int net = host & head[i].mask;
  828. if (n <= 0) return NULL;
  829. if (n == 1)
  830. {
  831. if ((host & head[0].mask) != (head[0].host & head[0].mask))
  832. return NULL;
  833. else return &(head[0]);
  834. }
  835. if (net < (head[i].host & head[i].mask)) return getnetwork (host, head, i);
  836. else if (net > (head[i].host & head[i].mask)) return getnetwork (host, head+i+1, n-i-1);
  837. else return &(head[i]);
  838. }
  840. //192.168.0.0/16/1
  841. //host/mask/netid
  842. int readNETBLOCK (char *fname)
  843. {
  844. int i = 0;
  845. char buffer[MAX_LINE],*p, *q;
  846. struct in_addr inp;
  847. FILE *f = fopen (fname, "r");
  848. if (!f) return -1;
  849. while (fgets (buffer, sizeof (buffer), f))
  850. {
  851. p = index (buffer, '/');
  852. q = rindex (buffer, '/');
  853. if (p == q || (!p) || (!q)) continue;
  854. *p = 0;
  855. *q = 0;
  856. if (inet_aton (buffer, &inp) == 0) continue;
  857. NETBLOCKS[i].host = ntohl (inp.s_addr);
  858. NETBLOCKS[i].mask = MASKS[atoi(p+1)];
  859. NETBLOCKS[i].net = atoi (q+1);
  860. i++;
  861. if (i >= MAX_NET) break;
  862. }
  863. fclose (f);
  864. qsort (NETBLOCKS, i, sizeof (struct networks), compareNet);
  865. return i;
  866. }
  868. int Net = 0;
  869. int Layer = 0;
  870. int compareSession (const void *a, const void *b)
  871. {
  872. struct Session *p = *(struct Session **)a;
  873. struct Session *q = *(struct Session **)b;
  874. if (p->u.p.p.layer == q->u.p.p.layer)
  875. {
  876. if  (p->net == Net) return -1;
  877. else if (q->net == Net) return 1;
  878. else if (p->net > q->net) return 1;
  879. else return -1;
  880. } else if (p->u.p.p.layer <= Layer)
  881. return -1;
  882. else if (q->u.p.p.layer <= Layer)
  883. return 1;
  884. else return (p->u.p.p.layer - q->u.p.p.layer);
  885. }
  887. #endif
  890. int init_udpserver (struct SessionCluster *c, char *host, int *port, unsigned int max, int numbind)
  891. {
  892. int i;
  893. struct Session *head;
  894. // c->port = port;
  895. c->cur = 0; // current count
  896. c->max = max; // maximum count of sessions
  897. c->maxbuf = 0;
  898. if (max > 0)
  899. {
  900. c->head = (struct Session*)calloc (sizeof (struct Session), max); // allocate continuous session pool space
  901. c->hash = (struct Session**)calloc (sizeof (struct Session *), max); // allocate session hash table
  902. head = c->head;
  903. // chain the sessions in session pool as a freelist
  904. for (i=0; i<max-1; i++)
  905. {
  906. head[i].hnext = &(head[i+1]);
  907. }
  909. // initialize hash table
  910. for (i=1; i<max; i++) c->hash[i] = 0;
  911. // this is really ugly!
  912. c->hash[0] = &(head[0]); // c->hash[0] points to the empty list
  913. } // HOW if max <= 0 ? return with error
  915. for (i=0; i<MAX_BIND && i<numbind; i++)
  916. {
  917. switch (BINDALL) // means 'BIND ANY HOST'
  918. {
  919. case 0:
  920. if ((c->sock[i] = init_udp (host, port[i])) < 0)
  921. return -1;
  922. break;
  923. default:
  924. if ((c->sock[i] = init_udp (NULL, port[i])) < 0)
  925. return -1;
  926. break;
  927. }
  928. // highsock indicates the largest active fd. used in select() Used to speed up search
  929. if (c->sock[i] > highsock) highsock = c->sock[i];
  930. FD_SET(c->sock[i], &osocks);
  931. }
  933. return 0;
  934. }
  937. int init_ts ()
  938. {
  939. struct rlimit rl;
  940. #ifdef DEBUG
  941. system ("ulimit -a");
  942. if (getrlimit (RLIMIT_CORE, &rl) != 0)
  943. {
  944. perror ("getrlimit");
  945. return -1;
  946. }
  947. fprintf (stderr, "Get core limit %d:%dn", (int)rl.rlim_cur, (int)rl.rlim_max);
  948. rl.rlim_cur = rl.rlim_max = (rlim_t )10240000;
  949. if (setrlimit (RLIMIT_CORE, &rl) != 0)
  950. {
  951. perror ("getrlimit");
  952. return -1;
  953. }
  954. if (getrlimit (RLIMIT_CORE, &rl) != 0)
  955. {
  956. perror ("getrlimit");
  957. return -1;
  958. }
  959. fprintf (stderr, "Set core limit to %d:%dn", (int)rl.rlim_cur, (int)rl.rlim_max);
  960. system ("ulimit -a");
  961. #endif
  962. if (init_udpserver (&NPTRACKER, LOCALHOST, (int*)cfgTS4NP_PORT.ptr, MAX_NP, cfgTS4NP_PORT.size) < 0
  963. #ifdef HAVE_RM
  964. || init_udpserver (&RMTRACKER, LOCALHOST, (int*)cfgTS4RM_PORT.ptr, MAX_RM, cfgTS4RM_PORT.size) < 0
  965. #endif
  966. || init_udpserver (&CPTRACKER, LOCALHOST, (int*)cfgTS4CP_PORT.ptr, MAX_CP, cfgTS4CP_PORT.size) < 0)
  967. {
  968. return -1;
  969. }
  970. #ifdef SORT_NET
  971. maxNet = readNETBLOCK (NETFN);
  972. #endif
  974. statlog = fopen("./stat.log", "a+");// create file "stat.log" automatically if it does not exist. 
  975. if (statlog == NULL) {
  976. perror ("error opening statistics log file: stat.log.n");
  977. }
  979. return 0;
  980. }
  983. int main(int argc, char **argv)
  984. {
  985. int i, mode = 1;
  986. // struct itimerval t, ot;
  988. signal (SIGPIPE, SIG_IGN); // SIGPIPE is raised when the client closes the socket exceptionally
  989.                            // if not handled, SIGPIPE would cause unexpected termination.
  990. // signal (SIGINT, my_exit);
  991. // argv[1]: daemon mode, not clear
  992. // argv[2]: output status, not used and not clear
  993. if (argc > 1)
  994. {
  995. mode = atoi (argv[1]);
  996. if (argc > 2) OUTPUT_STAT = atoi (argv[2]);
  997. }
  998. if (mode == 0)
  999. daemon(1,1); // run in the background
  1001. // read configuration file. just ignored right now.
  1002. // 参数为文件名, 一个struct NamVal *, 以及该struct NamVal的项数
  1003. read_config (CONFIG, ConfigParameters, sizeof(ConfigParameters)/sizeof (struct NamVal));
  1004. readconfig("ip.list");
  1006. for (i=0; i<10; i++) // retry 10 times. not useful here.
  1007. {
  1008. FD_ZERO(&osocks);
  1009. if (init_ts () < 0)
  1010. {
  1011. PDEBUG ("Error in initialization.n");
  1012. exit (1);
  1013. }
  1014. #ifdef HAVE_MYSQL
  1015. if ((local_mysql = init_mysql (MYSQL_HOST, MYSQL_USER, MYSQL_PASS, MYSQL_DB, "/var/run/mysqld/mysqld.sock")) == 0)
  1016. {
  1017. PDEBUG ("Error in init_mysql.n");
  1018. exit (1);
  1019. }
  1020. #endif
  1021. process_child ();
  1022. }
  1023. return 0;
  1024. }
  1027. //===============================================
  1028. //===== Here begin the message process part =====
  1029. //===============================================
  1031. int init_NP (struct Session *p)
  1032. {
  1033. // maxid: maximum session index currently in the list. for optimization of search
  1034. int listnum = p - NPTRACKER.head; // this is the index!
  1035. if (listnum > NPTRACKER.maxid)
  1036. NPTRACKER.maxid = listnum;
  1037. NPTRACKER.cur ++; // cur is in fact the counter of sessions
  1038. if (p->u.p.cur) // if there is an edge, then the client is already in a channel
  1039. PDEBUG ("NP %d in %d enter Channel %.32s(%d clients).n", p-NPTRACKER.head, NPTRACKER.cur, p->u.p.cur->head->name, p->u.p.cur->head->numclient);
  1040. else
  1041. PDEBUG ("NP %d in %d no default channel.n", p-NPTRACKER.head, NPTRACKER.cur);
  1042. return 0;
  1043. }
  1045. int process_NP (int idsock)
  1046. {
  1047. int len, listnum;
  1048. struct Session *p;
  1049. struct TSMessage *m = &UDPMsg;
  1050. #ifdef MEASUREMENT
  1051. //struct timeval tm;
  1052. //long long msec;
  1053. #endif
  1054. socklen_t addr_len = sizeof (UDPCLIENT);
  1056. memset ((char *)&UDPCLIENT, 0, sizeof (UDPCLIENT));
  1057. if ((len = (recvfrom (NPTRACKER.sock[idsock], &UDPMsg, MAX_DATA, 0, (struct sockaddr *)&UDPCLIENT, &addr_len))) < 0)
  1058. {
  1059. perror ("recvfrom:");
  1060. return -1;
  1061. }
  1063. #ifdef MEASUREMENT
  1064. //gettimeofday (&tm, NULL);
  1065. //msec = ((long long)tm.tv_sec) * 1000000l + tm.tv_usec;
  1066. #endif
  1067. PDEBUG ("type %d len %d.", m->type, m->len);
  1069. if (m->type == NP2TS_LOGIN)
  1070. {
  1071. process_NP2TS_LOGIN ((struct Message *)m);
  1072. ++np2tsLoginCount;
  1073. } else
  1074. {
  1075. listnum = m->authcode1; // index of session object
  1076. p = NPTRACKER.head+listnum;
  1077. // check the session: 1. bad index; 2. uninitialized or cleared; 3. not match
  1078. if (listnum >= NPTRACKER.max || p->socket == 0
  1079. || p->auth != m->authcode2)
  1080. {
  1081. if (m->type != NP2TS_LOGOUT)
  1082. SEND_NPMSG(NPTRACKER.sock[idsock],TS2NP_MSG,ERR_INTERNAL,1,&UDPCLIENT);
  1083. return -1;
  1084. }
  1085. switch (m->type)
  1086. {
  1087. case NP2TS_REPORT: // 报告Interval信息,如果refresh为true, 则重置, 否则则先增加后删除.
  1088. process_NP2TS_REPORT (p, m);
  1089. ++np2tsReportCount;
  1090. break;
  1091. case NP2TS_NEED_PEERS:
  1092. PDEBUG("Need peers!!n");
  1093. process_NP2TS_NEED_PEERS (p, m);
  1094. ++np2tsNeedPeerCount;
  1095. break;
  1096. case NP2TS_LOGOUT: // 退出
  1097. closure_NP (p);
  1098. ++np2tsLogoutCount;
  1099. break;
  1100. case NP2TS_RES_LIST: /* 发送当前NP的所有RESOURCE,使用addSession来进行处理, 
  1101.         如果还没有这条边, 就添加. */
  1103. process_NP2TS_RES_LIST (p, m);
  1104. ++np2tsResListCount;
  1105. break;
  1106. case NP2TS_REQ_RES: // 添加RES, 并返回Peers
  1107. process_NP2TS_REQ_RES (p, m);
  1108. ++np2tsReqResCount;
  1109. break;
  1110. case NP2TS_DEL_RES: // 删除RES
  1111. process_NP2TS_DEL_RES (p, m);
  1112. ++np2tsDelResCount;
  1113. break;
  1114. case NP2TS_QUERY_RES: //查询RES
  1115. process_NP2TS_QUERY_RES (p, m);
  1116. break;
  1117. case NP2TS_REPORT2:
  1118. process_NP2TS_REPORT2 (p, m);
  1119. break;
  1120. default:
  1121. SEND_NPMSG(NPTRACKER.sock[idsock],TS2NP_MSG,ERR_INTERNAL,1,&UDPCLIENT);
  1122. closure_NP (p);
  1123. break;
  1124. }
  1125. p->last_access = CurTimeSec;
  1126. }
  1127. PDEBUG ("donen");
  1128. #ifdef MEASUREMENT
  1129. //gettimeofday (&tm, NULL);
  1130. //PDEBUG ("msg type %d, len %d: %lld msec.n", m->type, m->len, ((long long)tm.tv_sec) * 1000000l + tm.tv_usec - msec);
  1131. #endif
  1132. return 0;
  1133. }
  1135. void logStat(struct Session *p)
  1136. {
  1137. // 打印记录的时间和客户端的版本号
  1138. fprintf(statlog, "%u %f", time(NULL), p->clientVer);
  1139.        fprintf(statlog, "%d:%d | %d:%dn", p->host, p->port, p->intra, p->npport);
  1140.        fprintf(statlog, "%ut%ut%ut%ut%ut%ut%ut%ut%ut%fn", 
  1141.                        p->u.p.s.playingBlock - p->u.p.startBlock,
  1142.                        p->u.p.s.currBufferTime,
  1143.                        p->u.p.s.bufferCount,
  1144.                        p->u.p.s.bufferTime,
  1145.                        p->u.p.s.connFailCount,
  1146.                        p->u.p.s.inConnections,
  1147.                        p->u.p.s.outConnections,
  1148.                        p->u.p.s.avgInConnTime,
  1149.                        p->u.p.s.avgOutConnTime,
  1150.                        p->u.p.s.messagePercent);
  1151.        fprintf(statlog, "%lldt%lldt%ft%ft%ft%fn",
  1152.                        p->u.p.t.totalDownBytes,
  1153.                        p->u.p.t.totalUpBytes,
  1154.                        p->u.p.t.currDownSpeed,
  1155.                        p->u.p.t.currUpSpeed,
  1156.                        p->u.p.t.avgDownSpeed,
  1157.                        p->u.p.t.avgUpSpeed);
  1158.                        
  1159. current_log_count ++;
  1160. if(current_log_count == MAX_LOG_COUNT)
  1161. {
  1162. fflush(statlog);
  1163. current_log_count = 0;
  1164. }
  1165. }
  1167. int closure_NP (struct Session *p)
  1168. {
  1169. int i, id;
  1170. struct Session *q;
  1172.         // write statistics to log file
  1173.         logStat(p);
  1175. // 1. decrease <maxid> if current session is the last
  1176. if ((i = p - NPTRACKER.head) == NPTRACKER.maxid && i > 0)
  1177. {
  1178. for (i--; NPTRACKER.head[i].socket == 0 && i> 0; i--);
  1179. NPTRACKER.maxid = i;
  1180. }
  1182. // 2. delete corresponding session in the session-channel map
  1183. delSession (p);
  1185. // 3. remove session from the hash table
  1186. id = hash_np (p->host, p->npport);
  1187. if ((q = NPTRACKER.hash[id]) != p) // not head of chain
  1188. {
  1189. for (; q && q->hnext != p; q=q->hnext); // search through the chain for the parent of <p>
  1190. assert (q);
  1191. if (q) q->hnext = p->hnext; // remove
  1192. } else NPTRACKER.hash[id] = p->hnext; // head of chain, got it
  1194. // 4. clear and free session object to the freelist
  1195. memset (p, 0, sizeof (struct Session)); // clear session
  1196. p->hnext = NPTRACKER.hash[0];
  1197. NPTRACKER.hash[0] = p;
  1198. Polluted ++;
  1199. NPTRACKER.cur --;
  1200. return 0;
  1201. }
  1203. int init_CP (struct Session *p)
  1204. {
  1205. const char* servicetype;
  1208. servicetype = find_cp_service_type(p->host);
  1209. if(servicetype == NULL)
  1210. servicetype = "UNKNOWN";
  1211. strcpy(p->u.cp.servicetype, servicetype);
  1213. PDEBUG("n******************************************************************ninit_CP: cp service type is %sn", servicetype);
  1215. int listnum = p - CPTRACKER.head;
  1216. if (listnum > CPTRACKER.maxid)
  1217. CPTRACKER.maxid = listnum;
  1218. CPTRACKER.cur ++;
  1219. GCPCHOICE = p;
  1221. // add_cp_to_list((void*)p);
  1223. return 0;
  1224. }
  1226. int process_CP (int idsock)
  1227. {
  1228. int len, listnum;
  1229. struct Session *p;
  1230. struct TSMessage *m = &UDPMsg;
  1231. #ifdef MEASUREMENT
  1232. struct timeval tm;
  1233. long long msec;
  1234. #endif
  1235. socklen_t addr_len = sizeof (UDPCLIENT);
  1237. memset ((char *)&UDPCLIENT, 0, sizeof (UDPCLIENT));
  1238. if ((len = (recvfrom (CPTRACKER.sock[idsock], &UDPMsg, MAX_DATA, 0, (struct sockaddr *)&UDPCLIENT, &addr_len))) < 0)
  1239. {
  1240. perror ("recvfrom:");
  1241. return -1;
  1242. }
  1244. #ifdef MEASUREMENT
  1245. gettimeofday (&tm, NULL);
  1246. msec = ((long long)tm.tv_sec) * 1000000l + tm.tv_usec;
  1247. #endif
  1249. if (m->type == CP2TS_REGISTER)
  1250. {
  1251. process_CP2TS_REGISTER ((struct Message *)m);
  1252. } else
  1253. {
  1254. listnum = m->authcode1;
  1255. p = CPTRACKER.head+listnum;
  1256. if (listnum >= CPTRACKER.max || p->socket == 0
  1257. || p->auth != m->authcode2)
  1258. {
  1259. if (m->type != CP2TS_LOGOUT)
  1260. {
  1261. PDEBUG("CP error. listnum=%d/%d. socket=%d auth=%d/%dn", listnum, CPTRACKER.max, p->socket, p->auth, m->authcode2);
  1262. SEND_NPMSG(CPTRACKER.sock[idsock],TS2CP_MSG,ERR_INTERNAL,1,&UDPCLIENT);
  1263. }
  1264. return -1;
  1265. }
  1266. switch (m->type)
  1267. {
  1268. case CP2TS_NEED_PEERS: // ECP查询用, 目前尚未使用
  1269. process_CP2TS_NEED_PEERS (p, m);
  1270. break;
  1271. case CP2TS_UPDATE: // 报告CP负载
  1272. process_CP2TS_UPDATE (p, m);
  1273. break;
  1274. case CP2TS_LOGOUT:
  1275. closure_CP (p);
  1276. break;
  1277. default:
  1278. closure_CP (p);
  1279. break;
  1280. }
  1281. p->last_access = CurTimeSec;
  1282. }
  1283. return 0;
  1284. }
  1286. int closure_CP (struct Session *p)
  1287. {
  1288. int i, id;
  1289. struct Session *q;
  1291. //PDEBUG("closure_CP.n");
  1292. //remove_cp_from_list((void*)p);
  1293. //PDEBUG("closure_CP OK.n");
  1295. if ((i = p - CPTRACKER.head) == CPTRACKER.maxid && i > 0)
  1296. {
  1297. for (i--; CPTRACKER.head[i].socket == 0 && i> 0; i--);
  1298. CPTRACKER.maxid = i;
  1299. }
  1300. if (GCPCHOICE == p)
  1301. {
  1302. for (i=CPTRACKER.maxid; i>=0; i--)
  1303. {
  1304. if (CPTRACKER.head[i].socket != 0 && CPTRACKER.head[i].u.cp.type == CT_GENERAL)
  1305. {
  1306. GCPCHOICE = &(CPTRACKER.head[i]);
  1307. break;
  1308. }
  1309. }
  1310. }
  1311. id = hash_cp (p->host, p->npport);
  1312. if ((q = CPTRACKER.hash[id]) != p)
  1313. {
  1314. for (; q && q->hnext != p; q=q->hnext);
  1315. assert (q);
  1316. if (q) q->hnext = p->hnext;
  1317. } else CPTRACKER.hash[id] = p->hnext;
  1318. memset (p, 0, sizeof (struct Session));
  1319. p->hnext = CPTRACKER.hash[0];
  1320. CPTRACKER.hash[0] = p;
  1321. Polluted ++;
  1322. CPTRACKER.cur --;
  1323. return 0;
  1324. }
  1326. #ifdef HAVE_RM
  1327. int getChannelInfo (char *md5, char **buf)
  1328. {
  1329. struct Channel *pc;
  1330. int i, total=0;
  1331. if (strcmp (md5, "*") == 0)
  1332. {
  1333. for (i=0; i<MAX_CHANNEL; i++)
  1334. {
  1335. for (pc=ChannelHash[i]; pc; pc=pc->next)
  1336. {
  1337. memcpy (*buf, pc->name, MD5_LEN);
  1338. *buf += MD5_LEN;
  1339. *(int *)(*buf) = pc->numclient;
  1340. *buf += sizeof (int);
  1341. total ++;
  1342. }
  1343. }
  1344. return total;
  1345. }
  1346. if ((pc=findChannel (md5, MD5_LEN)) != NULL)
  1347. {
  1348. memcpy (*buf, md5, MD5_LEN);
  1349. *buf += MD5_LEN;
  1350. *(int *)(*buf) = pc->numclient;
  1351. *buf += sizeof (int);
  1352. return 1;
  1353. }
  1355. for (i=0; i<MAX_CHANNEL; i++)
  1356. {
  1357. for (pc=ChannelHash[i]; pc; pc=pc->next)
  1358. {
  1359. if (strstr (pc->name, md5) != NULL)
  1360. {
  1361. memcpy (*buf, pc->name, MD5_LEN);
  1362. *buf += MD5_LEN;
  1363. *(int *)(*buf) = pc->numclient;
  1364. *buf += sizeof (int);
  1365. total ++;
  1366. }
  1367. }
  1368. }
  1369. return total;
  1370. }
  1372. int init_RM (struct Session *p)
  1373. {
  1374. int listnum = p - RMTRACKER.head;
  1375. if (listnum > RMTRACKER.maxid)
  1376. RMTRACKER.maxid = listnum;
  1377. RMTRACKER.cur ++;
  1378. return 0;
  1379. }
  1380. #define RM2TS_STAT_QUERY 0x20
  1381. #define TS2RM_STAT_RESPONSE 0x30
  1382. int process_RM (int idsock)
  1383. {
  1384. char buffer[MAX_DATA];
  1385. char *p, *buf = buffer;
  1386. int * psize;
  1387. int querynum;
  1388. int len, total, i;
  1389. struct Message Msg, *m=&Msg;
  1390. #ifdef MEASUREMENT
  1391. struct timeval tm;
  1392. long long msec;
  1393. #endif
  1394. socklen_t addr_len = sizeof (UDPCLIENT);
  1396. memset ((char *)&UDPCLIENT, 0, sizeof (UDPCLIENT));
  1397. if ((len = (recvfrom (RMTRACKER.sock[idsock], m, MAX_DATA, 0, (struct sockaddr *)&UDPCLIENT, &addr_len))) < 0)
  1398. {
  1399. perror ("recvfrom:");
  1400. return -1;
  1401. }
  1403. #ifdef MEASUREMENT
  1404. gettimeofday (&tm, NULL);
  1405. msec = ((long long)tm.tv_sec) * 1000000l + tm.tv_usec;
  1406. #endif
  1407. PDEBUG("got RM msg, len %d. n", len);
  1408. switch (m->type)
  1409. {
  1410. case RM2TS_STAT_QUERY:
  1411. querynum = *(int*)m->buffer;
  1412. if(querynum > 100)
  1413. break; 
  1414. if(querynum > 0)
  1415. {
  1416. buf += sizeof (int);
  1417. *(unsigned char *)buf = TS2RM_STAT_RESPONSE;
  1418. buf += sizeof (char);
  1419. psize = (int*)buf;
  1420. buf += sizeof (int);
  1421. total = 0;
  1423. p = m->buffer + sizeof(int);
  1424. for(i = 0; i < querynum; ++i)
  1425. {
  1426. PDEBUG("query %s. n", p);
  1427. total += getChannelInfo(p, &buf);
  1428. p += MD5_LEN;
  1429. }
  1430. *(int *)buffer = buf - buffer;
  1431. *psize = total;
  1432. if(*psize > 0)
  1433. sendMessage(RMTRACKER.sock[idsock], buffer, &UDPCLIENT);
  1434. }
  1435. break;
  1436. default:
  1437. break;
  1438. }
  1439. #ifdef MEASUREMENT
  1440. gettimeofday (&tm, NULL);
  1441. PDEBUG ("msg type %d, len %d: %lld msec.n", m->type, m->len, ((long long)tm.tv_sec) * 1000000l + tm.tv_usec - msec);
  1442. #endif
  1443. return 0;
  1444. }
  1446. int closure_RM (struct Session *p)
  1447. {
  1448. int i;
  1449. if ((i = p - RMTRACKER.head) == RMTRACKER.maxid && i > 0)
  1450. {
  1451. for (i--; RMTRACKER.head[i].socket == 0 && i> 0; i--);
  1452. RMTRACKER.maxid = i;
  1453. }
  1454. return 0;
  1455. }
  1456. #endif
  1459. //------------------------------------------------
  1460. //- Here begin the specific message process part -
  1461. //------------------------------------------------
  1463. // | login id(UINT32) | md5 password(MD5_LEN) |
  1464. // | version of client(float) | listening port(USHORT) |
  1465. // | size of local ip list(UINT8) | first ip addr(in_addr) |... |
  1467. /* NP向TS登录, 按照来源IP地址和所报告的npport进行hash, 如果距离上次
  1468.    发送NP2TS_LOGIN消息的时间小于SILENCE_TIME, 则直接返回,否则发送WELCOME消息. */
  1469. int process_NP2TS_LOGIN (struct Message *m)
  1470. {
  1471. struct Session *p;
  1472. // struct ChannelInfo tmpch;
  1473. char md5[MD5_LEN+1];
  1474. unsigned int host, myhost, intra;
  1475. unsigned short port, npport;
  1476. int i,id, userID;
  1477. float clientVer;
  1478. unsigned int num;
  1479. char *buf;
  1480. struct P2PAddress *addr;
  1482. buf = m->buffer;
  1483. userID = *(int *)buf;
  1484. buf += sizeof (int);
  1485. memcpy (md5, buf, MD5_LEN);
  1486. md5[MD5_LEN] = 0;
  1487. buf += MD5_LEN;
  1488. #ifdef HAVE_MYSQL
  1489. if (authUser (userID, md5, local_mysql, NULL) == 0)
  1490. {
  1491. SEND_NPMSG(NPTRACKER.sock[CurrentSock],TS2NP_MSG,ERR_INTERNAL,1,&UDPCLIENT);
  1492. closure_NP (p);
  1493. return -1;
  1494. }
  1495. #endif
  1496. // check protocol version
  1497. clientVer = *(float*)buf;
  1498. buf += sizeof (float);
  1499. if(clientVer < MIN_CLIENT_VERSION) {
  1500. SEND_NPMSG(NPTRACKER.sock[CurrentSock], TS2NP_MSG, ERR_LOW_VERSION, 1, &UDPCLIENT);
  1501. return -1;
  1502. }
  1504. npport = ntohs (*(unsigned short *)buf);
  1505. buf += sizeof (short);
  1507. host = ntohl (UDPCLIENT.sin_addr.s_addr);
  1508. port = ntohs (UDPCLIENT.sin_port);
  1509. // find client session in the session cluster (hash table)
  1510. id = hash_np (host, npport);
  1511. for (p=NPTRACKER.hash[id]; p; p=p->hnext)
  1512. if (p->host == host && p->port == port && p->npport == npport)
  1513. break;
  1514. if (!p) // not found, allocate and add new session object
  1515. {
  1516. // error adding session: TS full or uninitialized
  1517. if (NPTRACKER.cur >= NPTRACKER.max || NPTRACKER.hash[0] == 0)
  1518. {
  1519. SEND_NPMSG(NPTRACKER.sock[CurrentSock],TS2NP_MSG,ERR_INTERNAL,1,&UDPCLIENT);
  1520. return -1;
  1521. }
  1522. // allocate session object from the freelist indicated by hash[0]
  1523. // allocate the FIRST free object in hash[0] and insert it into HEAD of corresponding bucket
  1524. p = NPTRACKER.hash[0];
  1525. NPTRACKER.hash[0] = p->hnext;
  1526. p->hnext = NPTRACKER.hash[id];
  1527. NPTRACKER.hash[id] = p;
  1528. // fill in the object
  1529. p->socket = NPTRACKER.sock[CurrentSock];
  1530. p->type = TYPE_NP;
  1531. p->port = port;
  1532. p->npport = npport;
  1533. p->host = host;
  1534. p->clientVer = clientVer;
  1535. #ifdef SORT_NET
  1536. {
  1537. struct networks *pnetworks = getnetwork (host, NETBLOCKS, maxNet);
  1538. if (pnetworks) p->net = pnetworks->net;
  1539. else p->net = 0;
  1540. }
  1541. #endif
  1542. p->time_sec = CurTimeSec;
  1544. // process the ip list
  1545. num = *(unsigned char *)buf; // size of local ip list
  1546. buf += sizeof (char);
  1547. intra = 0;
  1548. for (i=0; i<num; i++)
  1549. {
  1550. myhost = ntohl (*(unsigned int *)buf);
  1551. buf += sizeof (int);
  1552. if (intra == 0xffffffff)
  1553. continue; // Must use continue to modify buf to right place
  1554. if ((myhost >> 16) == 0xc0a8) // 0xc0a8 == 192.168
  1555. {
  1556. intra = myhost;
  1557. if(host == intra)
  1558. intra = 0xffffffff;
  1559. }
  1560. else if (myhost == host)
  1561. intra = 0xffffffff;
  1562. }
  1563. p->intra = intra;
  1564. p->auth = random ();
  1566. //init statistics
  1567. p->u.p.startBlock = -1;
  1569. init_NP (p);
  1570. } else if (CurTimeSec < p->last_access + SILENCE_TIME)
  1571. return 0; // still active, do NOTHING
  1572. if (buf - m->buffer + NORMAL_HEADER > m->len)
  1573. {
  1574. PDEBUG ("Invalid message NP2TS_LOGIN, length %d not enough.n", m->len);
  1575. return -1;
  1576. }
  1577. // Time to send back welcome message to NP
  1578. UDPMsg.type = TS2NP_WELCOME;
  1579. UDPMsg.len = 12+sizeof(struct P2PAddress);
  1580. UDPMsg.authcode1 = p-NPTRACKER.head;
  1581. UDPMsg.authcode2 = p->auth;
  1582. addr = (struct P2PAddress *)(UDPMsg.buffer);
  1583. addr->outerIP.sin_family = addr->subnetIP.sin_family = AF_INET;
  1584. addr->outerIP.sin_addr.s_addr = htonl (p->host);
  1585. addr->subnetIP.sin_addr.s_addr = htonl (p->intra);
  1586. addr->outerIP.sin_port = ntohs (port);
  1587. addr->subnetIP.sin_port = ntohs (npport);
  1589. ++ts2npWelcomeCount;
  1591. if (sendMessage(p->socket, (char *)&UDPMsg, &UDPCLIENT) < 0)
  1592. {
  1593. closure_NP (p);
  1594. return -1;
  1595. }
  1596. // process_NEED_PEERS_real (p, NULL, 1, 0, 0);
  1598. return 0;
  1599. }
  1601. // |---CHECK DIGITS(7 BYTEs)---|res count(UINT8)|RESOURCE MD5(MD5_LEN)|...|
  1602. int process_NP2TS_QUERY_RES (struct Session *p, struct TSMessage *m)
  1603. {
  1604. int i;
  1605. char *buf = m->buffer;
  1606. char buffer[MAX_DATA];
  1607. char *resultMsg, *prescount;
  1608. struct Channel *pc;
  1609. unsigned char num = *(unsigned char *)buf; //# of queried resources
  1610. buf += sizeof (char);
  1611. resultMsg = buffer+sizeof(int);
  1612. *(unsigned char *)resultMsg = TS2NP_RESINFO;
  1613. resultMsg += sizeof (char);
  1614. prescount = resultMsg;
  1615. *(unsigned char *)prescount = 0; // counter of resources
  1616. for (i=0; i<num; i++)
  1617. {
  1618. if ((pc = findChannel (buf, MD5_LEN)) != NULL)
  1619. {
  1620. (*(unsigned char *)prescount) ++;
  1621. memcpy (resultMsg, buf, MD5_LEN);
  1622. resultMsg += MD5_LEN;
  1623. *(unsigned short *)resultMsg = pc->numclient;
  1624. resultMsg += sizeof (short);
  1625. }
  1626. buf += MD5_LEN;
  1627. }
  1628. if (buf - m->buffer + AUTH_HEADER > m->len)
  1629. {
  1630. PDEBUG ("Invalid message NP2TS_LOGIN, length %d not enough.n", m->len);
  1631. closure_NP (p);
  1632. return -1;
  1633. }
  1634. *(unsigned int *)buffer = resultMsg - buffer; // The length of the message
  1635. if (sendMessage(p->socket,buffer,&UDPCLIENT) < 0)
  1636. {
  1637. // closure_NP (p);
  1638. return -1;
  1639. }
  1640. return 0;
  1641. }
  1643. // new version of REPORT with TransferInfo, ignoring TransferInfo
  1644. // | ---CHECK DIGITS(7 BYTEs)--- |
  1645. // | info of peer(CorePeerInfo) | refresh(bool) |
  1646. // | Interval count(UINT8) | first BlockInterval | ... |
  1647. // | transfer Info(TransferInfo) |
  1648. int process_NP2TS_REPORT (struct Session *p, struct TSMessage *m)
  1649. {
  1650. unsigned char type;
  1651. char *buf = m->buffer;
  1653. // 1. extract CorePeerInfo
  1654. memcpy (&(p->u.p.p), buf, sizeof (struct CorePeerInfo));
  1655. buf += sizeof (struct CorePeerInfo);
  1657. if (p->u.p.cur)
  1658. {
  1659. type = *(unsigned char *)buf;
  1660. buf += sizeof (char);
  1661. if (type) // refresh==true
  1662. {
  1663. p->u.p.cur->numinter = *(unsigned char *)buf;
  1664. PDEBUG ("Set %d intervalsn", p->u.p.cur->numinter);
  1665. buf += sizeof (char);
  1666. if (p->u.p.cur->numinter > 0 && p->u.p.cur->numinter < MAX_INTERVAL)
  1667. memcpy (p->u.p.cur->inter, buf, p->u.p.cur->numinter*sizeof (struct Interval));
  1668. else p->u.p.cur->numinter = 0;
  1669. } else // refresh==false incremental update
  1670. {
  1671. type = *(unsigned char *)buf; // # of newly added intervals
  1672. buf += sizeof (char);
  1673. PDEBUG ("Add %d intervals,", type);
  1674. if (type > 0) // merge & delete are not good name!
  1675. {
  1676. p->u.p.cur->numinter = merge (p->u.p.cur->inter, p->u.p.cur->numinter, (struct Interval *)buf, type);
  1677. buf += type*sizeof(struct Interval);
  1678. }
  1679. type = *(unsigned char *)buf;
  1680. buf += sizeof (char);
  1681. if (type > 0)
  1682. {
  1683. p->u.p.cur->numinter = delete_interval (p->u.p.cur->inter, p->u.p.cur->numinter, (struct Interval *)buf, type);
  1684. buf += type*sizeof(struct Interval);
  1685. }
  1686. PDEBUG ("del %d intervals, now is %d.n", type, p->u.p.cur->numinter);
  1687. }
  1688. }
  1689. // ignore TransferInfo, but still count the size for integrity check
  1690. // --DELETED-- copy transferinfo of client
  1691. // --DELETED-- memcpy (&(p->u.p.t), buf, sizeof (struct TransferInfo));
  1692. buf += sizeof(struct TransferInfo);
  1693. if (buf - m->buffer + AUTH_HEADER > m->len)
  1694. {
  1695. PDEBUG ("Invalid message NP2TS_LOGIN, length %d not enough.n", m->len);
  1696. closure_NP (p);
  1697. return -1;
  1698. }
  1699. return 0;
  1700. }
  1702. int process_NP2TS_REPORT2 (struct Session *p, struct TSMessage *m)
  1703. {
  1704. char *buf = m->buffer;
  1706. // TODO: 
  1707. //  extract current playing block and count them
  1708. memcpy (&(p->u.p.s), buf, sizeof (struct StatInfo));
  1709. buf += sizeof (struct StatInfo);
  1710. memcpy (&(p->u.p.t), buf, sizeof (struct TransferInfo));
  1711. buf += sizeof (struct TransferInfo);
  1713. // check if this is the first report and record starting block of NP
  1714. if ( p->u.p.startBlock == -1 ) {
  1715. p->u.p.startBlock = p->u.p.s.playingBlock;
  1716. }
  1718. if (buf - m->buffer + AUTH_HEADER > m->len)
  1719. {
  1720. PDEBUG ("Invalid message NP2TS_LOGIN, length %d not enough.n", m->len);
  1721. closure_NP (p);
  1722. return -1;
  1723. }
  1724. return 0;
  1725. }
  1726. // |---CHECK DIGITS(7 BYTEs)---|
  1727. // |resource list size(UINT8)|
  1728. // |RESOURCE MD5(MD5_LEN)|Interval count(UINT8)|first BlockInterval|...|
  1729. int process_NP2TS_RES_LIST (struct Session *p, struct TSMessage *m)
  1730. {
  1731. unsigned int num, i;
  1732. struct ChannelInfo c;
  1733. char *buf = m->buffer;
  1734. // unsigned char needcp;
  1735. num = *(unsigned char *)buf;
  1736. buf += sizeof (char);
  1737. for (i=0; i<num; i++)
  1738. {
  1739. memcpy (c.md5, buf, MD5_LEN);
  1740. c.md5[MD5_LEN] = 0;
  1741. buf += MD5_LEN;
  1742. c.numinter = *(unsigned char *)buf;
  1743. buf += sizeof (char);
  1744. if (c.numinter > MAX_INTERVAL) return -1;
  1745. memcpy (c.inter, buf, c.numinter*sizeof(struct Interval));
  1746. buf += sizeof(struct Interval) * c.numinter;
  1747. addSession (p, &c, 1, 0);
  1748. }
  1749. if (buf - m->buffer + AUTH_HEADER > m->len)
  1750. {
  1751. PDEBUG ("Invalid message NP2TS_LOGIN, length %d not enough.n", m->len);
  1752. closure_NP (p);
  1753. return -1;
  1754. }
  1755. return 0;
  1756. }
  1758. // |---CHECK DIGITS(7 BYTEs)---|
  1759. // |RESOURCE MD5(MD5_LEN)|Interval count(UINT8)|first BlockInterval|...|
  1760. // |current block(UINT32)|need CP(bool)|
  1761. int process_NP2TS_REQ_RES (struct Session *p, struct TSMessage *m)
  1762. {
  1763. struct ChannelInfo c;
  1764. unsigned int cur;
  1765. char *buf = m->buffer;
  1766. unsigned char needcp;
  1767. memcpy (c.md5, buf, MD5_LEN);
  1768. c.md5[MD5_LEN] = 0;
  1769. buf += MD5_LEN;
  1770. c.numinter = *(unsigned char *)buf;
  1771. buf += sizeof (char);
  1772. if (c.numinter > MAX_INTERVAL) return -1;
  1773. memcpy (c.inter, buf, c.numinter*sizeof(struct Interval));
  1774. buf += sizeof(struct Interval) * c.numinter;
  1775. cur = *(unsigned int *)buf;
  1776. buf += sizeof (int);
  1777. needcp = *(unsigned char *)buf;
  1778. buf += sizeof (char);
  1779. if (buf - m->buffer + AUTH_HEADER > m->len)
  1780. {
  1781. PDEBUG ("Invalid message NP2TS_LOGIN, length %d not enough.n", m->len);
  1782. closure_NP (p);
  1783. return -1;
  1784. }
  1785. addSession (p, &c, 1, cur);
  1786. process_NEED_PEERS_real (p, c.md5, needcp, cur, 0);
  1787. SEND_NPMSG(NPTRACKER.sock[CurrentSock],TS2NP_MSG,ERR_ADD_RES_OK,0,&UDPCLIENT);
  1788. return 0;
  1789. }
  1790. // NP has quit from one channel
  1791. // |---CHECK DIGITS(7 BYTEs)---|RESOURCE MD5(MD5_LEN)|
  1792. int process_NP2TS_DEL_RES (struct Session *p, struct TSMessage *m)
  1793. {
  1794. struct Edge *pedge;
  1795. struct Channel *pc;
  1796. char *buf = m->buffer;
  1797. if ((pc=findChannel (buf, MD5_LEN)) == NULL)
  1798. return -1;
  1799. buf += MD5_LEN;
  1800. if (buf - m->buffer + AUTH_HEADER > m->len)
  1801. {
  1802. PDEBUG ("Invalid message NP2TS_LOGIN, length %d not enough.n", m->len);
  1803. closure_NP (p);
  1804. return -1;
  1805. }
  1806. for (pedge=p->u.p.header; pedge && pedge->head != pc; pedge=pedge->enext);
  1807. if (pedge) delEdge (pedge);
  1808. return 0;
  1809. }
  1811. int process_NEED_PEERS_real (struct Session *p, char *md5, int needcp, unsigned int cur, unsigned char layer)
  1812. {
  1813. char buffer[MAX_DATA], buffer1[MAX_DATA]; /* buffer is used to hold TS2CP/NP_PEERS message, 
  1814.      buffer1 is used to hold TS2NP_CONNECT_TO message */
  1815. int num, numcp=0, numnp, conn = p->socket;
  1816. char *buf, *psize;
  1817. struct Channel *pc;
  1818. struct P2PAddress *addr;
  1820. PDEBUG ("NEED_PEERS type %d needcp %d cur %dn", p->type, needcp, cur);
  1821. buf = buffer+sizeof (int);
  1822. if (p->type == TYPE_CP)
  1823. *(unsigned char *)buf = TS2CP_PEERS;
  1824. else
  1825. *(unsigned char *)buf = TS2NP_PEERS;
  1826. buf += sizeof (char);
  1828. ++ts2npPeersCount;
  1830. if (md5) // md5 of the channel
  1831. {
  1832. pc = findChannel (md5, MD5_LEN);
  1833. if (p->type == TYPE_CP)
  1834. {
  1835. memcpy (buf, md5, MD5_LEN);
  1836. buf += MD5_LEN;
  1837. }
  1838. } else if (p->u.p.cur) // Session->NPInfo->CurrentEdge
  1839. pc = p->u.p.cur->head;
  1840. else
  1841. pc = NULL;
  1843. if (pc == NULL) 
  1844. {
  1845. PDEBUG("no channeln");
  1846. return -1;
  1847. }
  1849. num = MAX_PEER;
  1850. if (needcp)
  1851. {
  1852. psize = buf;
  1853. buf += sizeof (char);
  1854. if (p->type == TYPE_CP)
  1855. *(unsigned char *)psize = numcp = findCPPeers (0, pc->name, &buf);
  1856. else
  1857. //*(unsigned char *)psize = numcp = findCPPeers (ntohs (p->host & 0xffff0000), pc->name, &buf);
  1858. *(unsigned char *)psize = numcp = findCPPeers (p->host, pc->name, &buf);
  1859. } else
  1860. {
  1861. *(unsigned char *)buf = 0;
  1862. buf += sizeof (char);
  1863. }
  1864. psize = buf;
  1865. buf += sizeof (char);
  1866. if (p->intra == 0xffffffff) //NP is in the public network, ask NP to contact other NPs
  1867. {
  1868. // |len(INT32) | type(INT8)|target addr(P2PAddress)|connect for free(bool)|
  1869. *(int *)buffer1 = sizeof(struct P2PAddress) + sizeof (int) + 2*sizeof (char);
  1870. *(unsigned char *)(buffer1+sizeof (int)) = TS2NP_CONNECT_TO;
  1871. addr = (struct P2PAddress *)(buffer1+sizeof(int)+sizeof(char));
  1872. addr->outerIP.sin_family = PF_INET;
  1873. addr->subnetIP.sin_family = PF_INET;
  1874. addr->outerIP.sin_port = htons (p->port);
  1875. addr->outerIP.sin_addr.s_addr = htonl (p->host);
  1876. addr->subnetIP.sin_port = htons (p->npport);
  1877. addr->subnetIP.sin_addr.s_addr = htonl (p->intra);
  1878. *(unsigned char *)(addr + 1) = 0; // connect for free(bool)
  1880. PDEBUG("Send ConnectTo %sn", inet_ntoa(addr->outerIP.sin_addr));
  1882. numnp = findNPPeers (pc, p, cur, num, &buf, buffer1);
  1883. } else
  1884. {
  1885. numnp = findNPPeers (pc, p, cur, num, &buf, NULL);
  1886. }
  1887. *psize = numnp; // set # of NP
  1888. *(int *)buffer = buf - buffer;
  1889. if (sendMessage(conn,buffer,&UDPCLIENT) < 0)
  1890. {
  1891. // closure_NP (p);
  1892. PDEBUG("send msg errn");
  1893. return -1;
  1894. }
  1895. PDEBUG ("find %d NP and %d CPn", numnp, numcp);
  1896. return numnp+numcp;
  1897. }
  1900. /* 查询Peer信息, 使用findCPPeer寻找合适的CP, 使用findNPPeers寻找合适的NP. 
  1901.    NP寻找时, 找到结果后按照networks来排序,保证在同一个网络中的排在前面. */
  1902. int process_NP2TS_NEED_PEERS (struct Session *p, struct TSMessage *m)
  1903. {
  1904. char *buf = m->buffer;
  1905. int needcp;
  1907. needcp = *(unsigned char *)buf;
  1908. buf += sizeof (char);
  1909. if(p->u.p.cur == NULL)
  1910. {
  1911. PDEBUG("no current in NPInfon");
  1912. return -1;
  1913. }
  1914. p->u.p.cur->current = *(unsigned int *)buf;
  1915. buf += sizeof (int);
  1916. p->u.p.p.layer= *(unsigned char *)buf;
  1917. buf += sizeof (char);
  1918. if (buf - m->buffer + AUTH_HEADER > m->len)
  1919. {
  1920. PDEBUG ("Invalid message NP2TS_LOGIN, length %d not enough.n", m->len);
  1921. closure_NP (p);
  1922. return -1;
  1923. }
  1925. return process_NEED_PEERS_real (p, NULL, needcp, p->u.p.cur->current, p->u.p.p.layer);
  1927. }
  1929. int findCPPeers (unsigned long host, char *md5, char **buffer)
  1930. {
  1931. // float minband = -1;
  1932. struct Session* p = NULL, *choice1 = NULL, *choice2 = NULL;
  1933. struct NormalAddress *addr1;
  1934. struct NormalAddress *addr2;
  1935. int i;
  1936. // We should start with a random address
  1938. p = CPTRACKER.head + rand() % (CPTRACKER.maxid+1);
  1940. //use pure random address instead of comparing the band
  1942. PDEBUG("In findCPPeers.n");
  1943. int minpriority = -1;
  1945. for( i=0; i <= CPTRACKER.maxid && host != 0; i++, p++)
  1946. {
  1947. PDEBUG("begin to call findcppeers.n");
  1948. if( p-CPTRACKER.head > CPTRACKER.maxid ) // round back to head
  1949. {
  1951. p = CPTRACKER.head;
  1952. }
  1954. if(p->u.cp.maxConn == 1 || p->socket == 0)
  1955. {
  1956. PDEBUG("Invalid CP.n");
  1957. continue;
  1958. }
  1961. int priority = findcppeers(host, (void*)p);
  1963. //如果不是对应的CP，查找下一个
  1964. if(priority == -1)
  1965. {
  1966. PDEBUG("not found.n");
  1967. continue;
  1968. }
  1970. PDEBUG("found: host : %d  servicetype: %s.n", host, p->u.cp.servicetype);
  1971. if(minpriority == -1)
  1972. minpriority = priority;
  1974. if(priority == 1)// priority - 1~n
  1975. {
  1976. if(choice1 == NULL)
  1977. {
  1978. choice1 = p;
  1979. minpriority = -1;
  1980. }
  1981. else
  1982. {
  1983. choice2 = p;
  1984. break;
  1985. }
  1986. }
  1987. else
  1988. {
  1989. if(priority < minpriority)
  1990. {
  1991. if(choice1 == NULL)
  1992. {
  1993. choice1 = p;
  1994. minpriority = -1;
  1995. }
  1996. else
  1997. {
  1998. choice2 = p;
  1999. break;
  2000. }
  2001. }
  2002. }
  2003. }
  2005. if(i > CPTRACKER.maxid)//find nothing via findcppeers, return CP directly
  2006. {
  2007. PDEBUG("begin to find sequencely.n");
  2008. for( i=0; i <= CPTRACKER.maxid; i++, p++)
  2009. {
  2010. if( p-CPTRACKER.head > CPTRACKER.maxid ) // round back to head
  2011. {
  2012. p = CPTRACKER.head;
  2013. }
  2015. if(p->u.cp.maxConn == 1 || p->socket == 0)
  2016. continue;
  2018. // 始终保证choice1比choice2先获得值，如果两个都有了就break
  2019. if(choice1 == NULL)
  2020. {
  2021. choice1 = p;
  2022. }
  2023. else if(p->host == choice1->host)// 避免返回两个相同的CP
  2024. continue;
  2025. else if(choice2 == NULL)
  2026. {
  2027. choice2 = p;
  2028. }
  2029. else
  2030. break;
  2031. }
  2032. }
  2034. int found_cp_count = 0;
  2035. if (choice1)
  2036. {
  2037. addr1 = (struct NormalAddress *)*buffer;
  2038. addr1->sin_family = PF_INET;
  2039. addr1->sin_port = htons (choice1->npport);
  2040. addr1->sin_addr.s_addr = htonl (choice1->host);
  2041. *buffer += sizeof (*addr1);
  2043. found_cp_count ++; // one CP is found
  2044. }
  2046. if(choice2)
  2047. {
  2048. addr2 = (struct NormalAddress *)*buffer;
  2049. addr2->sin_family = PF_INET;
  2050. addr2->sin_port = htons (choice2->npport);
  2051. addr2->sin_addr.s_addr = htonl (choice2->host);
  2052. *buffer += sizeof (*addr2);
  2054. found_cp_count ++;
  2055. }
  2057. return found_cp_count;// 返回最终找到的CP的个数
  2058. }
  2060. #ifndef SORT_NET
  2061. int findNPPeers (struct Channel *pc, struct Session *me, int playing, int num, char **buffer, char *buffer1)
  2062. {
  2063. struct Edge *pedge = NULL;
  2064. struct Session *ps;
  2065. struct P2PAddress addr;
  2066. int j, k;
  2067. unsigned int randstart;
  2068. struct sockaddr_in client;
  2070. if (pc == NULL || pc->numclient <= 0) return 0;
  2071. if (me->cachepeer != NULL && me->cachepeer->u.p.header != NULL && me->type == TYPE_NP)
  2072. {
  2073. for (pedge=me->cachepeer->u.p.header; pedge && pedge->head != pc; pedge=pedge->enext);
  2074. }
  2075. if (pedge == NULL)
  2076. {
  2077. randstart = rand () % pc->numclient;
  2078. for (pedge=pc->PeerHead; pedge && randstart > 0; pedge=pedge->cnext, randstart--);
  2079. }
  2081. for (j=0,k=0; j<num && k<pc->numclient; pedge=pedge->cnext)
  2082. {
  2083. k++;
  2084. if (pedge == NULL) pedge = pc->PeerHead;
  2085. if ((ps = pedge->me) == me) continue; // exclude myself
  2086. if (CurTimeSec > pedge->me->last_access+60) continue; // don't bother it too often
  2087. if (ps->u.p.p.isMaxIn == 0 && ((me->type == TYPE_CP) || playing == 0xffffffff || check_valid (pedge, playing)))
  2088. {
  2089. if (ps->intra == 0xffffffff || ps->host == me->host) // 0xffffffff means ps is on the public network, ps->host == me->host means in the same private network
  2090. {
  2091. memcpy (*buffer, &(ps->u.p.p), sizeof (struct CorePeerInfo));
  2092. addr = (*buffer) + sizeof (struct CorePeerInfo);
  2093. addr->outerIP.sin_port = htons (ps->port);
  2094. addr->outerIP.sin_addr.s_addr = htonl (ps->host);
  2095. addr->subnetIP.sin_port = htons (ps->npport);
  2096. addr->subnetIP.sin_addr.s_addr = htonl (ps->intra);
  2097. addr->outerIP.sin_family = PF_INET;
  2098. addr->subnetIP.sin_family = PF_INET;
  2099. *buffer = (struct char *)(addr + 1);
  2100. j++;
  2101. } else if (buffer1)
  2102. {
  2103. memset ((char *)&client, 0, sizeof (client));
  2104. client.sin_port = htons (ps->port);
  2105. client.sin_family = AF_INET;
  2106. client.sin_addr.s_addr = htonl (ps->host);
  2107. if (sendMessage(ps->socket,buffer1, &client) < 0)
  2108. {
  2109. closure_NP (ps);
  2110. }
  2111. PDEBUG("Send ConnecTo to %sn", inet_ntoa(client.sin_addr));
  2112. }
  2113. }
  2114. }
  2115. if (pedge != NULL) me->cachepeer = pedge->me;
  2116. return j;
  2117. }
  2118. #else
  2119. int findNPPeers (struct Channel *pc, struct Session *me, int playing, int num, char **buffer, char *buffer1)
  2120. {
  2121. int i, j, k, m, mnetnum;
  2122. unsigned int randstart;
  2123. struct Session *result[MAX_PEER];
  2124. struct Session *ps;
  2125. struct sockaddr_in client;
  2126. struct Edge *pedge;
  2127. struct P2PAddress *addr;
  2129. if (pc == NULL || pc->numclient <= 0) return 0;
  2130. Net = me->net;
  2131. for (j=0,m=0; m<MAX_NET_NUM; m++) { mnetnum = (Net + m) % MAX_NET_NUM;
  2132. k = 0;
  2133. if (pc->nclient_net[mnetnum] <= 0) continue;
  2134. pedge = NULL;
  2135. if (me->cachepeer[mnetnum] != NULL && me->cachepeer[mnetnum]->u.p.header != NULL && me->type == TYPE_NP)
  2136. {
  2137. for (pedge=me->cachepeer[mnetnum]->u.p.header; pedge && pedge->head != pc; pedge=pedge->enext);
  2138. }
  2139. if (pedge == NULL)
  2140. {
  2141. randstart = rand () % pc->nclient_net[mnetnum];
  2142. for (pedge=pc->PeerHead[mnetnum]; pedge && randstart > 0; pedge=pedge->cnext, randstart--);
  2143. }
  2144. for (; j<MAX_PEER && k < pc->nclient_net[mnetnum]; pedge=pedge->cnext)
  2145. {
  2146. k++;
  2147. if (pedge == NULL) pedge = pc->PeerHead[mnetnum];
  2148. if ((ps = pedge->me) == me) continue;
  2149. if (CurTimeSec > pedge->me->last_access+60) continue;
  2150. if (ps->u.p.p.isMaxIn == 0 && ((me->type == TYPE_CP) || playing == 0xffffffff || check_valid (pedge, playing)))
  2151. {
  2152. if (ps->intra == 0xffffffff || ps->host == me->host)
  2153. {
  2154. result[j] = ps;
  2155. j++;
  2156. } else if (buffer1)
  2157. {
  2158. memset ((char *)&client, 0, sizeof (client));
  2159. client.sin_port = htons (ps->port);
  2160. client.sin_family = AF_INET;
  2161. client.sin_addr.s_addr = htonl (ps->host);
  2162. if (sendMessage(ps->socket,buffer1, &client) < 0)
  2163. {
  2164. closure_NP (ps);
  2165. }
  2166. PDEBUG("Send ConnecTo to %sn", inet_ntoa(client.sin_addr));
  2167. }
  2168. }
  2169. }
  2170. if (pedge != NULL) me->cachepeer[mnetnum] = pedge->me;
  2171. }
  2172. if (j > 1)
  2173. qsort (result, j, sizeof (struct Session *), compareSession);
  2174. if (num > 0 && j > num) j = num;
  2175. if (me->type == TYPE_NP)
  2176. PDEBUG ("NP %d find %d NP:", me-NPTRACKER.head, j);
  2177. else
  2178. PDEBUG ("CP %d find %d CP:", me-CPTRACKER.head, j);
  2179. for (i=0; i<j; i++)
  2180. {
  2181. ps = result[i];
  2182. PDEBUG ("%dt", ps-NPTRACKER.head);
  2183. memcpy (*buffer, &(ps->u.p.p), sizeof (struct CorePeerInfo));
  2184. addr = (struct P2PAddress *)((*buffer) + sizeof (struct CorePeerInfo));
  2185. addr->outerIP.sin_port = htons (ps->port);
  2186. addr->outerIP.sin_addr.s_addr = htonl (ps->host);
  2187. addr->subnetIP.sin_port = htons (ps->npport);
  2188. addr->subnetIP.sin_addr.s_addr = htonl (ps->intra);
  2189. addr->outerIP.sin_family = PF_INET;
  2190. addr->subnetIP.sin_family = PF_INET;
  2191. *buffer = (char *)(addr + 1);
  2192. }
  2193. PDEBUG ("n");
  2194. return j;
  2195. }
  2196. #endif
  2198. int process_CP2TS_UPDATE (struct Session *p, struct TSMessage *m)
  2199. {
  2200. char *buf = m->buffer;
  2201. p->u.cp.resnum = *(int *)buf;
  2202. buf += sizeof (int);
  2203. p->u.cp.connnum = *(unsigned short *)buf;
  2204. buf += sizeof (unsigned short);
  2205. p->u.cp.band = *(float *)buf;
  2206. buf += sizeof(float);
  2207. p->u.cp.maxConn = *(unsigned char *)buf;
  2208. buf += sizeof(char);
  2209. if (buf - m->buffer + AUTH_HEADER > m->len)
  2210. {
  2211. PDEBUG ("Invalid message NP2TS_LOGIN, length %d not enough.n", m->len);
  2212. closure_CP (p);
  2213. return -1;
  2214. }
  2215. if (p->u.cp.type == CT_GENERAL &&
  2216. (GCPCHOICE == NULL || p->u.cp.resnum < GCPCHOICE->u.cp.resnum))
  2217. GCPCHOICE = p;
  2218. if (p->u.cp.resnum < 0) p->u.cp.resnum = 0;
  2219. if (p->u.cp.band < 0) p->u.cp.band = 0.001;
  2220. return 0;
  2221. }
  2223. int process_CP2TS_NEED_PEERS (struct Session *p, struct TSMessage *m)
  2224. {
  2225. return process_NEED_PEERS_real (p, m->buffer/*md5*/, 1/*needcp*/, 0/*cur*/, 0/*layer*/);
  2226. }
  2228. /* 登录, CP向TS登录, 按照来源IP地址和所报告的npport进行hash,
  2229.    如果距离上次发送CP2TS_REGISTER消息的时间小于SILENCE_TIME, 则直接返回,否则发送WELCOME消息. */
  2230. int process_CP2TS_REGISTER (struct Message *m)
  2231. {
  2232. char md5[MD5_LEN+1];
  2233. unsigned int host, cur = 0;
  2234. unsigned short port, npport;
  2235. int id, userID;
  2236. char *buf;
  2237. struct Session *p;
  2239. buf = m->buffer;
  2240. userID = *(int *)buf;
  2241. buf += sizeof (int);
  2242. memcpy (md5, buf, MD5_LEN);
  2243. md5[MD5_LEN] = 0;
  2244. buf += MD5_LEN;
  2246. #ifdef HAVE_MYSQL
  2247. if (authUser (userID, md5, local_mysql, NULL) == 0)
  2248. {
  2249. }
  2250. #endif
  2252. npport = ntohs(*(unsigned short *)buf);
  2253. buf += sizeof (short);
  2254. host = ntohl (UDPCLIENT.sin_addr.s_addr);
  2255. port = ntohs (UDPCLIENT.sin_port);
  2256. id = hash_cp (host, npport);
  2257. for (p=CPTRACKER.hash[id]; p; p=p->hnext)
  2258. if (p->host == host && p->port == port && p->npport == npport)
  2259. break;
  2260. if (!p)
  2261. {
  2262. if (CPTRACKER.cur >= CPTRACKER.max || CPTRACKER.hash[0] == 0)
  2263. {
  2264. PDEBUG("CP reg err, wrong cp index in array.n");
  2265. SEND_NPMSG(CPTRACKER.sock[CurrentSock],TS2CP_MSG,ERR_INTERNAL,1,&UDPCLIENT);
  2266. return -1;
  2267. }
  2268. p = CPTRACKER.hash[0];
  2269. CPTRACKER.hash[0] = p->hnext;
  2270. p->hnext = CPTRACKER.hash[id];
  2271. CPTRACKER.hash[id] = p;
  2272. p->socket = CPTRACKER.sock[CurrentSock];
  2273. p->type = TYPE_CP;
  2274. p->port = port;
  2275. p->npport = npport;
  2276. p->host = host;
  2279. #ifdef SORT_NET
  2280. {
  2281. struct networks *pnetworks = getnetwork (host, NETBLOCKS, maxNet);
  2282. if (pnetworks) p->net = pnetworks->net;
  2283. else p->net = 0;
  2284. }
  2285. #endif
  2286. p->time_sec = cur;
  2288. p->u.cp.userid = userID;
  2289. p->u.cp.type = *(unsigned char *)buf;
  2290. buf += sizeof (char);
  2292. if (p->u.cp.type == CT_EDGE)
  2293. {
  2294. p->u.cp.numHeads = *(unsigned short *)buf;
  2295. buf += sizeof (short);
  2296. if (((char *)m) +m->len - buf <= sizeof (p->u.cp.parameter))
  2297. memcpy (p->u.cp.parameter, buf, (char *)m+m->len-buf);
  2298. else
  2299. memcpy (p->u.cp.parameter, buf,sizeof (p->u.cp.parameter));
  2300. if (p->u.cp.numHeads > sizeof(p->u.cp.parameter)/2)
  2301. p->u.cp.numHeads = sizeof (p->u.cp.parameter)/2;
  2302. } else if (p->u.cp.type == CT_SPECIFIED_RES)
  2303. {
  2304. memcpy (p->u.cp.parameter, buf, MD5_LEN);
  2305. buf += MD5_LEN;
  2306. }
  2308. p->auth = random ();
  2310. init_CP (p);
  2311. } else if (CurTimeSec < p->last_access + SILENCE_TIME)
  2312. return 0;
  2313. if (buf - m->buffer + NORMAL_HEADER > m->len)
  2314. {
  2315. PDEBUG ("Invalid message NP2TS_LOGIN, length %d not enough.n", m->len);
  2316. closure_CP (p);
  2317. return -1;
  2318. }
  2320. PDEBUG("CP register. socket=%d listnum=%d auth=%d.n", p->socket, p-CPTRACKER.head, p->auth);
  2322. UDPMsg.len = 12;
  2323. UDPMsg.type=TS2CP_WELCOME;
  2324. UDPMsg.authcode1 = p-CPTRACKER.head;
  2325. UDPMsg.authcode2 = p->auth;
  2327. if (sendMessage(p->socket,(char *)&UDPMsg, &UDPCLIENT) < 0)
  2328. {
  2329. closure_CP (p);
  2330. return -1;
  2331. }
  2332. return 0;
  2333. }
  2335. int compareInter (const void *a, const void *b)
  2336. {
  2337. struct Interval *p = (struct Interval *) a;
  2338. struct Interval *q = (struct Interval *) b;
  2339. if ((p->start >= q->start && p->start+p->len <= q->start + q->len)
  2340. || (q->start >= p->start && q->start+q->len <= p->start + p->len))
  2341. return 0;
  2342. return (p->start - q->start);
  2343. }
  2345. // delete用于从原有的Interval当中去掉新的.
  2346. int delete_interval (struct Interval *head, int total, struct Interval *_new, int num)
  2347. {
  2348. int i,j,k;
  2349. struct Interval tmp[MAX_INTERVAL*2];
  2350. for (i=0,j=0,k=0; i<total && j<num;)
  2351. {
  2352. if (_new[j].start+_new[j].len <= head[i].start)
  2353. j++;
  2354. else if (head[i].start+head[i].len <= _new[j].start)
  2355. {
  2356. tmp[k].start = head[i].start;
  2357. tmp[k].len = head[i].len;
  2358. i++;
  2359. k++;
  2360. } else
  2361. {
  2362. if (_new[j].start <= head[i].start)
  2363. {
  2364. if (_new[j].start+_new[j].len >= head[i].start+head[i].len)
  2365. i++;
  2366. else
  2367. {
  2368. tmp[k].start = _new[j].start+_new[j].len;
  2369. tmp[k].len = head[i].start+head[i].len-tmp[k].start;
  2370. i++;
  2371. if (tmp[k].len > 0) k++;
  2372. }
  2373. } else
  2374. {
  2375. tmp[k].start = head[i].start;
  2376. tmp[k].len = _new[j].start-tmp[k].start;
  2377. i++;
  2378. if (tmp[k].len > 0) k++;
  2379. if (_new[j].start+_new[j].len < head[i].start+head[i].len)
  2380. {
  2381. tmp[k].start = _new[j].start+_new[j].len;
  2382. tmp[k].len = head[i].start+head[i].len-tmp[k].start;
  2383. i++;
  2384. j++;
  2385. if (tmp[k].len > 0) k++;
  2386. }
  2387. }
  2388. }
  2389. }
  2390. if (i<total)
  2391. {
  2392. memcpy (tmp+k, head+i, (total-i)*sizeof(struct Interval));
  2393. k += (total-i);
  2394. }
  2395. if (k > MAX_INTERVAL) k = MAX_INTERVAL;
  2396. if (k>0) memcpy (head, tmp, k*sizeof(struct Interval));
  2397. return k;
  2398. }
  2400. // merge用于将原有的Interval和新的Interval列表合在一起 
  2401. int merge (struct Interval *head, int total, struct Interval *_new, int num)
  2402. {
  2403. int i,j,k;
  2404. struct Interval tmp[MAX_INTERVAL*2];
  2405. for (i=0,j=0,k=0; i<total && j<num;)
  2406. {
  2407. if (head[i].start <= _new[j].start)
  2408. {
  2409. tmp[k].start = head[i].start;
  2410. tmp[k].len = head[i].len;
  2411. i++;
  2412. } else
  2413. {
  2414. tmp[k].start = _new[j].start;
  2415. tmp[k].len = _new[j].len;
  2416. j++;
  2417. }
  2418. for (; i<total || j<num;)
  2419. {
  2420. if (i<total && head[i].start <= tmp[k].start+tmp[k].len)
  2421. {
  2422. if (head[i].start+head[i].len > tmp[k].start+tmp[k].len)
  2423. tmp[k].len = head[i].start+head[i].len-tmp[k].start;
  2424. i++;
  2425. } else if (j<num && _new[j].start <= tmp[k].start+tmp[k].len)
  2426. {
  2427. if (_new[j].start+_new[j].len > tmp[k].start+tmp[k].len)
  2428. tmp[k].len = _new[j].start+_new[j].len-tmp[k].start;
  2429. j++;
  2430. } else
  2431. break;
  2432. }
  2433. k++;
  2434. }
  2435. if (i<total)
  2436. {
  2437. memcpy (tmp+k, head+i, (total-i)*sizeof(struct Interval));
  2438. k += (total-i);
  2439. } else if (j<num)
  2440. {
  2441. memcpy (tmp+k, _new+j, (num-j)*sizeof(struct Interval));
  2442. k += (num-j);
  2443. }
  2444. if (k > MAX_INTERVAL) k = MAX_INTERVAL;
  2445. if (k>0) memcpy (head, tmp, k*sizeof(struct Interval));
  2446. return k;
  2447. }
  2450. int check_valid (struct Edge *e, int play)
  2451. {
  2452. struct Interval tmp;
  2453. struct Interval *i = e->inter, *result;
  2454. int num = e->numinter;
  2455. if (num <= 0) return 0;
  2456. if (i[0].start > play || i[num-1].start+i[num-1].len <= play) return 0;
  2457. tmp.start = play;
  2458. tmp.len = 1;
  2459. if ((result = (struct Interval*)bsearch ((void*)&tmp, i, num, sizeof (struct Interval), compareInter)) == NULL)
  2460. return 0;
  2461. return 1;
  2462. }
  2464. int logto_xml (long channelcount, unsigned int totalclient, long long totalstay)
  2465. {
  2466. FILE *logf = fopen (LOGXML, "w");
  2467. if(!logf)
  2468. {
  2469. PDEBUG("Couldn't open log xml!.n");
  2470. return -1;
  2471. }
  2472. fprintf(logf, "<?xml version="1.0" encoding="iso-8859-1"?>n");
  2473. fprintf(logf, "<TS>n");
  2474. fprintf(logf, "<ElapsedTime>%ld</ElapsedTime>n", CurTimeSec - startTime);
  2475. fprintf(logf, "<TotalLogin>%lld</TotalLogin>n", np2tsLoginCount);
  2476. fprintf(logf, "<TotalMsg>%lld</TotalMsg>n", np2tsLoginCount + np2tsResListCount + np2tsReqResCount + np2tsDelResCount + np2tsReportCount + np2tsNeedPeerCount + np2tsLogoutCount);
  2477. fprintf(logf, "<ActiveChannel>%ld</ActiveChannel>n", channelcount);
  2478. fprintf(logf, "<OnlineUser>%d</OnlineUser>n", totalclient);
  2479. fprintf(logf, "<AvgTime>%.1f</AvgTime>n", totalclient==0?0:(totalstay/((double)totalclient)));
  2480. fprintf(logf, "</TS>n");
  2481. fclose(logf);
  2482. return 0;
  2483. }