Ignore peer which possibly sent bad data in next few minuts.

In aria2, one piece is downloaded from several peers, so when hash
check failed for that piece, it cannot determine which peer(s) sent
bad data. So, we ignore peer who sent last block of data in random
minutes. We use randomized timeout because all peers get waken up at
the almost same time.

8 files changed, 94 insertions, 0 deletions

diff --git a/src/BtPieceMessage.cc b/src/BtPieceMessage.cc
index 2e0f0b3a..25b628b7 100644
--- a/src/BtPieceMessage.cc
+++ b/src/BtPieceMessage.cc
@@ -55,6 +55,7 @@
 #include "PeerConnection.h"
 #include "fmt.h"
 #include "DownloadContext.h"
+#include "PeerStorage.h"
 
 namespace aria2 {
 
@@ -132,6 +133,7 @@ void BtPieceMessage::doReceivedAction()
         onNewPiece(piece);
       } else {
         onWrongPiece(piece);
+        peerStorage_->addBadPeer(getPeer()->getIPAddress());
         throw DL_ABORT_EX("Bad piece hash.");
       }
     }
@@ -320,4 +322,10 @@ void BtPieceMessage::setDownloadContext
   downloadContext_ = downloadContext;
 }
 
+void BtPieceMessage::setPeerStorage
+(const SharedHandle<PeerStorage>& peerStorage)
+{
+  peerStorage_ = peerStorage;
+}
+
 } // namespace aria2
diff --git a/src/BtPieceMessage.h b/src/BtPieceMessage.h
index e9eaff3c..2aadc31f 100644
--- a/src/BtPieceMessage.h
+++ b/src/BtPieceMessage.h
@@ -42,6 +42,7 @@ namespace aria2 {
 class Piece;
 class BtPieceMessage;
 class DownloadContext;
+class PeerStorage;
 
 typedef SharedHandle<BtPieceMessage> BtPieceMessageHandle;
 
@@ -53,6 +54,7 @@ private:
   unsigned char* block_;
   unsigned char* rawData_;
   SharedHandle<DownloadContext> downloadContext_;
+  SharedHandle<PeerStorage> peerStorage_;
 
   static size_t MESSAGE_HEADER_LENGTH;
 
@@ -95,6 +97,8 @@ public:
 
   void setDownloadContext(const SharedHandle<DownloadContext>& downloadContext);
 
+  void setPeerStorage(const SharedHandle<PeerStorage>& peerStorage);
+
   static BtPieceMessageHandle create
   (const unsigned char* data, size_t dataLength);
 
diff --git a/src/DefaultBtMessageFactory.cc b/src/DefaultBtMessageFactory.cc
index 7f1a733e..61c5e3a0 100644
--- a/src/DefaultBtMessageFactory.cc
+++ b/src/DefaultBtMessageFactory.cc
@@ -340,6 +340,7 @@ DefaultBtMessageFactory::createPieceMessage
                                  pieceStorage_->getPieceLength(index)));
   msg->setBtMessageValidator(validator);
   msg->setDownloadContext(downloadContext_);
+  msg->setPeerStorage(peerStorage_);
   setCommonProperty(msg);
   return msg;
 }
diff --git a/src/DefaultPeerStorage.cc b/src/DefaultPeerStorage.cc
index 900a6b33..af68de2a 100644
--- a/src/DefaultPeerStorage.cc
+++ b/src/DefaultPeerStorage.cc
@@ -47,6 +47,7 @@
 #include "wallclock.h"
 #include "a2functional.h"
 #include "fmt.h"
+#include "SimpleRandomizer.h"
 
 namespace aria2 {
 
@@ -100,6 +101,11 @@ bool DefaultPeerStorage::addPeer(const SharedHandle<Peer>& peer) {
                      peer->getIPAddress().c_str(), peer->getPort()));
     return false;
   }
+  if(isBadPeer(peer->getIPAddress())) {
+    A2_LOG_DEBUG(fmt("Adding %s:%u is rejected because it is marked bad.",
+                     peer->getIPAddress().c_str(), peer->getPort()));
+    return false;
+  }
   const size_t peerListSize = peers_.size();
   if(peerListSize >= maxPeerListSize_) {
     deleteUnusedPeer(peerListSize-maxPeerListSize_+1);
@@ -122,6 +128,10 @@ void DefaultPeerStorage::addPeer(const std::vector<SharedHandle<Peer> >& peers)
                        " added.",
                        peer->getIPAddress().c_str(), peer->getPort()));
       continue;
+    } else if(isBadPeer(peer->getIPAddress())) {
+      A2_LOG_DEBUG(fmt("Adding %s:%u is rejected because it is marked bad.",
+                       peer->getIPAddress().c_str(), peer->getPort()));
+      continue;
     } else {
       A2_LOG_DEBUG(fmt(MSG_ADDING_PEER,
                        peer->getIPAddress().c_str(), peer->getPort()));
@@ -301,6 +311,40 @@ TransferStat DefaultPeerStorage::getTransferStatFor
   }
 }
 
