2006-05-24 Tatsuhiro Tsujikawa <tujikawa at rednoah dot com>

	To attempt to handle EINTR:
	
	* src/SocketCore.cc
	(isReadable): Added EINTR handling.
	(isWritable): Added EINTR handling.

	To improve the performance:
	
	* src/AbstractCommand.cc
	(isTimeoutDetected): Use Util::difftvsec() instead of 
Util::difftv().
	* src/PeerInteractionCommand.h
	(receiveMessage): Renamed as receiveMessages().
	(receiveMessages): New function.
	* src/PeerInteractionCommand.cc
	(executeInternal): receiveMessage loop is moved to new 
receiveMessages
	() function. detectMessageFlooding() is moved here.
	(detectMessageFlooding): Use Util::difftvsec() instead of 
Util::difftv
	(). The flooding detection for have message is comment out.
	(checkLongTimePeerChoking): Use Util::difftvsec() instead of
	Util::difftv().
	(receiveMessage): Renamed as receiveMessages().
	(receiveMessages): New function.
	(keepAlive): Use Util::difftvsec() instead of Util::difftv().
	(beforeSocketCheck): Commented out checkLongTimePeerChoking().
	* src/SleepCommand.cc
	(execute): Use Util::difftvsec() instead of Util::difftv().
	* src/BitfieldMan.h
	(getNthBitIndex): New function.
	(hasMissingPiece): New function.
	(getAllMissingIndexes): New function.
	* src/BitfieldMan.cc
	(countSetBit): Rewritten.
	(getNthBitIndex): New function.
	(getMissingIndexRandomly): Rewritten.
	(hasMissingPiece): New function.
	(getAllMissingIndexes): New function.
	* src/TorrentMan.cc
	(hasMissingPiece): New function.
	(deleteUsedPiece): Rewritten using STL.
	* src/PeerInteraction.cc
	(getNewPieceAndSendInterest): Use TorrentMan::hasMissingPiece(), 
which
	is a little bit faster then TorrentMan::getMissingPieceIndex().
	(addRequests): Updated the number of pending requests.
	* src/PeerAbstractCommand.cc
	(isTimeoutDetected): Use Util::difftvsec() instead of 
Util::difftv().
	(execute): Returns true if TorrentMan::isHalt() is true.
	Corrected the condition of upload limit checking.
	* src/Util.h
	(countBit): New function.
	* src/Util.cc
	(nbits): New variable.
	(countBit): New function.
	* src/DownloadEngine.h
	(SockCmdMap): New type definition.
	(waitData): Added an argument.
	(addSocket): Added an argument.
	(addSocketForReadCheck): Added an argument.
	(addSocketForWriteCheck): Added an argument.
	* src/DownloadEngine.cc
	(run): Executes only commands whose sockets are ready to read or 
write.
	All commands are executed in every 1 second.
	(waitData): Calls select() again if it returned
	EINTR.
	(addSocket): Saves socket and command object pair to the map.
	(deleteSocket): Deletes socket and command object pair from the 
map.
	(addSocketForReadCheck): Added an argument.
	(addSocketForWriteCheck): Added an argument.
	* src/DownloadCommand.cc
	(executeInternal): Use Util::difftvsec() instead of 
Util::difftv().
	
	To send "event=stopped" to the tracker when the application 
finishes:
	
	* src/PeerChokeCommand.h
	(checkPoint): New variable.
	* src/PeerChokeCommand.cc
	(PeerChokeCommand): Added the initialization of checkPoint.
	(execute): The interval check is now done by checkPoint, not
	SleepCommand. Return true if TorrentMan::isHalt() is true.
	* src/TorrentMan.h
	(halt): New function.
	(hasMissingPiece): New function.
	(isHalt): New function.
	(setHalt): New function.
	* src/TorrentMan.cc
	(TorrentMan): Added the initialization of halt.
	* src/TorrentAutoSaveCommand.h
	(checkPoint): New variable.
	* src/PeerListenCommand.cc
	(execute): Returns true if TorrentMan::isHalt() is true.
	* src/main.cc
	(setSignalHander): Added flags argument.
	(torrentHandler): Just calls TorrentMan::setHalt(true) and 
returns.
	(main): Set SA_ONESHOT flag of the signal hander of SIGINT and 
SIGTERM.
	Removed printDownloadAbortMessage() after torrent downloading 
loop.
	* src/TorrentAutoSaveCommand.cc
	(TorrentAutoSaveCommand): Added the initialization of 
checkPoint.
	(execute): Returns true if TorrentMan::isHalt() is true.
	The interval check is now done by checkPoint, not SleepCommand.
	* src/TrackerWatcherCommand.cc
	(execute): If TorrentMan::isHalt(), then create a tracker 
request with
	event=stopped.
	* src/TrackerUpdateCommand.cc
	(prepareForRetry): Do not use SleepCommand here.
	(execute): Returns true if TorrentMan::isHalt() is true.
	
	Others:
	
	* src/TorrentMan.cc
	(getMissingPieceIndex): Updated log message.
	* src/PeerInteraction.cc
	(createPeerMessag): Updated log message.

1 files changed, 47 insertions, 47 deletions

