[hist] Implement RHistEngine::SnapshotAtomic

It returns a consistent copy of the histogram, during concurrent
filling. The implementation is based on a successful double collect
as described by Afek et al in their 1993 paper "Atomic Snapshots of
Shared Memory." The idea is to iterate twice over all bins until no
change is observed.

This is guaranteed to work for unweighted filling because bin contents
are monotonically increasing. With weighted filling and potentially
negative weights, there could be the situation where a bin goes back
to its previous content and a change is missed. This is probably
unlikely in practice, or can be entirely avoided by using RBinWithError
that also tracks the sum of squares, which is monotonically increasing
even for negative weights.

Author: hahnjo

hist/histv7/inc/ROOT/RHistEngine.hxx

@@ -20,6 +20,7 @@
 #include <cassert>
 #include <cstddef>
 #include <cstdint>
+#include <cstring>
 #include <stdexcept>
 #include <tuple>
 #include <type_traits>
@@ -811,6 +812,43 @@ public:
       return Slice(sliceSpecs);
    }
 
+   /// Create an atomic snapshot of this histogram engine.
+   ///
+   /// A snapshot is a consistent copy of the histogram, during concurrent filling. It is guaranteed that the returned
+   /// copy represents a state between the begin and end of the snapshot operation.
+   ///
+   /// Snapshotting a histogram engine with many bins can be an expensive operation.
+   ///
+   /// \return the atomic snapshot
+   RHistEngine SnapshotAtomic() const
+   {
+      static_assert(std::is_trivially_copyable_v<BinContentType>,
+                    "snapshotting requires a trivially copyable bin content type");
+
+      RHistEngine snapshot(fAxes.Get());
+      // Do a first collect.
+      for (std::size_t i = 0; i < fBinContents.size(); i++) {
+         Internal::AtomicLoad(&fBinContents[i], &snapshot.fBinContents[i]);
+      }
+
+      // Now do another collect. If no change is detected, the snapshot is consistent. Otherwise update the bin contents
+      // and try again.
+      BinContentType tmp;
+      bool changed;
+      do {
+         changed = false;
+         for (std::size_t i = 0; i < fBinContents.size(); i++) {
+            Internal::AtomicLoad(&fBinContents[i], &tmp);
+            if (std::memcmp(&tmp, &snapshot.fBinContents[i], sizeof(BinContentType))) {
+               std::memcpy(&snapshot.fBinContents[i], &tmp, sizeof(BinContentType));
+               changed = true;
+            }
+         }
+      } while (changed);
+
+      return snapshot;
+   }
+
    /// \}
 
    /// %ROOT Streamer function to throw when trying to store an object of this class.

hist/histv7/test/hist_engine_atomic.cxx

@@ -267,6 +267,64 @@ TEST(RHistEngine, StressFillAddAtomicWeight)
    EXPECT_EQ(engineA.GetBinContent(0), NOps * Weight);
 }
 
+TEST(RHistEngine, SnapshotAtomic)
+{
+   static constexpr std::size_t Bins = 20;
+   const RRegularAxis axis(Bins, {0, Bins});
+   RHistEngine<int> engineA({axis});
+
+   engineA.Fill(-100);
+   for (std::size_t i = 0; i < Bins; i++) {
+      engineA.Fill(i + 0.5);
+   }
+   engineA.Fill(100);
+
+   RHistEngine<int> engineB = engineA.SnapshotAtomic();
+   ASSERT_EQ(engineB.GetNDimensions(), 1);
+   ASSERT_EQ(engineB.GetTotalNBins(), Bins + 2);
+
+   EXPECT_EQ(engineB.GetBinContent(RBinIndex::Underflow()), 1);
+   for (auto index : axis.GetNormalRange()) {
+      EXPECT_EQ(engineB.GetBinContent(index), 1);
+   }
+   EXPECT_EQ(engineB.GetBinContent(RBinIndex::Overflow()), 1);
+}
+
+TEST(RHistEngine, StressSnapshotAtomic)
+{
+   static constexpr std::size_t Bins = 20;
+   static constexpr std::size_t NThreads = 4;
+   static constexpr std::size_t NFillsPerThread = 100000;
+   static constexpr std::size_t NSnapshots = 10000;
+
+   // Create a histogram with some bins that takes a bit of time to snapshot. The idea of this stress test is then to
+   // fill the first and last bin from multiple threads. If the snapshot is consistent, it must never have a bigger bin
+   // content in the last bin.
+   RHistEngine<int> engine(Bins, {0, Bins});
+   int first = 0, last = 0;
+
+   std::atomic_flag snapshotter;
+   StressInParallel(NThreads, [&] {
+      if (!snapshotter.test_and_set()) {
+         for (std::size_t i = 0; i < NSnapshots; i++) {
+            RHistEngine<int> snapshot = engine.SnapshotAtomic();
+            first = snapshot.GetBinContent(0);
+            last = snapshot.GetBinContent(Bins - 1);
+            if (last > first) {
+               return;
+            }
+         }
+      } else {
+         for (std::size_t i = 0; i < NFillsPerThread; i++) {
+            engine.FillAtomic(0.5);
+            engine.FillAtomic(Bins - 0.5);
+         }
+      }
+   });
+
+   EXPECT_GE(first, last);
+}
+
 TEST(RHistEngine_RBinWithError, AddAtomic)
 {
    static constexpr std::size_t Bins = 20;