connectionmanager.cpp

p2p技术C源代码.rar

/* 
 * Copyright (C) 2001-2003 Jacek Sieka, [e-mail]
 *
 * 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.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 */

#include "stdinc.h"
#include "DCPlusPlus.h"

#include "ConnectionManager.h"
#include "DownloadManager.h"
#include "UploadManager.h"

#include "UserConnection.h"
#include "CryptoManager.h"
#include "ClientManager.h"
#include "QueueManager.h"

ConnectionManager* Singleton<ConnectionManager>::instance = NULL;

//#define WITH_GETZBLOCK 1

ConnectionManager::ConnectionManager() : shuttingDown(false) {
	TimerManager::getInstance()->addListener(this);
	socket.addListener(this);

	features.push_back("BZList");

#ifdef WITH_GETZBLOCK
	features.push_back("GetZBlock");
#endif
};


/**
 * Request a connection for downloading.
 * DownloadManager::addConnection will be called as soon as the connection is ready
 * for downloading.
 * @param aUser The user to connect to.
 */
void ConnectionManager::getDownloadConnection(const User::Ptr& aUser) {
	dcassert((bool)aUser);
	ConnectionQueueItem* cqi = NULL;
	{
		Lock l(cs);

		// Check the download pool
		if(find(downPool.begin(), downPool.end(), aUser) != downPool.end()) {
			if(find(pendingAdd.begin(), pendingAdd.end(), aUser) == pendingAdd.end())
				pendingAdd.push_back(aUser);
			return;
		}
		
		// See if we're already trying to connect
		for(ConnectionQueueItem::TimeIter i = pendingDown.begin(); i != pendingDown.end(); ++i) {
			if(i->first->getUser() == aUser) {
				return;
			}
		}

		// Check if we have an active download connection already
		for(ConnectionQueueItem::Iter j = active.begin(); j != active.end(); ++j) {
			dcassert((*j)->getConnection());
			if((*j == aUser) && ((*j)->getConnection()->isSet(UserConnection::FLAG_DOWNLOAD)))
				return;
		}
		
		// Add it to the pending...
		cqi = new ConnectionQueueItem(aUser);
		cqi->setState(ConnectionQueueItem::WAITING);
		pendingDown.insert(make_pair(cqi, 0));

		fire(ConnectionManagerListener::ADDED, cqi);
	}
}

void ConnectionManager::putDownloadConnection(UserConnection* aSource, bool reuse /* = false */) {
	// Pool it for later usage...
	if(reuse) {
		aSource->addListener(this);
		{
			Lock l(cs);
			aSource->getCQI()->setState(ConnectionQueueItem::IDLE);

			dcassert(find(active.begin(), active.end(), aSource->getCQI()) != active.end());
			active.erase(find(active.begin(), active.end(), aSource->getCQI()));
			
			downPool.push_back(aSource->getCQI());
		}
		dcdebug("ConnectionManager::putDownloadConnection Pooling reusable connection %p to %s\n", aSource, aSource->getUser()->getNick().c_str());
		
	} else {
		if(QueueManager::getInstance()->hasDownload(aSource->getCQI()->getUser())) {
			aSource->removeListeners();
			aSource->disconnect();
			Lock l(cs);

			ConnectionQueueItem* cqi = aSource->getCQI();
			dcassert(cqi);
			
			// Remove the userconnection, don't need it any more
			dcassert(find(userConnections.begin(), userConnections.end(), aSource) != userConnections.end());
			userConnections.erase(find(userConnections.begin(), userConnections.end(), aSource));
			pendingDelete.push_back(aSource);

			cqi->setConnection(NULL);
			cqi->setState(ConnectionQueueItem::WAITING);
			
			dcassert(find(active.begin(), active.end(), aSource->getCQI()) != active.end());
			active.erase(find(active.begin(), active.end(), aSource->getCQI()));
			
			pendingDown[cqi] = GET_TICK();
		} else {
			{
				Lock l(cs);
				dcassert(find(active.begin(), active.end(), aSource->getCQI()) != active.end());
				active.erase(find(active.begin(), active.end(), aSource->getCQI()));

			}
			putConnection(aSource);
		}
	}
}

void ConnectionManager::putUploadConnection(UserConnection* aSource) {
	{
		Lock l(cs);
		dcassert(find(active.begin(), active.end(), aSource->getCQI()) != active.end());
		active.erase(find(active.begin(), active.end(), aSource->getCQI()));
	}
	putConnection(aSource);
}

void ConnectionManager::putConnection(UserConnection* aConn) {
	aConn->removeListeners();
	aConn->disconnect();
	ConnectionQueueItem* cqi = NULL;
	{
		Lock l(cs);
		cqi = aConn->getCQI();
		
		dcassert(find(userConnections.begin(), userConnections.end(), aConn) != userConnections.end());
		userConnections.erase(find(userConnections.begin(), userConnections.end(), aConn));
		pendingDelete.push_back(aConn);
	}
	if(cqi) {
		fire(ConnectionManagerListener::REMOVED, cqi);
		delete cqi;
	}
}

