2008-09-28 Tatsuhiro Tsujikawa <tujikawa at rednoah dot com>

Fixed the bug that aria2 may send duplicate data to BitTorrent 
peer if
	whole message data is not sent in one send() call.
	* src/BtPieceMessage.cc
	* src/BtPieceMessage.h
	* src/PeerConnection.cc
	* src/PeerConnection.h
	* src/SimpleBtMessage.cc
	* src/SimpleBtMessage.h

ChangeLog

@@ -1,3 +1,14 @@
+2008-09-28  Tatsuhiro Tsujikawa  <tujikawa at rednoah dot com>
+
+	Fixed the bug that aria2 may send duplicate data to BitTorrent peer if
+	whole message data is not sent in one send() call.
+	* src/BtPieceMessage.cc
+	* src/BtPieceMessage.h
+	* src/PeerConnection.cc
+	* src/PeerConnection.h
+	* src/SimpleBtMessage.cc
+	* src/SimpleBtMessage.h
+	
 2008-09-28  Tatsuhiro Tsujikawa  <tujikawa at rednoah dot com>
 
 	Fixed the bug that aria2 downloads faster than max-download-limit when

src/BtPieceMessage.cc

@@ -53,6 +53,7 @@
 #include "StringFormat.h"
 #include <cstring>
 #include <cstdlib>
