Merge histograms on the fly (if necessary)

include/diagnostics/simple_charge_stats.hpp

@@ -22,6 +22,8 @@
 
 #pragma once
 
+#include <mutex>
+
 #include <math/histogram.hpp>
 #include <iact_data/event_visitor.hpp>
 #include <iact_data/waveform_treatment_event_visitor.hpp>
@@ -96,6 +98,8 @@ class SimpleChargeStatsParallelEventVisitor:
       delete ped_wf_q2_sum_vs_time;
     }
 
+    void insert_from_and_clear(SingleGainChannelHists* from);
+
     calin::math::histogram::Histogram1D* all_pedwin_1_sum_vs_time;
     calin::math::histogram::Histogram1D* all_pedwin_q_sum_vs_time;
     calin::math::histogram::Histogram1D* all_pedwin_q2_sum_vs_time;
@@ -208,6 +212,8 @@ class SimpleChargeStatsParallelEventVisitor:
     const SingleGainChannelHists& hists,
     calin::ix::diagnostics::simple_charge_stats::PartialOneGainCameraSimpleChargeStats* partials);
 
+  void merge_time_histograms_if_necessary();
+
   void record_one_gain_channel_data(const calin::ix::iact_data::telescope_event::TelescopeEvent* event,
     const calin::iact_data::waveform_treatment_event_visitor::OptimalWindowSumWaveformTreatmentParallelEventVisitor* sum_visitor,
     unsigned ichan, double elapsed_event_time,
@@ -227,6 +233,7 @@ class SimpleChargeStatsParallelEventVisitor:
   void record_dual_visitor_data(uint64_t seq_index);
 
   SimpleChargeStatsParallelEventVisitor* parent_ = nullptr;
+  std::mutex on_the_fly_merge_lock_;
   calin::ix::diagnostics::simple_charge_stats::SimpleChargeStatsConfig config_;
   calin::iact_data::waveform_treatment_event_visitor::OptimalWindowSumWaveformTreatmentParallelEventVisitor* high_gain_visitor_ = nullptr;
   calin::iact_data::waveform_treatment_event_visitor::OptimalWindowSumWaveformTreatmentParallelEventVisitor* low_gain_visitor_ = nullptr;

src/diagnostics/simple_charge_stats.cpp

@@ -473,6 +473,57 @@ void SimpleChargeStatsParallelEventVisitor::dump_single_gain_camera_hists_to_par
   delete hp;
 }
 
+void SimpleChargeStatsParallelEventVisitor::SingleGainChannelHists::insert_from_and_clear(
+  SimpleChargeStatsParallelEventVisitor::SingleGainChannelHists* from)
+{
+  all_pedwin_1_sum_vs_time->insert_hist(*from->all_pedwin_1_sum_vs_time);
+  from->all_pedwin_1_sum_vs_time->clear();
+
+  all_pedwin_q_sum_vs_time->insert_hist(*from->all_pedwin_q_sum_vs_time);
+  from->all_pedwin_q_sum_vs_time->clear();
+
+  all_pedwin_q2_sum_vs_time->insert_hist(*from->all_pedwin_q2_sum_vs_time);
+  from->all_pedwin_q2_sum_vs_time->clear();
+
+  ped_wf_1_sum_vs_time->insert_hist(*from->ped_wf_1_sum_vs_time);
+  from->ped_wf_1_sum_vs_time->clear();
+
+  ped_wf_q_sum_vs_time->insert_hist(*from->ped_wf_q_sum_vs_time);
+  from->ped_wf_q_sum_vs_time->clear();
+
+  ped_wf_q2_sum_vs_time->insert_hist(*from->ped_wf_q2_sum_vs_time);
+  from->ped_wf_q2_sum_vs_time->clear();
+}
+
+void SimpleChargeStatsParallelEventVisitor::merge_time_histograms_if_necessary()
+{
+  // This function added to do on-the-fly merging of the time histograms if they get very
+  // large. Only used on very long runs with more than 1000 bins (corresponding to 5000 
+  // seconds by default), to avoid memory problems in multi-threaded envrionment, since these
+  // histograms are duplicated across all the threads
+
+  if(parent_ == nullptr or camera_hists_->high_gain->all_pedwin_1_sum_vs_time->size()<1000) {
+    return;
+  }
+
+  std::lock_guard<std::mutex> lock { parent_->on_the_fly_merge_lock_ };
+
+  for(int ichan = 0; ichan<chan_hists_.size(); ichan++) {
+    if(chan_hists_[ichan]->high_gain) {
+      parent_->chan_hists_[ichan]->high_gain->insert_from_and_clear(chan_hists_[ichan]->high_gain);
+    }
+    if(chan_hists_[ichan]->low_gain) {
+      parent_->chan_hists_[ichan]->low_gain->insert_from_and_clear(chan_hists_[ichan]->low_gain);
+    }
+  }
+  if(camera_hists_->high_gain) {
+    parent_->camera_hists_->high_gain->insert_from_and_clear(camera_hists_->high_gain);
+  }
+  if(camera_hists_->low_gain) {
+    parent_->camera_hists_->low_gain->insert_from_and_clear(camera_hists_->low_gain);
+  }
+}
+
 bool SimpleChargeStatsParallelEventVisitor::leave_telescope_run(
   calin::ix::provenance::chronicle::ProcessingRecord* processing_record)
 {