void ConnectionManager::onTimerSecond(u_int32_t aTick) {
	ConnectionQueueItem::List failPassive;
	ConnectionQueueItem::List connecting;
	ConnectionQueueItem::List removed;
	User::List getDown;
	{
		Lock l(cs);
		{
			for(User::Iter k = pendingAdd.begin(); k != pendingAdd.end(); ++k) {
				ConnectionQueueItem::Iter i = find(downPool.begin(), downPool.end(), *k);
				if(i == downPool.end()) {
					// Hm, connection must have failed before it could be collected...
					getDown.push_back(*k);
				} else {
					ConnectionQueueItem* cqi = *i;
					downPool.erase(i);
					dcassert(find(active.begin(), active.end(), cqi) == active.end());
					active.push_back(cqi);
					dcassert(cqi->getConnection());
					dcassert(cqi->getConnection()->getCQI() == cqi);
					cqi->getConnection()->removeListener(this);
					DownloadManager::getInstance()->addConnection(cqi->getConnection());
				}
			}
			pendingAdd.clear();
		}

		bool tooMany = ((SETTING(DOWNLOAD_SLOTS) != 0) && DownloadManager::getInstance()->getDownloads() >= SETTING(DOWNLOAD_SLOTS));
		bool tooFast = ((SETTING(MAX_DOWNLOAD_SPEED) != 0 && DownloadManager::getInstance()->getAverageSpeed() >= (SETTING(MAX_DOWNLOAD_SPEED)*1024)));
		
		bool startDown = !tooMany && !tooFast;

		int attempts = 0;
		
		ConnectionQueueItem::TimeIter i = pendingDown.begin();
		while(i != pendingDown.end()) {
			ConnectionQueueItem* cqi = i->first;
			dcassert(cqi->getUser());

			if(!cqi->getUser()->isOnline()) {
				// Not online anymore...remove him from the pending...
				pendingDown.erase(i++);
				removed.push_back(cqi);
				continue;
			}

			if( ((i->second + 60*1000) < aTick) && (attempts < 2) ) {
				i->second = aTick;

				if(!QueueManager::getInstance()->hasDownload(cqi->getUser())) {
					pendingDown.erase(i++);
					removed.push_back(cqi);
					continue;
				}

				if(cqi->getUser()->isSet(User::PASSIVE) && (SETTING(CONNECTION_TYPE) != SettingsManager::CONNECTION_ACTIVE)) {
					pendingDown.erase(i++);
					failPassive.push_back(cqi);
					continue;
				}

				// Always start high-priority downloads...
				if(!startDown) {
					startDown = QueueManager::getInstance()->hasDownload(cqi->getUser(), QueueItem::HIGHEST);
				}
				if(cqi->getState() == ConnectionQueueItem::WAITING) {
					if(startDown) {
						cqi->setState(ConnectionQueueItem::CONNECTING);
						cqi->getUser()->connect();
						fire(ConnectionManagerListener::STATUS_CHANGED, cqi);
						attempts++;
					} else {
						cqi->setState(ConnectionQueueItem::NO_DOWNLOAD_SLOTS);
						fire(ConnectionManagerListener::FAILED, cqi, STRING(ALL_DOWNLOAD_SLOTS_TAKEN));
					}
				} else if(cqi->getState() == ConnectionQueueItem::NO_DOWNLOAD_SLOTS && startDown) {
					cqi->setState(ConnectionQueueItem::WAITING);
				}
			} else if(((i->second + 50*1000) < aTick) && (cqi->getState() == ConnectionQueueItem::CONNECTING)) {
				fire(ConnectionManagerListener::FAILED, cqi, STRING(CONNECTION_TIMEOUT));
				cqi->setState(ConnectionQueueItem::WAITING);
			}
			++i;
		}
	}

	ConnectionQueueItem::Iter m;
	for(m = removed.begin(); m != removed.end(); ++m) {
		fire(ConnectionManagerListener::REMOVED, *m);
		delete *m;
	}
	for(m = failPassive.begin(); m != failPassive.end(); ++m) {
		QueueManager::getInstance()->removeSources((*m)->getUser(), QueueItem::Source::FLAG_PASSIVE);
		fire(ConnectionManagerListener::REMOVED, *m);
		delete *m;
	}
	for(User::Iter n = getDown.begin(); n != getDown.end(); ++n) {
		getDownloadConnection(*n);
	}
}

void ConnectionManager::onTimerMinute(u_int32_t aTick) {
	Lock l(cs);

	{
		for(UserConnection::Iter j = userConnections.begin(); j != userConnections.end(); ++j) {
			if(((*j)->getLastActivity() + 180*1000) < aTick) {
				(*j)->disconnect();
			}
		}
	}

	for_each(pendingDelete.begin(), pendingDelete.end(), DeleteFunction<UserConnection*>());
	pendingDelete.clear();
}

/**
 * Someone's connecting, accept the connection and wait for identification...
 * It's always the other fellow that starts sending if he made the connection.
 */
void ConnectionManager::onIncomingConnection() throw() {
	UserConnection* uc = NULL;
	
	try { 
		uc = getConnection();
		uc->setFlag(UserConnection::FLAG_INCOMING);
		uc->setState(UserConnection::STATE_NICK);
		uc->setLastActivity(GET_TICK());
		uc->accept(socket);
	} catch(const SocketException& e) {
		dcdebug("ConnectionManager::OnIncomingConnection caught: %s\n", e.getError().c_str());
		if(uc)
			putConnection(uc);
	}
}

void ConnectionManager::connect(const string& aServer, short aPort, const string& aNick) {
	if(shuttingDown)
		return;

	UserConnection* uc = NULL;
	try {
		uc = getConnection();