2009-03-19 Tatsuhiro Tsujikawa <t-tujikawa@users.sourceforge.net>

Reduced the number of calls to PeerStorage::calculateStat() and
	SegmentMan::calculateDownloadSpeed() to lower CPU usage on
	higher transfer rate.	
	* src/DefaultBtInteractive.cc
	* src/DefaultBtMessageDispatcher.cc
	* src/DefaultPeerStorage.cc
	* src/DefaultPeerStorage.h
	* src/DownloadCommand.cc
	* src/DownloadEngine.cc
	* src/PeerStorage.h
	* src/SegmentMan.cc
	* src/SegmentMan.h
	* test/MockPeerStorage.h

ChangeLog

@@ -1,3 +1,19 @@
+2009-03-19  Tatsuhiro Tsujikawa  <t-tujikawa@users.sourceforge.net>
+
+	Reduced the number of calls to PeerStorage::calculateStat() and
+	SegmentMan::calculateDownloadSpeed() to lower CPU usage on higher
+	transfer rate.	
+	* src/DefaultBtInteractive.cc
+	* src/DefaultBtMessageDispatcher.cc
+	* src/DefaultPeerStorage.cc
+	* src/DefaultPeerStorage.h
+	* src/DownloadCommand.cc
+	* src/DownloadEngine.cc
+	* src/PeerStorage.h
+	* src/SegmentMan.cc
+	* src/SegmentMan.h
+	* test/MockPeerStorage.h
+	
 2009-03-13  Tatsuhiro Tsujikawa  <t-tujikawa@users.sourceforge.net>
 
 	Prepended _dir+"/" after joining path elements in torrent file.

src/DefaultBtInteractive.cc

@@ -265,8 +265,10 @@ size_t DefaultBtInteractive::receiveMessages() {
 	floodingStat.incChokeUnchokeCount();
       }
       break;
-    case BtRequestMessage::ID:
     case BtPieceMessage::ID:
+      _peerStorage->updateTransferStatFor(peer);
+      // pass through
+    case BtRequestMessage::ID:
       inactiveCheckPoint.reset();
       break;
     }

src/DefaultBtMessageDispatcher.cc

@@ -93,6 +93,9 @@ void DefaultBtMessageDispatcher::sendMessages() {
       }
     }
     msg->send();
