connectionmanager.cpp

软件是使用VC70

/* 
 * Copyright (C) 2001-2005 Jacek Sieka, arnetheduck on gmail point com
 *
 * 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 "ResourceManager.h"
#include "DownloadManager.h"
#include "UploadManager.h"
#include "CryptoManager.h"
#include "ClientManager.h"
#include "QueueManager.h"

#include "UserConnection.h"

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

	features.push_back(UserConnection::FEATURE_MINISLOTS);
	features.push_back(UserConnection::FEATURE_XML_BZLIST);
	features.push_back(UserConnection::FEATURE_ADCGET);
	features.push_back(UserConnection::FEATURE_TTHL);
	features.push_back(UserConnection::FEATURE_TTHF);

	adcFeatures.push_back("+BASE");
}

/**
 * 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);
	{
		Lock l(cs);
		ConnectionQueueItem::Iter i = find(downloads.begin(), downloads.end(), aUser);
		if(i == downloads.end()) {
			getCQI(aUser, true);
		} else {
			ConnectionQueueItem* cqi = *i;
			if(cqi->getState() == ConnectionQueueItem::IDLE) {
				if(find(pendingAdd.begin(), pendingAdd.end(), aUser) == pendingAdd.end())
					pendingAdd.push_back(aUser);
				return;
			}
		}
	}
}

ConnectionQueueItem* ConnectionManager::getCQI(const User::Ptr& aUser, bool download) {
	ConnectionQueueItem* cqi = new ConnectionQueueItem(aUser, download);
	if(download) {
		dcassert(find(downloads.begin(), downloads.end(), aUser) == downloads.end());
		downloads.push_back(cqi);

	} else {
		dcassert(find(uploads.begin(), uploads.end(), aUser) == uploads.end());
		uploads.push_back(cqi);
	}

	fire(ConnectionManagerListener::Added(), cqi);
	
	return cqi;
}

void ConnectionManager::putCQI(ConnectionQueueItem* cqi) {
	fire(ConnectionManagerListener::Removed(), cqi);
	if(cqi->getDownload()) {
		dcassert(find(downloads.begin(), downloads.end(), cqi) != downloads.end());
		downloads.erase(find(downloads.begin(), downloads.end(), cqi));
	} else {
		dcassert(find(uploads.begin(), uploads.end(), cqi) != uploads.end());
		uploads.erase(find(uploads.begin(), uploads.end(), cqi));
	}
	delete cqi;
}

void ConnectionManager::putDownloadConnection(UserConnection* aSource, bool reuse /* = false */, bool ntd /* = false */) {
	ConnectionQueueItem* cqi = aSource->getCQI();
	dcassert(cqi);

	if(reuse) {
		dcdebug("ConnectionManager::putDownloadConnection Pooling reusable connection %p to %s\n", aSource, aSource->getUser()->getNick().c_str());
		// Pool it for later usage...
		aSource->addListener(this);
		{
			Lock l(cs);
			cqi->setState(ConnectionQueueItem::IDLE);
		}
	} else {
		// Disassociate the two...
		aSource->setCQI(NULL);

		bool hasDown = QueueManager::getInstance()->hasDownload(aSource->getUser());

		{
			Lock l(cs);
			cqi->setConnection(NULL);
			if(hasDown) {
				cqi->setLastAttempt(GET_TICK());
				cqi->setState(ConnectionQueueItem::WAITING);
			} else {
				putCQI(cqi);
			}
		}

		if(ntd) {
			aSource->unsetFlag(UserConnection::FLAG_DOWNLOAD);
			addUploadConnection(aSource);
		} else {
			putConnection(aSource);
		}
	}
}

void ConnectionManager::putUploadConnection(UserConnection* aSource, bool ntd) {
	ConnectionQueueItem* cqi = aSource->getCQI();
	aSource->setCQI(NULL);

	if(ntd) {
		// We should pass it to the download manager...
		aSource->unsetFlag(UserConnection::FLAG_UPLOAD);
		aSource->setFlag(UserConnection::FLAG_DOWNLOAD);
		addDownloadConnection(aSource, false);
	} else {
		putConnection(aSource);
	}

	{
		Lock l(cs);
		putCQI(cqi);
	}
}

UserConnection* ConnectionManager::getConnection(bool aNmdc) throw(SocketException) {
	UserConnection* uc = new UserConnection();
	uc->addListener(this);
	{
		Lock l(cs);
		userConnections.push_back(uc);
	}
	if(aNmdc)
		uc->setFlag(UserConnection::FLAG_NMDC);
	return uc;
}

void ConnectionManager::putConnection(UserConnection* aConn) {
	aConn->removeListeners();
	aConn->disconnect();
	{
		Lock l(cs);
		
		dcassert(find(userConnections.begin(), userConnections.end(), aConn) != userConnections.end());
		userConnections.erase(find(userConnections.begin(), userConnections.end(), aConn));
		pendingDelete.push_back(aConn);
	}
}

