2008-06-08 Tatsuhiro Tsujikawa <tujikawa at rednoah dot com>

Limited the number files opened in each BitTorrent download. The 
default
	value is 100. The option to change this value will be added.
	* src/MultiDiskAdaptor.cc
	* src/MultiDiskAdaptor.h
	* src/MultiFileAllocationIterator.cc

ChangeLog

@@ -1,3 +1,11 @@
+2008-06-08  Tatsuhiro Tsujikawa  <tujikawa at rednoah dot com>
+
+	Limited the number files opened in each BitTorrent download. The default
+	value is 100. The option to change this value will be added.
+	* src/MultiDiskAdaptor.cc
+	* src/MultiDiskAdaptor.h
+	* src/MultiFileAllocationIterator.cc
+
 2008-06-08  Tatsuhiro Tsujikawa  <tujikawa at rednoah dot com>
 
 	Fixed the bug that HTTP downloads don't start with old libraries.

src/MultiDiskAdaptor.cc

@@ -42,12 +42,14 @@
 #include "DlAbortEx.h"
 #include "File.h"
 #include "StringFormat.h"
+#include "Logger.h"
+#include "SimpleRandomizer.h"
 #include <algorithm>
 
 namespace aria2 {
 
 DiskWriterEntry::DiskWriterEntry(const SharedHandle<FileEntry>& fileEntry):
-  fileEntry(fileEntry) {}
+  fileEntry(fileEntry), _open(false), _directIO(false) {}
 
 DiskWriterEntry::~DiskWriterEntry() {}
 
@@ -59,21 +61,41 @@ std::string DiskWriterEntry::getFilePath(const std::string& topDir) const
 void DiskWriterEntry::initAndOpenFile(const std::string& topDir)
 {
   diskWriter->initAndOpenFile(getFilePath(topDir), fileEntry->getLength());
+  if(_directIO) {
+    diskWriter->enableDirectIO();
+  }
+  _open = true;
 }
 
 void DiskWriterEntry::openFile(const std::string& topDir)
 {
   diskWriter->openFile(getFilePath(topDir), fileEntry->getLength());
+  if(_directIO) {
+    diskWriter->enableDirectIO();
+  }
+  _open = true;
 }
 
 void DiskWriterEntry::openExistingFile(const std::string& topDir)
 {
   diskWriter->openExistingFile(getFilePath(topDir), fileEntry->getLength());
+  if(_directIO) {
+    diskWriter->enableDirectIO();
+  }
+  _open = true;
+}
+
+bool DiskWriterEntry::isOpen() const
+{
+  return _open;
 }
 
 void DiskWriterEntry::closeFile()
 {
+  if(_open) {
     diskWriter->closeFile();
+    _open = false;
+  }
 }
 
 bool DiskWriterEntry::fileExists(const std::string& topDir)
@@ -106,6 +128,22 @@ bool DiskWriterEntry::operator<(const DiskWriterEntry& entry) const
   return fileEntry < entry.fileEntry;
 }
 
+void DiskWriterEntry::enableDirectIO()
+{
+  if(_open) {
+    diskWriter->enableDirectIO();
+  }
+  _directIO = true;
+}
+
+void DiskWriterEntry::disableDirectIO()
+{
+  if(_open) {
+    diskWriter->disableDirectIO();
+  }
+  _directIO = false;
+}
+
 MultiDiskAdaptor::MultiDiskAdaptor():
   pieceLength(0) {}
 
@@ -141,36 +179,60 @@ void MultiDiskAdaptor::mkdir(const std::string& topDirPath) const
   }
 }
 
