2007-10-18 Tatsuhiro Tsujikawa <tujikawa at rednoah dot com>

	Added HTTP/1.1 keep alive and pipelining support.
	See --enable-http-keep-alive and --enable-http-pipelining 
option.
	* src/AbstractCommand.{h, cc}: Now it has one-to-many relation 
to
	Segment.
	* src/HttpDownloadCommand.{h, cc}
	* src/OptionHandlerFactory.cc
	* src/HttpConnection.{h, cc}
	* src/version_usage.cc
	* src/HttpInitiateConnectionCommand.cc
	* src/FtpInitiateConnectionCommand.cc
	* src/Segment.h
	* src/HttpRequestCommand.{h, cc}
	* src/option_processing.cc
	* src/prefs.h
	* src/HttpResponseCommand.cc
	* src/SegmentMan.{h, cc}
	* src/FtpNegotiateCommand.cc
	* src/HttpProxyResponseCommand.cc
	* src/Request.cc
	* src/HttpRequest.cc
	* src/DownloadCommand.cc
	* test/HttpRequestTest.cc
	* test/RequestTest.cc

23 files changed, 185 insertions, 49 deletions

diff --git a/src/AbstractCommand.cc b/src/AbstractCommand.cc
index e4bd954b..9705386f 100644
--- a/src/AbstractCommand.cc
+++ b/src/AbstractCommand.cc
@@ -65,7 +65,6 @@ AbstractCommand::AbstractCommand(int32_t cuid,
 				 const SocketHandle& s):
   Command(cuid), RequestGroupAware(requestGroup),
   req(req), e(e), socket(s),
