github
/
aria2
mirror of https://github.com/aria2/aria2
1
0
Fork 0
Code Issues Releases Wiki Activity

2010-06-12 Tatsuhiro Tsujikawa <t-tujikawa@users.sourceforge.net> 

Renamed member variables. Removed member variable peerConnection.
	* src/DefaultBtInteractive.cc
	* src/DefaultBtInteractive.h
	* src/PeerInteractionCommand.cc
pull/1/head
Tatsuhiro Tsujikawa 2010-06-12 03:30:56 +00:00
parent 465502299b
commit d8d141e5f7
4 changed files with 155 additions and 167 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
ChangeLog
View File

@ -1,3 +1,10 @@
2010-06-12 Tatsuhiro Tsujikawa <t-tujikawa@users.sourceforge.net>
Renamed member variables. Removed member variable peerConnection.
* src/DefaultBtInteractive.cc
* src/DefaultBtInteractive.h
* src/PeerInteractionCommand.cc
2010-06-12 Tatsuhiro Tsujikawa <t-tujikawa@users.sourceforge.net> 2010-06-12 Tatsuhiro Tsujikawa <t-tujikawa@users.sourceforge.net>
Renamed member variables Renamed member variables

278
src/DefaultBtInteractive.cc
View File

@ -63,7 +63,6 @@
#include "BtMessageDispatcher.h" #include "BtMessageDispatcher.h"
#include "BtMessageFactory.h" #include "BtMessageFactory.h"
#include "BtRequestFactory.h" #include "BtRequestFactory.h"
#include "PeerConnection.h"
#include "Logger.h" #include "Logger.h"
#include "LogFactory.h" #include "LogFactory.h"
#include "StringFormat.h" #include "StringFormat.h"
@ -81,17 +80,17 @@ DefaultBtInteractive::DefaultBtInteractive
const SharedHandle<Peer>& peer) const SharedHandle<Peer>& peer)
: :
_downloadContext(downloadContext), _downloadContext(downloadContext),
peer(peer), _peer(peer),
_metadataGetMode(false), _metadataGetMode(false),
logger(LogFactory::getInstance()), _logger(LogFactory::getInstance()),
allowedFastSetSize(10), _allowedFastSetSize(10),
haveTimer(global::wallclock), _haveTimer(global::wallclock),
keepAliveTimer(global::wallclock), _keepAliveTimer(global::wallclock),
floodingTimer(global::wallclock), _floodingTimer(global::wallclock),
inactiveTimer(global::wallclock), _inactiveTimer(global::wallclock),
_pexTimer(global::wallclock), _pexTimer(global::wallclock),
_perSecTimer(global::wallclock), _perSecTimer(global::wallclock),
keepAliveInterval(120), _keepAliveInterval(120),
_utPexEnabled(false), _utPexEnabled(false),
_dhtEnabled(false), _dhtEnabled(false),
_numReceivedMessage(0), _numReceivedMessage(0),
@ -102,15 +101,15 @@ DefaultBtInteractive::~DefaultBtInteractive() {}
void DefaultBtInteractive::initiateHandshake() { void DefaultBtInteractive::initiateHandshake() {
SharedHandle<BtMessage> message = SharedHandle<BtMessage> message =
messageFactory->createHandshakeMessage _messageFactory->createHandshakeMessage
(bittorrent::getInfoHash(_downloadContext), bittorrent::getStaticPeerId()); (bittorrent::getInfoHash(_downloadContext), bittorrent::getStaticPeerId());
dispatcher->addMessageToQueue(message); _dispatcher->addMessageToQueue(message);
dispatcher->sendMessages(); _dispatcher->sendMessages();
} }
BtMessageHandle DefaultBtInteractive::receiveHandshake(bool quickReply) { BtMessageHandle DefaultBtInteractive::receiveHandshake(bool quickReply) {
SharedHandle<BtHandshakeMessage> message = SharedHandle<BtHandshakeMessage> message =
btMessageReceiver->receiveHandshake(quickReply); _btMessageReceiver->receiveHandshake(quickReply);
if(message.isNull()) { if(message.isNull()) {
return SharedHandle<BtMessage>(); return SharedHandle<BtMessage>();
} }
@ -119,7 +118,7 @@ BtMessageHandle DefaultBtInteractive::receiveHandshake(bool quickReply) {
throw DL_ABORT_EX throw DL_ABORT_EX
(StringFormat (StringFormat
("CUID#%s - Drop connection from the same Peer ID", ("CUID#%s - Drop connection from the same Peer ID",
util::itos(cuid).c_str()).str()); util::itos(_cuid).c_str()).str());
} }
std::vector<SharedHandle<Peer> > activePeers; std::vector<SharedHandle<Peer> > activePeers;
_peerStorage->getActivePeers(activePeers); _peerStorage->getActivePeers(activePeers);
@ -129,36 +128,36 @@ BtMessageHandle DefaultBtInteractive::receiveHandshake(bool quickReply) {
throw DL_ABORT_EX throw DL_ABORT_EX
(StringFormat (StringFormat
("CUID#%s - Same Peer ID has been already seen.", ("CUID#%s - Same Peer ID has been already seen.",
util::itos(cuid).c_str()).str()); util::itos(_cuid).c_str()).str());
} }
} }
peer->setPeerId(message->getPeerId()); _peer->setPeerId(message->getPeerId());
if(message->isFastExtensionSupported()) { if(message->isFastExtensionSupported()) {
peer->setFastExtensionEnabled(true); _peer->setFastExtensionEnabled(true);
if(logger->info()) { if(_logger->info()) {
logger->info(MSG_FAST_EXTENSION_ENABLED, util::itos(cuid).c_str()); _logger->info(MSG_FAST_EXTENSION_ENABLED, util::itos(_cuid).c_str());
} }
} }
if(message->isExtendedMessagingEnabled()) { if(message->isExtendedMessagingEnabled()) {
peer->setExtendedMessagingEnabled(true); _peer->setExtendedMessagingEnabled(true);
if(!_utPexEnabled) { if(!_utPexEnabled) {
_extensionMessageRegistry->removeExtension("ut_pex"); _extensionMessageRegistry->removeExtension("ut_pex");
} }
if(logger->info()) { if(_logger->info()) {
logger->info(MSG_EXTENDED_MESSAGING_ENABLED, util::itos(cuid).c_str()); _logger->info(MSG_EXTENDED_MESSAGING_ENABLED, util::itos(_cuid).c_str());
} }
} }
if(message->isDHTEnabled()) { if(message->isDHTEnabled()) {
peer->setDHTEnabled(true); _peer->setDHTEnabled(true);
if(logger->info()) { if(_logger->info()) {
logger->info(MSG_DHT_ENABLED_PEER, util::itos(cuid).c_str()); _logger->info(MSG_DHT_ENABLED_PEER, util::itos(_cuid).c_str());
} }
} }
if(logger->info()) { if(_logger->info()) {
logger->info(MSG_RECEIVE_PEER_MESSAGE, util::itos(cuid).c_str(), _logger->info(MSG_RECEIVE_PEER_MESSAGE, util::itos(_cuid).c_str(),
peer->ipaddr.c_str(), peer->port, _peer->ipaddr.c_str(), _peer->port,
message->toString().c_str()); message->toString().c_str());
} }
return message; return message;
@ -170,17 +169,17 @@ BtMessageHandle DefaultBtInteractive::receiveAndSendHandshake() {
void DefaultBtInteractive::doPostHandshakeProcessing() { void DefaultBtInteractive::doPostHandshakeProcessing() {
// Set time 0 to haveTimer to cache http/ftp download piece completion // Set time 0 to haveTimer to cache http/ftp download piece completion
haveTimer.reset(0); _haveTimer.reset(0);
keepAliveTimer = global::wallclock; _keepAliveTimer = global::wallclock;
floodingTimer = global::wallclock; _floodingTimer = global::wallclock;
_pexTimer.reset(0); _pexTimer.reset(0);
if(peer->isExtendedMessagingEnabled()) { if(_peer->isExtendedMessagingEnabled()) {
addHandshakeExtendedMessageToQueue(); addHandshakeExtendedMessageToQueue();
} }
if(!_metadataGetMode) { if(!_metadataGetMode) {
addBitfieldMessageToQueue(); addBitfieldMessageToQueue();
} }
if(peer->isDHTEnabled() && _dhtEnabled) { if(_peer->isDHTEnabled() && _dhtEnabled) {
addPortMessageToQueue(); addPortMessageToQueue();
} }
if(!_metadataGetMode) { if(!_metadataGetMode) {
@ -191,8 +190,8 @@ void DefaultBtInteractive::doPostHandshakeProcessing() {
void DefaultBtInteractive::addPortMessageToQueue() void DefaultBtInteractive::addPortMessageToQueue()
{ {
dispatcher->addMessageToQueue _dispatcher->addMessageToQueue
(messageFactory->createPortMessage(_localNode->getPort())); (_messageFactory->createPortMessage(_localNode->getPort()));
} }
void DefaultBtInteractive::addHandshakeExtendedMessageToQueue() void DefaultBtInteractive::addHandshakeExtendedMessageToQueue()
@ -206,123 +205,124 @@ void DefaultBtInteractive::addHandshakeExtendedMessageToQueue()
if(attrs.containsKey(bittorrent::METADATA)) { if(attrs.containsKey(bittorrent::METADATA)) {
m->setMetadataSize(attrs[bittorrent::METADATA_SIZE].i()); m->setMetadataSize(attrs[bittorrent::METADATA_SIZE].i());
} }
SharedHandle<BtMessage> msg = messageFactory->createBtExtendedMessage(m); SharedHandle<BtMessage> msg = _messageFactory->createBtExtendedMessage(m);
dispatcher->addMessageToQueue(msg); _dispatcher->addMessageToQueue(msg);
} }
void DefaultBtInteractive::addBitfieldMessageToQueue() { void DefaultBtInteractive::addBitfieldMessageToQueue() {
if(peer->isFastExtensionEnabled()) { if(_peer->isFastExtensionEnabled()) {
if(_pieceStorage->allDownloadFinished()) { if(_pieceStorage->allDownloadFinished()) {
dispatcher->addMessageToQueue(messageFactory->createHaveAllMessage()); _dispatcher->addMessageToQueue(_messageFactory->createHaveAllMessage());
} else if(_pieceStorage->getCompletedLength() > 0) { } else if(_pieceStorage->getCompletedLength() > 0) {
dispatcher->addMessageToQueue(messageFactory->createBitfieldMessage()); _dispatcher->addMessageToQueue(_messageFactory->createBitfieldMessage());
} else { } else {
dispatcher->addMessageToQueue(messageFactory->createHaveNoneMessage()); _dispatcher->addMessageToQueue(_messageFactory->createHaveNoneMessage());
} }
} else { } else {
if(_pieceStorage->getCompletedLength() > 0) { if(_pieceStorage->getCompletedLength() > 0) {
dispatcher->addMessageToQueue(messageFactory->createBitfieldMessage()); _dispatcher->addMessageToQueue(_messageFactory->createBitfieldMessage());
} }
} }
} }
void DefaultBtInteractive::addAllowedFastMessageToQueue() { void DefaultBtInteractive::addAllowedFastMessageToQueue() {
if(peer->isFastExtensionEnabled()) { if(_peer->isFastExtensionEnabled()) {
std::vector<size_t> fastSet; std::vector<size_t> fastSet;
bittorrent::computeFastSet(fastSet, peer->ipaddr, bittorrent::computeFastSet(fastSet, _peer->ipaddr,
_downloadContext->getNumPieces(), _downloadContext->getNumPieces(),
bittorrent::getInfoHash(_downloadContext), bittorrent::getInfoHash(_downloadContext),
allowedFastSetSize); _allowedFastSetSize);
for(std::vector<size_t>::const_iterator itr = fastSet.begin(), for(std::vector<size_t>::const_iterator itr = fastSet.begin(),
eoi = fastSet.end(); itr != eoi; ++itr) { eoi = fastSet.end(); itr != eoi; ++itr) {
dispatcher->addMessageToQueue _dispatcher->addMessageToQueue
(messageFactory->createAllowedFastMessage(*itr)); (_messageFactory->createAllowedFastMessage(*itr));
} }
} }
} }
void DefaultBtInteractive::decideChoking() { void DefaultBtInteractive::decideChoking() {
if(peer->shouldBeChoking()) { if(_peer->shouldBeChoking()) {
if(!peer->amChoking()) { if(!_peer->amChoking()) {
dispatcher->addMessageToQueue(messageFactory->createChokeMessage()); _dispatcher->addMessageToQueue(_messageFactory->createChokeMessage());
} }
} else { } else {
if(peer->amChoking()) { if(_peer->amChoking()) {
dispatcher->addMessageToQueue(messageFactory->createUnchokeMessage()); _dispatcher->addMessageToQueue(_messageFactory->createUnchokeMessage());
} }
} }
} }
void DefaultBtInteractive::checkHave() { void DefaultBtInteractive::checkHave() {
std::vector<size_t> indexes; std::vector<size_t> indexes;
_pieceStorage->getAdvertisedPieceIndexes(indexes, cuid, haveTimer); _pieceStorage->getAdvertisedPieceIndexes(indexes, _cuid, _haveTimer);
haveTimer = global::wallclock; _haveTimer = global::wallclock;
if(indexes.size() >= 20) { if(indexes.size() >= 20) {
if(peer->isFastExtensionEnabled() && _pieceStorage->allDownloadFinished()) { if(_peer->isFastExtensionEnabled() &&
dispatcher->addMessageToQueue(messageFactory->createHaveAllMessage()); _pieceStorage->allDownloadFinished()) {
_dispatcher->addMessageToQueue(_messageFactory->createHaveAllMessage());
} else { } else {
dispatcher->addMessageToQueue(messageFactory->createBitfieldMessage()); _dispatcher->addMessageToQueue(_messageFactory->createBitfieldMessage());
} }
} else { } else {
for(std::vector<size_t>::const_iterator itr = indexes.begin(), for(std::vector<size_t>::const_iterator itr = indexes.begin(),
eoi = indexes.end(); itr != eoi; ++itr) { eoi = indexes.end(); itr != eoi; ++itr) {
dispatcher->addMessageToQueue(messageFactory->createHaveMessage(*itr)); _dispatcher->addMessageToQueue(_messageFactory->createHaveMessage(*itr));
} }
} }
} }
void DefaultBtInteractive::sendKeepAlive() { void DefaultBtInteractive::sendKeepAlive() {
if(keepAliveTimer.difference(global::wallclock) >= keepAliveInterval) { if(_keepAliveTimer.difference(global::wallclock) >= _keepAliveInterval) {
dispatcher->addMessageToQueue(messageFactory->createKeepAliveMessage()); _dispatcher->addMessageToQueue(_messageFactory->createKeepAliveMessage());
dispatcher->sendMessages(); _dispatcher->sendMessages();
keepAliveTimer = global::wallclock; _keepAliveTimer = global::wallclock;
} }
} }
size_t DefaultBtInteractive::receiveMessages() { size_t DefaultBtInteractive::receiveMessages() {
size_t countOldOutstandingRequest = dispatcher->countOutstandingRequest(); size_t countOldOutstandingRequest = _dispatcher->countOutstandingRequest();
size_t msgcount = 0; size_t msgcount = 0;
for(int i = 0; i < 50; ++i) { for(int i = 0; i < 50; ++i) {
if(_requestGroupMan->doesOverallDownloadSpeedExceed() || if(_requestGroupMan->doesOverallDownloadSpeedExceed() ||
_downloadContext->getOwnerRequestGroup()->doesDownloadSpeedExceed()) { _downloadContext->getOwnerRequestGroup()->doesDownloadSpeedExceed()) {
break; break;
} }
BtMessageHandle message = btMessageReceiver->receiveMessage(); BtMessageHandle message = _btMessageReceiver->receiveMessage();
if(message.isNull()) { if(message.isNull()) {
break; break;
} }
++msgcount; ++msgcount;
if(logger->info()) { if(_logger->info()) {
logger->info(MSG_RECEIVE_PEER_MESSAGE, util::itos(cuid).c_str(), _logger->info(MSG_RECEIVE_PEER_MESSAGE, util::itos(_cuid).c_str(),
peer->ipaddr.c_str(), peer->port, _peer->ipaddr.c_str(), _peer->port,
message->toString().c_str()); message->toString().c_str());
} }
message->doReceivedAction(); message->doReceivedAction();
switch(message->getId()) { switch(message->getId()) {
case BtKeepAliveMessage::ID: case BtKeepAliveMessage::ID:
floodingStat.incKeepAliveCount(); _floodingStat.incKeepAliveCount();
break; break;
case BtChokeMessage::ID: case BtChokeMessage::ID:
if(!peer->peerChoking()) { if(!_peer->peerChoking()) {
floodingStat.incChokeUnchokeCount(); _floodingStat.incChokeUnchokeCount();
} }
break; break;
case BtUnchokeMessage::ID: case BtUnchokeMessage::ID:
if(peer->peerChoking()) { if(_peer->peerChoking()) {
floodingStat.incChokeUnchokeCount(); _floodingStat.incChokeUnchokeCount();
} }
break; break;
case BtPieceMessage::ID: case BtPieceMessage::ID:
_peerStorage->updateTransferStatFor(peer); _peerStorage->updateTransferStatFor(_peer);
// pass through // pass through
case BtRequestMessage::ID: case BtRequestMessage::ID:
inactiveTimer = global::wallclock; _inactiveTimer = global::wallclock;
break; break;
} }
} }
if(countOldOutstandingRequest > 0 && if(countOldOutstandingRequest > 0 &&
dispatcher->countOutstandingRequest() == 0){ _dispatcher->countOutstandingRequest() == 0){
_maxOutstandingRequest = std::min((size_t)UB_MAX_OUTSTANDING_REQUEST, _maxOutstandingRequest = std::min((size_t)UB_MAX_OUTSTANDING_REQUEST,
_maxOutstandingRequest*2); _maxOutstandingRequest*2);
} }
@ -330,42 +330,42 @@ size_t DefaultBtInteractive::receiveMessages() {
} }
void DefaultBtInteractive::decideInterest() { void DefaultBtInteractive::decideInterest() {
if(_pieceStorage->hasMissingPiece(peer)) { if(_pieceStorage->hasMissingPiece(_peer)) {
if(!peer->amInterested()) { if(!_peer->amInterested()) {
if(logger->debug()) { if(_logger->debug()) {
logger->debug(MSG_PEER_INTERESTED, util::itos(cuid).c_str()); _logger->debug(MSG_PEER_INTERESTED, util::itos(_cuid).c_str());
} }
dispatcher-> _dispatcher->
addMessageToQueue(messageFactory->createInterestedMessage()); addMessageToQueue(_messageFactory->createInterestedMessage());
} }
} else { } else {
if(peer->amInterested()) { if(_peer->amInterested()) {
if(logger->debug()) { if(_logger->debug()) {
logger->debug(MSG_PEER_NOT_INTERESTED, util::itos(cuid).c_str()); _logger->debug(MSG_PEER_NOT_INTERESTED, util::itos(_cuid).c_str());
} }
dispatcher-> _dispatcher->
addMessageToQueue(messageFactory->createNotInterestedMessage()); addMessageToQueue(_messageFactory->createNotInterestedMessage());
} }
} }
} }
void DefaultBtInteractive::fillPiece(size_t maxMissingBlock) { void DefaultBtInteractive::fillPiece(size_t maxMissingBlock) {
if(_pieceStorage->hasMissingPiece(peer)) { if(_pieceStorage->hasMissingPiece(_peer)) {
size_t numMissingBlock = btRequestFactory->countMissingBlock(); size_t numMissingBlock = _btRequestFactory->countMissingBlock();
if(peer->peerChoking()) { if(_peer->peerChoking()) {
if(peer->isFastExtensionEnabled()) { if(_peer->isFastExtensionEnabled()) {
std::vector<size_t> excludedIndexes; std::vector<size_t> excludedIndexes;
excludedIndexes.reserve(btRequestFactory->countTargetPiece()); excludedIndexes.reserve(_btRequestFactory->countTargetPiece());
btRequestFactory->getTargetPieceIndexes(excludedIndexes); _btRequestFactory->getTargetPieceIndexes(excludedIndexes);
while(numMissingBlock < maxMissingBlock) { while(numMissingBlock < maxMissingBlock) {
SharedHandle<Piece> piece = SharedHandle<Piece> piece =
_pieceStorage->getMissingFastPiece(peer, excludedIndexes); _pieceStorage->getMissingFastPiece(_peer, excludedIndexes);
if(piece.isNull()) { if(piece.isNull()) {
break; break;
} else { } else {
btRequestFactory->addTargetPiece(piece); _btRequestFactory->addTargetPiece(piece);
numMissingBlock += piece->countMissingBlock(); numMissingBlock += piece->countMissingBlock();
excludedIndexes.push_back(piece->getIndex()); excludedIndexes.push_back(piece->getIndex());
} }
@ -373,15 +373,15 @@ void DefaultBtInteractive::fillPiece(size_t maxMissingBlock) {
} }
} else { } else {
std::vector<size_t> excludedIndexes; std::vector<size_t> excludedIndexes;
excludedIndexes.reserve(btRequestFactory->countTargetPiece()); excludedIndexes.reserve(_btRequestFactory->countTargetPiece());
btRequestFactory->getTargetPieceIndexes(excludedIndexes); _btRequestFactory->getTargetPieceIndexes(excludedIndexes);
while(numMissingBlock < maxMissingBlock) { while(numMissingBlock < maxMissingBlock) {
SharedHandle<Piece> piece = SharedHandle<Piece> piece =
_pieceStorage->getMissingPiece(peer, excludedIndexes); _pieceStorage->getMissingPiece(_peer, excludedIndexes);
if(piece.isNull()) { if(piece.isNull()) {
break; break;
} else { } else {
btRequestFactory->addTargetPiece(piece); _btRequestFactory->addTargetPiece(piece);
numMissingBlock += piece->countMissingBlock(); numMissingBlock += piece->countMissingBlock();
excludedIndexes.push_back(piece->getIndex()); excludedIndexes.push_back(piece->getIndex());
} }
@ -393,29 +393,29 @@ void DefaultBtInteractive::fillPiece(size_t maxMissingBlock) {
void DefaultBtInteractive::addRequests() { void DefaultBtInteractive::addRequests() {
fillPiece(_maxOutstandingRequest); fillPiece(_maxOutstandingRequest);
size_t reqNumToCreate = size_t reqNumToCreate =
_maxOutstandingRequest <= dispatcher->countOutstandingRequest() ? _maxOutstandingRequest <= _dispatcher->countOutstandingRequest() ?
0 : _maxOutstandingRequest-dispatcher->countOutstandingRequest(); 0 : _maxOutstandingRequest-_dispatcher->countOutstandingRequest();
if(reqNumToCreate > 0) { if(reqNumToCreate > 0) {
std::vector<SharedHandle<BtMessage> > requests; std::vector<SharedHandle<BtMessage> > requests;
requests.reserve(reqNumToCreate); requests.reserve(reqNumToCreate);
if(_pieceStorage->isEndGame()) { if(_pieceStorage->isEndGame()) {
btRequestFactory->createRequestMessagesOnEndGame(requests,reqNumToCreate); _btRequestFactory->createRequestMessagesOnEndGame(requests,reqNumToCreate);
} else { } else {
btRequestFactory->createRequestMessages(requests, reqNumToCreate); _btRequestFactory->createRequestMessages(requests, reqNumToCreate);
} }
dispatcher->addMessageToQueue(requests); _dispatcher->addMessageToQueue(requests);
} }
} }
void DefaultBtInteractive::cancelAllPiece() { void DefaultBtInteractive::cancelAllPiece() {
btRequestFactory->removeAllTargetPiece(); _btRequestFactory->removeAllTargetPiece();
if(_metadataGetMode && _downloadContext->getTotalLength() > 0) { if(_metadataGetMode && _downloadContext->getTotalLength() > 0) {
std::vector<size_t> metadataRequests = std::vector<size_t> metadataRequests =
_utMetadataRequestTracker->getAllTrackedIndex(); _utMetadataRequestTracker->getAllTrackedIndex();
for(std::vector<size_t>::const_iterator i = metadataRequests.begin(), for(std::vector<size_t>::const_iterator i = metadataRequests.begin(),
eoi = metadataRequests.end(); i != eoi; ++i) { eoi = metadataRequests.end(); i != eoi; ++i) {
if(logger->debug()) { if(_logger->debug()) {
logger->debug("Cancel metadata: piece=%lu", _logger->debug("Cancel metadata: piece=%lu",
static_cast<unsigned long>(*i)); static_cast<unsigned long>(*i));
} }
_pieceStorage->cancelPiece(_pieceStorage->getPiece(*i)); _pieceStorage->cancelPiece(_pieceStorage->getPiece(*i));
@ -424,30 +424,30 @@ void DefaultBtInteractive::cancelAllPiece() {
} }
void DefaultBtInteractive::sendPendingMessage() { void DefaultBtInteractive::sendPendingMessage() {
dispatcher->sendMessages(); _dispatcher->sendMessages();
} }
void DefaultBtInteractive::detectMessageFlooding() { void DefaultBtInteractive::detectMessageFlooding() {
if(floodingTimer. if(_floodingTimer.
difference(global::wallclock) >= FLOODING_CHECK_INTERVAL) { difference(global::wallclock) >= FLOODING_CHECK_INTERVAL) {
if(floodingStat.getChokeUnchokeCount() >= 2 || if(_floodingStat.getChokeUnchokeCount() >= 2 ||
floodingStat.getKeepAliveCount() >= 2) { _floodingStat.getKeepAliveCount() >= 2) {
throw DL_ABORT_EX(EX_FLOODING_DETECTED); throw DL_ABORT_EX(EX_FLOODING_DETECTED);
} else { } else {
floodingStat.reset(); _floodingStat.reset();
} }
floodingTimer = global::wallclock; _floodingTimer = global::wallclock;
} }
} }
void DefaultBtInteractive::checkActiveInteraction() void DefaultBtInteractive::checkActiveInteraction()
{ {
time_t inactiveTime = inactiveTimer.difference(global::wallclock); time_t inactiveTime = _inactiveTimer.difference(global::wallclock);
// To allow aria2 to accept mutially interested peer, disconnect unintersted // To allow aria2 to accept mutially interested peer, disconnect unintersted
// peer. // peer.
{ {
const time_t interval = 30; const time_t interval = 30;
if(!peer->amInterested() && !peer->peerInterested() && if(!_peer->amInterested() && !_peer->peerInterested() &&
inactiveTime >= interval) { inactiveTime >= interval) {
// TODO change the message // TODO change the message
throw DL_ABORT_EX throw DL_ABORT_EX
@ -472,13 +472,13 @@ void DefaultBtInteractive::addPeerExchangeMessage()
if(_pexTimer. if(_pexTimer.
difference(global::wallclock) >= UTPexExtensionMessage::DEFAULT_INTERVAL) { difference(global::wallclock) >= UTPexExtensionMessage::DEFAULT_INTERVAL) {
UTPexExtensionMessageHandle m UTPexExtensionMessageHandle m
(new UTPexExtensionMessage(peer->getExtensionMessageID("ut_pex"))); (new UTPexExtensionMessage(_peer->getExtensionMessageID("ut_pex")));
const std::deque<SharedHandle<Peer> >& peers = _peerStorage->getPeers(); const std::deque<SharedHandle<Peer> >& peers = _peerStorage->getPeers();
{ {
for(std::deque<SharedHandle<Peer> >::const_iterator i = for(std::deque<SharedHandle<Peer> >::const_iterator i =
peers.begin(), eoi = peers.end(); peers.begin(), eoi = peers.end();
i != eoi && !m->freshPeersAreFull(); ++i) { i != eoi && !m->freshPeersAreFull(); ++i) {
if(peer->ipaddr != (*i)->ipaddr) { if(_peer->ipaddr != (*i)->ipaddr) {
m->addFreshPeer(*i); m->addFreshPeer(*i);
} }
} }
@ -488,13 +488,13 @@ void DefaultBtInteractive::addPeerExchangeMessage()
peers.rbegin(), eoi = peers.rend(); peers.rbegin(), eoi = peers.rend();
i != eoi && !m->droppedPeersAreFull(); i != eoi && !m->droppedPeersAreFull();
++i) { ++i) {
if(peer->ipaddr != (*i)->ipaddr) { if(_peer->ipaddr != (*i)->ipaddr) {
m->addDroppedPeer(*i); m->addDroppedPeer(*i);
} }
} }
} }
BtMessageHandle msg = messageFactory->createBtExtendedMessage(m); BtMessageHandle msg = _messageFactory->createBtExtendedMessage(m);
dispatcher->addMessageToQueue(msg); _dispatcher->addMessageToQueue(msg);
_pexTimer = global::wallclock; _pexTimer = global::wallclock;
} }
} }
@ -507,14 +507,13 @@ void DefaultBtInteractive::doInteractionProcessing() {
// HandshakeExtensionMessage::doReceivedAction(). // HandshakeExtensionMessage::doReceivedAction().
_pieceStorage = _pieceStorage =
_downloadContext->getOwnerRequestGroup()->getPieceStorage(); _downloadContext->getOwnerRequestGroup()->getPieceStorage();
if(_peer->getExtensionMessageID("ut_metadata") &&
if(peer->getExtensionMessageID("ut_metadata") &&
_downloadContext->getTotalLength() > 0) { _downloadContext->getTotalLength() > 0) {
size_t num = _utMetadataRequestTracker->avail(); size_t num = _utMetadataRequestTracker->avail();
if(num > 0) { if(num > 0) {
std::vector<SharedHandle<BtMessage> > requests; std::vector<SharedHandle<BtMessage> > requests;
_utMetadataRequestFactory->create(requests, num, _pieceStorage); _utMetadataRequestFactory->create(requests, num, _pieceStorage);
dispatcher->addMessageToQueue(requests); _dispatcher->addMessageToQueue(requests);
} }
if(_perSecTimer.difference(global::wallclock) >= 1) { if(_perSecTimer.difference(global::wallclock) >= 1) {
_perSecTimer = global::wallclock; _perSecTimer = global::wallclock;
@ -534,29 +533,22 @@ void DefaultBtInteractive::doInteractionProcessing() {
} }
} else { } else {
checkActiveInteraction(); checkActiveInteraction();
decideChoking(); decideChoking();
detectMessageFlooding(); detectMessageFlooding();
if(_perSecTimer.difference(global::wallclock) >= 1) { if(_perSecTimer.difference(global::wallclock) >= 1) {
_perSecTimer = global::wallclock; _perSecTimer = global::wallclock;
dispatcher->checkRequestSlotAndDoNecessaryThing(); _dispatcher->checkRequestSlotAndDoNecessaryThing();
} }
checkHave(); checkHave();
sendKeepAlive(); sendKeepAlive();
_numReceivedMessage = receiveMessages(); _numReceivedMessage = receiveMessages();
_btRequestFactory->removeCompletedPiece();
btRequestFactory->removeCompletedPiece();
decideInterest(); decideInterest();
if(!_pieceStorage->downloadFinished()) { if(!_pieceStorage->downloadFinished()) {
addRequests(); addRequests();
} }
} }
if(peer->getExtensionMessageID("ut_pex") && _utPexEnabled) { if(_peer->getExtensionMessageID("ut_pex") && _utPexEnabled) {
addPeerExchangeMessage(); addPeerExchangeMessage();
} }
@ -570,12 +562,12 @@ void DefaultBtInteractive::setLocalNode(const WeakHandle<DHTNode>& node)
size_t DefaultBtInteractive::countPendingMessage() size_t DefaultBtInteractive::countPendingMessage()
{ {
return dispatcher->countMessageInQueue(); return _dispatcher->countMessageInQueue();
} }
bool DefaultBtInteractive::isSendingMessageInProgress() bool DefaultBtInteractive::isSendingMessageInProgress()
{ {
return dispatcher->isSendingInProgress(); return _dispatcher->isSendingInProgress();
} }
size_t DefaultBtInteractive::countReceivedMessageInIteration() const size_t DefaultBtInteractive::countReceivedMessageInIteration() const
@ -588,7 +580,7 @@ size_t DefaultBtInteractive::countOutstandingRequest()
if(_metadataGetMode) { if(_metadataGetMode) {
return _utMetadataRequestTracker->count(); return _utMetadataRequestTracker->count();
} else { } else {
return dispatcher->countOutstandingRequest(); return _dispatcher->countOutstandingRequest();
} }
} }
@ -612,31 +604,25 @@ void DefaultBtInteractive::setPeerStorage
void DefaultBtInteractive::setPeer(const SharedHandle<Peer>& peer) void DefaultBtInteractive::setPeer(const SharedHandle<Peer>& peer)
{ {
this->peer = peer; _peer = peer;
} }
void DefaultBtInteractive::setBtMessageReceiver void DefaultBtInteractive::setBtMessageReceiver
(const SharedHandle<BtMessageReceiver>& receiver) (const SharedHandle<BtMessageReceiver>& receiver)
{ {
this->btMessageReceiver = receiver; _btMessageReceiver = receiver;
} }
void DefaultBtInteractive::setDispatcher void DefaultBtInteractive::setDispatcher
(const SharedHandle<BtMessageDispatcher>& dispatcher) (const SharedHandle<BtMessageDispatcher>& dispatcher)
{ {
this->dispatcher = dispatcher; _dispatcher = dispatcher;
} }
void DefaultBtInteractive::setBtRequestFactory void DefaultBtInteractive::setBtRequestFactory
(const SharedHandle<BtRequestFactory>& factory) (const SharedHandle<BtRequestFactory>& factory)
{ {
this->btRequestFactory = factory; _btRequestFactory = factory;
}
void DefaultBtInteractive::setPeerConnection
(const SharedHandle<PeerConnection>& peerConnection)
{
this->peerConnection = peerConnection;
} }
void DefaultBtInteractive::setExtensionMessageFactory void DefaultBtInteractive::setExtensionMessageFactory
@ -648,7 +634,7 @@ void DefaultBtInteractive::setExtensionMessageFactory
void DefaultBtInteractive::setBtMessageFactory void DefaultBtInteractive::setBtMessageFactory
(const SharedHandle<BtMessageFactory>& factory) (const SharedHandle<BtMessageFactory>& factory)
{ {
this->messageFactory = factory; _messageFactory = factory;
} }
void DefaultBtInteractive::setRequestGroupMan void DefaultBtInteractive::setRequestGroupMan

36
src/DefaultBtInteractive.h
View File

@ -54,7 +54,6 @@ class BtMessageReceiver;
class BtMessageDispatcher; class BtMessageDispatcher;
class BtMessageFactory; class BtMessageFactory;
class BtRequestFactory; class BtRequestFactory;
class PeerConnection;
class ExtensionMessageFactory; class ExtensionMessageFactory;
class ExtensionMessageRegistry; class ExtensionMessageRegistry;
class DHTNode; class DHTNode;
@ -98,7 +97,7 @@ public:
class DefaultBtInteractive : public BtInteractive { class DefaultBtInteractive : public BtInteractive {
private: private:
cuid_t cuid; cuid_t _cuid;
SharedHandle<DownloadContext> _downloadContext; SharedHandle<DownloadContext> _downloadContext;
@ -108,13 +107,12 @@ private:
SharedHandle<PeerStorage> _peerStorage; SharedHandle<PeerStorage> _peerStorage;
SharedHandle<Peer> peer; SharedHandle<Peer> _peer;
SharedHandle<BtMessageReceiver> btMessageReceiver; SharedHandle<BtMessageReceiver> _btMessageReceiver;
SharedHandle<BtMessageDispatcher> dispatcher; SharedHandle<BtMessageDispatcher> _dispatcher;
SharedHandle<BtRequestFactory> btRequestFactory; SharedHandle<BtRequestFactory> _btRequestFactory;
SharedHandle<PeerConnection> peerConnection; SharedHandle<BtMessageFactory> _messageFactory;
SharedHandle<BtMessageFactory> messageFactory;
SharedHandle<ExtensionMessageFactory> _extensionMessageFactory; SharedHandle<ExtensionMessageFactory> _extensionMessageFactory;
SharedHandle<ExtensionMessageRegistry> _extensionMessageRegistry; SharedHandle<ExtensionMessageRegistry> _extensionMessageRegistry;
SharedHandle<UTMetadataRequestFactory> _utMetadataRequestFactory; SharedHandle<UTMetadataRequestFactory> _utMetadataRequestFactory;
@ -124,16 +122,16 @@ private:
WeakHandle<DHTNode> _localNode; WeakHandle<DHTNode> _localNode;
Logger* logger; Logger* _logger;
size_t allowedFastSetSize; size_t _allowedFastSetSize;
Timer haveTimer; Timer _haveTimer;
Timer keepAliveTimer; Timer _keepAliveTimer;
Timer floodingTimer; Timer _floodingTimer;
FloodingStat floodingStat; FloodingStat _floodingStat;
Timer inactiveTimer; Timer _inactiveTimer;
Timer _pexTimer; Timer _pexTimer;
Timer _perSecTimer; Timer _perSecTimer;
time_t keepAliveInterval; time_t _keepAliveInterval;
bool _utPexEnabled; bool _utPexEnabled;
bool _dhtEnabled; bool _dhtEnabled;
@ -191,7 +189,7 @@ public:
void setCuid(cuid_t cuid) void setCuid(cuid_t cuid)
{ {
this->cuid = cuid; _cuid = cuid;
} }
void setBtRuntime(const SharedHandle<BtRuntime>& btRuntime); void setBtRuntime(const SharedHandle<BtRuntime>& btRuntime);
@ -208,8 +206,6 @@ public:
void setBtRequestFactory(const SharedHandle<BtRequestFactory>& factory); void setBtRequestFactory(const SharedHandle<BtRequestFactory>& factory);
void setPeerConnection(const SharedHandle<PeerConnection>& peerConnection);
void setBtMessageFactory(const SharedHandle<BtMessageFactory>& factory); void setBtMessageFactory(const SharedHandle<BtMessageFactory>& factory);
void setExtensionMessageFactory void setExtensionMessageFactory
@ -222,7 +218,7 @@ public:
} }
void setKeepAliveInterval(time_t keepAliveInterval) { void setKeepAliveInterval(time_t keepAliveInterval) {
this->keepAliveInterval = keepAliveInterval; _keepAliveInterval = keepAliveInterval;
} }
void setUTPexEnabled(bool f) void setUTPexEnabled(bool f)

1
src/PeerInteractionCommand.cc
View File

@ -185,7 +185,6 @@ PeerInteractionCommand::PeerInteractionCommand
btInteractive->setBtMessageReceiver(receiver); btInteractive->setBtMessageReceiver(receiver);
btInteractive->setDispatcher(dispatcher); btInteractive->setDispatcher(dispatcher);
btInteractive->setBtRequestFactory(reqFactory); btInteractive->setBtRequestFactory(reqFactory);
btInteractive->setPeerConnection(peerConnection);
btInteractive->setExtensionMessageFactory(extensionMessageFactory); btInteractive->setExtensionMessageFactory(extensionMessageFactory);
btInteractive->setExtensionMessageRegistry(exMsgRegistry); btInteractive->setExtensionMessageRegistry(exMsgRegistry);
btInteractive->setKeepAliveInterval btInteractive->setKeepAliveInterval