servmgr.cpp

这是和p2p相关的一份源码

// ------------------------------------------------
// File : servmgr.cpp
// Date: 4-apr-2002
// Author: giles
// Desc: 
//		Management class for handling multiple servent connections.
//
// (c) 2002 peercast.org
// ------------------------------------------------
// This program is free software; you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation; either version 2 of the License, or
// (at your option) any later version.

// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
// ------------------------------------------------

#include <stdlib.h>
#include "servent.h"
#include "servmgr.h"
#include "inifile.h"
#include "stats.h"
#include "peercast.h"
#include "pcp.h"
#include "atom.h"

static int debugtest=0;



// -----------------------------------
ServMgr::ServMgr()
:replyIDs(500)
{
	authType = AUTH_COOKIE;
	cookieList.init();

	serventNum = 0;

	startTime = sys->getTime();

	allowServer1 = Servent::ALLOW_ALL;
	allowServer2 = Servent::ALLOW_BROADCAST;

	clearHostCache(ServHost::T_NONE);
	password[0]=0;

	allowGnutella = false;
	useFlowControl = true;

	minGnuIncoming = 10;
	maxGnuIncoming = 50;

	maxBitrate = MAX_BITRATE;
	maxStreams = 2;
	refreshHTML = 5;

	networkID.clear();

	notifyMask = 0xffff;

	tryoutDelay = 10;
	numVersions = 0;

	sessionID.generate();

	isRoot = false;

	forceIP[0]	= 0;		// force server IP

	strcpy(connectHost,"connect1.peercast.org");

	rootHost = "yp.peercast.org";

	serverHost.fromStrIP("127.0.0.1",DEFAULT_PORT);

	firewalled = FW_UNKNOWN;		
	allowDirect = true;
	autoConnect = true;
	forceLookup = true;
	autoServe = true;
	forceNormal = false;

	maxControl = 3;

	queryTTL = 7;

	totalStreams = 0;
	firewallTimeout = 30;
	pauseLog = false;
	showLog = 0;

	shutdownTimer = 0;

	downloadURL[0] = 0;
	rootMsg.clear();


	restartServer=false;

	setFilterDefaults();

	servents = NULL;

}
// -----------------------------------
void	ServMgr::connectBroadcaster()
{
	if (!rootHost.isEmpty())
	{
		if (!numUsed(Servent::T_COUT))
		{
			Servent *sv = allocServent();
			if (sv)
			{
				sv->initOutgoing(Servent::T_COUT);
				sys->sleep(3000);
			}
		}
	}
}
// -----------------------------------
void	ServMgr::addVersion(unsigned int ver)
{
	for(int i=0; i<numVersions; i++)
		if (clientVersions[i] == ver)
		{
			clientCounts[i]++;
			return;
		}

	if (numVersions < MAX_VERSIONS)
	{
		clientVersions[numVersions] = ver;
		clientCounts[numVersions] = 1;
		numVersions++;
	}
}

// -----------------------------------
void ServMgr::setFilterDefaults()
{
	numFilters = 0;

	filters[numFilters].host.fromStrIP("255.255.255.255",0);
	filters[numFilters].flags = ServFilter::F_NETWORK;
	numFilters++;

	filters[numFilters].host.fromStrIP("127.0.0.1",0);
	filters[numFilters].flags = ServFilter::F_PRIVATE|ServFilter::F_NETWORK|ServFilter::F_DIRECT;
	numFilters++;

}
// -----------------------------------
void	ServMgr::setPassiveSearch(unsigned int t)
{
//	if ((t > 0) && (t < 60))
//		t = 60;
//	passiveSearch = t;
}
// -----------------------------------
bool ServMgr::seenHost(Host &h, ServHost::TYPE type,unsigned int time)
{
	time = sys->getTime()-time;

	for(int i=0; i<MAX_HOSTCACHE; i++)
		if (hostCache[i].type == type)
			if (hostCache[i].host.ip == h.ip)
				if (hostCache[i].time >= time)
					return true;
	return false;
}

