TerraSync: make tarball extraction asynchronous

simgear/io/HTTPRepository.cxx

@@ -416,6 +416,66 @@ public:
         return _relativePath;
     }
 
+    class ArchiveExtractTask {
+    public:
+      ArchiveExtractTask(SGPath p, const std::string &relPath)
+          : relativePath(relPath), file(p), extractor(p.dir()) {
+        if (!file.open(SG_IO_IN)) {
+          SG_LOG(SG_TERRASYNC, SG_ALERT,
+                 "Unable to open " << p << " to extract");
+          return;
+        }
+
+        buffer = (uint8_t *)malloc(bufferSize);
+      }
+
+      ArchiveExtractTask(const ArchiveExtractTask &) = delete;
+
+      HTTPRepoPrivate::ProcessResult run(HTTPRepoPrivate *repo) {
+        size_t rd = file.read((char *)buffer, bufferSize);
+        extractor.extractBytes(buffer, rd);
+
+        if (file.eof()) {
+          extractor.flush();
+          file.close();
+
+          if (!extractor.isAtEndOfArchive()) {
+            SG_LOG(SG_TERRASYNC, SG_ALERT, "Corrupt tarball " << relativePath);
+            repo->failedToUpdateChild(relativePath,
+                                      HTTPRepository::REPO_ERROR_IO);
+            return HTTPRepoPrivate::ProcessFailed;
+          }
+
+          if (extractor.hasError()) {
+            SG_LOG(SG_TERRASYNC, SG_ALERT, "Error extracting " << relativePath);
+            repo->failedToUpdateChild(relativePath,
+                                      HTTPRepository::REPO_ERROR_IO);
+            return HTTPRepoPrivate::ProcessFailed;
+          }
+
+          return HTTPRepoPrivate::ProcessDone;
+        }
+
+        return HTTPRepoPrivate::ProcessContinue;
+      }
+
+      ~ArchiveExtractTask() { free(buffer); }
+
+    private:
+      // intentionally small so we extract incrementally on Windows
+      // where Defender throttles many small files, sorry
+      // if you make this bigger we will be more efficient but stall for
+      // longer when extracting the Airports_archive
+      const int bufferSize = 1024 * 64;
+
+      std::string relativePath;
+      uint8_t *buffer = nullptr;
+      SGBinaryFile file;
+      ArchiveExtractor extractor;
+    };
+
+    using ArchiveExtractTaskPtr = std::shared_ptr<ArchiveExtractTask>;
+
     void didUpdateFile(const std::string& file, const std::string& hash, size_t sz)
     {
         // check hash matches what we expected
@@ -454,35 +514,22 @@ public:
                   }
 
                   if (pathAvailable) {
-                    // If this is a tarball, then extract it.
+                    // we use a Task helper to extract tarballs incrementally.
-                    SGBinaryFile f(p);
+                    // without this, archive extraction blocks here, which
-                    if (! f.open(SG_IO_IN)) SG_LOG(SG_TERRASYNC, SG_ALERT, "Unable to open " << p << " to extract");
+                    // prevents other repositories downloading / updating.
+                    // Unfortunately due Windows AV (Defender, etc) we cna block
+                    // here for many minutes.
 
-                    SG_LOG(SG_TERRASYNC, SG_INFO, "Extracting " << absolutePath() << "/" << file << " to " << p.dir());
+                    // use a lambda to own this shared_ptr; this means when the
-                    SGPath extractDir = p.dir();
+                    // lambda is destroyed, the ArchiveExtraTask will get
-                    ArchiveExtractor ex(extractDir);
+                    // cleaned up.
-
+                    ArchiveExtractTaskPtr t =
-                    const size_t bufSize = 1024 * 1024;
+                        std::make_shared<ArchiveExtractTask>(p, _relativePath);
-                    uint8_t* buf = (uint8_t*)malloc(bufSize);
+                    auto cb = [t](HTTPRepoPrivate *repo) {
-                    while (!f.eof()) {
+                      return t->run(repo);
-                        size_t rd = f.read((char*)buf, bufSize);
+                    };
-                        ex.extractBytes(buf, rd);
-                    }
-
-                    ex.flush();
-                    free(buf);
-                    if (! ex.isAtEndOfArchive()) {
-                      SG_LOG(SG_TERRASYNC, SG_ALERT, "Corrupt tarball " << p);
-                      _repository->failedToUpdateChild(
-                          _relativePath, HTTPRepository::REPO_ERROR_IO);
-                    }
-
-                    if (ex.hasError()) {
-                      SG_LOG(SG_TERRASYNC, SG_ALERT, "Error extracting " << p);
-                      _repository->failedToUpdateChild(
-                          _relativePath, HTTPRepository::REPO_ERROR_IO);
-                    }
 
+                    _repository->addTask(cb);
                   } else {
                     SG_LOG(SG_TERRASYNC, SG_ALERT, "Unable to remove old file/directory " << removePath);
                   } // of pathAvailable