diff --git a/src/PeerInteractionCommand.cc b/src/PeerInteractionCommand.cc
index e7be9d55..3137602e 100644
--- a/src/PeerInteractionCommand.cc
+++ b/src/PeerInteractionCommand.cc
@@ -113,12 +113,10 @@ bool PeerInteractionCommand::executeInternal() {
   case WIRED:
     peerInteraction->syncPiece();
     decideChoking();
-    for(int i = 0; i < 50; i++) {
-      if(!socket->isReadable(0)) {
-	break;
-      }
-      receiveMessage();
-    }
+    receiveMessages();
+    detectMessageFlooding();
+    //checkLongTimePeerChoking();
+
     peerInteraction->deleteTimeoutRequestSlot();
     peerInteraction->deleteCompletedRequestSlot();
     peerInteraction->addRequests();
@@ -142,11 +140,12 @@ void PeerInteractionCommand::detectMessageFlooding() {
   if(freqCheckPoint.tv_sec == 0 && freqCheckPoint.tv_usec == 0) {
     freqCheckPoint = now;
   } else {
-    if(Util::difftv(now, freqCheckPoint) >= 5*1000000) {
-      if(chokeUnchokeCount*1.0/(Util::difftv(now, freqCheckPoint)/1000000) >= 0.4
-	 || haveCount*1.0/(Util::difftv(now, freqCheckPoint)/1000000) >= 20.0
-	 || keepAliveCount*1.0/(Util::difftv(now, freqCheckPoint)/1000000) >= 1.0) {
-	throw new DlAbortEx("flooding detected.");
+    int elapsed = Util::difftvsec(now, freqCheckPoint);
+    if(elapsed >= 5) {
+      if(chokeUnchokeCount*1.0/elapsed >= 0.4
+	 //|| haveCount*1.0/elapsed >= 20.0
+	 || keepAliveCount*1.0/elapsed >= 1.0) {
+	throw new DlAbortEx("Flooding detected.");
       } else {
 	chokeUnchokeCount = 0;
 	haveCount = 0;
@@ -169,8 +168,8 @@ void PeerInteractionCommand::checkLongTimePeerChoking() {
     }
   } else {
     if(peer->amInterested && peer->peerChoking) {
-      if(Util::difftv(now, chokeCheckPoint) >= MAX_PEER_CHOKING_INTERVAL*1000000) {
-	throw new DlAbortEx("too long choking");
+      if(Util::difftvsec(now, chokeCheckPoint) >= MAX_PEER_CHOKING_INTERVAL) {
+	throw new DlAbortEx("Too long choking.");
       }
     } else {
       chokeCheckPoint.tv_sec = 0;
@@ -191,39 +190,41 @@ void PeerInteractionCommand::decideChoking() {
   }
 }
 
-void PeerInteractionCommand::receiveMessage() {
-  PeerMessage* message = peerInteraction->receiveMessage();
-  if(message == NULL) {
-    return;
-  }
-  logger->info(MSG_RECEIVE_PEER_MESSAGE, cuid,
-	       peer->ipaddr.c_str(), peer->port,
-	       message->toString().c_str());
-  // to detect flooding
-  switch(message->getId()) {
-  case KeepAliveMessage::ID:
-    keepAliveCount++;
-    break;
-  case ChokeMessage::ID:
-    if(!peer->peerChoking) {
-      chokeUnchokeCount++;
+void PeerInteractionCommand::receiveMessages() {
+  for(int i = 0; i < 50; i++) {
+    PeerMessage* message = peerInteraction->receiveMessage();
+    if(message == NULL) {
+      return;
     }
-    break;
-  case UnchokeMessage::ID:
-    if(peer->peerChoking) {
-      chokeUnchokeCount++;
+    logger->info(MSG_RECEIVE_PEER_MESSAGE, cuid,
+		 peer->ipaddr.c_str(), peer->port,
+		 message->toString().c_str());
+    // to detect flooding
+    switch(message->getId()) {
+    case KeepAliveMessage::ID:
+      keepAliveCount++;
+      break;
+    case ChokeMessage::ID:
+      if(!peer->peerChoking) {
+	chokeUnchokeCount++;
+      }
+      break;
+    case UnchokeMessage::ID:
+      if(peer->peerChoking) {
+	chokeUnchokeCount++;
+      }
+      break;
+    case HaveMessage::ID:
+      haveCount++;
+      break;
+    }
+    try {
+      message->receivedAction();
+      delete message;
+    } catch(Exception* ex) {
+      delete message;
+      throw;
     }
-    break;
-  case HaveMessage::ID:
-    haveCount++;
-    break;
-  }
-  try {
-    message->receivedAction();
-    delete message;
-  } catch(Exception* ex) {
-    delete message;
-    throw;
   }
 }
 
@@ -255,7 +256,7 @@ void PeerInteractionCommand::keepAlive() {
   } else {
     struct timeval now;
     gettimeofday(&now, NULL);
-    if(Util::difftv(now, keepAliveCheckPoint) >= (long long int)120*1000000) {
+    if(Util::difftvsec(now, keepAliveCheckPoint) >= 120) {
       if(peerInteraction->countMessageInQueue() == 0) {
 	peerInteraction->addMessage(peerInteraction->createKeepAliveMessage());
 	peerInteraction->sendMessages(e->getUploadSpeed());
@@ -269,8 +270,7 @@ void PeerInteractionCommand::beforeSocketCheck() {
   if(sequence == WIRED) {
     e->torrentMan->unadvertisePiece(cuid);
     detectMessageFlooding();
-    checkLongTimePeerChoking();
-
+    //checkLongTimePeerChoking();
     PieceIndexes indexes = e->torrentMan->getAdvertisedPieceIndexes(cuid);
     if(indexes.size() >= 20) {
       if(peer->isFastExtensionEnabled()) {