// -----------------------------------
void ServMgr::addHost(Host &h, ServHost::TYPE type, unsigned int time)
{
	int i;
	if (!h.isValid())
		return;

	ServHost *sh=NULL;

	for(i=0; i<MAX_HOSTCACHE; i++)
		if (hostCache[i].type == type)
		{
			sh = &hostCache[i];
			break;
		}

	char str[64];
	h.toStr(str);

	if (!sh)
		LOG_DEBUG("New host: %s - %s",str,ServHost::getTypeStr(type));
	else
		LOG_DEBUG("Old host: %s - %s",str,ServHost::getTypeStr(type));

	h.value = 0;	// make sure dead count is zero
	if (!sh)
	{


		// find empty slot
		for(i=0; i<MAX_HOSTCACHE; i++)
			if (hostCache[i].type == ServHost::T_NONE)
			{
				sh = &hostCache[i];
				break;
			}

		// otherwise, find oldest host and replace
		if (!sh)
			for(i=0; i<MAX_HOSTCACHE; i++)
				if (hostCache[i].type != ServHost::T_NONE)
				{
					if (sh)
					{
						if (hostCache[i].time < sh->time)
							sh = &hostCache[i];
					}else{
						sh = &hostCache[i];
					}
				}
	}

	if (sh)
		sh->init(h,type,time);
}

// -----------------------------------
void ServMgr::deadHost(Host &h,ServHost::TYPE t)
{
	for(int i=0; i<MAX_HOSTCACHE; i++)
		if (hostCache[i].type == t)
			if (hostCache[i].host.ip == h.ip)
				if (hostCache[i].host.port == h.port)
					hostCache[i].init();
}
// -----------------------------------
void ServMgr::clearHostCache(ServHost::TYPE type)
{
	for(int i=0; i<MAX_HOSTCACHE; i++)
		if ((hostCache[i].type == type) || (type == ServHost::T_NONE))
			hostCache[i].init();
}
	
// -----------------------------------
unsigned int ServMgr::numHosts(ServHost::TYPE type)
{
	unsigned int cnt = 0;
	for(int i=0; i<MAX_HOSTCACHE; i++)
		if ((hostCache[i].type == type) || (type == ServHost::T_NONE))
			cnt++;
	return cnt;
}
// -----------------------------------
int ServMgr::getNewestServents(Host *hl,int max,Host &rh)
{
	int cnt=0;
	for(int i=0; i<max; i++)
	{
		// find newest host not in list
		ServHost *sh=NULL;
		for(int j=0; j<MAX_HOSTCACHE; j++)
		{
			// find newest servent
			if (hostCache[j].type == ServHost::T_SERVENT)
				if (!(rh.globalIP() && !hostCache[j].host.globalIP()))
				{
					// and not in list already
					bool found=false;
					for(int k=0; k<cnt; k++)
						if (hl[k].isSame(hostCache[j].host))
						{
							found=true; 
							break;
						}

					if (!found)
					{
						if (!sh)
						{
							sh = &hostCache[j];
						}else{
							if (hostCache[j].time > sh->time)
								sh = &hostCache[j];
						}
					}
				}
		}

		// add to list
		if (sh)
			hl[cnt++]=sh->host;
	}
	
	return cnt;
}
// -----------------------------------
ServHost ServMgr::getOutgoingServent(GnuID &netid)
{
	ServHost host;

	Host lh(ClientSocket::getIP(NULL),0);

	// find newest host not in list
	ServHost *sh=NULL;
	for(int j=0; j<MAX_HOSTCACHE; j++)
	{
		ServHost *hc=&hostCache[j];
		// find newest servent not already connected.
		if (hc->type == ServHost::T_SERVENT)
		{
			if (!((lh.globalIP() && !hc->host.globalIP()) || lh.isSame(hc->host)))
			{
#if 0
				if (!findServent(Servent::T_OUTGOING,hc->host,netid))
				{
					if (!sh)
					{
						sh = hc;
					}else{
						if (hc->time > sh->time)
							sh = hc;
					}
				}
#endif
			}
		}
	}

	if (sh)
		host = *sh;

	return host;
}
// -----------------------------------
Servent *ServMgr::findOldestServent(Servent::TYPE type, bool priv)
{
	Servent *oldest=NULL;

	Servent *s = servents;
	while (s)
	{
		if (s->type == type)
			if (s->thread.active)
				if (s->isOlderThan(oldest))
					if (s->isPrivate() == priv)
						oldest = s;
		s=s->next;
	}
	return oldest;
}
// -----------------------------------
Servent *ServMgr::findServent(Servent::TYPE type, Host &host, GnuID &netid)
{
	lock.on();
	Servent *s = servents;
	while (s)
	{
		if (s->type == type)
		{
			Host h = s->getHost();
			if (h.isSame(host) && s->networkID.isSame(netid))
			{
				lock.off();
				return s;
			}
		}
		s=s->next;
	}
	lock.off();
	return NULL;

}

// -----------------------------------
Servent *ServMgr::findServent(unsigned int ip, unsigned short port, GnuID &netid)
{
	lock.on();
	Servent *s = servents;
	while (s)
	{
		if (s->type != Servent::T_NONE)
		{
			Host h = s->getHost();
			if ((h.ip == ip) && (h.port == port) && (s->networkID.isSame(netid)))
			{
				lock.off();
				return s;
			}
		}
		s=s->next;
	}
	lock.off();
	return NULL;

}