+    if(msg->isUploading()) {
+      _peerStorage->updateTransferStatFor(peer);
+    }
     if(msg->isSendingInProgress()) {
       messageQueue.push_front(msg);
       break;

src/DefaultPeerStorage.cc

@@ -59,7 +59,8 @@ DefaultPeerStorage::DefaultPeerStorage(const BtContextHandle& btContext,
   removedPeerSessionDownloadLength(0),
   removedPeerSessionUploadLength(0),
   _seederStateChoke(new BtSeederStateChoke()),
-  _leecherStateChoke(new BtLeecherStateChoke())
+  _leecherStateChoke(new BtLeecherStateChoke()),
+  _lastTransferStatMapUpdated(0)
 {}
 
 DefaultPeerStorage::~DefaultPeerStorage()
@@ -191,31 +192,46 @@ void DefaultPeerStorage::getActivePeers(std::deque<SharedHandle<Peer> >& activeP
   std::for_each(peers.begin(), peers.end(), CollectActivePeer(activePeers));
 }
 
-class CalculateStat {
-private:
-  TransferStat _stat;
-  struct timeval _now;
-public:
-  CalculateStat()
-  {
-    gettimeofday(&_now, 0);
-  }
+static TransferStat caluclateStatFor(const SharedHandle<Peer>& peer)
+{
+  struct timeval now;
+  gettimeofday(&now, 0);
+  TransferStat s;
+  s.downloadSpeed = peer->calculateDownloadSpeed(now);
+  s.uploadSpeed = peer->calculateUploadSpeed(now);
+  s.sessionDownloadLength = peer->getSessionDownloadLength();
+  s.sessionUploadLength = peer->getSessionUploadLength();
+  return s;
+}
 
-  void operator()(const PeerHandle& peer)
-  {
-    if(peer->isActive()) {
-      _stat.downloadSpeed += peer->calculateDownloadSpeed(_now);
-      _stat.uploadSpeed += peer->calculateUploadSpeed(_now);
-      _stat.sessionDownloadLength += peer->getSessionDownloadLength();
-      _stat.sessionUploadLength += peer->getSessionUploadLength();    
+TransferStat DefaultPeerStorage::calculateStat()
+{
+  TransferStat stat;
+  if(_lastTransferStatMapUpdated.elapsedInMillis(250)) {
+    logger->debug("Updating TransferStat of PeerStorage");
+    _lastTransferStatMapUpdated.reset();
+    _peerTransferStatMap.clear();
+    std::deque<SharedHandle<Peer> > activePeers;
+    getActivePeers(activePeers);
+    struct timeval now;
+    gettimeofday(&now, 0);
+    for(std::deque<SharedHandle<Peer> >::const_iterator i = activePeers.begin();
+	i != activePeers.end(); ++i) {
+      TransferStat s;
+      s.downloadSpeed = (*i)->calculateDownloadSpeed(now);
+      s.uploadSpeed = (*i)->calculateUploadSpeed(now);
+      s.sessionDownloadLength = (*i)->getSessionDownloadLength();
+      s.sessionUploadLength = (*i)->getSessionUploadLength();
+
+      _peerTransferStatMap[(*i)->getID()] = caluclateStatFor(*i);
+      stat = stat+s;
+    }
+  } else {
+    for(std::map<std::string, TransferStat>::const_iterator i =
+	  _peerTransferStatMap.begin(); i != _peerTransferStatMap.end(); ++i) {
+      stat = stat+(*i).second;
     }
   }
-
-  const TransferStat& getTransferStat() { return _stat; }
-};
-
-TransferStat DefaultPeerStorage::calculateStat() {
-  TransferStat stat = std::for_each(peers.begin(), peers.end(), CalculateStat()).getTransferStat();
   stat.sessionDownloadLength += removedPeerSessionDownloadLength;
   stat.sessionUploadLength += removedPeerSessionUploadLength;
   stat.setAllTimeUploadLength(_btRuntime->getUploadLengthAtStartup()+
@@ -223,6 +239,12 @@ TransferStat DefaultPeerStorage::calculateStat() {
   return stat;
 }
 
+void DefaultPeerStorage::updateTransferStatFor(const SharedHandle<Peer>& peer)
+{
+  logger->debug("Updating TransferStat for peer %s", peer->getID().c_str());
+  _peerTransferStatMap[peer->getID()] = caluclateStatFor(peer);
+}
+
 void DefaultPeerStorage::deleteUnusedPeer(size_t delSize) {
   Peers temp;
   for(Peers::reverse_iterator itr = peers.rbegin();

src/DefaultPeerStorage.h

@@ -37,6 +37,10 @@
 
 #include "PeerStorage.h"
 
+#include <map>
+
+#include "TimeA2.h"
+
 namespace aria2 {
 
 class BtContext;
@@ -61,6 +65,10 @@ private:
   BtSeederStateChoke* _seederStateChoke;
   BtLeecherStateChoke* _leecherStateChoke;
 
+  std::map<std::string, TransferStat> _peerTransferStatMap;
+
+  Time _lastTransferStatMapUpdated;
+
   bool isPeerAlreadyAdded(const SharedHandle<Peer>& peer);
 public:
   DefaultPeerStorage(const SharedHandle<BtContext>& btContext,
@@ -86,6 +94,8 @@ public:
 
   virtual TransferStat calculateStat();
 
+  virtual void updateTransferStatFor(const SharedHandle<Peer>& peer);
+
   virtual void returnPeer(const SharedHandle<Peer>& peer);
 
   virtual bool chokeRoundIntervalElapsed();

src/DownloadCommand.cc

@@ -161,6 +161,8 @@ bool DownloadCommand::executeInternal() {
   
   peerStat->updateDownloadLength(bufSize);
 
+  _requestGroup->getSegmentMan()->updateDownloadSpeedFor(peerStat);
+
   if(_requestGroup->getTotalLength() != 0 && bufSize == 0 &&
      !socket->wantRead() && !socket->wantWrite()) {
     throw DlRetryEx(EX_GOT_EOF);

src/DownloadEngine.cc

@@ -204,6 +204,7 @@ void DownloadEngine::onEndOfRun()
 
 void DownloadEngine::afterEachIteration()
 {
+  _requestGroupMan->calculateStat();
   if(globalHaltRequested == 1) {
     logger->notice(_("Shutdown sequence commencing... Press Ctrl-C again for emergency shutdown."));
     requestHalt();

src/PeerStorage.h

@@ -85,6 +85,8 @@ public:
    */
   virtual TransferStat calculateStat() = 0;
 
+  virtual void updateTransferStatFor(const SharedHandle<Peer>& peer) = 0;
+
   /**
    * Tells PeerStorage object that peer is no longer used in the session.
    */

src/SegmentMan.cc

@@ -33,6 +33,11 @@
  */
 /* copyright --> */
 #include "SegmentMan.h"
+
+#include <cassert>
+#include <algorithm>
+#include <numeric>
+
 #include "Util.h"
 #include "message.h"
 #include "prefs.h"
@@ -45,9 +50,6 @@
 #include "Option.h"
 #include "DownloadContext.h"
 #include "Piece.h"
-#include <algorithm>
-#include <numeric>
-#include <cassert>
 
 namespace aria2 {
 
@@ -62,7 +64,8 @@ SegmentMan::SegmentMan(const Option* option,
   _option(option),
   logger(LogFactory::getInstance()),
   _downloadContext(downloadContext),
-  _pieceStorage(pieceStorage)
+  _pieceStorage(pieceStorage),
+  _lastPeerStatDlspdMapUpdated(0)
 {}
 
 SegmentMan::~SegmentMan() {}
@@ -296,17 +299,35 @@ const std::deque<SharedHandle<PeerStat> >& SegmentMan::getPeerStats() const
   return peerStats;
 }
 
-unsigned int SegmentMan::calculateDownloadSpeed() const {
+unsigned int SegmentMan::calculateDownloadSpeed()
+{
   unsigned int speed = 0;
-  for(std::deque<SharedHandle<PeerStat> >::const_iterator itr = peerStats.begin(); itr != peerStats.end(); itr++) {
-    const PeerStatHandle& peerStat = *itr;
-    if(peerStat->getStatus() == PeerStat::ACTIVE) {
-      speed += peerStat->calculateDownloadSpeed();
+  if(_lastPeerStatDlspdMapUpdated.elapsedInMillis(250)) {
+    _lastPeerStatDlspdMapUpdated.reset();
+    _peerStatDlspdMap.clear();
+    for(std::deque<SharedHandle<PeerStat> >::const_iterator i =
+	  peerStats.begin(); i != peerStats.end(); ++i) {
+      if((*i)->getStatus() == PeerStat::ACTIVE) {
+	unsigned int s = (*i)->calculateDownloadSpeed();
+	_peerStatDlspdMap[(*i)->getCuid()] = s;
+	speed += s;
+      }
+    }
+  } else {
+    for(std::map<int32_t, unsigned int>::const_iterator i =
+	  _peerStatDlspdMap.begin();
+	i != _peerStatDlspdMap.end(); ++i) {
+      speed += (*i).second;
     }
   }
   return speed;
 }
 
+void SegmentMan::updateDownloadSpeedFor(const SharedHandle<PeerStat>& pstat)
+{
+  _peerStatDlspdMap[pstat->getCuid()] = pstat->calculateDownloadSpeed();
+}
+
 class PeerStatDownloadLengthOperator {
 public:
   uint64_t operator()(uint64_t total, const SharedHandle<PeerStat>& ps)

src/SegmentMan.h

@@ -36,8 +36,13 @@
 #define _D_SEGMENT_MAN_H_
 
 #include "common.h"
-#include "SharedHandle.h"
+
 #include <deque>
+#include <map>
+
+#include "SharedHandle.h"
+#include "TimeA2.h"
+#include "Command.h"
 
 namespace aria2 {
 
@@ -79,6 +84,11 @@ private:
 
   std::deque<SharedHandle<PeerStat> > peerStats;
 
+  // key: PeerStat's cuid, value: its download speed
+  std::map<int32_t, unsigned int> _peerStatDlspdMap;
+
+  Time _lastPeerStatDlspdMapUpdated;
+
   SharedHandle<Segment> checkoutSegment(int32_t cuid,
 					const SharedHandle<Piece>& piece);
 
@@ -177,7 +187,9 @@ public:
   /**
    * Returns current download speed in bytes per sec. 
    */
-  unsigned int calculateDownloadSpeed() const;
+  unsigned int calculateDownloadSpeed();
+
+  void updateDownloadSpeedFor(const SharedHandle<PeerStat>& pstat);
 
   /**
    * Returns the downloaded bytes in this session.

test/MockPeerStorage.h

@@ -71,6 +71,8 @@ public:
     ++_numChokeExecuted;
   }
 
+  virtual void updateTransferStatFor(const SharedHandle<Peer>& peer) {}
+
   int getNumChokeExecuted() const
   {
     return _numChokeExecuted;