+void MultiDiskAdaptor::openIfNot
+(const SharedHandle<DiskWriterEntry>& entry,
+ void (DiskWriterEntry::*open)(const std::string&),
+ const std::string& topDirPath)
+{
+  if(!entry->isOpen()) {
+//     logger->debug("DiskWriterEntry: Cache MISS. offset=%s",
+// 		  Util::itos(entry->getFileEntry()->getOffset()).c_str());
+ 
+    size_t numOpened = _openedDiskWriterEntries.size();
+    (entry.get()->*open)(topDirPath);
+    if(numOpened >= OPEN_FILE_MAX) {
+      // Cache is full. 
+      // Choose one DiskWriterEntry randomly and close it.
+      size_t index = SimpleRandomizer::getInstance()->getRandomNumber(numOpened);
+      std::deque<SharedHandle<DiskWriterEntry> >::iterator i =
+	_openedDiskWriterEntries.begin();
+      std::advance(i, index);
+      (*i)->closeFile();
+      (*i) = entry;
+    } else {
+      _openedDiskWriterEntries.push_back(entry);
+    } 
+  } else {
+//     logger->debug("DiskWriterEntry: Cache HIT. offset=%s",
+// 		  Util::itos(entry->getFileEntry()->getOffset()).c_str());
+  }
+}
+
 void MultiDiskAdaptor::openFile()
 {
-  const std::string topDirPath = getTopDirPath();
+  _cachedTopDirPath = getTopDirPath();
-  mkdir(topDirPath);
+  mkdir(_cachedTopDirPath);
   resetDiskWriterEntries();
-  for(DiskWriterEntries::iterator itr = diskWriterEntries.begin();
+  // TODO we should call openIfNot here?
-      itr != diskWriterEntries.end(); ++itr) {
-    (*itr)->openFile(topDirPath);
-  }
 }
 
 void MultiDiskAdaptor::initAndOpenFile()
 {
-  const std::string topDirPath = getTopDirPath();
+  _cachedTopDirPath = getTopDirPath();
-  mkdir(topDirPath);
+  mkdir(_cachedTopDirPath);
   resetDiskWriterEntries();
+  // Call DiskWriterEntry::initAndOpenFile to make files truncated.
   for(DiskWriterEntries::iterator itr = diskWriterEntries.begin();
       itr != diskWriterEntries.end(); ++itr) {
-    (*itr)->initAndOpenFile(topDirPath);
+    openIfNot(*itr, &DiskWriterEntry::initAndOpenFile, _cachedTopDirPath);
   }
 }
 
 void MultiDiskAdaptor::openExistingFile()
 {
-  const std::string topDirPath = getTopDirPath();
+  _cachedTopDirPath = getTopDirPath();
   resetDiskWriterEntries();
-  for(DiskWriterEntries::iterator itr = diskWriterEntries.begin();
+  // Not need to call openIfNot here.
-      itr != diskWriterEntries.end(); ++itr) {
-    (*itr)->openExistingFile(topDirPath);
-  }
 }
 
 void MultiDiskAdaptor::closeFile()