// -----------------------------------
Servent *ServMgr::findServent(Servent::TYPE t)
{
	Servent *s = servents;
	while (s)
	{
		if (s->type == t)
			return s;
		s=s->next;
	}
	return NULL;
}
// -----------------------------------
Servent *ServMgr::findServentByID(int id)
{
	Servent *s = servents;
	while (s)
	{
		if (s->serventID == id)
			return s;
		s=s->next;
	}
	return NULL;
}

// -----------------------------------
Servent *ServMgr::allocServent()
{
	lock.on();

	Servent *s = servents;
	while (s)
	{
		if (s->status == Servent::S_DEAD)
			break;
		s=s->next;
	}

	if (!s)
	{
		s = new Servent(++serventNum);
		s->next = servents;
		servents = s;

		LOG_DEBUG("allocated servent %d",serventNum);
	}else
		LOG_DEBUG("reused servent %d",s->serventID);


	s->reset();

	lock.off();

	return s;
}
// -----------------------------------
unsigned int ServMgr::numConnected(int type,bool priv)
{
	unsigned int cnt=0;

	Servent *s = servents;
	while (s)
	{
		if (s->thread.active)
			if (s->isConnected())
				if (s->type == type)
					if (s->isPrivate()==priv)
						cnt++;

		s=s->next;
	}
	return cnt;
}
// -----------------------------------
unsigned int ServMgr::numConnected()
{
	unsigned int cnt=0;

	Servent *s = servents;
	while (s)
	{
		if (s->thread.active)
			if (s->isConnected())
				cnt++;

		s=s->next;
	}
	return cnt;
}
// -----------------------------------
unsigned int ServMgr::numUsed(int type)
{
	unsigned int cnt=0;

	Servent *s = servents;
	while (s)
	{
		if (s->type == type)
			cnt++;
		s=s->next;
	}
	return cnt;
}

// -----------------------------------
unsigned int ServMgr::numStreams(bool all)
{
	unsigned int cnt=0;

	Servent *s = servents;
	while (s)
	{
		if (s->thread.active)
			if (s->isConnected())
				if (s->type == Servent::T_STREAM)
					if (all || !s->isPrivate())
						cnt++;
		s=s->next;
	}
	return cnt;
}
// -----------------------------------
unsigned int ServMgr::totalOutput(bool all)
{
	unsigned int tot = 0;
	Servent *s = servents;
	while (s)
	{
		if (s->isConnected())
			if (s->type == Servent::T_STREAM)
				if (all || !s->isPrivate())
					if (s->sock)
						tot += s->sock->bytesOutPerSec;
		s=s->next;
	}

	return tot;
}

// -----------------------------------
unsigned int ServMgr::numOutgoing()
{
	int cnt=0;

	Servent *s = servents;
	while (s)
	{
//		if ((s->type == Servent::T_INCOMING) ||
//		    (s->type == Servent::T_OUTGOING)) 
//			cnt++;
		s=s->next;
	}
	return cnt;
}

// -----------------------------------
bool ServMgr::seenPacket(GnuPacket &p)
{
	Servent *s = servents;
	while (s)
	{
		if (s->isConnected())
			if (s->seenIDs.contains(p.id))
				return true;
		s=s->next;
	}
	return false;
}

// -----------------------------------
void ServMgr::quit()
{
	Servent *s = servents;
	while (s)
	{
		if (s->isConnected())
		{
			try
			{
				s->thread.active=false;	// cant call abort because the connection may need to send quit
			}catch(StreamException &){}
		}
		s=s->next;
	}

}

// -----------------------------------
int ServMgr::broadcast(GnuPacket &pack,Servent *src)
{
	int cnt=0;
	if (pack.ttl)
	{
		Servent *s = servents;
		while (s)
		{

			if (s != src)
				if (s->isConnected())
					if (s->type == Servent::T_PGNU)
						if (!s->seenIDs.contains(pack.id))
						{

							if (src)
								if (!src->networkID.isSame(s->networkID))
									continue;

							if (s->outputPacket(pack,false))
								cnt++;
						}
			s=s->next;
		}
	}

	LOG_NETWORK("broadcast: %s (%d) to %d servents",GNU_FUNC_STR(pack.func),pack.ttl,cnt);

	return cnt;
}
// -----------------------------------
int ServMgr::route(GnuPacket &pack, GnuID &routeID, Servent *src)
{
	int cnt=0;
	if (pack.ttl)
	{
		Servent *s = servents;