2007-11-28 Tatsuhiro Tsujikawa <tujikawa at rednoah dot com>

	Added direct I/O support. The current implementation uses O_DIRECT,
	which is not posix standard and is tested on linux 2.6.21.
	Currently only file allocation uses direct I/O.
	* src/SingleFileAllocationIterator.{h, cc}
	* test/SingleFileAllocationIteratorTest.cc
	* src/MultiFileAllocationIterator.{h, cc}
	* test/MultiFileAllocationIteratorTest.cc
	* src/BinaryStream.h
	* src/DiskWriter.h
	* src/AbstractDiskWriter.{h, cc}
	* src/ByteArrayDiskWriter.h
	* src/DiskAdaptor.h
	* src/AbstractSingleDiskAdaptor.{h, cc}
	* src/MultiDiskAdaptor.{h, cc}
	* src/FileAllocationEntry.cc
	* src/Util.{h, cc}
	* src/OptionHandlerFactory.cc
	* src/prefs.h
	* src/version_usage.cc
	* src/option_processing.cc

	Moved FileAllocationMan::markCurrentFileAllocationEntryDone() to
	handleException.
	* src/MultiFileAllocationIterator.cc

	Added EINTR handling
	* src/SocketCore.cc

26 files changed, 262 insertions, 114 deletions

diff --git a/src/AbstractDiskWriter.cc b/src/AbstractDiskWriter.cc
index d27c3cab..5df2292e 100644
--- a/src/AbstractDiskWriter.cc
+++ b/src/AbstractDiskWriter.cc
@@ -163,3 +163,21 @@ int64_t AbstractDiskWriter::size() const
   }
   return fileStat.st_size;
 }
+
+void AbstractDiskWriter::enableDirectIO()
+{
+#ifdef ENABLE_DIRECT_IO
+  int32_t flg;
+  while((flg = fcntl(fd, F_GETFL)) == -1 && errno == EINTR);
+  while(fcntl(fd, F_SETFL, flg|O_DIRECT) == -1 && errno == EINTR);
+#endif // ENABLE_DIRECT_IO
+}
+
+void AbstractDiskWriter::disableDirectIO()
+{
+#ifdef ENABLE_DIRECT_IO
+  int32_t flg;
+  while((flg = fcntl(fd, F_GETFL)) == -1 && errno == EINTR);
+  while(fcntl(fd, F_SETFL, flg&(~O_DIRECT)) == -1 && errno == EINTR);
+#endif // ENABLE_DIRECT_IO
+}
diff --git a/src/AbstractDiskWriter.h b/src/AbstractDiskWriter.h
index a69ef3c0..8971484b 100644
--- a/src/AbstractDiskWriter.h
+++ b/src/AbstractDiskWriter.h
@@ -69,6 +69,10 @@ public:
   virtual void truncate(int64_t length);
 
   virtual int64_t size() const;
+  
+  virtual void enableDirectIO();
+
+  virtual void disableDirectIO();
 };
 
 #endif // _D_ABSTRACT_DISK_WRITER_H_
diff --git a/src/AbstractSingleDiskAdaptor.cc b/src/AbstractSingleDiskAdaptor.cc
index 490525ea..bb9efe0f 100644
--- a/src/AbstractSingleDiskAdaptor.cc
+++ b/src/AbstractSingleDiskAdaptor.cc
@@ -73,5 +73,20 @@ bool AbstractSingleDiskAdaptor::fileExists()
 
 FileAllocationIteratorHandle AbstractSingleDiskAdaptor::fileAllocationIterator()
 {
-  return new SingleFileAllocationIterator(this);
+  SingleFileAllocationIteratorHandle h =
+    new SingleFileAllocationIterator(this,
+				     size(),
+				     totalLength);
+  h->init();
+  return h;
+}
+
+void AbstractSingleDiskAdaptor::enableDirectIO()
+{
+  diskWriter->enableDirectIO();
+}
+
+void AbstractSingleDiskAdaptor::disableDirectIO()
+{
+  diskWriter->disableDirectIO();
 }
diff --git a/src/AbstractSingleDiskAdaptor.h b/src/AbstractSingleDiskAdaptor.h
index 4fcedc61..c7b2e089 100644
--- a/src/AbstractSingleDiskAdaptor.h
+++ b/src/AbstractSingleDiskAdaptor.h
@@ -74,6 +74,10 @@ public:
   
   virtual FileAllocationIteratorHandle fileAllocationIterator();
 
