Just-MR: Add progress reporting for fetch command

Co-authored-by: Sascha Roloff <sascha.roloff@huawei.com>

4 files changed, 71 insertions, 4 deletions

diff --git a/src/other_tools/just_mr/main.cpp b/src/other_tools/just_mr/main.cpp
index cffaf8e1..d1675bdd 100644
--- a/src/other_tools/just_mr/main.cpp
+++ b/src/other_tools/just_mr/main.cpp
@@ -514,6 +514,9 @@ void DefaultReachableRepositories(
 [[nodiscard]] auto MultiRepoFetch(std::shared_ptr<Configuration> const& config,
                                   CommandLineArguments const& arguments)
     -> int {
+    // provide report
+    Logger::Log(LogLevel::Info, "Performing repositories fetch");
+
     // find fetch dir
     auto fetch_dir = arguments.fetch.fetch_dir;
     if (not fetch_dir) {
@@ -715,11 +718,39 @@ void DefaultReachableRepositories(
             return kExitFetchError;
         }
     }
+
+    // report progress
+    auto nr = repos_to_fetch.size();
+    Logger::Log(LogLevel::Info,
+                "Found {} {} to fetch",
+                nr,
+                nr == 1 ? "archive" : "archives");
+
     // create async maps
     auto content_cas_map = CreateContentCASMap(arguments.common.just_mr_paths,
                                                arguments.common.jobs);
     auto repo_fetch_map =
         CreateRepoFetchMap(&content_cas_map, *fetch_dir, arguments.common.jobs);
+
+    // set up map for progress tracing
+    auto& repo_set = JustMRProgress::Instance().RepositorySet();
+    repo_set.clear();
+    repo_set.reserve(nr);
+    for (auto const& repo : repos_to_fetch) {
+        auto distfile = (repo.archive.distfile
+                             ? repo.archive.distfile.value()
+                             : std::filesystem::path(repo.archive.fetch_url)
+                                   .filename()
+                                   .string());
+        repo_set.emplace(std::move(distfile));
+    }
+    // set up progress observer
+    std::atomic<bool> done{false};
+    std::condition_variable cv{};
+    auto reporter = JustMRProgressReporter::Reporter();
+    auto observer =
+        std::thread([reporter, &done, &cv]() { reporter(&done, &cv); });
+
     // do the fetch
     bool failed{false};
     {
@@ -743,9 +774,17 @@ void DefaultReachableRepositories(
                 failed = failed or fatal;
             });
     }
+
+    // close progress observer
+    done = true;
+    cv.notify_all();
+    observer.join();
+
     if (failed) {
         return kExitFetchError;
     }
+    // report success
+    Logger::Log(LogLevel::Info, "Fetch completed");
     return kExitSuccess;
 }
 
diff --git a/src/other_tools/ops_maps/TARGETS b/src/other_tools/ops_maps/TARGETS
index 07869b59..c270978a 100644
--- a/src/other_tools/ops_maps/TARGETS
+++ b/src/other_tools/ops_maps/TARGETS
@@ -65,6 +65,8 @@
     , ["src/buildtool/crypto", "hasher"]
     , ["src/buildtool/execution_api/local", "local"]
     , ["src/buildtool/file_system", "file_storage"]
+    , ["src/other_tools/just_mr/progress_reporting", "statistics"]
+    , ["src/other_tools/just_mr/progress_reporting", "progress"]
     ]
   }
 , "repo_fetch_map":
@@ -78,6 +80,8 @@
     [ ["src/other_tools/just_mr", "utils"]
     , ["src/buildtool/execution_api/local", "local"]
     , ["src/buildtool/file_system", "file_storage"]
+    , ["src/other_tools/just_mr/progress_reporting", "statistics"]
+    , ["src/other_tools/just_mr/progress_reporting", "progress"]
     ]
   }
 }
diff --git a/src/other_tools/ops_maps/content_cas_map.cpp b/src/other_tools/ops_maps/content_cas_map.cpp
index 7ed1f3df..f9b16588 100644
--- a/src/other_tools/ops_maps/content_cas_map.cpp
+++ b/src/other_tools/ops_maps/content_cas_map.cpp
@@ -17,6 +17,8 @@
 #include "src/buildtool/crypto/hasher.hpp"
 #include "src/buildtool/execution_api/local/local_cas.hpp"
 #include "src/buildtool/file_system/file_storage.hpp"