+#include <cassert>
 
 namespace aria2 {
 
@@ -140,61 +141,34 @@ void BtPieceMessage::send() {
   if(invalidate) {
     return;
   }
-  if(!headerSent) {
+  if(!sendingInProgress) {
-    if(!sendingInProgress) {
+    logger->info(MSG_SEND_PEER_MESSAGE,
-      logger->info(MSG_SEND_PEER_MESSAGE,
+		 cuid, peer->ipaddr.c_str(), peer->port,
-		   cuid, peer->ipaddr.c_str(), peer->port,
+		 toString().c_str());
-		   toString().c_str());
+    getMessageHeader();
-      getMessageHeader();
+    peerConnection->sendMessage(msgHeader, getMessageHeaderLength());
-      leftDataLength = getMessageHeaderLength();
-      sendingInProgress = true;
-    }
-    size_t writtenLength
-      = peerConnection->sendMessage(msgHeader+getMessageHeaderLength()-leftDataLength,
-				    leftDataLength);
-    if(writtenLength == leftDataLength) {
-      headerSent = true;
-      leftDataLength = blockLength;
-    } else {
-      leftDataLength -= writtenLength;
-    }
-  }
-  if(headerSent) {
-    sendingInProgress = false;
     off_t pieceDataOffset =
-      (off_t)index*btContext->getPieceLength()+begin+blockLength-leftDataLength;
+      (off_t)index*btContext->getPieceLength()+begin;
-    size_t writtenLength = sendPieceData(pieceDataOffset, leftDataLength);
+    size_t writtenLength = sendPieceData(pieceDataOffset, blockLength);
+    logger->debug("msglength = %zu bytes",
+		  getMessageHeaderLength()+blockLength);
+    peer->updateUploadLength(writtenLength);
+  } else {
+    ssize_t writtenLength = peerConnection->sendPendingData();
     peer->updateUploadLength(writtenLength);
-    if(writtenLength < leftDataLength) {
-      sendingInProgress = true;
-    }
-    leftDataLength -= writtenLength;
   }
+  sendingInProgress = !peerConnection->sendBufferIsEmpty();
 }
 
 size_t BtPieceMessage::sendPieceData(off_t offset, size_t length) const {
-  size_t BUF_SIZE = 256;
+  assert(length <= 16*1024);
-  unsigned char buf[BUF_SIZE];
+  unsigned char buf[16*1024];
-  div_t res = div(length, BUF_SIZE);
+  if(pieceStorage->getDiskAdaptor()->readData(buf, length, offset) ==
-  size_t writtenLength = 0;
+     static_cast<ssize_t>(length)) {
-  for(int i = 0; i < res.quot; i++) {
+    return peerConnection->sendMessage(buf, length);
-    if((size_t)pieceStorage->getDiskAdaptor()->readData(buf, BUF_SIZE, offset+i*BUF_SIZE) < BUF_SIZE) {
+  } else {
-      throw DlAbortEx(EX_DATA_READ);
+    throw DlAbortEx(EX_DATA_READ);
-    }
-    size_t ws = peerConnection->sendMessage(buf, BUF_SIZE);
-    writtenLength += ws;
-    if(ws != BUF_SIZE) {
-      return writtenLength;
-    }
   }
-  if(res.rem > 0) {
-    if(pieceStorage->getDiskAdaptor()->readData(buf, res.rem, offset+res.quot*BUF_SIZE) < res.rem) {
-      throw DlAbortEx(EX_DATA_READ);
-    }
-    size_t ws = peerConnection->sendMessage(buf, res.rem);
-    writtenLength += ws;
-  }
-  return writtenLength;
 }
 
 std::string BtPieceMessage::toString() const {

src/BtPieceMessage.h

@@ -52,8 +52,6 @@ private:
   uint32_t begin;
   uint32_t blockLength;
   unsigned char* block;
-  size_t leftDataLength;
-  bool headerSent;
   unsigned char* msgHeader;
 
   static size_t MESSAGE_HEADER_LENGTH;
@@ -99,8 +97,6 @@ public:
      begin(begin),
      blockLength(blockLength),
      block(0),
-     leftDataLength(0),
-     headerSent(false),
      msgHeader(0)
   {
     uploading = true;

src/PeerConnection.cc

@@ -69,11 +69,9 @@ PeerConnection::~PeerConnection() {}
 ssize_t PeerConnection::sendMessage(const unsigned char* data,
 				    size_t dataLength)
 {
-  if(socket->isWritable(0)) {
+  ssize_t writtenLength = sendData(data, dataLength, _encryptionEnabled);
-    return sendData(data, dataLength, _encryptionEnabled);
+  logger->debug("sent %d byte(s).", writtenLength);
-  } else {
+  return writtenLength;
-    return 0;
-  }
 }
 
 bool PeerConnection::receiveMessage(unsigned char* data, size_t& dataLength) {
@@ -231,4 +229,16 @@ void PeerConnection::presetBuffer(const unsigned char* data, size_t length)
   resbufLength = length;
 }
 
+bool PeerConnection::sendBufferIsEmpty() const
+{
+  return _socketBuffer.sendBufferIsEmpty();
+}
+
+ssize_t PeerConnection::sendPendingData()
+{
+  ssize_t writtenLength = _socketBuffer.send();
+  logger->debug("sent %d byte(s).", writtenLength);
+  return writtenLength;
+}
+
 } // namespace aria2

src/PeerConnection.h

@@ -99,6 +99,10 @@ public:
 			const SharedHandle<ARC4Decryptor>& decryptor);
 
   void presetBuffer(const unsigned char* data, size_t length);
+
+  bool sendBufferIsEmpty() const;
+  
+  ssize_t sendPendingData();
 };
 
 typedef SharedHandle<PeerConnection> PeerConnectionHandle;

src/SimpleBtMessage.cc

@@ -41,7 +41,7 @@
 
 namespace aria2 {
 
-SimpleBtMessage::SimpleBtMessage():leftDataLength(0) {}
+SimpleBtMessage::SimpleBtMessage() {}//:leftDataLength(0) {}
 
 SimpleBtMessage::~SimpleBtMessage() {}
 
@@ -49,22 +49,22 @@ void SimpleBtMessage::send() {
   if(invalidate) {
     return;
   }
-  if(sendPredicate() || sendingInProgress) {
+  if(!sendPredicate() && !sendingInProgress) {
+    return;
+  }
+  if(!sendingInProgress) {
     const unsigned char* msg = getMessage();
     size_t msgLength = getMessageLength();
-    if(!sendingInProgress) {
+    logger->info(MSG_SEND_PEER_MESSAGE,
-      logger->info(MSG_SEND_PEER_MESSAGE,
+		 cuid, peer->ipaddr.c_str(), peer->port, toString().c_str());
-		   cuid, peer->ipaddr.c_str(), peer->port, toString().c_str());
+    logger->debug("msglength = %zu bytes", msgLength);
-      leftDataLength = getMessageLength();
+    peerConnection->sendMessage(msg, msgLength);
-    }
+  } else {
-    sendingInProgress = false;
+    peerConnection->sendPendingData();
-    size_t writtenLength = peerConnection->sendMessage(msg+msgLength-leftDataLength, leftDataLength);
+  }
-    if(writtenLength == leftDataLength) {
+  sendingInProgress = !peerConnection->sendBufferIsEmpty();
-      onSendComplete();
+  if(!sendingInProgress) {
-    } else {
+    onSendComplete();
-      leftDataLength -= writtenLength;
-      sendingInProgress = true;
-    }
   }
 }
 

src/SimpleBtMessage.h

@@ -40,8 +40,6 @@
 namespace aria2 {
 
 class SimpleBtMessage : public AbstractBtMessage {
-private:
-  size_t leftDataLength;
 public:
   SimpleBtMessage();