servmgr.cpp

这是和p2p相关的一份源码

				{
					chanMgr->createRelay(info,stayConnected);
				}else
				{
					Channel *c = chanMgr->createChannel(info,NULL);
					if (c)
						c->startURL(sourceURL.cstr());
				}
			}


			// old host cache
			else if (iniFile.isName("[HostCache]") || iniFile.isName("[Servents]"))
			{
				while (iniFile.readNext())
				{
					if (iniFile.isName("[End]"))
						break;
					Host h;
					h.fromStr(iniFile.getName(),DEFAULT_PORT);
					GnuID netid;
					netid.clear();
					if (h.isValid())
					{
						char tt[64];
						h.toStr(tt);
						servMgr->addHost(h,ServHost::T_SERVENT,0,netid);
					}
					
				}

			// host 
			}else if (iniFile.isName("[Host]"))
			{
				GnuID netid;
				netid.clear();
				Host h;
				ServHost::TYPE type=ServHost::T_NONE;
				unsigned int time=0;
				while (iniFile.readNext())
				{
					if (iniFile.isName("[End]"))
						break;
					else if (iniFile.isName("address"))
						h.fromStr(iniFile.getStrValue(),DEFAULT_PORT);
					else if (iniFile.isName("type"))
						type = ServHost::getTypeFromStr(iniFile.getStrValue());
					else if (iniFile.isName("time"))
						time = iniFile.getIntValue();
					else if (iniFile.isName("networkID"))
						netid.fromStr(iniFile.getStrValue());
				}
				servMgr->addHost(h,type,time,netid);

			}


		}
	}

	if (!numFilters)
		setFilterDefaults();

}
// --------------------------------------------------
int ServMgr::findChannel(ChanInfo &info)
{
	char idStr[64];
	info.id.toStr(idStr);

	LOG_NETWORK("Querying network: %s %s",info.name.cstr(),idStr);

	if (info.id.isSet())
	{
		// if we have an ID then try and connect to known hosts carrying channel.
		ServHost sh = getOutgoingServent(info.id);
		addOutgoing(sh.host,info.id,true);
	}

	GnuPacket pack;

	XML xml;
	XML::Node *n = info.createQueryXML();
	xml.setRoot(n);
	pack.initFind(NULL,&xml);

	addReplyID(pack.id);
	int cnt = broadcast(pack,NULL);

	return cnt;
}
// --------------------------------------------------
// add outgoing network connection from string (ip:port format)
bool ServMgr::addOutgoing(Host h, GnuID &netid, bool pri)
{
	if (h.ip)
	{
		if (!findServent(h.ip,h.port,netid))
		{
			Servent *sv = allocServent();
			if (sv)
			{
				if (pri)
					sv->priorityConnect = true;
				sv->networkID = netid;
				sv->initOutgoing(h,Servent::T_OUTGOING);
				return true;
			}
		}
	}
	return false;
}
// --------------------------------------------------
void ServMgr::procConnectArgs(char *str,GnuID &netid)
{
	char arg[512];
	char curr[256];
	char *cmd = strstr(str,"?");

	GnuID noid;
	noid.clear();

	if (cmd)
	{
		*cmd++ = 0;	// remove "?"

		while (cmd=nextCGIarg(cmd,curr,arg))
		{
			// ip - add direct network connection to client with channel
			if (strcmp(curr,"ip")==0)
			{
				Host h;
				h.fromStr(arg,DEFAULT_PORT);
				if (servMgr->addOutgoing(h,noid,true))
					LOG_NETWORK("Added connection: %s",arg);

			}else if (strcmp(curr,"pip")==0)
			{
				Host h;
				h.fromStr(arg,DEFAULT_PORT);
				if (servMgr->addOutgoing(h,netid,true))
					LOG_NETWORK("Added private connection: %s",arg);
			}

		}
	}
}

