tsnew.c

mysee网络直播源代码Mysee Lite是Mysee独立研发的网络视频流媒体播放系统。在应有了P2P技术和一系列先进流媒体技术之后

		for (i=0; i<MAX_BIND && CPTRACKER.sock[i] != 0; i++)
		{
			if (FD_ISSET (CPTRACKER.sock[i], &socks))
			{
				CurrentSock = i;
				process_CP (i);
			}
		}
#ifdef HAVE_RM
		for (i=0; i<MAX_BIND && RMTRACKER.sock[i] != 0; i++)
		{
			if (FD_ISSET(RMTRACKER.sock[i], &socks))
			{
				CurrentSock = i;
				process_RM (i);
			}
		}
#endif
	}
}

void my_exit ()
{
	int max, listnum;
	struct Session *head;
	int (*closure) (struct Session *);

#ifdef HAVE_RM
	FREE_TRACKER(RMTRACKER);
#endif

	FREE_TRACKER(NPTRACKER);
	FREE_TRACKER(CPTRACKER);

	// free configuration file.
	// 参数为和read_config相同的struct NamVal *以及项数.
	free_config (ConfigParameters, sizeof(ConfigParameters)/sizeof(struct NamVal));
#ifdef HAVE_MYSQL
	mysql_close (local_mysql);
#endif
	PDEBUG ("exit...\n");
	exit (0);
}

inline void my_memmove (char *dst, char *src, int len)
{
	int i;
	if (len <= 0 || dst == src) return;
	for (i=0; i<len; i++)
		*dst++ = *src++;
}

struct Channel *findChannel (char *name, int len)
{
	int id = hash_str ((unsigned char*)name, len);
	struct Channel *p;
	for (p=ChannelHash[id]; p; p=p->next)
	{
		if (strncmp (name, p->name, len) == 0)
			return p;
	}
	return NULL;
}

struct Edge *newEdge (struct Channel *head, struct Session *me)
{
	struct Edge *result = (struct Edge *)malloc (sizeof (struct Edge));
	result->head = head;
	result->me = me;
#ifndef SORT_NET
	result->cnext = head->PeerHead;
	head->PeerHead = result;
#else
	result->cnext = head->PeerHead[me->net];
	head->PeerHead[me->net] = result;
	head->nclient_net[me->net] ++;
#endif
	result->enext = me->u.p.header;
	me->u.p.header = result;
	head->numclient ++;
	head->accumclient ++;
	return result;
}

int delEdge (struct Edge *e)
{
	struct Channel *pchannel=e->head;
	struct Session *psession=e->me;
	struct Edge *pedge;

	if (pchannel)
	{
#ifndef SORT_NET
		if (pchannel->PeerHead == e) pchannel->PeerHead = e->cnext;
#else
		pchannel->nclient_net[psession->net] --;
		if (pchannel->PeerHead[psession->net] == e) pchannel->PeerHead[psession->net] = e->cnext;
#endif
		else
		{
#ifndef SORT_NET
			for (pedge=pchannel->PeerHead; pedge && pedge->cnext && pedge->cnext != e; pedge = pedge->cnext);
#else
			for (pedge=pchannel->PeerHead[psession->net]; pedge && pedge->cnext && pedge->cnext != e; pedge = pedge->cnext);
#endif
			if (pedge && pedge->cnext) pedge->cnext = e->cnext;
		}
		pchannel->numclient --;
		if (pchannel->numclient == 0)
			freeChannel (pchannel);
	}
	if (psession)
	{
		if (psession->u.p.header == e) psession->u.p.header = e->enext;
		else
		{
			for (pedge=psession->u.p.header; pedge && pedge->enext && pedge->enext != e; pedge = pedge->enext);
			if (pedge && pedge->enext) pedge->enext = e->enext;
		}
	}
	if (psession->u.p.cur == e) psession->u.p.cur = NULL;
	free (e);
	return 0;
}


