2007-04-06 Tatsuhiro Tsujikawa <tujikawa at rednoah dot com>

* src/PeerAbstractCommand.cc (onAbort): Call PeerStorage::returnPeer() * src/DefaultPeerStorage.h, src/DefaultPeerStorage.cc (incomingPeers): New variable. (addIncomingPeer): New function. (returnPeer): New function. (onErasingPeer): New function. (addPeer): push_back -> push_front (getActivePeers): Rewritten. (calculateStat): Rewritten. * src/PeerStorage.h (TransferStat::copy): New function. (TransferStat::TransferStat): New function. (TransferStat::operator=): New function. (addIncomingPeer): New function. (returnPeer): New function. * src/PeerListenCommand.cc (execute): Use PeerStorage::addIncomingPeer() instead of Peer::addPeer().
2007-04-05 16:47:59 +00:00 · 2007-04-05 16:47:59 +00:00 · dd37b7be0a
parent f030408d8e
commit dd37b7be0a
10 changed files with 217 additions and 38 deletions
--- a/22
+++ b/22
@ -1,3 +1,25 @@
+2007-04-06  Tatsuhiro Tsujikawa  <tujikawa at rednoah dot com>
+
+	* src/PeerAbstractCommand.cc
+	(onAbort): Call PeerStorage::returnPeer()
+	* src/DefaultPeerStorage.h, src/DefaultPeerStorage.cc
+	(incomingPeers): New variable.
+	(addIncomingPeer): New function.
+	(returnPeer): New function.
+	(onErasingPeer): New function.
+	(addPeer): push_back -> push_front
+	(getActivePeers): Rewritten.
+	(calculateStat): Rewritten.
+	* src/PeerStorage.h
+	(TransferStat::copy): New function.
+	(TransferStat::TransferStat): New function.
+	(TransferStat::operator=): New function.
+	(addIncomingPeer): New function.
+	(returnPeer): New function.
+	* src/PeerListenCommand.cc
+	(execute): Use PeerStorage::addIncomingPeer() instead of
+	Peer::addPeer().
+
 2007-04-03  Tatsuhiro Tsujikawa  <tujikawa at rednoah dot com>

 	Connect to a peer actively when download speed is lower than specified