@@ -240,6 +302,9 @@ void MultiDiskAdaptor::writeData(const unsigned char* data, size_t len,
   off_t fileOffset = offset-(*first)->getFileEntry()->getOffset();
   for(DiskWriterEntries::const_iterator i = first; i != diskWriterEntries.end(); ++i) {
     size_t writeLength = calculateLength(*i, fileOffset, rem);
+
+    openIfNot(*i, &DiskWriterEntry::openFile, _cachedTopDirPath);
+
     (*i)->getDiskWriter()->writeData(data+(len-rem), writeLength, fileOffset);
     rem -= writeLength;
     fileOffset = 0;
@@ -258,6 +323,9 @@ ssize_t MultiDiskAdaptor::readData(unsigned char* data, size_t len, off_t offset
   off_t fileOffset = offset-(*first)->getFileEntry()->getOffset();
   for(DiskWriterEntries::const_iterator i = first; i != diskWriterEntries.end(); ++i) {
     size_t readLength = calculateLength(*i, fileOffset, rem);
+
+    openIfNot(*i, &DiskWriterEntry::openFile, _cachedTopDirPath);
+
     totalReadLength +=
       (*i)->getDiskWriter()->readData(data+(len-rem), readLength, fileOffset);
     rem -= readLength;
@@ -274,9 +342,12 @@ bool MultiDiskAdaptor::fileExists()
   if(diskWriterEntries.empty()) {
     resetDiskWriterEntries();
   }
+  // Don't use _cachedTopDirPath because they are initialized after opening files.
+  // This method could be called before opening files.
+  std::string topDirPath = getTopDirPath();
   for(DiskWriterEntries::iterator itr = diskWriterEntries.begin();
-      itr != diskWriterEntries.end(); itr++) {
+      itr != diskWriterEntries.end(); ++itr) {
-    if((*itr)->fileExists(getTopDirPath())) {
+    if((*itr)->fileExists(topDirPath)) {
       return true;
     }				   
   }
@@ -288,7 +359,7 @@ uint64_t MultiDiskAdaptor::size() const
 {
   uint64_t size = 0;
   for(DiskWriterEntries::const_iterator itr = diskWriterEntries.begin();
-      itr != diskWriterEntries.end(); itr++) {
+      itr != diskWriterEntries.end(); ++itr) {
     size += (*itr)->size();
   }
   return size;
@@ -303,7 +374,7 @@ void MultiDiskAdaptor::enableDirectIO()
 {
   for(DiskWriterEntries::const_iterator itr = diskWriterEntries.begin();
       itr != diskWriterEntries.end(); ++itr) {
-    (*itr)->getDiskWriter()->enableDirectIO();
+    (*itr)->enableDirectIO();
   }
 }
 
@@ -311,7 +382,7 @@ void MultiDiskAdaptor::disableDirectIO()
 {
   for(DiskWriterEntries::const_iterator itr = diskWriterEntries.begin();
       itr != diskWriterEntries.end(); ++itr) {
-    (*itr)->getDiskWriter()->disableDirectIO();
+    (*itr)->disableDirectIO();
   }
 }
 

src/MultiDiskAdaptor.h

@@ -47,6 +47,8 @@ class DiskWriterEntry {
 private:
   SharedHandle<FileEntry> fileEntry;
   SharedHandle<DiskWriter> diskWriter;
+  bool _open;
+  bool _directIO;
 public:
   DiskWriterEntry(const SharedHandle<FileEntry>& fileEntry);
 
@@ -62,6 +64,8 @@ public:
 
   void closeFile();
 
+  bool isOpen() const;
+
   bool fileExists(const std::string& topDir);
 
   uint64_t size() const;
@@ -73,6 +77,16 @@ public:
   SharedHandle<DiskWriter> getDiskWriter() const;
 
   bool operator<(const DiskWriterEntry& entry) const;
+
+  // Set _directIO to true.
+  // Additionally, if diskWriter is opened, diskWriter->enableDirectIO() is
+  // called.
+  void enableDirectIO();
+
+  // Set _directIO to false.
+  // Additionally, if diskWriter is opened, diskWriter->disableDirectIO() is
+  // called.
+  void disableDirectIO();
 };
 
 typedef SharedHandle<DiskWriterEntry> DiskWriterEntryHandle;
@@ -86,6 +100,10 @@ private:
   size_t pieceLength;
   DiskWriterEntries diskWriterEntries;
 
+  std::string _cachedTopDirPath;
+
+  std::deque<SharedHandle<DiskWriterEntry> > _openedDiskWriterEntries;
+
   bool _directIOAllowed;
 
   void resetDiskWriterEntries();
@@ -93,6 +111,13 @@ private:
   void mkdir(const std::string& topDirPath) const;
 
   std::string getTopDirPath() const;
+
+  void openIfNot(const SharedHandle<DiskWriterEntry>& entry,
+		 void (DiskWriterEntry::*f)(const std::string&),
+		 const std::string& topDirPath);
+ 
+  static const size_t OPEN_FILE_MAX = 100;
+
 public:
   MultiDiskAdaptor();
 

src/MultiFileAllocationIterator.cc

@@ -59,8 +59,12 @@ void MultiFileAllocationIterator::allocateChunk()
     DiskWriterEntryHandle entry = _entries.front();
     _entries.pop_front();
     FileEntryHandle fileEntry = entry->getFileEntry();
+    // Open file before calling DiskWriterEntry::size()
+    _diskAdaptor->openIfNot(entry, &DiskWriterEntry::openFile,
+			    _diskAdaptor->getTopDirPath());
+    entry->enableDirectIO();
     if(entry->size() < fileEntry->getLength()) {
-      entry->getDiskWriter()->enableDirectIO();
+      // Calling private function of MultiDiskAdaptor.
       _fileAllocationIterator.reset
 	(new SingleFileAllocationIterator(entry->getDiskWriter().get(),
 					  entry->size(),