Merge pull request #70 from albertziegenhagel/progress-support

Initial progress and cancelation support

snail/analysis/analysis.cpp

@@ -184,9 +184,11 @@ const call_tree_node& stacks_analysis::get_call_tree_node(call_tree_node::id_t i
     return id == call_tree_root.id ? call_tree_root : call_tree_nodes.at(id);
 }
 
-stacks_analysis snail::analysis::analyze_stacks(const samples_provider& provider,
-                                                unique_process_id       process_id,
-                                                const sample_filter&    filter)
+stacks_analysis snail::analysis::analyze_stacks(const samples_provider&           provider,
+                                                unique_process_id                 process_id,
+                                                const sample_filter&              filter,
+                                                const common::progress_listener*  progress_listener,
+                                                const common::cancellation_token* cancellation_token)
 {
     std::unordered_map<std::string, module_info::id_t>                               modules_by_name;
     std::unordered_map<std::pair<module_info::id_t, std::string>, module_info::id_t> functions_by_name;
@@ -217,10 +219,36 @@ stacks_analysis snail::analysis::analyze_stacks(const samples_provider& provider
         .children    = {},
     };
 
+    std::size_t total_work = 0;
+    if(progress_listener)
+    {
+        for(const auto& source_info : provider.sample_sources())
+        {
+            total_work += provider.count_samples(source_info.id, process_id, filter);
+        }
+    }
+    else
+    {
+        total_work = std::numeric_limits<std::size_t>::max();
+    }
+
+    common::progress_reporter progress(progress_listener, total_work);
+
+    bool cancel = false;
+
     for(const auto& source_info : provider.sample_sources())
     {
+        if(cancel) break;
+
         for(const auto& sample : provider.samples(source_info.id, process_id, filter))
         {
+            if(cancellation_token && cancellation_token->is_canceled())
+            {
+                cancel = true;
+                break;
+            }
+            progress.progress(1);
+
             if(sample.has_stack())
             {
                 ++result.call_tree_root.hits.get(source_info.id).total;
@@ -333,5 +361,7 @@ stacks_analysis snail::analysis::analyze_stacks(const samples_provider& provider
         }
     }
 
+    if(!cancel) progress.finish();
+
     return result;
 }

snail/analysis/analysis.hpp

@@ -2,6 +2,8 @@
 
 #include <limits>
 
+#include <snail/common/progress.hpp>
+
 #include <snail/analysis/data/call_tree.hpp>
 #include <snail/analysis/data/file.hpp>
 #include <snail/analysis/data/functions.hpp>
@@ -18,9 +20,11 @@ class samples_provider;
 
 struct stacks_analysis;
 
-stacks_analysis analyze_stacks(const samples_provider& provider,
-                               unique_process_id       process_id,
-                               const sample_filter&    filter = {});
+stacks_analysis analyze_stacks(const samples_provider&           provider,
+                               unique_process_id                 process_id,
+                               const sample_filter&              filter             = {},
+                               const common::progress_listener*  progress_listener  = nullptr,
+                               const common::cancellation_token* cancellation_token = nullptr);
 
 struct stacks_analysis
 {
@@ -41,9 +45,11 @@ struct stacks_analysis
     const std::vector<file_info>&     all_files() const;
 
 private:
-    friend stacks_analysis analyze_stacks(const samples_provider& provider,
-                                          unique_process_id       process_id,
-                                          const sample_filter&    filter);
+    friend stacks_analysis analyze_stacks(const samples_provider&           provider,
+                                          unique_process_id                 process_id,
+                                          const sample_filter&              filter,
+                                          const common::progress_listener*  progress_listener,
+                                          const common::cancellation_token* cancellation_token);
 
     // FIXME: This is a workaround: we would like to use the max of std::size_t, but since we will
     //        serialize those values to JSON and eventually handle them in JavaScript which cannot

snail/analysis/data_provider.hpp

@@ -6,6 +6,7 @@
 #include <string>
 
 #include <snail/common/generator.hpp>
+#include <snail/common/progress.hpp>
 
 #include <snail/analysis/data/ids.hpp>
 #include <snail/analysis/data/process.hpp>
@@ -78,7 +79,9 @@ class file_processor
 public:
     virtual ~file_processor() = default;
 
-    virtual void process(const std::filesystem::path& file_path) = 0;
+    virtual void process(const std::filesystem::path&      file_path,
+                         const common::progress_listener*  progress_listener  = nullptr,
+                         const common::cancellation_token* cancellation_token = nullptr) = 0;
 };
 
 class data_provider :

snail/analysis/diagsession_data_provider.cpp

@@ -96,7 +96,9 @@ diagsession_data_provider::~diagsession_data_provider()
     try_cleanup();
 }
 
-void diagsession_data_provider::process(const std::filesystem::path& file_path)
+void diagsession_data_provider::process(const std::filesystem::path&      file_path,
+                                        const common::progress_listener*  progress_listener,
+                                        const common::cancellation_token* cancellation_token)
 {
     static constexpr std::array<std::uint8_t, 4> zip_magic           = {0x50, 0x4b, 0x03, 0x04};
     static constexpr std::array<std::uint8_t, 8> compound_file_magic = {0xd0, 0xcf, 0x11, 0xe0, 0xa1, 0xb1, 0x1a, 0xe1};
@@ -180,10 +182,37 @@ void diagsession_data_provider::process(const std::filesystem::path& file_path)
         }
 
         std::ofstream etl_file_stream(*temp_etl_file_path_, std::ios::binary);
-        etl_file_entry.readContent(etl_file_stream);
+
+        common::progress_reporter progress(progress_listener,
+                                           etl_file_entry.getSize());
+
+        const auto result = archive.readEntry(
+            etl_file_entry,
+            [&etl_file_stream, &progress, cancellation_token](const void* data, libzippp_uint64 size) -> bool
+            {
+                // threat cancellation as failure to write
+                if(cancellation_token && cancellation_token->is_canceled()) return false;
+
+                etl_file_stream.write((char*)data, size);
+                progress.progress(size);
+                return bool(etl_file_stream);
+            });
+
+        // If this was canceled, there is no need to check the error code.
+        if(cancellation_token && cancellation_token->is_canceled()) return;
+
+        if(result != LIBZIPPP_OK)
+        {
+            // TODO: get more meaningful error message
+            throw std::runtime_error(std::format("Failed to extract ETL file from '{}' to '{}'", *archive_etl_file_path, temp_etl_file_path_->string()));
+        }
+
+        progress.finish();
     }
 