+  virtual void enableDirectIO();
+
+  virtual void disableDirectIO();
+
   void setDiskWriter(const DiskWriterHandle diskWriter) {
     this->diskWriter = diskWriter;
   }
diff --git a/src/BinaryStream.h b/src/BinaryStream.h
index 1d11d8e5..d16ac728 100644
--- a/src/BinaryStream.h
+++ b/src/BinaryStream.h
@@ -45,6 +45,8 @@ public:
   virtual void writeData(const unsigned char* data, int32_t len, int64_t offset) = 0;
 
   virtual int32_t readData(unsigned char* data, int32_t len, int64_t offset) = 0;
+
+  virtual void truncate(int64_t length) = 0;
 };
 
 typedef SharedHandle<BinaryStream> BinaryStreamHandle;
diff --git a/src/ByteArrayDiskWriter.h b/src/ByteArrayDiskWriter.h
index f35b4366..47d99946 100644
--- a/src/ByteArrayDiskWriter.h
+++ b/src/ByteArrayDiskWriter.h
@@ -65,6 +65,10 @@ public:
   {
     return buf.str().size();
   }
+
+  virtual void enableDirectIO() {}
+
+  virtual void disableDirectIO() {}
 };
 
 #endif // _D_BYTE_ARRAY_DISK_WRITER_H_
diff --git a/src/DefaultPieceStorage.cc b/src/DefaultPieceStorage.cc
index 0189d47e..c652426c 100644
--- a/src/DefaultPieceStorage.cc
+++ b/src/DefaultPieceStorage.cc
@@ -394,6 +394,7 @@ void DefaultPieceStorage::setFileFilter(const Strings& filePaths)
 void DefaultPieceStorage::setFileFilter(IntSequence seq)
 {
   Integers fileIndexes = seq.flush();
+  // TODO Is sorting necessary?
   sort(fileIndexes.begin(), fileIndexes.end());
   fileIndexes.erase(unique(fileIndexes.begin(), fileIndexes.end()), fileIndexes.end());
   Strings filePaths;
diff --git a/src/DiskAdaptor.h b/src/DiskAdaptor.h
index ab690955..3cb7467e 100644
--- a/src/DiskAdaptor.h
+++ b/src/DiskAdaptor.h
@@ -89,6 +89,10 @@ public:
   const string& getStoreDir() const { return this->storeDir; }
 
   virtual FileAllocationIteratorHandle fileAllocationIterator() = 0;
+
+  virtual void enableDirectIO() {};
+
+  virtual void disableDirectIO() {};
 };
 
 typedef SharedHandle<DiskAdaptor> DiskAdaptorHandle;
diff --git a/src/DiskWriter.h b/src/DiskWriter.h
index 3d1bb69c..80484175 100644
--- a/src/DiskWriter.h
+++ b/src/DiskWriter.h
@@ -71,6 +71,10 @@ public:
 
   // Returns file length
   virtual int64_t size() const = 0;
+
+  virtual void enableDirectIO() = 0;
+
+  virtual void disableDirectIO() = 0;
 };
 
 typedef SharedHandle<DiskWriter> DiskWriterHandle;
diff --git a/src/FileAllocationCommand.cc b/src/FileAllocationCommand.cc
index 01a24d65..16f3c697 100644
--- a/src/FileAllocationCommand.cc
+++ b/src/FileAllocationCommand.cc
@@ -51,29 +51,25 @@ bool FileAllocationCommand::executeInternal()
   if(_e->isHaltRequested()) {
     return true;
   }