+#include "src/other_tools/just_mr/progress_reporting/progress.hpp"
+#include "src/other_tools/just_mr/progress_reporting/statistics.hpp"
 #include "src/other_tools/utils/curl_easy_handle.hpp"
 
 namespace {
@@ -50,6 +52,11 @@ auto CreateContentCASMap(JustMR::PathsPtr const& just_mr_paths,
                                          auto logger,
                                          auto /*unused*/,
                                          auto const& key) {
+        // start work reporting, but only if part of a distdir
+        if (key.origin_from_distdir) {
+            JustMRProgress::Instance().TaskTracker().Start(key.origin);
+            JustMRStatistics::Instance().IncrementQueuedCounter();
+        }
         // check if content already in CAS
         auto const& casf = LocalCAS<ObjectType::File>::Instance();
         auto digest = ArtifactDigest(key.content, 0, false);
@@ -122,6 +129,11 @@ auto CreateContentCASMap(JustMR::PathsPtr const& just_mr_paths,
             return;
         }
         (*setter)(true);
+        // report work done, but only if part of a distdir
+        if (key.origin_from_distdir) {
+            JustMRProgress::Instance().TaskTracker().Stop(key.origin);
+            JustMRStatistics::Instance().IncrementExecutedCounter();
+        }
     };
     return AsyncMapConsumer<ArchiveContent, bool>(ensure_in_cas, jobs);
 }
diff --git a/src/other_tools/ops_maps/repo_fetch_map.cpp b/src/other_tools/ops_maps/repo_fetch_map.cpp
index 890e8518..39474b12 100644
--- a/src/other_tools/ops_maps/repo_fetch_map.cpp
+++ b/src/other_tools/ops_maps/repo_fetch_map.cpp
@@ -16,6 +16,8 @@
 
 #include "src/buildtool/execution_api/local/local_cas.hpp"
 #include "src/buildtool/file_system/file_storage.hpp"
+#include "src/other_tools/just_mr/progress_reporting/progress.hpp"
+#include "src/other_tools/just_mr/progress_reporting/statistics.hpp"
 #include "src/other_tools/just_mr/utils.hpp"
 
 auto CreateRepoFetchMap(gsl::not_null<ContentCASMap*> const& content_cas_map,
@@ -26,16 +28,20 @@ auto CreateRepoFetchMap(gsl::not_null<ContentCASMap*> const& content_cas_map,
                                                    auto logger,
                                                    auto /* unused */,
                                                    auto const& key) {
-        // if archive available as a git tree ID stored to file,
-        // that's good enough, as it means it needs no fetching
-        auto tree_id_file = JustMR::Utils::GetArchiveTreeIDFile(
-            key.repo_type, key.archive.content);
+        // get corresponding distfile
         auto distfile =
             (key.archive.distfile ? key.archive.distfile.value()
                                   : std::filesystem::path(key.archive.fetch_url)
                                         .filename()
                                         .string());
+        // if archive available as a git tree ID stored to file,
+        // that's good enough, as it means it needs no fetching
+        auto tree_id_file = JustMR::Utils::GetArchiveTreeIDFile(
+            key.repo_type, key.archive.content);
         if (not FileSystemManager::Exists(tree_id_file)) {
+            // start work reporting
+            JustMRProgress::Instance().TaskTracker().Start(key.archive.origin);
+            JustMRStatistics::Instance().IncrementQueuedCounter();
             // make sure content is in CAS
             content_cas_map->ConsumeAfterKeysReady(
                 ts,
@@ -43,6 +49,7 @@ auto CreateRepoFetchMap(gsl::not_null<ContentCASMap*> const& content_cas_map,
                 [fetch_dir,
                  content = key.archive.content,
                  distfile,
+                 origin = key.archive.origin,
                  setter,
                  logger]([[maybe_unused]] auto const& values) {
                     // content is now in CAS
@@ -70,6 +77,9 @@ auto CreateRepoFetchMap(gsl::not_null<ContentCASMap*> const& content_cas_map,
                         }
                         // success
                         (*setter)(true);
+                        // report work done
+                        JustMRProgress::Instance().TaskTracker().Stop(origin);
+                        JustMRStatistics::Instance().IncrementExecutedCounter();
                     }
                     else {
                         (*logger)(
@@ -112,6 +122,8 @@ auto CreateRepoFetchMap(gsl::not_null<ContentCASMap*> const& content_cas_map,
                 }
                 // success
                 (*setter)(true);
+                // report cache hit
+                JustMRStatistics::Instance().IncrementCacheHitsCounter();
             }
             else {
                 (*logger)(fmt::format("Content {} could not be found in CAS",