-    etl_data_provider::process(*temp_etl_file_path_);
+    etl_data_provider::process(*temp_etl_file_path_,
+                               progress_listener,
+                               cancellation_token);
 }
 
 void diagsession_data_provider::try_cleanup() noexcept

snail/analysis/diagsession_data_provider.hpp

@@ -15,7 +15,9 @@ class diagsession_data_provider : public etl_data_provider
 
     virtual ~diagsession_data_provider();
 
-    virtual void process(const std::filesystem::path& file_path) override;
+    virtual void process(const std::filesystem::path&      file_path,
+                         const common::progress_listener*  progress_listener,
+                         const common::cancellation_token* cancellation_token) override;
 
 private:
     std::optional<std::filesystem::path> temp_etl_file_path_;

snail/analysis/etl_data_provider.cpp

@@ -182,12 +182,16 @@ etl_data_provider::etl_data_provider(pdb_symbol_find_options find_options,
 
 etl_data_provider::~etl_data_provider() = default;
 
-void etl_data_provider::process(const std::filesystem::path& file_path)
+void etl_data_provider::process(const std::filesystem::path&      file_path,
+                                const common::progress_listener*  progress_listener,
+                                const common::cancellation_token* cancellation_token)
 {
     process_context_ = std::make_unique<detail::etl_file_process_context>();
 
     etl::etl_file file(file_path);
-    file.process(process_context_->observer());
+    file.process(process_context_->observer(),
+                 progress_listener,
+                 cancellation_token);
 
     process_context_->finish();
 

snail/analysis/etl_data_provider.hpp

@@ -26,7 +26,9 @@ class etl_data_provider : public data_provider
 
     virtual ~etl_data_provider();
 
-    virtual void process(const std::filesystem::path& file_path) override;
+    virtual void process(const std::filesystem::path&      file_path,
+                         const common::progress_listener*  progress_listener,
+                         const common::cancellation_token* cancellation_token) override;
 
     virtual const analysis::session_info& session_info() const override;
 

snail/analysis/perf_data_data_provider.cpp

@@ -155,12 +155,16 @@ perf_data_data_provider::perf_data_data_provider(dwarf_symbol_find_options find_
 
 perf_data_data_provider::~perf_data_data_provider() = default;
 
-void perf_data_data_provider::process(const std::filesystem::path& file_path)
+void perf_data_data_provider::process(const std::filesystem::path&      file_path,
+                                      const common::progress_listener*  progress_listener,
+                                      const common::cancellation_token* cancellation_token)
 {
     process_context_ = std::make_unique<detail::perf_data_file_process_context>();
 
     perf_data::perf_data_file file(file_path);
-    file.process(process_context_->observer());
+    file.process(process_context_->observer(),
+                 progress_listener,
+                 cancellation_token);
 
     process_context_->finish();
 

snail/analysis/perf_data_data_provider.hpp

@@ -29,7 +29,9 @@ class perf_data_data_provider : public data_provider
 
     virtual ~perf_data_data_provider();
 
-    virtual void process(const std::filesystem::path& file_path) override;
+    virtual void process(const std::filesystem::path&      file_path,
+                         const common::progress_listener*  progress_listener,
+                         const common::cancellation_token* cancellation_token) override;
 
     virtual const analysis::session_info& session_info() const override;
 

snail/common/CMakeLists.txt

@@ -11,6 +11,7 @@ target_sources(common
     trim.cpp
     wildcard.cpp
     ms_xca_decompression.cpp
+    progress.cpp
 )
 
 target_link_libraries(common 

snail/common/progress.cpp

@@ -0,0 +1,75 @@
+
+#include <snail/common/progress.hpp>
+
+#include <algorithm>
+#include <cmath>
+
+using namespace snail::common;
+
+progress_listener::progress_listener(double resolution) :
+    resolution_(resolution)
+{}
+
+double progress_listener::resolution() const
+{
+    return resolution_;
+}
+
+void cancellation_token::cancel()
+{
+    cancel_ = true;
+}
+
+bool cancellation_token::is_canceled() const
+{
+    return cancel_.load();
+}
+
+progress_reporter::progress_reporter(const progress_listener* listener,
+                                     work_type                total_work) :
+    listener_(listener),
+    total_work_(total_work),
+    step_work_(((listener != nullptr) ? listener->resolution() : 1.0) * total_work_),
+    current_work_(0),
+    next_report_step_(1),
+    next_report_work_(static_cast<work_type>(std::ceil(step_work_ * next_report_step_)))
+{
+    if(listener_ != nullptr)
+    {
+        listener_->start();
+        listener_->report(0.0);
+    }
+}
+
+void progress_reporter::progress(work_type work)
+{
+    current_work_ += work;
+    if(current_work_ < next_report_work_) return;
+
+    const auto current_progress = double(current_work_) / total_work_;
+    if(listener_ != nullptr) listener_->report(current_progress);
+
+    const auto current_step = static_cast<unsigned int>(double(current_work_) / step_work_);
+
+    next_report_step_ = std::max(next_report_step_ + 1u, current_step + 1u);
+
+    next_report_work_ = static_cast<work_type>(std::ceil(step_work_ * next_report_step_));
+
+    if(current_work_ < total_work_ && next_report_work_ > total_work_)
+    {
+        // Make sure we always report at 100%
+        // This does also fix issues with rounding around 100%.
+        next_report_work_ = total_work_;
+        --next_report_step_; // Since the value we will report at next is not a real
+                             // step, we should not count it as one.
+                             // The easiest thing is to simply decrement the step counter
+                             // here.
+    }
+}
+
+void progress_reporter::finish()
+{
+    if(listener_ == nullptr) return;
+    if(current_work_ < total_work_) listener_->report(1.0);
+    listener_->finish();
+}

snail/common/progress.hpp

@@ -0,0 +1,58 @@
+#pragma once
+
+#include <atomic>
+#include <cstddef>
+
+namespace snail::common {
+
+class progress_listener
+{
+public:
+    progress_listener(double resolution = 0.01);
+
+    double resolution() const;
+
+    virtual void start() const                 = 0;
+    virtual void report(double progress) const = 0;
+    virtual void finish() const                = 0;
+
+private:
+    double resolution_;
+};
+
+class cancellation_token
+{
+public:
+    void cancel();
+    bool is_canceled() const;
+
+private:
+    std::atomic<bool> cancel_ = false;
+};
+
+class progress_reporter
+{
+public:
+    using work_type = std::size_t;
+
+    progress_reporter(const progress_listener* listener,
+                      work_type                total_work);
+
+    void start();
+    void progress(work_type work);
+    void finish();
+
+private:
+    const progress_listener* listener_;
+
+    const work_type total_work_;
+
+    const double step_work_;
+
+    work_type current_work_;
+
+    unsigned int next_report_step_;
+    work_type    next_report_work_;
+};
+
+} // namespace snail::common