void ConnectionManager::on(TimerManagerListener::Second, u_int32_t aTick) throw() {
	User::List passiveUsers;
	ConnectionQueueItem::List removed;
	UserConnection::List added;
	UserConnection::List penDel;

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

	{
		Lock l(cs);

		int attempts = 0;

		for(ConnectionQueueItem::Iter i = downloads.begin(); i != downloads.end(); ++i) {
			ConnectionQueueItem* cqi = *i;

			if(cqi->getState() == ConnectionQueueItem::ACTIVE) {
				// Do nothing
			} else if(cqi->getState() == ConnectionQueueItem::IDLE) {
				User::Iter it = find(pendingAdd.begin(), pendingAdd.end(), cqi->getUser());
				if(it != pendingAdd.end()) {
					dcassert(cqi->getConnection());
					dcassert(cqi->getConnection()->getCQI() == cqi);
					cqi->setState(ConnectionQueueItem::ACTIVE);
					cqi->getConnection()->removeListener(this);
					added.push_back(cqi->getConnection());

					pendingAdd.erase(it);
				}
			} else {
				
				if(!cqi->getUser()->isOnline()) {
					// Not online anymore...remove it from the pending...
					removed.push_back(cqi);
					continue;
				} 
				
				if(cqi->getUser()->isSet(User::PASSIVE) && (SETTING(CONNECTION_TYPE) != SettingsManager::CONNECTION_ACTIVE)) {
					passiveUsers.push_back(cqi->getUser());
					removed.push_back(cqi);
					continue;
				}

				if( ((cqi->getLastAttempt() + 60*1000) < aTick) && (attempts < 2) ) {
					cqi->setLastAttempt(aTick);

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

					// Always start high-priority downloads unless we have 3 more than maxdownslots already...
					bool startDown = !tooMany && !tooFast;

					if(!startDown) {
						bool extraFull = (SETTING(DOWNLOAD_SLOTS) != 0) && (DownloadManager::getInstance()->getDownloadCount() >= (size_t)(SETTING(DOWNLOAD_SLOTS)+3));
						startDown = !extraFull && QueueManager::getInstance()->hasDownload(cqi->getUser(), QueueItem::HIGHEST);
					}

					if(cqi->getState() == ConnectionQueueItem::WAITING) {
						if(startDown) {
							cqi->setState(ConnectionQueueItem::CONNECTING);
							cqi->getUser()->connect();
							fire(ConnectionManagerListener::StatusChanged(), 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(((cqi->getLastAttempt() + 50*1000) < aTick) && (cqi->getState() == ConnectionQueueItem::CONNECTING)) {
					fire(ConnectionManagerListener::Failed(), cqi, STRING(CONNECTION_TIMEOUT));
					cqi->setState(ConnectionQueueItem::WAITING);
				}
			}
		}

		pendingAdd.clear();

		for(ConnectionQueueItem::Iter m = removed.begin(); m != removed.end(); ++m) {
			putCQI(*m);
		}

		penDel = pendingDelete;
		pendingDelete.clear();

	}

	for_each(penDel.begin(), penDel.end(), DeleteFunction<UserConnection*>());

	for(User::Iter ui = passiveUsers.begin(); ui != passiveUsers.end(); ++ui) {
		QueueManager::getInstance()->removeSources(*ui, QueueItem::Source::FLAG_PASSIVE);
	}

	for(UserConnection::Iter i = added.begin(); i != added.end(); ++i) {
		DownloadManager::getInstance()->addConnection(*i);
	}
}

void ConnectionManager::on(TimerManagerListener::Minute, u_int32_t aTick) throw() {	
	Lock l(cs);

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

static const u_int32_t FLOOD_TRIGGER = 20000;
static const u_int32_t FLOOD_ADD = 2000;

/**
 * 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::on(ServerSocketListener::IncomingConnection) throw() {
	UserConnection* uc = NULL;
	u_int32_t now = GET_TICK();

	if(now > floodCounter) {
		floodCounter = now + FLOOD_ADD;
	} else {
		if(now + FLOOD_TRIGGER < floodCounter) {
			Socket s;
			try {
				s.accept(socket);
			} catch(const SocketException&) {
				// ...
			}
			dcdebug("Connection flood detected!\n");
			return;
		} else {
			floodCounter += FLOOD_ADD;
		}
	}

	try { 
		uc = getConnection(false);
		uc->setFlag(UserConnection::FLAG_INCOMING);
		uc->setState(UserConnection::STATE_SUPNICK);
		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::nmdcConnect(const string& aServer, short aPort, const string& aNick) {
	if(shuttingDown)
		return;

	UserConnection* uc = NULL;
	try {
		uc = getConnection(true);
		uc->setNick(aNick);
		uc->setState(UserConnection::STATE_CONNECT);
		uc->setFlag(UserConnection::FLAG_NMDC);
		uc->connect(aServer, aPort);
	} catch(const SocketException&) {
		if(uc)
			putConnection(uc);
	}
}