+bool DefaultPeerStorage::isBadPeer(const std::string& ipaddr)
+{
+  std::map<std::string, time_t>::iterator i = badPeers_.find(ipaddr);
+  if(i == badPeers_.end()) {
+    return false;
+  } else if(global::wallclock().getTime() >= (*i).second) {
+    badPeers_.erase(i);
+    return false;
+  } else {
+    return true;
+  }
+}
+
+void DefaultPeerStorage::addBadPeer(const std::string& ipaddr)
+{
+  if(lastBadPeerCleaned_.difference(global::wallclock()) >= 3600) {
+    for(std::map<std::string, time_t>::iterator i = badPeers_.begin(),
+          eoi = badPeers_.end(); i != eoi;) {
+      if(global::wallclock().getTime() >= (*i).second) {
+        A2_LOG_DEBUG(fmt("Purge %s from bad peer", (*i).first.c_str()));
+        badPeers_.erase(i++);
+        eoi = badPeers_.end();
+      } else {
+        ++i;
+      }
+    }
+    lastBadPeerCleaned_ = global::wallclock();
+  }
+  A2_LOG_DEBUG(fmt("Added %s as bad peer", ipaddr.c_str()));
+  // We use variable timeout to avoid many bad peers wake up at once.
+  badPeers_[ipaddr] = global::wallclock().getTime()+
+    std::max(SimpleRandomizer::getInstance()->getRandomNumber(601), 120L);
+}
+
 void DefaultPeerStorage::deleteUnusedPeer(size_t delSize) {
   std::deque<SharedHandle<Peer> > temp;
   for(std::deque<SharedHandle<Peer> >::const_reverse_iterator itr =
diff --git a/src/DefaultPeerStorage.h b/src/DefaultPeerStorage.h
index 2089831e..5411eac7 100644
--- a/src/DefaultPeerStorage.h
+++ b/src/DefaultPeerStorage.h
@@ -68,6 +68,9 @@ private:
 
   TransferStat cachedTransferStat_;
 
+  std::map<std::string, time_t> badPeers_;
+  Timer lastBadPeerCleaned_;
+
   bool isPeerAlreadyAdded(const SharedHandle<Peer>& peer);
 
   void addDroppedPeer(const SharedHandle<Peer>& peer);
@@ -100,6 +103,10 @@ public:
 
   virtual TransferStat getTransferStatFor(const SharedHandle<Peer>& peer);
 
+  virtual bool isBadPeer(const std::string& ipaddr);
+
+  virtual void addBadPeer(const std::string& ipaddr);
+
   virtual void returnPeer(const SharedHandle<Peer>& peer);
 
   virtual bool chokeRoundIntervalElapsed();
diff --git a/src/PeerStorage.h b/src/PeerStorage.h
index b3bcce88..7e75a11f 100644
--- a/src/PeerStorage.h
+++ b/src/PeerStorage.h
@@ -104,6 +104,17 @@ public:
   virtual TransferStat getTransferStatFor(const SharedHandle<Peer>& peer) = 0;
 
   /**
+   * Returns true if peer with ipaddr should be ignored because, for
+   * example, it sends bad data.
+   */
+  virtual bool isBadPeer(const std::string& ipaddr) = 0;
+
+  /**
+   * Adds peer with ipaddr in bad peer set.
+   */
+  virtual void addBadPeer(const std::string& ipaddr) = 0;
+
+  /**
    * Tells PeerStorage object that peer is no longer used in the session.
    */
   virtual void returnPeer(const SharedHandle<Peer>& peer) = 0;
diff --git a/test/DefaultPeerStorageTest.cc b/test/DefaultPeerStorageTest.cc
index b38cf5cd..aa740128 100644
--- a/test/DefaultPeerStorageTest.cc
+++ b/test/DefaultPeerStorageTest.cc
@@ -25,6 +25,7 @@ class DefaultPeerStorageTest:public CppUnit::TestFixture {
   CPPUNIT_TEST(testCalculateStat);
   CPPUNIT_TEST(testReturnPeer);
   CPPUNIT_TEST(testOnErasingPeer);
+  CPPUNIT_TEST(testAddBadPeer);
   CPPUNIT_TEST_SUITE_END();
 private:
   SharedHandle<BtRuntime> btRuntime;
@@ -48,6 +49,7 @@ public:
   void testCalculateStat();
   void testReturnPeer();
   void testOnErasingPeer();
+  void testAddBadPeer();
 };
 
 
@@ -262,4 +264,12 @@ void DefaultPeerStorageTest::testOnErasingPeer()
   // test this
 }
 
+void DefaultPeerStorageTest::testAddBadPeer()
+{
+  DefaultPeerStorage ps;
+  ps.addBadPeer("192.168.0.1");
+  CPPUNIT_ASSERT(ps.isBadPeer("192.168.0.1"));
+  CPPUNIT_ASSERT(!ps.isBadPeer("192.168.0.2"));
+}
+
 } // namespace aria2
diff --git a/test/MockPeerStorage.h b/test/MockPeerStorage.h
index 91e04d39..fbb7ccf1 100644
--- a/test/MockPeerStorage.h
+++ b/test/MockPeerStorage.h
@@ -71,6 +71,15 @@ public:
     this->stat = stat;
   }
 
+  virtual bool isBadPeer(const std::string& ipaddr)
+  {
+    return false;
+  }
+
+  virtual void addBadPeer(const std::string& ipaddr)
+  {
+  }
+
   virtual void returnPeer(const SharedHandle<Peer>& peer)
   {
   }