@@ -820,14 +867,19 @@ void HTTPRepository::process()
     const int maxToProcess = 16;
 
     while (processedCount < maxToProcess) {
-        if (_d->pendingUpdateOfChildren.empty()) {
+      if (_d->pendingTasks.empty()) {
-            break;
+        break;
-        }
+      }
 
-        auto dirToUpdate = _d->pendingUpdateOfChildren.front();
+      auto task = _d->pendingTasks.front();
-        _d->pendingUpdateOfChildren.pop_front();
+      auto result = task(_d.get());
-        dirToUpdate->updateChildrenBasedOnHash();
+      if (result == HTTPRepoPrivate::ProcessContinue) {
-        ++processedCount;
+        // assume we're not complete
+        return;
+      }
+
+      _d->pendingTasks.pop_front();
+      ++processedCount;
     }
 
     _d->checkForComplete();
@@ -835,9 +887,10 @@ void HTTPRepository::process()
 
 void HTTPRepoPrivate::checkForComplete()
 {
-    if (pendingUpdateOfChildren.empty() && activeRequests.empty() && queuedRequests.empty()) {
+  if (pendingTasks.empty() && activeRequests.empty() &&
-        isUpdating = false;
+      queuedRequests.empty()) {
-    }
+    isUpdating = false;
+  }
 }
 
 size_t HTTPRepository::bytesToDownload() const
@@ -1309,12 +1362,16 @@ HTTPRepository::failure() const
 
     void HTTPRepoPrivate::scheduleUpdateOfChildren(HTTPDirectory* dir)
     {
-        auto it = std::find(pendingUpdateOfChildren.begin(), pendingUpdateOfChildren.end(), dir);
+      auto updateChildTask = [dir](const HTTPRepoPrivate *) {
-        if (it != pendingUpdateOfChildren.end()) {
+        dir->updateChildrenBasedOnHash();
-            return; // duplicate add, skip
+        return ProcessDone;
-        }
+      };
 
-        pendingUpdateOfChildren.push_back(dir);
+      addTask(updateChildTask);
+    }
+
+    void HTTPRepoPrivate::addTask(RepoProcessTask task) {
+      pendingTasks.push_back(task);
     }
 
     int HTTPRepoPrivate::countDirtyHashCaches() const

simgear/io/HTTPRepository_private.hxx

@@ -20,6 +20,7 @@
 #pragma once
 
 #include <deque>
+#include <functional>
 #include <memory>
 #include <string>
 #include <unordered_map>
@@ -50,7 +51,7 @@ protected:
   size_t _contentSize = 0;
 };
 
-typedef SGSharedPtr<HTTPRepoGetRequest> RepoRequestPtr;
+using RepoRequestPtr = SGSharedPtr<HTTPRepoGetRequest>;
 
 class HTTPRepoPrivate {
 public:
@@ -102,20 +103,24 @@ public:
   HTTPDirectory *getOrCreateDirectory(const std::string &path);
   bool deleteDirectory(const std::string &relPath, const SGPath &absPath);
 
-  void scheduleUpdateOfChildren(HTTPDirectory* dir);
 
   typedef std::vector<HTTPDirectory_ptr> DirectoryVector;
   DirectoryVector directories;
 
-  // list of directories to be locally processed
+  void scheduleUpdateOfChildren(HTTPDirectory *dir);
-  // this is used to avoid deep recursion when receiving the .dirIndex;
-  // we don't want to recurse over a large repo in a single call
-  std::deque<HTTPDirectory*> pendingUpdateOfChildren;
 
   SGPath installedCopyPath;
 
   int countDirtyHashCaches() const;
   void flushHashCaches();
+
+  enum ProcessResult { ProcessContinue, ProcessDone, ProcessFailed };
+
+  using RepoProcessTask = std::function<ProcessResult(HTTPRepoPrivate *repo)>;
+
+  void addTask(RepoProcessTask task);
+
+  std::deque<RepoProcessTask> pendingTasks;
 };
 
 } // namespace simgear