[RDF] Refactor progress bar and shorten its output.

This removes redundant information from the line printed to the
terminal and shortens the output. Otherwise, the progress bar
frequently overflows the terminal.

When the progress bar prints to a file, reduce the frequency to
every 10 seconds, and limit its width to 60 chars.

Furthermore, significantly improve how completion is estimated.
The following two heuristics are employed:
- Files that have been opened count as
  fractionOfFilesAlreadyOpened * eventsProcessed/totalEvents
  This tracks the progress of all files for which the number of events
  has been seen.
- Files that have not been opened count as 1/totalFiles until they have
  been opened. This means that the progress bar e.g. can't reach 50% if
  half of the files haven't been opened yet.
This change significantly reduces the jumps of the progress bar when a
new file is opened.

Finally, significantly refactor the code:
- Handle locking logic for updates with a single RAII, update locks to C++17.
- Remove a lock and mutex that didn't have any effect.
- Reduce repeated function calls to functions that hold locks.
- Simplify computation of average number of events.
- Relax memory order of the atomics to what's necessary.
- Outline as many functions as possible. Since RDFHelpers.hxx goes
  through JITting and pcms, it gets compiled frequently, so outlining is
  probably beneficial.
- Make a lot of members const to avoid unintenional modifications. This
  might be interesting because it's an MT context.
- Collect helper functions in one anonymous namespace.
- Remove a constructor argument that was ignored.
- Don't unconditionally clear cout at the end. The bar might be going to
  a different stream.

Author: hageboeck

tree/dataframe/inc/ROOT/RDFHelpers.hxx