struct Channel *newChannel (char *name, int len)
{
	int id = hash_str ((unsigned char*)name, len);
	struct Channel *result = (struct Channel *)calloc (1, sizeof (struct Channel));
	if (!result) return NULL;
	if (len != MD5_LEN)
	{
		free (result);
		return NULL;
	}
	memcpy (result->name, name, len);
	result->name[len] = 0;
	result->latest_time = CurTimeSec;
	result->next = ChannelHash[id];
	ChannelHash[id] = result;
	return result;
}

void freeChannel (struct Channel *p)
{
	int id = hash_str ((unsigned char*)(p->name), MD5_LEN);
	struct Channel *q;
	if (ChannelHash[id] == p)
		ChannelHash[id] = p->next;
	else
	{
		for (q=ChannelHash[id]; q && q->next != p; q=q->next);
		if (q) q->next = p->next;
	}
	free (p);
}

// join in a channel for session *p
// Add a session-channel mapping (aka. an edge)
int addSession (struct Session *p, struct ChannelInfo *c, int isdefault, unsigned int cur)
{
	struct Channel *pc;
	struct Edge *pedge = NULL;
	if ((pc=findChannel (c->md5, MD5_LEN)) == NULL) /* Since the channel do not exist, the edge does not exist. 
							   So there is no need to search the list. */
	{
		if ((pc = newChannel (c->md5, MD5_LEN)) == NULL)
			return -1;
	} else{		
		for (pedge=p->u.p.header; pedge && pedge->head != pc; pedge=pedge->enext);
	}
	if (pedge == NULL)
	{
		pedge = newEdge (pc, p);
	}
	pedge->numinter = c->numinter;
	memcpy (pedge->inter, c->inter, c->numinter*sizeof(struct Interval));
	pedge->current = cur;
	if (isdefault) p->u.p.cur = pedge;
	return 0;
}

// delete corresponding session in the session-channel map
int delSession (struct Session *p)
{
	struct Edge *pedge, *prevedge;
	for (pedge=p->u.p.header; pedge; pedge=prevedge)
	{
		prevedge = pedge->enext;
		delEdge (pedge);
	}
	return 0;
}

void periodLOG (int s)
{
	int i;
	struct Session *p;
	struct tm result;

	localtime_r (&CurTimeSec, &result);
	if (s)
	{
		for (i=0,p=NPTRACKER.head; i<=NPTRACKER.maxid; i++, p++)
		{
			if (p->socket > 0 && CurTimeSec - p->last_access > MAX_IDLE)
			{
				closure_NP (p);
			}
		}
		for (i=0,p=CPTRACKER.head; i<=CPTRACKER.maxid; i++, p++)
		{
			if (p->socket > 0 && CurTimeSec - p->last_access > MAX_IDLE)
			{
				closure_CP (p);
			}
		}
	}

	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);
	PDEBUG ("STAT: \"%ld小时%ld分钟%ld秒\"\n", CurTimeSec/3600, (CurTimeSec%3600)/60, CurTimeSec%60);
	return;
}