-  try {
-    _fileAllocationEntry->allocateChunk();
-    if(_fileAllocationEntry->finished()) {
-      logger->debug(MSG_ALLOCATION_COMPLETED,
-		    _timer.difference(),
-		    Util::llitos(_requestGroup->getTotalLength(), true).c_str());
-      _e->_fileAllocationMan->markCurrentFileAllocationEntryDone();
-      
-      _e->addCommand(_fileAllocationEntry->prepareForNextAction(_e));
-      
-      return true;
-    } else {
-      _e->commands.push_back(this);
-      return false;
-    }
-  } catch(Exception* e) {
+  _fileAllocationEntry->allocateChunk();
+  if(_fileAllocationEntry->finished()) {
+    logger->debug(MSG_ALLOCATION_COMPLETED,
+		  _timer.difference(),
+		  Util::llitos(_requestGroup->getTotalLength(), true).c_str());
     _e->_fileAllocationMan->markCurrentFileAllocationEntryDone();
-    throw;
+    
+    _e->addCommand(_fileAllocationEntry->prepareForNextAction(_e));
+    
+    return true;
+  } else {
+    _e->commands.push_back(this);
+    return false;
   }
 }
 
 bool FileAllocationCommand::handleException(Exception* e)
 {
+  _e->_fileAllocationMan->markCurrentFileAllocationEntryDone();
   logger->error(MSG_FILE_ALLOCATION_FAILURE, e, cuid);
   logger->error(MSG_DOWNLOAD_NOT_COMPLETE, cuid, _requestGroup->getFilePath().c_str());
   return true;
diff --git a/src/FileAllocationEntry.cc b/src/FileAllocationEntry.cc
index af9c67c1..5bc39ee9 100644
--- a/src/FileAllocationEntry.cc
+++ b/src/FileAllocationEntry.cc
@@ -38,13 +38,24 @@
 #include "RequestGroup.h"
 #include "PieceStorage.h"
 #include "DiskAdaptor.h"
+#include "prefs.h"
+#include "Option.h"
 
 FileAllocationEntry::FileAllocationEntry(RequestGroup* requestGroup, Command* nextCommand):
   RequestGroupEntry(requestGroup, nextCommand),
   _fileAllocationIterator(requestGroup->getPieceStorage()->getDiskAdaptor()->fileAllocationIterator())
-{}
+{
+  if(_requestGroup->getOption()->getAsBool(PREF_ENABLE_DIRECT_IO)) {
+    _requestGroup->getPieceStorage()->getDiskAdaptor()->enableDirectIO();
+  }
+}
 
-FileAllocationEntry:: ~FileAllocationEntry() {}
+FileAllocationEntry:: ~FileAllocationEntry()
+{
+  if(_requestGroup->getOption()->getAsBool(PREF_ENABLE_DIRECT_IO)) {
+    _requestGroup->getPieceStorage()->getDiskAdaptor()->disableDirectIO();
+  }
+}
 
 int64_t FileAllocationEntry::getCurrentLength()
 {
diff --git a/src/MultiDiskAdaptor.cc b/src/MultiDiskAdaptor.cc
index d79c21a7..87aa3e1a 100644
--- a/src/MultiDiskAdaptor.cc
+++ b/src/MultiDiskAdaptor.cc
@@ -207,3 +207,19 @@ FileAllocationIteratorHandle MultiDiskAdaptor::fileAllocationIterator()
 {
   return new MultiFileAllocationIterator(this);
 }
+
+void MultiDiskAdaptor::enableDirectIO()
+{
+  for(DiskWriterEntries::const_iterator itr = diskWriterEntries.begin();
+      itr != diskWriterEntries.end(); ++itr) {
+    (*itr)->getDiskWriter()->enableDirectIO();
+  }
+}
+
+void MultiDiskAdaptor::disableDirectIO()
+{
+  for(DiskWriterEntries::const_iterator itr = diskWriterEntries.begin();
+      itr != diskWriterEntries.end(); ++itr) {
+    (*itr)->getDiskWriter()->disableDirectIO();
+  }
+}
diff --git a/src/MultiDiskAdaptor.h b/src/MultiDiskAdaptor.h
index db9699a9..e5d40291 100644
--- a/src/MultiDiskAdaptor.h
+++ b/src/MultiDiskAdaptor.h
@@ -39,6 +39,8 @@
 #include "DiskWriter.h"
 #include "File.h"
 
+class MultiFileAllocationIterator;
+
 class DiskWriterEntry {
 private:
   FileEntryHandle fileEntry;
@@ -96,6 +98,11 @@ public:
   DiskWriterHandle getDiskWriter() const {
     return diskWriter;
   }
+
+  bool operator<(const DiskWriterEntry& entry) const
+  {
+    return fileEntry < entry.fileEntry;
+  }
 };
 
 typedef SharedHandle<DiskWriterEntry> DiskWriterEntryHandle;
@@ -103,6 +110,7 @@ typedef SharedHandle<DiskWriterEntry> DiskWriterEntryHandle;
 typedef deque<DiskWriterEntryHandle> DiskWriterEntries;
 
 class MultiDiskAdaptor : public DiskAdaptor {
+  friend class MultiFileAllocationIterator;
 private:
   string topDir;
   int32_t pieceLength;
@@ -151,6 +159,10 @@ public:
 
   virtual FileAllocationIteratorHandle fileAllocationIterator();
 
+  virtual void enableDirectIO();
+
+  virtual void disableDirectIO();
+
   void setTopDir(const string& topDir) {
     this->topDir = topDir;
   }
diff --git a/src/MultiFileAllocationIterator.cc b/src/MultiFileAllocationIterator.cc
index 7d8cdfcb..04961ae6 100644
--- a/src/MultiFileAllocationIterator.cc
+++ b/src/MultiFileAllocationIterator.cc
@@ -35,96 +35,92 @@
 #include "MultiFileAllocationIterator.h"
 #include "MultiDiskAdaptor.h"
 #include "FileEntry.h"
-
-#define BUFSIZE 16*1024
+#include "SingleFileAllocationIterator.h"
 
 MultiFileAllocationIterator::MultiFileAllocationIterator(MultiDiskAdaptor* diskAdaptor):
   _diskAdaptor(diskAdaptor),
-  _entries(makeFileEntries(diskAdaptor->getFileEntries(), diskAdaptor->getPieceLength())),
-  _currentEntry(0),
+  _entries(makeDiskWriterEntries(diskAdaptor->diskWriterEntries,
+				 diskAdaptor->getPieceLength())),
+  _fileAllocationIterator(0),
   _offset(0)
 {}
 
 MultiFileAllocationIterator::~MultiFileAllocationIterator() {}
 
-void MultiFileAllocationIterator::prepareNextEntry()
-{
-  _currentEntry = 0;
-  _offset = 0;
-  if(!_entries.empty()) {
-    FileEntryHandle entry = _entries.front();
-    _entries.pop_front();
-
-    _currentEntry = entry;
-    _offset = File(_diskAdaptor->getStoreDir()+"/"+
-		   _diskAdaptor->getTopDir()+"/"+
-		   _currentEntry->getPath()).size();
-  }
-}
-
-
 void MultiFileAllocationIterator::allocateChunk()
 {
-  while(_currentEntry.isNull() || _currentEntry->getLength() <= _offset) {
-    prepareNextEntry();
-    if(_currentEntry.isNull()) {
+  while(_fileAllocationIterator.isNull() || _fileAllocationIterator->finished()) {
+    if(_entries.empty()) {
       break;
     }
+    DiskWriterEntryHandle entry = _entries.front();
+    _entries.pop_front();
+    FileEntryHandle fileEntry = entry->getFileEntry();
+    if(entry->size() < fileEntry->getLength()) {
+      _fileAllocationIterator =
+	new SingleFileAllocationIterator(entry->getDiskWriter().get(),
+					 entry->size(),
+					 fileEntry->getLength());
+      _fileAllocationIterator->init();
+    }
   }
   if(finished()) {
     return;
   }
-  int32_t bufSize = BUFSIZE;
-  unsigned char buf[BUFSIZE];
-  memset(buf, 0, bufSize);
-  
-  int32_t wsize = _offset+bufSize > _currentEntry->getLength() ?
-    _currentEntry->getLength()-_offset:bufSize;
-  _diskAdaptor->writeData(buf, wsize, _offset+_currentEntry->getOffset());
-  _offset += wsize;  
+  _fileAllocationIterator->allocateChunk();
 }
 
 bool MultiFileAllocationIterator::finished()
 {
-  return _entries.empty() && _currentEntry.isNull();
+  return _entries.empty() && (_fileAllocationIterator.isNull() || _fileAllocationIterator->finished());
+}
+
+int64_t MultiFileAllocationIterator::getCurrentLength()
+{
+  if(_fileAllocationIterator.isNull()) {
+    return 0;
+  } else {
+    return _fileAllocationIterator->getCurrentLength();
+  }
 }
 
 int64_t MultiFileAllocationIterator::getTotalLength()
 {
-  if(_currentEntry.isNull()) {
+  if(_fileAllocationIterator.isNull()) {
     return 0;
   } else {
-    return _currentEntry->getLength();
+    return _fileAllocationIterator->getTotalLength();
   }
 }
 
-const FileEntries& MultiFileAllocationIterator::getFileEntries() const
+const DiskWriterEntries& MultiFileAllocationIterator::getDiskWriterEntries() const
 {
   return _entries;
 }
 
-FileEntries MultiFileAllocationIterator::makeFileEntries(const FileEntries& srcEntries, int32_t pieceLength) const
+DiskWriterEntries MultiFileAllocationIterator::makeDiskWriterEntries(const DiskWriterEntries& srcEntries, int32_t pieceLength) const
 {
   if(pieceLength == 0) {
-    FileEntries entries;
-    for(FileEntries::const_iterator itr = srcEntries.begin(); itr != srcEntries.end(); ++itr) {
-      if((*itr)->isRequested()) {
+    DiskWriterEntries entries;
+    for(DiskWriterEntries::const_iterator itr = srcEntries.begin(); itr != srcEntries.end(); ++itr) {
+      if((*itr)->getFileEntry()->isRequested()) {
 	entries.push_back(*itr);
       }
     }
     return entries;
   }
-  FileEntries temp(srcEntries);
-  temp.push_front(new FileEntry());
-  FileEntries entries;
-  FileEntries::const_iterator done = temp.begin();
-  for(FileEntries::const_iterator itr = temp.begin()+1; itr != temp.end(); ++itr) {
-    if(!(*itr)->isRequested()) {
+  DiskWriterEntries temp(srcEntries);
+  temp.push_front(new DiskWriterEntry(new FileEntry()));
+  DiskWriterEntries entries;
+  DiskWriterEntries::const_iterator done = temp.begin();
+  for(DiskWriterEntries::const_iterator itr = temp.begin()+1; itr != temp.end(); ++itr) {
+    FileEntryHandle fileEntry = (*itr)->getFileEntry();
+    if(!fileEntry->isRequested()) {
       continue;
     }
-    int64_t pieceStartOffset = ((*itr)->getOffset()/pieceLength)*pieceLength;
-    for(FileEntries::const_iterator i = itr-1; i != done; --i) {
-      if(pieceStartOffset < (*i)->getOffset()+(*i)->getLength()) {
+    int64_t pieceStartOffset = (fileEntry->getOffset()/pieceLength)*pieceLength;
+    for(DiskWriterEntries::const_iterator i = itr-1; i != done; --i) {
+      if(pieceStartOffset < (*i)->getFileEntry()->getOffset()+(*i)->getFileEntry()->getLength()) {
 	entries.push_back(*i);
       } else {
 	break;
diff --git a/src/MultiFileAllocationIterator.h b/src/MultiFileAllocationIterator.h
index bbed59b6..a9b4aa8a 100644
--- a/src/MultiFileAllocationIterator.h
+++ b/src/MultiFileAllocationIterator.h
@@ -38,38 +38,35 @@
 #include "FileAllocationIterator.h"
 
 class MultiDiskAdaptor;
-class FileEntry;
-typedef SharedHandle<FileEntry> FileEntryHandle;
-typedef deque<FileEntryHandle> FileEntries;
+class DiskWriterEntry;
+typedef SharedHandle<DiskWriterEntry> DiskWriterEntryHandle;
+typedef deque<DiskWriterEntryHandle> DiskWriterEntries;
+class SingleFileAllocationIterator;
+typedef SharedHandle<SingleFileAllocationIterator> SingleFileAllocationIteratorHandle;
 
 class MultiFileAllocationIterator:public FileAllocationIterator
 {
 private:
   MultiDiskAdaptor* _diskAdaptor;
-  FileEntries _entries;
-  FileEntryHandle _currentEntry;
+  DiskWriterEntries _entries;
+  SingleFileAllocationIteratorHandle _fileAllocationIterator;
   int64_t _offset;
 
-  FileEntries makeFileEntries(const FileEntries& srcEntries, int32_t pieceLength) const;
+  DiskWriterEntries makeDiskWriterEntries(const DiskWriterEntries& srcEntries, int32_t pieceLength) const;
 public:
   MultiFileAllocationIterator(MultiDiskAdaptor* diskAdaptor);
 
   virtual ~MultiFileAllocationIterator();
 
-  void prepareNextEntry();
-
   virtual void allocateChunk();
   
   virtual bool finished();
 
-  virtual int64_t getCurrentLength()
-  {
-    return _offset;
-  }
+  virtual int64_t getCurrentLength();
 
   virtual int64_t getTotalLength();
 
-  const FileEntries& getFileEntries() const;
+  const DiskWriterEntries& getDiskWriterEntries() const;
 };
 
 typedef SharedHandle<MultiFileAllocationIterator> MultiFileAllocationIteratorHandle;
diff --git a/src/OptionHandlerFactory.cc b/src/OptionHandlerFactory.cc
index aa4b7294..51e96612 100644
--- a/src/OptionHandlerFactory.cc
+++ b/src/OptionHandlerFactory.cc
@@ -100,6 +100,7 @@ OptionHandlers OptionHandlerFactory::createOptionHandlers()
   handlers.push_back(new BooleanOptionHandler(PREF_ENABLE_HTTP_KEEP_ALIVE));
   handlers.push_back(new BooleanOptionHandler(PREF_ENABLE_HTTP_PIPELINING));
   handlers.push_back(new UnitNumberOptionHandler(PREF_NO_FILE_ALLOCATION_LIMIT, 0));
+  handlers.push_back(new BooleanOptionHandler(PREF_ENABLE_DIRECT_IO));
 
   return handlers;
 }
diff --git a/src/RequestGroup.cc b/src/RequestGroup.cc
index 2ae83294..03cb791a 100644
--- a/src/RequestGroup.cc
+++ b/src/RequestGroup.cc
@@ -199,7 +199,7 @@ Commands RequestGroup::createInitialCommand(DownloadEngine* e)
 	     _option->get(PREF_ALLOW_OVERWRITE) != V_TRUE) {
 	    // TODO we need this->haltRequested = true?
 	    throw new DownloadFailureException(MSG_FILE_ALREADY_EXISTS,
-					       getFilePath().c_str());
+					       _pieceStorage->getDiskAdaptor()->getFilePath().c_str());
 	  } else {
 	    _pieceStorage->getDiskAdaptor()->openFile();
 	  }
@@ -237,7 +237,7 @@ Commands RequestGroup::createInitialCommand(DownloadEngine* e)
 Commands RequestGroup::processCheckIntegrityEntry(const CheckIntegrityEntryHandle& entry, DownloadEngine* e)
 {
 #ifdef ENABLE_MESSAGE_DIGEST
-  if(File(getFilePath()).size() > 0 &&
+  if(//File(getFilePath()).size() > 0 &&
      e->option->get(PREF_CHECK_INTEGRITY) == V_TRUE &&
      entry->isValidationReady()) {
     entry->initValidator();
diff --git a/src/SingleFileAllocationIterator.cc b/src/SingleFileAllocationIterator.cc
index 1384c163..620410fa 100644
--- a/src/SingleFileAllocationIterator.cc
+++ b/src/SingleFileAllocationIterator.cc
@@ -33,35 +33,42 @@
  */
 /* copyright --> */
 #include "SingleFileAllocationIterator.h"
-#include "AbstractSingleDiskAdaptor.h"
+#include "BinaryStream.h"
+#include "Util.h"
+#include "a2io.h"
 
-#define BUFSIZE 16*1024
+#define BUFSIZE (256*1024)
 
-SingleFileAllocationIterator::SingleFileAllocationIterator(AbstractSingleDiskAdaptor* diskAdaptor):_diskAdaptor(diskAdaptor), _offset(diskAdaptor->size()) {}
+SingleFileAllocationIterator::SingleFileAllocationIterator(BinaryStream* stream, int64_t offset, int64_t totalLength):_stream(stream), _offset(offset), _totalLength(totalLength), _buffer(0)
+{}
 
-SingleFileAllocationIterator::~SingleFileAllocationIterator() {}
+SingleFileAllocationIterator::~SingleFileAllocationIterator()
+{
+  delete [] _buffer;
+}
 
-void SingleFileAllocationIterator::allocateChunk()
+void SingleFileAllocationIterator::init()
 {
-  int32_t bufSize = BUFSIZE;
-  unsigned char buf[BUFSIZE];
-  memset(buf, 0, bufSize);
-  
-  _diskAdaptor->writeData(buf, bufSize, _offset);
-  _offset += bufSize;
-  
-  if(_diskAdaptor->getTotalLength() < _offset) {
-    _diskAdaptor->truncate(getTotalLength());
-    _offset = getTotalLength();
-  }
+#ifdef HAVE_POSIX_MEMALIGN
+  _buffer = (unsigned char*)Util::allocateAlignedMemory(512, BUFSIZE);
+#else
+  _buffer = new unsigned char[BUFSIZE];
+#endif // HAVE_POSIX_MEMALIGN
+  memset(_buffer, 0, BUFSIZE);
 }
 
-bool SingleFileAllocationIterator::finished()
+void SingleFileAllocationIterator::allocateChunk()
 {
-  return getCurrentLength() >= getTotalLength();
+  _stream->writeData(_buffer, BUFSIZE, _offset);
+  _offset += BUFSIZE;
+
+  if(_totalLength < _offset) {
+    _stream->truncate(_totalLength);
+    _offset = _totalLength;
+  }
 }
 
-int64_t SingleFileAllocationIterator::getTotalLength()
+bool SingleFileAllocationIterator::finished()
 {
-  return _diskAdaptor->getTotalLength();
+  return _offset >= _totalLength;
 }
diff --git a/src/SingleFileAllocationIterator.h b/src/SingleFileAllocationIterator.h
index 4b401f13..3690df88 100644
--- a/src/SingleFileAllocationIterator.h
+++ b/src/SingleFileAllocationIterator.h
@@ -37,15 +37,20 @@
 
 #include "FileAllocationIterator.h"
 
-class AbstractSingleDiskAdaptor;
+class BinaryStream;
 
 class SingleFileAllocationIterator:public FileAllocationIterator
 {
 private:
-  AbstractSingleDiskAdaptor* _diskAdaptor;
+  BinaryStream* _stream;
+
   int64_t _offset;
+
+  int64_t _totalLength;
+
+  unsigned char* _buffer;
 public:
-  SingleFileAllocationIterator(AbstractSingleDiskAdaptor* diskAdaptor);
+  SingleFileAllocationIterator(BinaryStream* stream, int64_t offset, int64_t totalLength);
 
   virtual ~SingleFileAllocationIterator();
 
@@ -58,7 +63,15 @@ public:
     return _offset;
   }
 
-  virtual int64_t getTotalLength();
+  virtual int64_t getTotalLength()
+  {
+    return _totalLength;
+  }
+
+  /**
+   * Must be called only once, before calling allocateChunk()
+   */
+  void init();
 };
 
 typedef SharedHandle<SingleFileAllocationIterator> SingleFileAllocationIteratorHandle;
diff --git a/src/SocketCore.cc b/src/SocketCore.cc
index 403dc470..6d256a66 100644
--- a/src/SocketCore.cc
+++ b/src/SocketCore.cc
@@ -213,9 +213,10 @@ void SocketCore::setNonBlockingMode()
     throw new DlAbortEx(EX_SOCKET_NONBLOCKING, errorMsg());
   }
 #else
-  int32_t flags = fcntl(sockfd, F_GETFL, 0);
+  int32_t flags;
+  while((flags = fcntl(sockfd, F_GETFL, 0)) == -1 && errno == EINTR);
   // TODO add error handling
-  fcntl(sockfd, F_SETFL, flags|O_NONBLOCK);
+  while(fcntl(sockfd, F_SETFL, flags|O_NONBLOCK) == -1 && errno == EINTR);
 #endif // __MINGW32__
   blocking = false;
 }
@@ -228,9 +229,10 @@ void SocketCore::setBlockingMode()
     throw new DlAbortEx(EX_SOCKET_BLOCKING, errorMsg());
   }
 #else
-  int32_t flags = fcntl(sockfd, F_GETFL, 0);
+  int32_t flags;
+  while((flags = fcntl(sockfd, F_GETFL, 0)) == -1 && errno == EINTR);
   // TODO add error handling
-  fcntl(sockfd, F_SETFL, flags&(~O_NONBLOCK));
+  while(fcntl(sockfd, F_SETFL, flags&(~O_NONBLOCK)) == -1 && errno == EINTR);
 #endif // __MINGW32__
   blocking = true;
 }
diff --git a/src/Util.cc b/src/Util.cc
index 5363e81a..5e161fd5 100644
--- a/src/Util.cc
+++ b/src/Util.cc
@@ -42,6 +42,7 @@
 #include "BitfieldMan.h"
 #include "DefaultDiskWriter.h"
 #include "BinaryStream.h"
+#include "FatalException.h"
 #include <ctype.h>
 #include <errno.h>
 #include <fcntl.h>
@@ -849,3 +850,18 @@ string Util::toString(const BinaryStreamHandle& binaryStream)
   }
   return strm.str();
 }
+
+#ifdef HAVE_POSIX_MEMALIGN
+/**
+ * In linux 2.6, alignment and size should be a multiple of 512.
+ */
+void* Util::allocateAlignedMemory(size_t alignment, size_t size)
+{
+  void* buffer;
+  int32_t res;
+  if((res = posix_memalign(&buffer, alignment, size)) != 0) {
+    throw new FatalException("Error in posix_memalign: %s", strerror(res));
+  }
+  return buffer;
+}
+#endif // HAVE_POSIX_MEMALIGN
diff --git a/src/Util.h b/src/Util.h
index 12120264..811050f7 100644
--- a/src/Util.h
+++ b/src/Util.h
@@ -168,6 +168,10 @@ public:
 
   // binaryStream has to be opened before calling this function.
   static string toString(const BinaryStreamHandle& binaryStream);
+
+#ifdef HAVE_POSIX_MEMALIGN
+  static void* allocateAlignedMemory(size_t alignment, size_t size);
+#endif // HAVE_POSIX_MEMALIGN
 };
 
 #endif // _D_UTIL_H_
diff --git a/src/a2io.h b/src/a2io.h
index 4434db0d..1b0dc27e 100644
--- a/src/a2io.h
+++ b/src/a2io.h
@@ -116,4 +116,8 @@
 # define a2mkdir(path, openMode) mkdir(path, openMode)
 #endif // __MINGW32__
 
+#if defined HAVE_POSIX_MEMALIGN && O_DIRECT
+//# define ENABLE_DIRECT_IO 1
+#endif // HAVE_POSIX_MEMALIGN && O_DIRECT
+#undef HAVE_POSIX_MEMALIGN
 #endif // _D_A2IO_H_
diff --git a/src/option_processing.cc b/src/option_processing.cc
index 61f98694..eae3e0aa 100644
--- a/src/option_processing.cc
+++ b/src/option_processing.cc
@@ -40,6 +40,7 @@
 #include "Util.h"
 #include "message.h"
 #include "Exception.h"
+#include "a2io.h"
 #include <fstream>
 #include <sstream>
 
@@ -125,6 +126,7 @@ Option* option_processing(int argc, char* const argv[])
   op->put(PREF_ENABLE_HTTP_PIPELINING, V_FALSE);
   op->put(PREF_MAX_HTTP_PIPELINING, "2");
   op->put(PREF_SEED_RATIO, "1.0");
+  op->put(PREF_ENABLE_DIRECT_IO, V_FALSE);
   while(1) {
     int optIndex = 0;
     int lopt;
@@ -173,6 +175,9 @@ Option* option_processing(int argc, char* const argv[])
       { "enable-http-keep-alive", optional_argument, &lopt, 207 },
       { "enable-http-pipelining", optional_argument, &lopt, 208 },
       { "no-file-allocation-limit", required_argument, &lopt, 209 },
+#ifdef ENABLE_DIRECT_IO
+      { PREF_ENABLE_DIRECT_IO, optional_argument, &lopt, 210 },
+#endif // ENABLE_DIRECT_IO
 #if defined ENABLE_BITTORRENT || ENABLE_METALINK
       { "show-files", no_argument, NULL, 'S' },
       { "select-file", required_argument, &lopt, 21 },
@@ -323,6 +328,9 @@ Option* option_processing(int argc, char* const argv[])
       case 209:
 	cmdstream << PREF_NO_FILE_ALLOCATION_LIMIT << "=" << optarg << "\n";
 	break;
+      case 210:
+	cmdstream << PREF_ENABLE_DIRECT_IO << "=" << toBoolArg(optarg) << "\n";
+	break;
       }
       break;
     }
diff --git a/src/prefs.h b/src/prefs.h
index a705f73f..3933b21f 100644
--- a/src/prefs.h
+++ b/src/prefs.h
@@ -115,6 +115,8 @@
 #define PREF_AUTO_FILE_RENAMING "auto-file-renaming"
 // value: true | false
 #define PREF_PARAMETERIZED_URI "parameterized-uri"
+// value: true | false
+#define PREF_ENABLE_DIRECT_IO "enable-direct-io"
 
 /**
  * FTP related preferences
diff --git a/src/version_usage.cc b/src/version_usage.cc
index 4a30a268..0775a846 100644
--- a/src/version_usage.cc
+++ b/src/version_usage.cc
@@ -33,6 +33,7 @@
  */
 /* copyright --> */
 #include "common.h"
+#include "a2io.h"
 #include "FeatureConfig.h"
 #ifdef ENABLE_MESSAGE_DIGEST
 # include "messageDigest.h"
@@ -161,9 +162,15 @@ void showUsage() {
 	    "                              Default: prealloc") << endl;
   cout << _(" --no-file-allocation-limit=SIZE No file allocation is made for files whose\n"
 	    "                              size is smaller than SIZE.\n"
-	    "                              You can append K or M(1K = 1024, 1M = 1024K).\n"
-	    "                              BitTorrent downloads ignore this option.\n"
-	    "                              Default: 5M") << endl;
+	    "                              You can append K or M(1K = 1024, 1M = 1024K).") << "\n"
+       << DEFAULT_MSG << "5M" << "\n";
+#ifdef ENABLE_DIRECT_IO
+  cout << _(" --enable-direct-io[=true|false] Enable directI/O, which lowers cpu usage while\n"
+	    "                              allocating files.\n"
+	    "                              Turn off if you encounter any error") << "\n"
+	    "                              Tested in Linux 2.6.21" << "\n"
+       << DEFAULT_MSG << "false" << "\n";
+#endif // ENABLE_DIRECT_IO
   cout << _(" --allow-overwrite=true|false If false, aria2 doesn't download a file which\n"
   		"                              already exists but the corresponding .aria2 file\n"
   		"                              doesn't exist.\n"