// --------------------------------------------------
bool ServMgr::start()
{
	static ThreadInfo serverThread,clientThread,idleThread;

	serverThread.func = ServMgr::serverProc;
	if (!sys->startThread(&serverThread))
		return false;

	clientThread.func = ServMgr::clientProc;
	if (!sys->startThread(&clientThread))
		return false;

	idleThread.func = ServMgr::idleProc;
	if (!sys->startThread(&idleThread))
		return false;

	return true;
}
// --------------------------------------------------
int ServMgr::clientProc(ThreadInfo *thread)
{

	thread->lock();

	while(thread->active)
	{
		if (servMgr->autoConnect)
		{
			if (servMgr->needConnections() || servMgr->forceLookup)
			{
				if (servMgr->needHosts() || servMgr->forceLookup)
				{
					// do lookup to find some hosts

					Host lh;
					lh.fromStr(servMgr->connectHost,DEFAULT_PORT);

					GnuID noid;
					noid.clear();

					if (!servMgr->findServent(lh.ip,lh.port,noid))
					{
						Servent *sv = servMgr->allocServent();
						if (sv)
						{
							LOG_DEBUG("Lookup: %s",servMgr->connectHost);
							sv->initOutgoing(lh,Servent::T_LOOKUP);
							servMgr->forceLookup = false;
						}
					}
				}

				for(int i=0; i<MAX_TRYOUT; i++)
				{
					if (servMgr->outUsedFull())
						break;
					if (servMgr->tryFull())
						break;


					GnuID noid;
					noid.clear();

					ServHost sh = servMgr->getOutgoingServent(noid);

					servMgr->addOutgoing(sh.host,sh.netid,false);
					sys->sleep(servMgr->tryoutDelay);
					break;
				}
			}
		}else{
			for(int i=0; i<ServMgr::MAX_SERVENTS; i++)
			{
				Servent *s = &servMgr->servents[i];
				if (s->type == Servent::T_OUTGOING) 
					s->thread.active = false;

			}
		}
		sys->sleepIdle();
	}
	thread->unlock();
	return 0;
}
// --------------------------------------------------
int ServMgr::idleProc(ThreadInfo *thread)
{

	thread->lock();

	unsigned int lastPasvFind=0;
	unsigned int lastBroadcast=0;


	// nothing much to do for the first couple of seconds, so just hang around.
	sys->sleep(2000);	

	unsigned int lastBWcheck=0;
	unsigned int bytesIn=0,bytesOut=0;

	while(thread->active)
	{
		stats.update();


		if (servMgr->relayBroadcast)
			if ((sys->getTime()-lastBroadcast) > servMgr->relayBroadcast)
			{
				chanMgr->broadcastRelays(chanMgr->broadcastTTL,NULL);
				lastBroadcast = sys->getTime();
			}

#if 0
		if (servMgr->passiveSearch)
			if ((sys->getTime()-lastPasvFind) > servMgr->passiveSearch)
			{
				LOG_DEBUG("Passive find");
				GnuPacket pack;
				pack.initFind("",NULL);
				servMgr->addActiveID(pack.id);
				servMgr->broadcast(pack,NULL);
				lastPasvFind = sys->getTime();
			}
#endif

#if 0
		if (chanMgr->searchActive)
		{
			if ((sys->getTime()-chanMgr->lastSearch) > 15)		
			{
				if (chanMgr->numFinds >= 2)
					chanMgr->searchActive = false;
				else
				{
					if (servMgr->findChannel(chanMgr->searchInfo))
					{
						LOG_DEBUG("Auto find: %d",chanMgr->numFinds);
						chanMgr->lastSearch = sys->getTime();
						chanMgr->numFinds++;
					}
				}
			}
		}
#endif
		// clear dead hits

		chanMgr->clearDeadHits();

		// kill old idle channels
		{
			int cnt=0;
			Channel *oldest=NULL;
			unsigned int last = (unsigned int )-1;
			for(int i=0; i<ChanMgr::MAX_CHANNELS; i++)
			{
				Channel *c = &chanMgr->channels[i];
				if (c->isIdle())
				{
					cnt++;
					if (c->lastIdleTime < last)
					{
						oldest = c;
						last = c->lastIdleTime;
					}
				}
			}
			if (cnt > ChanMgr::MAX_IDLE)
				if (oldest)
					oldest->thread.active = false;
		}


		// check for too many connections and kill some if needed.
		//if (!servMgr->isRelay)
		{
			char ipStr[64];
			if (servMgr->inOver())
			{
				Servent *s=servMgr->findOldestServent(Servent::T_INCOMING);
				if (s)
				{
					s->getHost().toStr(ipStr);
					LOG_NETWORK("Killing incoming (%d): %s",servMgr->numConnected(Servent::T_INCOMING),ipStr);
					s->thread.active = false;
				}
			}
			if (servMgr->outOver())
			{
				Servent *s=servMgr->findOldestServent(Servent::T_OUTGOING);
				if (s)
				{
					s->getHost().toStr(ipStr);
					LOG_NETWORK("Killing outgoing (%d): %s",servMgr->numConnected(Servent::T_OUTGOING),ipStr);
					s->thread.active = false;
				}
			}
		}

		if (servMgr->shutdownTimer)
		{
			if (--servMgr->shutdownTimer <= 0)
			{
				debugtest = 1;
				peercastInst->saveSettings();
				sys->exit();
			}
		}



		sys->sleep(500);
	}

	thread->unlock();
	return 0;
}

