Merge pull request #1644 from aliemjay/sync-caches

prevent corrupt downloads after app and/or system crash

src/AbstractDiskWriter.cc

@@ -590,4 +590,16 @@ void AbstractDiskWriter::dropCache(int64_t len, int64_t offset)
 #endif // HAVE_POSIX_FADVISE
 }
 
+void AbstractDiskWriter::flushOSBuffers()
+{
+  if (fd_ == A2_BAD_FD) {
+    return;
+  }
+#ifdef __MINGW32__
+  FlushFileBuffers(fd_);
+#else // !__MINGW32__
+  fsync(fd_);
+#endif // __MINGW32__
+}
+
 } // namespace aria2

src/AbstractDiskWriter.h

@@ -100,6 +100,8 @@ public:
   virtual void enableMmap() CXX11_OVERRIDE;
 
   virtual void dropCache(int64_t len, int64_t offset) CXX11_OVERRIDE;
+
+  virtual void flushOSBuffers() CXX11_OVERRIDE;
 };
 
 } // namespace aria2

src/AbstractSingleDiskAdaptor.cc

@@ -103,6 +103,11 @@ void AbstractSingleDiskAdaptor::writeCache(const WrDiskCacheEntry* entry)
   }
 }
 
+void AbstractSingleDiskAdaptor::flushOSBuffers()
+{
+  diskWriter_->flushOSBuffers();
+}
+
 bool AbstractSingleDiskAdaptor::fileExists()
 {
   return File(getFilePath()).exists();

src/AbstractSingleDiskAdaptor.h

@@ -72,6 +72,8 @@ public:
 
   virtual void writeCache(const WrDiskCacheEntry* entry) CXX11_OVERRIDE;
 
+  virtual void flushOSBuffers() CXX11_OVERRIDE;
+
   virtual bool fileExists() CXX11_OVERRIDE;
 
   virtual int64_t size() CXX11_OVERRIDE;

src/DefaultPieceStorage.cc

@@ -685,7 +685,7 @@ std::shared_ptr<DiskAdaptor> DefaultPieceStorage::getDiskAdaptor()
 
 WrDiskCache* DefaultPieceStorage::getWrDiskCache() { return wrDiskCache_; }
 
-void DefaultPieceStorage::flushWrDiskCacheEntry()
+void DefaultPieceStorage::flushWrDiskCacheEntry(bool releaseEntries)
 {
   if (!wrDiskCache_) {
     return;
@@ -697,9 +697,11 @@ void DefaultPieceStorage::flushWrDiskCacheEntry()
     auto ce = piece->getWrDiskCacheEntry();
     if (ce) {
       piece->flushWrCache(wrDiskCache_);
+      if (releaseEntries) {
         piece->releaseWrCache(wrDiskCache_);
       }
     }
+  }
 }
 
 int32_t DefaultPieceStorage::getPieceLength(size_t index)

src/DefaultPieceStorage.h

@@ -234,7 +234,7 @@ public:
 
   virtual WrDiskCache* getWrDiskCache() CXX11_OVERRIDE;
 
-  virtual void flushWrDiskCacheEntry() CXX11_OVERRIDE;
+  virtual void flushWrDiskCacheEntry(bool releaseEntries) CXX11_OVERRIDE;
 
   virtual int32_t getPieceLength(size_t index) CXX11_OVERRIDE;
 

src/DiskAdaptor.h

@@ -114,6 +114,9 @@ public:
   // Writes cached data to the underlying disk.
   virtual void writeCache(const WrDiskCacheEntry* entry) = 0;
 
+  // Force physical write of data from OS buffer cache.
+  virtual void flushOSBuffers() {};
+
   void setFileAllocationMethod(FileAllocationMethod method)
   {
     fileAllocationMethod_ = method;

src/DiskWriter.h

@@ -85,6 +85,9 @@ public:
 
   // Drops cache in range [offset, offset + len)
   virtual void dropCache(int64_t len, int64_t offset) {}
+
+  // Force physical write of data from OS buffer cache.
+  virtual void flushOSBuffers() {}
 };
 
 } // namespace aria2

src/MultiDiskAdaptor.cc

@@ -419,6 +419,17 @@ void MultiDiskAdaptor::writeCache(const WrDiskCacheEntry* entry)
   }
 }
 
+void MultiDiskAdaptor::flushOSBuffers()
+{
+  for (auto& dwent : openedDiskWriterEntries_) {
+    auto& dw = dwent->getDiskWriter();
+    if (!dw) {
+      continue;
+    }
+    dw->flushOSBuffers();
+  }
+}
+
 bool MultiDiskAdaptor::fileExists()
 {
   return std::find_if(std::begin(getFileEntries()), std::end(getFileEntries()),

src/MultiDiskAdaptor.h

@@ -135,6 +135,8 @@ public:
 
   virtual void writeCache(const WrDiskCacheEntry* entry) CXX11_OVERRIDE;
 
+  virtual void flushOSBuffers() CXX11_OVERRIDE;
+
   virtual bool fileExists() CXX11_OVERRIDE;
 
   virtual int64_t size() CXX11_OVERRIDE;

src/PieceStorage.h

@@ -226,8 +226,9 @@ public:
 
   virtual WrDiskCache* getWrDiskCache() = 0;
 
-  // Flushes write disk cache for in-flight piece and evicts them.
+  // Flushes write disk cache for in-flight piece
-  virtual void flushWrDiskCacheEntry() = 0;
+  // and optionally releases the associated cache entries.
+  virtual void flushWrDiskCacheEntry(bool releaseEntries) = 0;
 
   virtual int32_t getPieceLength(size_t index) = 0;
 

src/RequestGroup.cc

@@ -210,7 +210,8 @@ std::pair<error_code::Value, std::string> RequestGroup::downloadResult() const
 void RequestGroup::closeFile()
 {
   if (pieceStorage_) {
-    pieceStorage_->flushWrDiskCacheEntry();
+    pieceStorage_->flushWrDiskCacheEntry(true);
+    pieceStorage_->getDiskAdaptor()->flushOSBuffers();
     pieceStorage_->getDiskAdaptor()->closeFile();
   }
 }
@@ -1290,6 +1291,10 @@ bool RequestGroup::doesUploadSpeedExceed()
 void RequestGroup::saveControlFile() const
 {
   if (saveControlFile_) {
+    if (pieceStorage_) {
+      pieceStorage_->flushWrDiskCacheEntry(false);
+      pieceStorage_->getDiskAdaptor()->flushOSBuffers();
+    }
     progressInfoFile_->save();
   }
 }

src/UnknownLengthPieceStorage.h

@@ -216,7 +216,7 @@ public:
 
   virtual WrDiskCache* getWrDiskCache() CXX11_OVERRIDE { return nullptr; }
 
-  virtual void flushWrDiskCacheEntry() CXX11_OVERRIDE {}
+  virtual void flushWrDiskCacheEntry(bool releaseEntries) CXX11_OVERRIDE {}
 
   virtual int32_t getPieceLength(size_t index) CXX11_OVERRIDE;
 

test/MockPieceStorage.h

@@ -228,7 +228,7 @@ public:
 
   virtual WrDiskCache* getWrDiskCache() CXX11_OVERRIDE { return 0; }
 
-  virtual void flushWrDiskCacheEntry() CXX11_OVERRIDE {}
+  virtual void flushWrDiskCacheEntry(bool releaseEntries) CXX11_OVERRIDE {}
 
   void setDiskAdaptor(const std::shared_ptr<DiskAdaptor>& adaptor)
   {