void makeSnapShot(int count)
{
	FILE *f = fopen ("./ts.log", "a");

	if (!f)
	{
		PDEBUG("Couldn't open log file!.\n");
		return;
	}
	// seek to the end for write
	fseek(f, 0, SEEK_END);

	// 1. start SnapShot 
	struct tm result;
	localtime_r (&CurTimeSec, &result);
	fprintf (f, "\n\n********************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);
	
	// 2. log cpu & mem state
	//memlog(BASEDIR,"./log.sh");
	
	// 3. log message count
	fprintf(f, "Login: %lld. ResList: %lld. ReqRes: %lld. DelRes: %lld. Report: %lld. NeedPeers: %lld. Logout: %lld. \n", np2tsLoginCount, np2tsResListCount, np2tsReqResCount, np2tsDelResCount, np2tsReportCount, np2tsNeedPeerCount, np2tsLogoutCount);
	fprintf(f, "Welcome: %lld. Peers: %lld. ConnectTo: %lld. Msg: %lld.\n", ts2npWelcomeCount, ts2npPeersCount, ts2npConnectToCount, ts2npMsgCount);
	fflush(f);
			
	// 4. log channel state
	struct Channel *pChannel;
	int i, total=0, j, watchingCount, noIdleCount;
	struct Edge* pedge;
	int layerarray[OBSERVE_LAYER], noconnectioncount, higherlayercount;
	long long totalup, totaldown;
	long long totalcurrblock;
	long ts_ChannelCount = 0;
	long long ts_stay, totalstay = 0;
	for (i=0; i<MAX_CHANNEL; i++)
	{
		for (pChannel=ChannelHash[i]; pChannel; pChannel=pChannel->next)
		{
			++ts_ChannelCount;
			total += pChannel->numclient;
			watchingCount = 0;
			noIdleCount = 0;
			ts_stay = 0;
			totalup = totaldown = totalcurrblock = 0;
			memset(layerarray, 0, OBSERVE_LAYER*sizeof(int));
			noconnectioncount = higherlayercount = 0;
#ifndef SORT_NET
			for (j=0, pedge=pChannel->PeerHead; pedge && j < pChannel->numclient; pedge=pedge->cnext)
#else
{
			int mnetnum;
			for (mnetnum = 0; mnetnum < MAX_NET_NUM; mnetnum ++)
			{
			for (j=0, pedge=pChannel->PeerHead[mnetnum]; pedge && j < pChannel->nclient_net[mnetnum]; pedge=pedge->cnext)
#endif
			{
				if(pedge->me->socket <= 0)
					continue;
				if(pedge->me->u.p.p.layer == 0xff)
					noconnectioncount++;
				else if(pedge->me->u.p.p.layer < OBSERVE_LAYER)
					layerarray[pedge->me->u.p.p.layer]++;
				else
					higherlayercount++;
				ts_stay += CurTimeSec - pedge->me->time_sec;
				totalup += pedge->me->u.p.t.totalDownBytes;
				totaldown += pedge->me->u.p.t.totalUpBytes;
				if(pedge->current != 0xffffffff)
				{
					totalcurrblock += pedge->current;
					watchingCount++;
					if(CurTimeSec < pedge->me->last_access + 40)
						noIdleCount++;
				}
			}
#ifdef SORT_NET
			}
}
#endif
			totalstay += ts_stay;
			fprintf(f, "%s: accumulated %d clients, current %d clients and  %d/%d is watchings. Average Stay Time : %f\n", pChannel->name, pChannel->accumclient, pChannel->numclient, watchingCount,noIdleCount,pChannel->numclient==0?0:(ts_stay/(double)pChannel->numclient));
			fprintf(f, "total down: %llu, total up: %llu, down-up: %llu. Avg currBlock: %llu \n", totaldown, totalup, (totaldown-totalup), watchingCount==0?0:(totalcurrblock/watchingCount));
			fprintf(f, "No Connect: %d. Higher: %d. \n Detail: ", noconnectioncount, higherlayercount);
			for(j = 0; j < OBSERVE_LAYER; j++)
			{
				if(layerarray[j] != 0)
					fprintf(f, "%d: %d(%.1f), ", j, layerarray[j ], pChannel->numclient==0?0:(layerarray[j]/(double)pChannel->numclient));
			}
			fprintf(f, "\n");
		}
	}
	fprintf(f,"ChannelCount: %ld \n",ts_ChannelCount);
	fprintf(f, "\n ********************END SnapShot.\n\n");
	fclose(f);
	logto_xml (ts_ChannelCount, total, totalstay);
}
#if 0
int memlog(char *pwd, char *cmd)
{
	int pid = fork();
	if (pid == 0)
	{
		char buffer[10];
		daemon(0,1);
		snprintf(buffer,10,"%s",cmd);
		chdir(pwd);
		execlp("/bin/sh", "sh" ,"-c",buffer,NULL);
	}
	else if(pid < 0)
	{
		perror("fork error!!!");
		return -1;
	}
	return 0;
}
#endif

#ifdef SORT_NET
int compareNet (const void *a, const void *b)
{
	struct networks *p = (struct networks *)a;
	struct networks *q = (struct networks *)b;
	if (p->host > q->host) return 1;
	else if (p->host < q->host) return -1;
	return 0;
}

// This is implemented with recursion
struct networks *getnetwork (unsigned int host, struct networks *head, int n)
{
	int i = n/2-1;
	unsigned int net = host & head[i].mask;
	if (n <= 0) return NULL;
	if (n == 1)
	{
		if ((host & head[0].mask) != (head[0].host & head[0].mask))
			return NULL;
		else return &(head[0]);
	}
	if (net < (head[i].host & head[i].mask)) return getnetwork (host, head, i);
	else if (net > (head[i].host & head[i].mask)) return getnetwork (host, head+i+1, n-i-1);
	else return &(head[i]);
}

//192.168.0.0/16/1
//host/mask/netid
int readNETBLOCK (char *fname)
{
	int i = 0;
	char buffer[MAX_LINE],*p, *q;
	struct in_addr inp;
	FILE *f = fopen (fname, "r");
	if (!f) return -1;
	while (fgets (buffer, sizeof (buffer), f))
	{
		p = index (buffer, '/');
		q = rindex (buffer, '/');
		if (p == q || (!p) || (!q)) continue;
		*p = 0;
		*q = 0;
		if (inet_aton (buffer, &inp) == 0) continue;
		NETBLOCKS[i].host = ntohl (inp.s_addr);
		NETBLOCKS[i].mask = MASKS[atoi(p+1)];
		NETBLOCKS[i].net = atoi (q+1);
		i++;
		if (i >= MAX_NET) break;
	}
	fclose (f);
	qsort (NETBLOCKS, i, sizeof (struct networks), compareNet);
	return i;
}

int Net = 0;
int Layer = 0;
int compareSession (const void *a, const void *b)
{
	struct Session *p = *(struct Session **)a;
	struct Session *q = *(struct Session **)b;
	if (p->u.p.p.layer == q->u.p.p.layer)
	{
		if  (p->net == Net) return -1;
		else if (q->net == Net) return 1;
		else if (p->net > q->net) return 1;
		else return -1;
	} else if (p->u.p.p.layer <= Layer)
		return -1;
	else if (q->u.p.p.layer <= Layer)
		return 1;
	else return (p->u.p.p.layer - q->u.p.p.layer);
}

#endif


int init_udpserver (struct SessionCluster *c, char *host, int *port, unsigned int max, int numbind)
{
	int i;
	struct Session *head;
//	c->port = port;
	c->cur = 0; // current count
	c->max = max; // maximum count of sessions
	c->maxbuf = 0;
	if (max > 0)
	{
		c->head = (struct Session*)calloc (sizeof (struct Session), max); // allocate continuous session pool space
		c->hash = (struct Session**)calloc (sizeof (struct Session *), max); // allocate session hash table
		head = c->head;
		// chain the sessions in session pool as a freelist
		for (i=0; i<max-1; i++)
		{
			head[i].hnext = &(head[i+1]);
		}

		// initialize hash table
		for (i=1; i<max; i++) c->hash[i] = 0;
		// this is really ugly!
		c->hash[0] = &(head[0]);	// c->hash[0] points to the empty list
	} // HOW if max <= 0 ? return with error

	for (i=0; i<MAX_BIND && i<numbind; i++)
	{
		switch (BINDALL) // means 'BIND ANY HOST'
		{
			case 0:
				if ((c->sock[i] = init_udp (host, port[i])) < 0)
					return -1;
				break;
			default:
				if ((c->sock[i] = init_udp (NULL, port[i])) < 0)
					return -1;
				break;
		}
		// highsock indicates the largest active fd. used in select()	Used to speed up search
		if (c->sock[i] > highsock) highsock = c->sock[i];
		FD_SET(c->sock[i], &osocks);
	}

	return 0;
}


int init_ts ()
{
	struct rlimit rl;
#ifdef DEBUG