// --------------------------------------------------
int ServMgr::serverProc(ThreadInfo *thread)
{
	thread->lock();

	Servent *serv = servMgr->allocServent();
	serv->permAlloc = true;

	Servent *serv2 = servMgr->allocServent();
	serv2->permAlloc = true;

	unsigned int lastLookupTime=0;


	while (thread->active)
	{
		if (servMgr->restartServer)
		{
			serv->abort();		// force close

			serv2->abort();		// force close

			servMgr->restartServer = false;
		}

		if (servMgr->autoServe)
		{
			serv->allow = servMgr->allowServer1;
			serv2->allow = servMgr->allowServer2;

			if ((serv->lastConnect) && ((sys->getTime()-serv->lastConnect) > servMgr->firewallTimeout))
			{
				if (servMgr->getFirewall()==ServMgr::FW_UNKNOWN)
				{
					LOG_DEBUG("Server has had no incoming connections");
					servMgr->setFirewall(ServMgr::FW_ON);
				}
			}


			// force lookup every 24 hours to check for new messages
			if ((sys->getTime()-lastLookupTime) > 60*60*24)
			{
				servMgr->forceLookup = true;
				lastLookupTime = sys->getTime();
			}


			if ((serv->status == Servent::S_NONE) || (serv2->status == Servent::S_NONE))
			{
				LOG_DEBUG("Starting servers");
				servMgr->forceLookup = true;

				//if (servMgr->serverHost.ip != 0)
				{

					if (servMgr->forceNormal)
						servMgr->setFirewall(ServMgr::FW_OFF);
					else
						servMgr->setFirewall(ServMgr::FW_UNKNOWN);

					Host h = servMgr->serverHost;

					if (serv->status == Servent::S_NONE)
						serv->initServer(h);

					h.port++;
					if (serv2->status == Servent::S_NONE)
						serv2->initServer(h);


				}
			}
		}else{
			// stop server
			serv->abort();		// force close
			serv2->abort();		// force close

			// cancel incoming connectuions
			for(int i=0; i<ServMgr::MAX_SERVENTS; i++)
			{
				Servent *s = &servMgr->servents[i];
				if (s->type == Servent::T_INCOMING)
					s->thread.active = false;
			}

			servMgr->setFirewall(ServMgr::FW_ON);
		}
			
		sys->sleepIdle();

	}

	thread->unlock();
	return 0;
}

// -----------------------------------
XML::Node *ServMgr::createServentXML()
{

	return new XML::Node("servent agent=\"%s\" preview=\"%d\"",
		PCX_AGENT,maxPreviewTime
		);
}

// --------------------------------------------------
const char *ServHost::getTypeStr(TYPE t)
{
	switch(t)
	{
		case T_NONE: return "NONE";
		case T_STREAM: return "STREAM";
		case T_CHANNEL: return "CHANNEL";
		case T_SERVENT: return "SERVENT";
	}
	return "UNKNOWN";
}
// --------------------------------------------------
ServHost::TYPE ServHost::getTypeFromStr(const char *s)
{
	if (stricmp(s,"NONE")==0)
		return T_NONE;
	else if (stricmp(s,"SERVENT")==0)
		return T_SERVENT;
	else if (stricmp(s,"STREAM")==0)
		return T_STREAM;
	else if (stricmp(s,"CHANNEL")==0)
		return T_CHANNEL;

	return T_NONE;
}