@@ -333,66 +333,49 @@ class ProgressBarAction;
 /// ~~~
 class ProgressHelper {
 private:
+   std::size_t ComputeTotalEvents() const;
    double EvtPerSec() const;
    std::pair<std::size_t, std::chrono::seconds> RecordEvtCountAndTime();
-   void PrintStats(std::ostream &stream, std::size_t currentEventCount, std::chrono::seconds totalElapsedSeconds) const;
+   void PrintProgressAndStats(std::ostream &stream, std::size_t currentEventCount,
+                              std::chrono::seconds totalElapsedSeconds) const;
    void PrintStatsFinal(std::ostream &stream, std::chrono::seconds totalElapsedSeconds) const;
-   void PrintProgressBar(std::ostream &stream, std::size_t currentEventCount) const;
 
-   std::chrono::time_point<std::chrono::system_clock> fBeginTime = std::chrono::system_clock::now();
-   std::chrono::time_point<std::chrono::system_clock> fLastPrintTime = fBeginTime;
-   std::chrono::seconds fPrintInterval{1};
+   bool const fIsTTY;
+   bool const fUseShellColours;
 
    std::atomic<std::size_t> fProcessedEvents{0};
    std::size_t fLastProcessedEvents{0};
-   std::size_t fIncrement;
-
-   mutable std::mutex fSampleNameToEventEntriesMutex;
-   std::map<std::string, ULong64_t> fSampleNameToEventEntries; // Filename, events in the file
+   std::size_t const fIncrement;
+   unsigned int const fNColumns;
+   unsigned int const fTotalFiles;
 
-   std::array<double, 20> fEventsPerSecondStatistics;
-   std::size_t fEventsPerSecondStatisticsIndex{0};
+   std::array<double, 10> fEventsPerSecondStatistics;
+   unsigned int fEventsPerSecondStatisticsCounter{0};
 
-   unsigned int fBarWidth;
-   unsigned int fTotalFiles;
+   std::chrono::time_point<std::chrono::system_clock> const fBeginTime = std::chrono::system_clock::now();
+   std::chrono::time_point<std::chrono::system_clock> fLastPrintTime = fBeginTime;
+   std::chrono::seconds const fPrintInterval;
 
-   std::mutex fPrintMutex;
-   bool fIsTTY;
-   bool fUseShellColours;
+   std::mutex fUpdateMutex; // Mutex to ensure that only one thread updates the progress bar
 
-   std::shared_ptr<TTree> fTree{nullptr};
+   mutable std::mutex fSampleNameToEventEntriesMutex;          // Mutex to protect access to the below map
+   std::map<std::string, ULong64_t> fSampleNameToEventEntries; // Filename, events in the file
 
 public:
    /// Create a progress helper.
    /// \param increment RDF callbacks are called every `n` events. Pass this `n` here.
-   /// \param totalFiles read total number of files in the RDF.
-   /// \param progressBarWidth Number of characters the progress bar will occupy.
-   /// \param printInterval Update every stats every `n` seconds.
+   /// \param totalFiles number of files read in the RDF.
+   /// \param printInterval Update stats every `n` seconds.
    /// \param useColors Use shell colour codes to colour the output. Automatically disabled when
    /// we are not writing to a tty.
-   ProgressHelper(std::size_t increment, unsigned int totalFiles = 1, unsigned int progressBarWidth = 40,
-                  unsigned int printInterval = 1, bool useColors = true);
+   ProgressHelper(std::size_t increment, unsigned int totalFiles, unsigned int printInterval = 0,
+                  bool useColors = true);
 
    ~ProgressHelper() = default;
 
    friend class ProgressBarAction;
 
-   /// Register a new sample for completion statistics.
-   /// \see ROOT::RDF::RInterface::DefinePerSample().
-   /// The *id.AsString()* refers to the name of the currently processed file.
-   /// The idea is to populate the  event entries in the *fSampleNameToEventEntries* map
-   /// by selecting the greater of the two values:
-   /// *id.EntryRange().second* which is the upper event entry range of the processed sample
-   /// and the current value of the event entries in the *fSampleNameToEventEntries* map.
-   /// In the single threaded case, the two numbers are the same as the entry range corresponds
-   /// to the number of events in an individual file (each sample is simply a single file).
-   /// In the multithreaded case, the idea is to accumulate the higher event entry value until
-   /// the total number of events in a given file is reached.
-   void registerNewSample(unsigned int /*slot*/, const ROOT::RDF::RSampleInfo &id)
-   {
-      std::lock_guard<std::mutex> lock(fSampleNameToEventEntriesMutex);
-      fSampleNameToEventEntries[id.AsString()] = std::max(id.NEntriesTotal(), fSampleNameToEventEntries[id.AsString()]);
-   }
+   void registerNewSample(unsigned int /*slot*/, const ROOT::RDF::RSampleInfo &id);
 
    /// Thread-safe callback for RDataFrame.
    /// It will record elapsed times and event statistics, and print a progress bar every n seconds (set by the
@@ -414,7 +397,7 @@ public:
       // ***************************************************
       // Warning: Here, everything needs to be thread safe:
       // ***************************************************
-      fProcessedEvents += fIncrement;
+      fProcessedEvents.fetch_add(fIncrement, std::memory_order_relaxed);
 
       // We only print every n seconds.
       if (duration_cast<seconds>(system_clock::now() - fLastPrintTime) < fPrintInterval) {
@@ -424,40 +407,22 @@ public:
       // ***************************************************
       // Protected by lock from here:
       // ***************************************************
-      if (!fPrintMutex.try_lock())
+      std::unique_lock<std::mutex> lockGuard(fUpdateMutex, std::try_to_lock);
+      if (!lockGuard)
          return;
-      std::lock_guard<std::mutex> lockGuard(fPrintMutex, std::adopt_lock);
 
-      std::size_t eventCount;
-      seconds elapsedSeconds;
-      std::tie(eventCount, elapsedSeconds) = RecordEvtCountAndTime();
+      auto const [eventCount, elapsedSeconds] = RecordEvtCountAndTime();
 
       if (fIsTTY)
          std::cout << "\r";
 
-      PrintProgressBar(std::cout, eventCount);
-      PrintStats(std::cout, eventCount, elapsedSeconds);
+      PrintProgressAndStats(std::cout, eventCount, elapsedSeconds);
 
       if (fIsTTY)
          std::cout << std::flush;
       else
          std::cout << std::endl;
    }
-
-   std::size_t ComputeNEventsSoFar() const
-   {
-      std::unique_lock<std::mutex> lock(fSampleNameToEventEntriesMutex);
-      std::size_t result = 0;
-      for (const auto &item : fSampleNameToEventEntries)
-         result += item.second;
-      return result;
-   }
-
-   unsigned int ComputeCurrentFileIdx() const
-   {
-      std::unique_lock<std::mutex> lock(fSampleNameToEventEntriesMutex);
-      return fSampleNameToEventEntries.size();
-   }
 };
 } // namespace Experimental
 } // namespace RDF