--- a/src/ActivePeerConnectionCommand.cc
+++ b/src/ActivePeerConnectionCommand.cc
@ -39,7 +39,7 @@ bool ActivePeerConnectionCommand::execute() {
  if(btRuntime->isHalt()) {
    return true;
  }
-  if(checkPoint.elapsed(interval)) {
+  if(!pieceStorage->downloadFinished() && checkPoint.elapsed(interval)) {
    checkPoint.reset();

    TransferStat tstat = peerStorage->calculateStat();
--- a/src/DefaultPeerStorage.cc
+++ b/src/DefaultPeerStorage.cc
@ -56,7 +56,7 @@ bool DefaultPeerStorage::addPeer(const PeerHandle& peer) {
    if(peers.size() >= (size_t)maxPeerListSize) {
      deleteUnusedPeer(peers.size()-maxPeerListSize+1);
    }
-    peers.push_back(peer);
+    peers.push_front(peer);
    return true;
  } else {
    const PeerHandle& peer = *itr;
@ -69,6 +69,12 @@ bool DefaultPeerStorage::addPeer(const PeerHandle& peer) {
  }
 }

+bool DefaultPeerStorage::addIncomingPeer(const PeerHandle& peer)
+{
+  incomingPeers.push_back(peer);
+  return true;
+}
+
 void DefaultPeerStorage::addPeer(const Peers& peers) {
  for(Peers::const_iterator itr = peers.begin();
      itr != peers.end(); itr++) {
@ -124,7 +130,7 @@ PeerHandle DefaultPeerStorage::getPeer(const string& ipaddr,
  }
 }

-int DefaultPeerStorage::countPeer() const {
+int32_t DefaultPeerStorage::countPeer() const {
  return peers.size();
 }

@ -132,45 +138,94 @@ bool DefaultPeerStorage::isPeerAvailable() {
  return !getUnusedPeer().isNull();
 }

-Peers DefaultPeerStorage::getActivePeers() {
-  Peers activePeers;
-  for(Peers::iterator itr = peers.begin(); itr != peers.end(); itr++) {
-    PeerHandle& peer = *itr;
+class CollectActivePeer {
+private:
+  Peers _activePeers;
+public:
+  void operator()(const PeerHandle& peer)
+  {
    if(peer->isActive()) {
-      activePeers.push_back(peer);
+      _activePeers.push_back(peer);
    }
  }
-  return activePeers;
+
+  const Peers& getActivePeers() { return _activePeers; }
+};
+
+Peers DefaultPeerStorage::getActivePeers() {
+  CollectActivePeer funcObj;
+  funcObj = for_each(peers.begin(), peers.end(), funcObj);
+  funcObj = for_each(incomingPeers.begin(), incomingPeers.end(), funcObj);
+  return funcObj.getActivePeers();
 }

-TransferStat DefaultPeerStorage::calculateStat() {
-  TransferStat stat;
-  for(Peers::iterator itr = peers.begin(); itr != peers.end(); itr++) {
-    PeerHandle& peer = *itr;
+class CalculateStat {
+private:
+  TransferStat _stat;
+public:
+  void operator()(const PeerHandle& peer)
+  {
    if(peer->isActive()) {
-      stat.downloadSpeed += peer->calculateDownloadSpeed();
-      stat.uploadSpeed += peer->calculateUploadSpeed();
+      _stat.downloadSpeed += peer->calculateDownloadSpeed();
+      _stat.uploadSpeed += peer->calculateUploadSpeed();
    }
-    stat.sessionDownloadLength += peer->getSessionDownloadLength();
-    stat.sessionUploadLength += peer->getSessionUploadLength();
+    _stat.sessionDownloadLength += peer->getSessionDownloadLength();
+    _stat.sessionUploadLength += peer->getSessionUploadLength();    
  }
+
+  const TransferStat& getTransferStat() { return _stat; }
+};
+
+TransferStat DefaultPeerStorage::calculateStat() {
+  CalculateStat calStat;
+  calStat = for_each(peers.begin(), peers.end(), calStat);
+  calStat = for_each(incomingPeers.begin(), incomingPeers.end(), calStat);
+
+  TransferStat stat = calStat.getTransferStat();
  stat.sessionDownloadLength += removedPeerSessionDownloadLength;
  stat.sessionUploadLength += removedPeerSessionUploadLength;
  return stat;
 }

 void DefaultPeerStorage::deleteUnusedPeer(int delSize) {
-  for(Peers::iterator itr = peers.begin();
-      itr != peers.end() && delSize > 0;) {
+  Peers temp;
+  for(Peers::reverse_iterator itr = peers.rbegin();
+      itr != peers.rend(); ++itr) {
    const PeerHandle& p = *itr;
-    if(p->cuid == 0) {
+    if(p->cuid == 0 && delSize > 0) {
      // Update removedPeerSession******Length
-      removedPeerSessionDownloadLength += p->getSessionDownloadLength();
-      removedPeerSessionUploadLength += p->getSessionUploadLength();
-      itr = peers.erase(itr);
+      onErasingPeer(p);
      delSize--;
    } else {
-      itr++;
+      temp.push_front(p);
    }
  }
+  peers = temp;
+}
+
+void DefaultPeerStorage::onErasingPeer(const PeerHandle& peer)
+{
+  removedPeerSessionDownloadLength += peer->getSessionDownloadLength();
+  removedPeerSessionUploadLength += peer->getSessionUploadLength();
+}
+
+void DefaultPeerStorage::returnPeer(const PeerHandle& peer)
+{
+  Peers::iterator itr = find(peers.begin(), peers.end(), peer);
+  if(itr == peers.end()) {
+    itr = find(incomingPeers.begin(), incomingPeers.end(), peer);
+    if(itr == peers.end()) {
+      // do nothing
+    } else {
+      // erase incoming peer because we cannot connect to it with port number
+      // (*itr)->port. It is not the listening port.
+      onErasingPeer(*itr);
+      incomingPeers.erase(itr);
+    }
+  } else {
+    peer->startBadCondition();
+    peer->resetStatus();
+    peers.erase(itr);
+    peers.push_back(peer);
+  }
 }
--- a/src/DefaultPeerStorage.h
+++ b/src/DefaultPeerStorage.h
@ -49,6 +49,7 @@ private:
  BtContextHandle btContext;
  const Option* option;
  Peers peers;
+  Peers incomingPeers;
  int maxPeerListSize;
  Logger* logger;
  BtRuntimeHandle btRuntime;
@ -65,7 +66,9 @@ public:

  virtual bool addPeer(const PeerHandle& peer);

-  int countPeer() const;
+  virtual bool addIncomingPeer(const PeerHandle& peer);
+
+  int32_t countPeer() const;

  virtual PeerHandle getUnusedPeer();

@ -81,11 +84,17 @@ public:

  virtual TransferStat calculateStat();

+  virtual void returnPeer(const PeerHandle& peer);
+
  void setMaxPeerListSize(int size) { this->maxPeerListSize = size; }
 
  int getMaxPeerListSize() const { return maxPeerListSize; }

  void deleteUnusedPeer(int delSize);
+  
+  void onErasingPeer(const PeerHandle& peer);
+
+  const Peers& getIncomingPeers() const { return incomingPeers; }
 };

 #endif // _D_DEFAULT_PEER_STORAGE_H_
--- a/src/PeerAbstractCommand.cc
+++ b/src/PeerAbstractCommand.cc
@ -97,8 +97,7 @@ bool PeerAbstractCommand::prepareForRetry(int wait) {
 }

 void PeerAbstractCommand::onAbort(RecoverableException* ex) {
-  peer->startBadCondition();
-  peer->resetStatus();
+  peerStorage->returnPeer(peer);
 }

 void PeerAbstractCommand::disableReadCheckSocket() {
--- a/src/PeerListenCommand.cc
+++ b/src/PeerListenCommand.cc
@ -80,7 +80,7 @@ bool PeerListenCommand::execute() {
 	PeerHandle peer = PeerHandle(new Peer(peerInfo.first, peerInfo.second,
 					      btContext->getPieceLength(),
 					      btContext->getTotalLength()));
-	if(peerStorage->addPeer(peer)) {
+	if(peerStorage->addIncomingPeer(peer)) {
 	  int newCuid =  btRuntime->getNewCuid();
 	  peer->cuid = newCuid;
 	  PeerInteractionCommand* command =
--- a/src/PeerStorage.h
+++ b/src/PeerStorage.h
@ -44,10 +44,31 @@ public:
  int32_t uploadSpeed;
  int64_t sessionDownloadLength;
  int64_t sessionUploadLength;
+
+  void copy(const TransferStat& stat)
+  {
+    downloadSpeed = stat.downloadSpeed;
+    uploadSpeed = stat.uploadSpeed;
+    sessionDownloadLength = stat.sessionDownloadLength;
+    sessionUploadLength = stat.sessionUploadLength;
+  }
 public:
  TransferStat():downloadSpeed(0), uploadSpeed(0),
 		 sessionDownloadLength(0), sessionUploadLength(0) {}

+  TransferStat(const TransferStat& stat)
+  {
+    copy(stat);
+  }
+
+  TransferStat& operator=(const TransferStat& stat)
+  {
+    if(this != &stat) {
+      copy(stat);
+    }
+    return *this;
+  }
+
  int32_t getDownloadSpeed() const {
    return downloadSpeed;
  }
@ -86,11 +107,17 @@ public:
  virtual ~PeerStorage() {}

  /**
-   * Adds new peer to internal peer list.
+   * Adds new peer to the internal peer list.
   * If the peer is added successfully, returns true. Otherwise returns false.
   */
  virtual bool addPeer(const PeerHandle& peer) = 0;

+  /**
+   * Adds new incoming peer to the internal peer list.
+   * If the peer is added successfully, returns true. Otherwise returns false.
+   */
+  virtual bool addIncomingPeer(const PeerHandle& peer) = 0;
+
  /**
   * Adds all peers in peers to internal peer list.
   */
@ -121,6 +148,11 @@ public:
   * Calculates current download/upload statistics.
   */
  virtual TransferStat calculateStat() = 0;
+
+  /**
+   * Tells PeerStorage object that peer is no longer used in the session.
+   */
+  virtual void returnPeer(const PeerHandle& peer) = 0;
 };

 typedef SharedHandle<PeerStorage> PeerStorageHandle;
--- a/test/DefaultPeerStorageTest.cc
+++ b/test/DefaultPeerStorageTest.cc
@ -16,6 +16,9 @@ class DefaultPeerStorageTest:public CppUnit::TestFixture {
  CPPUNIT_TEST(testIsPeerAvailable);
  CPPUNIT_TEST(testActivatePeer);
  CPPUNIT_TEST(testCalculateStat);
+  CPPUNIT_TEST(testAddIncomingPeer);
+  CPPUNIT_TEST(testReturnPeer);
+  CPPUNIT_TEST(testOnErasingPeer);
  CPPUNIT_TEST_SUITE_END();
 private:
  BtContextHandle btContext;
@ -38,6 +41,9 @@ public:
  void testIsPeerAvailable();
  void testActivatePeer();
  void testCalculateStat();
+  void testAddIncomingPeer();
+  void testReturnPeer();
+  void testOnErasingPeer();
 };


@ -118,13 +124,25 @@ void DefaultPeerStorageTest::testAddPeer() {
 			    btContext->getTotalLength()));

  CPPUNIT_ASSERT(ps.addPeer(peer4));
-  // peer3 was deleted.
+  // peer1 was deleted.
  CPPUNIT_ASSERT_EQUAL(3, ps.countPeer());
  
  peer4->startBadCondition();

+  PeerHandle peer5(new Peer("192.168.0.4", 6889, btContext->getPieceLength(),
+			    btContext->getTotalLength()));
+
  // this returns false, because peer4 in the container has error.
-  CPPUNIT_ASSERT_EQUAL(false, ps.addPeer(peer4));
+  CPPUNIT_ASSERT_EQUAL(false, ps.addPeer(peer5));
+
+  peer3->cuid = 1;
+
+  PeerHandle peer6(new Peer("192.168.0.3", 6889, btContext->getPieceLength(),
+			    btContext->getTotalLength()));
+
+  // this is false, because peer3's cuid is not zero
+  CPPUNIT_ASSERT_EQUAL(false, ps.addPeer(peer6));
+  
 }

 void DefaultPeerStorageTest::testGetPeer() {
@ -194,3 +212,47 @@ void DefaultPeerStorageTest::testActivatePeer() {

 void DefaultPeerStorageTest::testCalculateStat() {
 }
+
+
+void DefaultPeerStorageTest::testAddIncomingPeer()
+{
+  DefaultPeerStorage ps(btContext, option);
+ 
+  CPPUNIT_ASSERT(ps.addIncomingPeer(new Peer("192.168.0.1", 6889,
+					     btContext->getPieceLength(),
+					     btContext->getTotalLength())));
+
+  CPPUNIT_ASSERT_EQUAL((size_t)1, ps.getIncomingPeers().size());
+}
+
+void DefaultPeerStorageTest::testReturnPeer()
+{
+  DefaultPeerStorage ps(btContext, option);
+
+  PeerHandle peer1(new Peer("192.168.0.1", 6889, btContext->getPieceLength(),
+			   btContext->getTotalLength()));
+  PeerHandle peer2(new Peer("192.168.0.2", 6889, btContext->getPieceLength(),
+			   btContext->getTotalLength()));
+  ps.addPeer(peer1);
+  ps.addPeer(peer2);
+
+  PeerHandle peer3(new Peer("192.168.0.3", 6889, btContext->getPieceLength(),
+			   btContext->getTotalLength()));
+  
+  ps.addIncomingPeer(peer3);
+
+  CPPUNIT_ASSERT_EQUAL(string("192.168.0.2"),
+		       ps.getPeers().front()->ipaddr);
+  ps.returnPeer(peer2);
+  CPPUNIT_ASSERT_EQUAL(string("192.168.0.1"),
+		       ps.getPeers().front()->ipaddr);
+
+  ps.returnPeer(peer3);
+  CPPUNIT_ASSERT_EQUAL((size_t)0, ps.getIncomingPeers().size());
+
+}
+
+void DefaultPeerStorageTest::testOnErasingPeer()
+{
+  // test this
+}
--- a/test/Makefile.am
+++ b/test/Makefile.am
@ -1,6 +1,8 @@
 TESTS = aria2c
 check_PROGRAMS = $(TESTS)
 aria2c_SOURCES = AllTest.cc\
+	PeerTest.cc\
+	DefaultPeerStorageTest.cc\
 	RequestFactoryTest.cc\
 	NetrcAuthResolverTest.cc\
 	DefaultAuthResolverTest.cc\
@ -41,13 +43,11 @@ aria2c_SOURCES = AllTest.cc\
 	TrackerWatcherCommandTest.cc\
 	DefaultBtContextTest.cc\
 	DefaultPieceStorageTest.cc\
-	DefaultPeerStorageTest.cc\
 	DefaultBtAnnounceTest.cc\
 	BtRegistryTest.cc\
 	DefaultBtMessageDispatcherTest.cc\
 	MockPeerStorage.h\
 	DefaultBtRequestFactoryTest.cc\
-	PeerTest.cc\
 	BtAllowedFastMessageTest.cc\
 	BtBitfieldMessageTest.cc\
 	BtCancelMessageTest.cc\
--- a/test/Makefile.in
+++ b/test/Makefile.in
@ -57,7 +57,8 @@ mkinstalldirs = $(SHELL) $(top_srcdir)/mkinstalldirs
 CONFIG_HEADER = $(top_builddir)/config.h
 CONFIG_CLEAN_FILES =
 am__EXEEXT_1 = aria2c$(EXEEXT)
-am_aria2c_OBJECTS = AllTest.$(OBJEXT) RequestFactoryTest.$(OBJEXT) \
+am_aria2c_OBJECTS = AllTest.$(OBJEXT) PeerTest.$(OBJEXT) \
+	DefaultPeerStorageTest.$(OBJEXT) RequestFactoryTest.$(OBJEXT) \
 	NetrcAuthResolverTest.$(OBJEXT) \
 	DefaultAuthResolverTest.$(OBJEXT) RequestTest.$(OBJEXT) \
 	HttpRequestTest.$(OBJEXT) UtilTest.$(OBJEXT) \
@ -80,10 +81,9 @@ am_aria2c_OBJECTS = AllTest.$(OBJEXT) RequestFactoryTest.$(OBJEXT) \
 	AnnounceListTest.$(OBJEXT) TrackerWatcherCommandTest.$(OBJEXT) \
 	DefaultBtContextTest.$(OBJEXT) \
 	DefaultPieceStorageTest.$(OBJEXT) \
-	DefaultPeerStorageTest.$(OBJEXT) \
 	DefaultBtAnnounceTest.$(OBJEXT) BtRegistryTest.$(OBJEXT) \
 	DefaultBtMessageDispatcherTest.$(OBJEXT) \
-	DefaultBtRequestFactoryTest.$(OBJEXT) PeerTest.$(OBJEXT) \
+	DefaultBtRequestFactoryTest.$(OBJEXT) \
 	BtAllowedFastMessageTest.$(OBJEXT) \
 	BtBitfieldMessageTest.$(OBJEXT) BtCancelMessageTest.$(OBJEXT) \
 	BtChokeMessageTest.$(OBJEXT) BtHaveAllMessageTest.$(OBJEXT) \
@ -262,6 +262,8 @@ sysconfdir = @sysconfdir@
 target_alias = @target_alias@
 TESTS = aria2c
 aria2c_SOURCES = AllTest.cc\
+	PeerTest.cc\
+	DefaultPeerStorageTest.cc\
 	RequestFactoryTest.cc\
 	NetrcAuthResolverTest.cc\
 	DefaultAuthResolverTest.cc\
@ -302,13 +304,11 @@ aria2c_SOURCES = AllTest.cc\
 	TrackerWatcherCommandTest.cc\
 	DefaultBtContextTest.cc\
 	DefaultPieceStorageTest.cc\
-	DefaultPeerStorageTest.cc\
 	DefaultBtAnnounceTest.cc\
 	BtRegistryTest.cc\
 	DefaultBtMessageDispatcherTest.cc\
 	MockPeerStorage.h\
 	DefaultBtRequestFactoryTest.cc\
-	PeerTest.cc\
 	BtAllowedFastMessageTest.cc\
 	BtBitfieldMessageTest.cc\
 	BtCancelMessageTest.cc\