-  segment(0),
   checkSocketIsReadable(false), checkSocketIsWritable(false),
   nameResolverCheck(false)
 { 
@@ -107,12 +106,23 @@ bool AbstractCommand::execute() {
        !checkSocketIsReadable && !checkSocketIsWritable && !nameResolverCheck) {
       checkPoint.reset();
       if(!_requestGroup->getPieceStorage().isNull()) {
-	if(segment.isNull()) {
-	  segment = _requestGroup->getSegmentMan()->getSegment(cuid);
+	_segments = _requestGroup->getSegmentMan()->getInFlightSegment(cuid);
+	int32_t maxSegments;
+	if(req->isKeepAlive() && e->option->get(PREF_ENABLE_HTTP_PIPELINING) == V_TRUE) {
+	  maxSegments = e->option->getAsInt(PREF_MAX_HTTP_PIPELINING);
+	} else {
+	  maxSegments = 1;
+	}
+	while((int32_t)_segments.size() < maxSegments) {
+	  SegmentHandle segment = _requestGroup->getSegmentMan()->getSegment(cuid);
 	  if(segment.isNull()) {
-	    logger->info(MSG_NO_SEGMENT_AVAILABLE, cuid);
-	    return prepareForRetry(1);
+	    break;
 	  }
+	  _segments.push_back(segment);
+	}
+	if(_segments.empty()) {
+	  logger->info(MSG_NO_SEGMENT_AVAILABLE, cuid);
+	  return prepareForRetry(1);
 	}
       }
       return executeInternal();
diff --git a/src/AbstractCommand.h b/src/AbstractCommand.h
index 26fd4ddb..eccdead1 100644
--- a/src/AbstractCommand.h
+++ b/src/AbstractCommand.h
@@ -45,6 +45,7 @@ extern typedef SharedHandle<Request> RequestHandle;
 class DownloadEngine;
 class Segment;
 extern typedef SharedHandle<Segment> SegmentHandle;
+extern typedef deque<SegmentHandle> Segments;
 class NameResolver;
 extern typedef SharedHandle<NameResolver> NameResolverHandle;
 
@@ -56,7 +57,7 @@ protected:
   RequestHandle req;
   DownloadEngine* e;
   SocketHandle socket;
-  SegmentHandle segment;
+  Segments _segments;
 
   void tryReserved();
   virtual bool prepareForRetry(int32_t wait);
diff --git a/src/DownloadCommand.cc b/src/DownloadCommand.cc
index 53a946de..b9ea50e2 100644
--- a/src/DownloadCommand.cc
+++ b/src/DownloadCommand.cc
@@ -47,6 +47,7 @@
 #include "Segment.h"
 #include "PieceStorage.h"
 #include "Option.h"
+#include "HttpRequestCommand.h"
 #ifdef ENABLE_MESSAGE_DIGEST
 #include "MessageDigestHelper.h"
 #endif // ENABLE_MESSAGE_DIGEST
@@ -82,8 +83,16 @@ bool DownloadCommand::executeInternal() {
     e->commands.push_back(this);
     return false;
   }
-  int32_t bufSize = 16*1024;
-  char buf[bufSize];
+  SegmentHandle segment = _segments.front();
+
+  int32_t BUFSIZE = 16*1024;
+  char buf[BUFSIZE];
+  int32_t bufSize;
+  if(segment->getLength()-segment->getWrittenLength() < BUFSIZE) {
+    bufSize = segment->getLength()-segment->getWrittenLength();
+  } else {
+    bufSize = BUFSIZE;
+  }
   socket->readData(buf, bufSize);
 
   if(transferDecoder.isNull()) {
@@ -149,7 +158,8 @@ bool DownloadCommand::prepareForNextSegment() {
     return true;
   } else {
     // Merge segment with next segment, if segment.index+1 == nextSegment.index
-    SegmentHandle tempSegment = segment;
+
+    SegmentHandle tempSegment = _segments.front();
     while(1) {
       SegmentHandle nextSegment =
 	_requestGroup->getSegmentMan()->getSegment(cuid,
@@ -165,12 +175,12 @@ bool DownloadCommand::prepareForNextSegment() {
 	  validatePieceHash(nextSegment);
 	  tempSegment = nextSegment;
 	} else {
-	  segment = nextSegment;
 	  e->commands.push_back(this);
 	  return false;
 	}
       }
     }
+
     return prepareForRetry(0);
   }
 }
diff --git a/src/FtpInitiateConnectionCommand.cc b/src/FtpInitiateConnectionCommand.cc
index 466bff9b..eb91a4de 100644
--- a/src/FtpInitiateConnectionCommand.cc
+++ b/src/FtpInitiateConnectionCommand.cc
@@ -45,6 +45,7 @@
 #include "message.h"
 #include "prefs.h"
 #include "Util.h"
+#include "HttpConnection.h"
 
 FtpInitiateConnectionCommand::FtpInitiateConnectionCommand(int cuid,
 							   const RequestHandle& req,
@@ -88,7 +89,7 @@ bool FtpInitiateConnectionCommand::executeInternal() {
 				e->option->getAsInt(PREF_HTTP_PROXY_PORT));
     
     if(useHttpProxyGet()) {
-      command = new HttpRequestCommand(cuid, req, _requestGroup, e, socket);
+      command = new HttpRequestCommand(cuid, req, _requestGroup, new HttpConnection(cuid, socket, e->option), e, socket);
     } else if(useHttpProxyConnect()) {
       command = new FtpTunnelRequestCommand(cuid, req, _requestGroup, e, socket);
     } else {
diff --git a/src/FtpNegotiationCommand.cc b/src/FtpNegotiationCommand.cc
index fed6658a..5e34fc73 100644
--- a/src/FtpNegotiationCommand.cc
+++ b/src/FtpNegotiationCommand.cc
@@ -62,7 +62,7 @@ FtpNegotiationCommand::~FtpNegotiationCommand() {
 }
 
 bool FtpNegotiationCommand::executeInternal() {
-  while(processSequence(segment));
+  while(processSequence(_segments.front()));
   if(sequence == SEQ_RETRY) {
     return prepareForRetry(0);
   } else if(sequence == SEQ_NEGOTIATION_COMPLETED) {
diff --git a/src/HttpConnection.cc b/src/HttpConnection.cc
index 74362823..7fb4cd4f 100644
--- a/src/HttpConnection.cc
+++ b/src/HttpConnection.cc
@@ -110,6 +110,10 @@ HttpResponseHandle HttpConnection::receiveResponse()
   logger->info(MSG_RECEIVE_RESPONSE, cuid, proc->getHeaderString().c_str());
 
   pair<string, HttpHeaderHandle> httpStatusHeader = proc->getHttpStatusHeader();
+  if(Util::toLower(httpStatusHeader.second->getFirst("Connection")).find("close") != string::npos) {
+    entry->getHttpRequest()->getRequest()->setKeepAlive(false);
+  }
+
   HttpResponseHandle httpResponse = new HttpResponse();
   httpResponse->setCuid(cuid);
   httpResponse->setStatus(strtol(httpStatusHeader.first.c_str(), 0, 10));
@@ -120,3 +124,17 @@ HttpResponseHandle HttpConnection::receiveResponse()
 
   return httpResponse;
 }
+
+bool HttpConnection::isIssued(const SegmentHandle& segment) const
+{
+  for(HttpRequestEntries::const_iterator itr = outstandingHttpRequests.begin();
+      itr != outstandingHttpRequests.end(); ++itr) {
+    HttpRequestHandle httpRequest = (*itr)->getHttpRequest();
+    // TODO fix this using operator==
+    if(httpRequest->getSegment().get() == segment.get()) {
+      return true;
+    }
+  }
+  return false;
+}
+
diff --git a/src/HttpConnection.h b/src/HttpConnection.h
index de723838..69b0a15d 100644
--- a/src/HttpConnection.h
+++ b/src/HttpConnection.h
@@ -123,6 +123,8 @@ public:
       return 0;
     }
   }
+
+  bool isIssued(const SegmentHandle& segment) const;
 };
 
 typedef SharedHandle<HttpConnection> HttpConnectionHandle;
diff --git a/src/HttpDownloadCommand.cc b/src/HttpDownloadCommand.cc
index 0993b94c..a0635f13 100644
--- a/src/HttpDownloadCommand.cc
+++ b/src/HttpDownloadCommand.cc
@@ -39,25 +39,24 @@
 #include "HttpRequestCommand.h"
 #include "Util.h"
 #include "message.h"
+#include "HttpConnection.h"
 
 HttpDownloadCommand::HttpDownloadCommand(int cuid,
 					 const RequestHandle req,
 					 RequestGroup* requestGroup,
+					 const HttpConnectionHandle& httpConnection,
 					 DownloadEngine* e,
 					 const SocketHandle& socket)
-  :DownloadCommand(cuid, req, requestGroup, e, socket) {}
+  :DownloadCommand(cuid, req, requestGroup, e, socket),
+   _httpConnection(httpConnection) {}
 
 HttpDownloadCommand::~HttpDownloadCommand() {}
 
 bool HttpDownloadCommand::prepareForNextSegment() {
-  if(!_requestGroup->downloadFinished()) {
-    if(req->isKeepAlive()) {
-      Command* command = new HttpRequestCommand(cuid, req, _requestGroup, e, socket);
-      e->commands.push_back(command);
-      return true;
-    } else {
-      return DownloadCommand::prepareForNextSegment();
-    }
+  if(!_requestGroup->downloadFinished() && req->isKeepAlive()) {
+    Command* command = new HttpRequestCommand(cuid, req, _requestGroup, _httpConnection, e, socket);
+    e->commands.push_back(command);
+    return true;
   } else {
     return DownloadCommand::prepareForNextSegment();
   }
diff --git a/src/HttpDownloadCommand.h b/src/HttpDownloadCommand.h
index 88953831..0d0345b7 100644
--- a/src/HttpDownloadCommand.h
+++ b/src/HttpDownloadCommand.h
@@ -37,13 +37,19 @@
 
 #include "DownloadCommand.h"
 
+class HttpConnection;
+extern typedef SharedHandle<HttpConnection> HttpConnectionHandle;
+
 class HttpDownloadCommand : public DownloadCommand {
+private:
+  HttpConnectionHandle _httpConnection;
 protected:
   virtual bool prepareForNextSegment();
 public:
   HttpDownloadCommand(int cuid,
 		      const RequestHandle req,
 		      RequestGroup* requestGroup,
+		      const HttpConnectionHandle& httpConnection,
 		      DownloadEngine* e,
 		      const SocketHandle& s);
   virtual ~HttpDownloadCommand();
diff --git a/src/HttpInitiateConnectionCommand.cc b/src/HttpInitiateConnectionCommand.cc
index a3680eb2..ece0b01f 100644
--- a/src/HttpInitiateConnectionCommand.cc
+++ b/src/HttpInitiateConnectionCommand.cc
@@ -44,6 +44,7 @@
 #include "DlRetryEx.h"
 #include "message.h"
 #include "prefs.h"
+#include "HttpConnection.h"
 
 HttpInitiateConnectionCommand::HttpInitiateConnectionCommand(int cuid,
 							     const RequestHandle& req,
@@ -88,7 +89,7 @@ bool HttpInitiateConnectionCommand::executeInternal() {
     if(useProxyTunnel()) {
       command = new HttpProxyRequestCommand(cuid, req, _requestGroup, e, socket);
     } else if(useProxyGet()) {
-      command = new HttpRequestCommand(cuid, req, _requestGroup, e, socket);
+      command = new HttpRequestCommand(cuid, req, _requestGroup, new HttpConnection(cuid, socket, e->option), e, socket);
     } else {
       // TODO
       throw new DlAbortEx("ERROR");
@@ -97,7 +98,7 @@ bool HttpInitiateConnectionCommand::executeInternal() {
     logger->info(MSG_CONNECTING_TO_SERVER, cuid, req->getHost().c_str(),
 		 req->getPort());
     socket->establishConnection(hostname, req->getPort());
-    command = new HttpRequestCommand(cuid, req, _requestGroup, e, socket);
+    command = new HttpRequestCommand(cuid, req, _requestGroup, new HttpConnection(cuid, socket, e->option), e, socket);
   }
   e->commands.push_back(command);
   return true;
diff --git a/src/HttpProxyResponseCommand.cc b/src/HttpProxyResponseCommand.cc
index 19edcb13..0c2b438a 100644
--- a/src/HttpProxyResponseCommand.cc
+++ b/src/HttpProxyResponseCommand.cc
@@ -47,5 +47,5 @@ HttpProxyResponseCommand::~HttpProxyResponseCommand() {}
 
 Command* HttpProxyResponseCommand::getNextCommand()
 {
-  return new HttpRequestCommand(cuid, req, _requestGroup, e, socket);
+  return new HttpRequestCommand(cuid, req, _requestGroup, httpConnection, e, socket);
 }
diff --git a/src/HttpRequest.cc b/src/HttpRequest.cc
index 942f6947..633217c4 100644
--- a/src/HttpRequest.cc
+++ b/src/HttpRequest.cc
@@ -101,7 +101,11 @@ string HttpRequest::createRequest() const
     requestLine += "\r\n";
   }
   if(proxyEnabled) {
-    requestLine += "Proxy-Connection: close\r\n";
+    if(request->isKeepAlive()) {
+      requestLine += "Proxy-Connection: Keep-Alive\r\n";
+    } else {
+      requestLine += "Proxy-Connection: close\r\n";
+    }
   }
   if(proxyEnabled && proxyAuthEnabled) {
     requestLine += getProxyAuthString();
@@ -135,8 +139,12 @@ string HttpRequest::createProxyRequest() const
     string("CONNECT ")+getHost()+":"+Util::itos(getPort())+
     string(" HTTP/1.1\r\n")+
     "User-Agent: "+Util::urlencode(userAgent)+"\r\n"+
-    "Proxy-Connection: close\r\n"+
     "Host: "+getHost()+":"+Util::itos(getPort())+"\r\n";
+  if(request->isKeepAlive()) {
+    requestLine += "Proxy-Connection: Keep-Alive\r\n";
+  }else {
+    requestLine += "Proxy-Connection: close\r\n";
+  }
   if(proxyAuthEnabled) {
     requestLine += getProxyAuthString();
   }
diff --git a/src/HttpRequestCommand.cc b/src/HttpRequestCommand.cc
index b5559715..f61ca2a5 100644
--- a/src/HttpRequestCommand.cc
+++ b/src/HttpRequestCommand.cc
@@ -38,13 +38,17 @@
 #include "HttpResponseCommand.h"
 #include "HttpConnection.h"
 #include "prefs.h"
+#include "SegmentMan.h"
 
 HttpRequestCommand::HttpRequestCommand(int cuid,
 				       const RequestHandle& req,
 				       RequestGroup* requestGroup,
+				       const HttpConnectionHandle& httpConnection,
 				       DownloadEngine* e,
 				       const SocketHandle& s)
-  :AbstractCommand(cuid, req, requestGroup, e, s) {
+  :AbstractCommand(cuid, req, requestGroup, e, s),
+   _httpConnection(httpConnection)
+{
   disableReadCheckSocket();
   setWriteCheckSocket(socket);
 }
@@ -56,21 +60,43 @@ bool HttpRequestCommand::executeInternal() {
   if(req->getProtocol() == "https") {
     socket->initiateSecureConnection();
   }
-  if(!e->option->getAsBool(PREF_HTTP_KEEP_ALIVE)) {
+  if(e->option->get(PREF_ENABLE_HTTP_PIPELINING) == V_TRUE) {
+    req->setKeepAlive(true);
+  } else if(e->option->get(PREF_ENABLE_HTTP_KEEP_ALIVE) == V_TRUE &&
+	    !_requestGroup->getSegmentMan().isNull() &&
+	    _requestGroup->getSegmentMan()->countFreePieceFrom(_segments.front()->getIndex()+1) <= 4) {
+    // TODO Do we need to consider the case where content-length is unknown?
+    // TODO parameterize the value which enables keep-alive, '4'
+    req->setKeepAlive(true);
+  } else {
     req->setKeepAlive(false);
   }
-  HttpRequestHandle httpRequest = new HttpRequest();
-  httpRequest->setUserAgent(e->option->get(PREF_USER_AGENT));
-  httpRequest->setRequest(req);
-  httpRequest->setSegment(segment);
-  httpRequest->setEntityLength(_requestGroup->getTotalLength());
-  httpRequest->configure(e->option);
 
-  HttpConnectionHandle httpConnection = new HttpConnection(cuid, socket, e->option);
+  if(_segments.empty()) {
+    HttpRequestHandle httpRequest = new HttpRequest();
+    httpRequest->setUserAgent(e->option->get(PREF_USER_AGENT));
+    httpRequest->setRequest(req);
+    httpRequest->setSegment(0);
+    httpRequest->setEntityLength(_requestGroup->getTotalLength());
+    httpRequest->configure(e->option);
+    
+    _httpConnection->sendRequest(httpRequest);
+  } else {
+    for(Segments::iterator itr = _segments.begin(); itr != _segments.end(); ++itr) {
+      SegmentHandle segment = *itr;
+      if(!_httpConnection->isIssued(segment)) {
+	HttpRequestHandle httpRequest = new HttpRequest();
+	httpRequest->setUserAgent(e->option->get(PREF_USER_AGENT));
+	httpRequest->setRequest(req);
+	httpRequest->setSegment(segment);
+	httpRequest->setEntityLength(_requestGroup->getTotalLength());
+	httpRequest->configure(e->option);
 
-  httpConnection->sendRequest(httpRequest);
-
-  Command* command = new HttpResponseCommand(cuid, req, _requestGroup, httpConnection, e, socket);
+	_httpConnection->sendRequest(httpRequest);
+      }
+    }
+  }
+  Command* command = new HttpResponseCommand(cuid, req, _requestGroup, _httpConnection, e, socket);
   e->commands.push_back(command);
   return true;
 }
diff --git a/src/HttpRequestCommand.h b/src/HttpRequestCommand.h
index 3d2d3d2f..00322ba1 100644
--- a/src/HttpRequestCommand.h
+++ b/src/HttpRequestCommand.h
@@ -37,13 +37,19 @@
 
 #include "AbstractCommand.h"
 
+class HttpConnection;
+extern typedef SharedHandle<HttpConnection> HttpConnectionHandle;
+
 class HttpRequestCommand:public AbstractCommand {
+private:
+  HttpConnectionHandle _httpConnection;
 protected:
   virtual bool executeInternal();
 public:
   HttpRequestCommand(int cuid,
 		     const RequestHandle& req,
 		     RequestGroup* requestGroup,
+		     const HttpConnectionHandle& httpConnection,
 		     DownloadEngine* e,
 		     const SocketHandle& s);
   virtual ~HttpRequestCommand();
diff --git a/src/HttpResponseCommand.cc b/src/HttpResponseCommand.cc
index 95dae918..9bc35758 100644
--- a/src/HttpResponseCommand.cc
+++ b/src/HttpResponseCommand.cc
@@ -122,6 +122,7 @@ bool HttpResponseCommand::handleDefaultEncoding(const HttpResponseHandle& httpRe
   //_requestGroup->getSegmentMan()->initDownloadContext(size);
 
   SingleFileDownloadContextHandle(_requestGroup->getDownloadContext())->setTotalLength(size);
+
   initPieceStorage();
 
   // quick hack for method 'head'
@@ -189,7 +190,7 @@ HttpDownloadCommand* HttpResponseCommand::createHttpDownloadCommand(const HttpRe
     enc->init();
   }
   HttpDownloadCommand* command =
-    new HttpDownloadCommand(cuid, req, _requestGroup, e, socket);
+    new HttpDownloadCommand(cuid, req, _requestGroup, httpConnection, e, socket);
   command->setMaxDownloadSpeedLimit(e->option->getAsInt(PREF_MAX_DOWNLOAD_LIMIT));
   command->setStartupIdleTime(e->option->getAsInt(PREF_STARTUP_IDLE_TIME));
   command->setLowestDownloadSpeedLimit(e->option->getAsInt(PREF_LOWEST_SPEED_LIMIT));
diff --git a/src/OptionHandlerFactory.cc b/src/OptionHandlerFactory.cc
index 1f3634bc..d264bbf8 100644
--- a/src/OptionHandlerFactory.cc
+++ b/src/OptionHandlerFactory.cc
@@ -97,6 +97,8 @@ OptionHandlers OptionHandlerFactory::createOptionHandlers()
   handlers.push_back(new BooleanOptionHandler(PREF_FORCE_SEQUENTIAL));  
   handlers.push_back(new BooleanOptionHandler(PREF_AUTO_FILE_RENAMING));  
   handlers.push_back(new BooleanOptionHandler(PREF_PARAMETERIZED_URI));  
+  handlers.push_back(new BooleanOptionHandler(PREF_ENABLE_HTTP_KEEP_ALIVE));
+  handlers.push_back(new BooleanOptionHandler(PREF_ENABLE_HTTP_PIPELINING));
 
   return handlers;
 }
diff --git a/src/Request.cc b/src/Request.cc
index fd40eede..132d5c9e 100644
--- a/src/Request.cc
+++ b/src/Request.cc
@@ -61,6 +61,7 @@ bool Request::resetUrl() {
 
 bool Request::redirectUrl(const string& url) {
   previousUrl = "";
+  keepAlive = false;
   return parseUrl(url);
 }
 
diff --git a/src/Segment.h b/src/Segment.h
index ef778de1..c6df8957 100644
--- a/src/Segment.h
+++ b/src/Segment.h
@@ -68,6 +68,7 @@ public:
 };
 
 typedef SharedHandle<Segment> SegmentHandle;
+typedef deque<SegmentHandle> Segments;
 
 #endif // _D_SEGMENT_H_
 
diff --git a/src/SegmentMan.cc b/src/SegmentMan.cc
index cef73132..7fe56caa 100644
--- a/src/SegmentMan.cc
+++ b/src/SegmentMan.cc
@@ -148,11 +148,20 @@ SegmentEntryHandle SegmentMan::findSlowerSegmentEntry(const PeerStatHandle& peer
   return slowSegmentEntry;
 }
 
-SegmentHandle SegmentMan::getSegment(int32_t cuid) {
-  SegmentEntryHandle segmentEntry = getSegmentEntryByCuid(cuid);
-  if(!segmentEntry.isNull()) {
-    return segmentEntry->segment;
+Segments SegmentMan::getInFlightSegment(int32_t cuid)
+{
+  Segments temp;
+  for(SegmentEntries::iterator itr = usedSegmentEntries.begin();
+      itr != usedSegmentEntries.end(); ++itr) {
+    const SegmentEntryHandle& segmentEntry = *itr;
+    if(segmentEntry->cuid == cuid) {
+      temp.push_back(segmentEntry->segment);
+    }
   }
+  return temp;
+}
+
+SegmentHandle SegmentMan::getSegment(int32_t cuid) {
   PieceHandle piece = _pieceStorage->getMissingPiece();
   if(piece.isNull()) {
     PeerStatHandle myPeerStat = getPeerStat(cuid);
@@ -186,11 +195,12 @@ SegmentHandle SegmentMan::getSegment(int32_t cuid, int32_t index) {
 
 void SegmentMan::cancelSegment(int32_t cuid) {
   for(SegmentEntries::iterator itr = usedSegmentEntries.begin();
-      itr != usedSegmentEntries.end(); ++itr) {
+      itr != usedSegmentEntries.end();) {
     if((*itr)->cuid == cuid) {
       _pieceStorage->cancelPiece((*itr)->segment->getPiece());
-      usedSegmentEntries.erase(itr);
-      break;
+      itr = usedSegmentEntries.erase(itr);
+    } else {
+      ++itr;
     }
   }
 }
@@ -314,3 +324,14 @@ SegmentEntries::iterator SegmentMan::getSegmentEntryIteratorByCuid(int32_t cuid)
   }
   return usedSegmentEntries.end();    
 }
+
+int32_t SegmentMan::countFreePieceFrom(int32_t index) const
+{
+  for(int32_t i = index; i < _downloadContext->getNumPieces(); ++i) {
+    if(_pieceStorage->hasPiece(i) || _pieceStorage->isPieceUsed(i)) {
+      return i-index;
+    }
+  }
+  return _downloadContext->getNumPieces()-index;
+}
+
diff --git a/src/SegmentMan.h b/src/SegmentMan.h
index d0dcecc6..433ebe60 100644
--- a/src/SegmentMan.h
+++ b/src/SegmentMan.h
@@ -39,6 +39,7 @@
 
 class Segment;
 extern typedef SharedHandle<Segment> SegmentHandle;
+extern typedef deque<SegmentHandle> Segments;
 class Logger;
 class Option;
 class PeerStat;
@@ -124,6 +125,7 @@ public:
    * If there is no vacant segment, then returns a segment instance whose
    * isNull call is true.
    */
+  Segments getInFlightSegment(int32_t cuid);
   SegmentHandle getSegment(int32_t cuid);
   /**
    * Returns a segment whose index is index. 
@@ -185,6 +187,8 @@ public:
   void markAllPiecesDone();
 
   void markPieceDone(int64_t length);
+
+  int32_t countFreePieceFrom(int32_t index) const;
 };
 
 typedef SharedHandle<SegmentMan> SegmentManHandle;
diff --git a/src/option_processing.cc b/src/option_processing.cc
index 92c4b2f0..bb8067d3 100644
--- a/src/option_processing.cc
+++ b/src/option_processing.cc
@@ -70,7 +70,6 @@ Option* option_processing(int argc, char* const argv[])
   op->put(PREF_SPLIT, "1");
   op->put(PREF_DAEMON, V_FALSE);
   op->put(PREF_SEGMENT_SIZE, Util::itos((int32_t)(1024*1024)));
-  op->put(PREF_HTTP_KEEP_ALIVE, V_FALSE);
   op->put(PREF_LISTEN_PORT, "-1");
   op->put(PREF_METALINK_SERVERS, "5");
   op->put(PREF_FOLLOW_TORRENT,
@@ -120,6 +119,9 @@ Option* option_processing(int argc, char* const argv[])
   op->put(PREF_FORCE_SEQUENTIAL, V_FALSE);
   op->put(PREF_AUTO_FILE_RENAMING, V_TRUE);
   op->put(PREF_PARAMETERIZED_URI, V_FALSE);
+  op->put(PREF_ENABLE_HTTP_KEEP_ALIVE, V_FALSE);
+  op->put(PREF_ENABLE_HTTP_PIPELINING, V_FALSE);
+  op->put(PREF_MAX_HTTP_PIPELINING, "2");
   while(1) {
     int optIndex = 0;
     int lopt;
@@ -165,6 +167,8 @@ Option* option_processing(int argc, char* const argv[])
       { "force-sequential", optional_argument, 0, 'Z' },
       { "auto-file-renaming", optional_argument, &lopt, 206 },
       { "parameterized-uri", optional_argument, 0, 'P' },
+      { "enable-http-keep-alive", optional_argument, &lopt, 207 },
+      { "enable-http-pipelining", optional_argument, &lopt, 208 },
 #if defined ENABLE_BITTORRENT || ENABLE_METALINK
       { "show-files", no_argument, NULL, 'S' },
       { "select-file", required_argument, &lopt, 21 },
@@ -306,6 +310,12 @@ Option* option_processing(int argc, char* const argv[])
       case 206:
 	cmdstream << PREF_AUTO_FILE_RENAMING << "=" << toBoolArg(optarg) << "\n";
 	break;
+      case 207:
+	cmdstream << PREF_ENABLE_HTTP_KEEP_ALIVE << "=" << toBoolArg(optarg) << "\n";
+	break;
+      case 208:
+	cmdstream << PREF_ENABLE_HTTP_PIPELINING << "=" << toBoolArg(optarg) << "\n";
+	break;
       }
       break;
     }
diff --git a/src/prefs.h b/src/prefs.h
index 99e5efc8..b9114f6f 100644
--- a/src/prefs.h
+++ b/src/prefs.h
@@ -140,12 +140,16 @@
 #  define V_BASIC "basic"
 // values: true | false
 #define PREF_HTTP_AUTH_ENABLED "http-auth-enabled"
-// values: true | false
-#define PREF_HTTP_KEEP_ALIVE "http-keep-alive"
 // values: string
 #define PREF_USER_AGENT "user-agent"
 // value: string that your file system recognizes as a file name.
 #define PREF_LOAD_COOKIES "load-cookies"
+// values: true | false
+#define PREF_ENABLE_HTTP_KEEP_ALIVE "enable-http-keep-alive"
+// values: true | false
+#define PREF_ENABLE_HTTP_PIPELINING "enable-http-pipelining"
+// value: 1*digit
+#define PREF_MAX_HTTP_PIPELINING "max-http-pipelining"
 
 /** 
  * HTTP proxy related preferences
diff --git a/src/version_usage.cc b/src/version_usage.cc
index b246e720..d67086b9 100644
--- a/src/version_usage.cc
+++ b/src/version_usage.cc
@@ -166,6 +166,10 @@ void showUsage() {
 	    "                              as the second example above, -Z option is\n"
 	    "                              required.\n"
 	    "                              Default: false") << endl;
+  cout << _(" --enable-http-keep-alive[=true|false] Enable HTTP/1.1 persistant connection.\n"
+	    "                              Default: false") << endl;
+  cout << _(" --enable-http-pipelining[=true|false] Enable HTTP/1.1 pipelining.\n"
+	    "                              Default: false") << endl;
 #ifdef ENABLE_MESSAGE_DIGEST
   cout << _(" --check-integrity=true|false  Check file integrity by validating piece hash.\n"
 	    "                              This option only affects in BitTorrent downloads\n"