[hist] Pause filling during snapshot

FillAtomic is much faster than SnapshotAtomic, so in case of heavy
contention it would starve the snapshot. By adding a new atomic flag
FillAtomic can pause while a Snapshot is running. The successful
double collect is still needed if a thread is already in FillAtomic.

If SnapshotAtomic is not used, the performance impact of the added
check is minimal / not measurable because fSnapshotInProgress is
always false and already in the cache line of the RHistEngine.

Author: hahnjo

hist/histv7/inc/ROOT/RHistEngine.hxx

@@ -84,6 +84,9 @@ class RHistEngine final {
    /// The bin contents for this histogram
    std::vector<BinContentType> fBinContents;
 
+   /// Flag to pause filling while a snapshot is ongoing
+   mutable std::atomic<bool> fSnapshotInProgress{false}; //!
+
 public:
    /// Construct a histogram engine.
    ///
@@ -134,7 +137,7 @@ public:
    /// Efficiently move construct a histogram engine.
    ///
    /// After this operation, the moved-from object is invalid.
-   RHistEngine(RHistEngine &&) = default;
+   RHistEngine(RHistEngine &&rhs) noexcept : fAxes(std::move(rhs.fAxes)), fBinContents(std::move(rhs.fBinContents)) {}
 
    /// The copy assignment operator is deleted.
    ///
@@ -144,7 +147,12 @@ public:
    /// Efficiently move a histogram engine.
    ///
    /// After this operation, the moved-from object is invalid.
-   RHistEngine &operator=(RHistEngine &&) = default;
+   RHistEngine &operator=(RHistEngine &&rhs) noexcept
+   {
+      std::swap(fAxes, rhs.fAxes);
+      std::swap(fBinContents, rhs.fBinContents);
+      return *this;
+   }
 
    ~RHistEngine() = default;
 
@@ -513,6 +521,10 @@ public:
    template <typename... A>
    void FillAtomic(const std::tuple<A...> &args)
    {
+      while (fSnapshotInProgress.load(std::memory_order_relaxed)) {
+         // Spin while a snapshot is running
+      }
+
       // We could rely on RAxes::ComputeGlobalIndex to check the number of arguments, but its exception message might
       // be confusing for users.
       if (sizeof...(A) != GetNDimensions()) {
@@ -537,6 +549,10 @@ public:
    {
       static_assert(SupportsWeightedFilling, "weighted filling is not supported for integral bin content types");
 
+      while (fSnapshotInProgress.load(std::memory_order_relaxed)) {
+         // Spin while a snapshot is running
+      }
+
       // We could rely on RAxes::ComputeGlobalIndex to check the number of arguments, but its exception message might
       // be confusing for users.
       if (sizeof...(A) != GetNDimensions()) {
@@ -562,6 +578,10 @@ public:
       static_assert(std::is_class_v<BinContentType>,
                     "user-defined weight types are only supported for user-defined bin content types");
 
+      while (fSnapshotInProgress.load(std::memory_order_relaxed)) {
+         // Spin while a snapshot is running
+      }
+
       // We could rely on RAxes::ComputeGlobalIndex to check the number of arguments, but its exception message might
       // be confusing for users.
       if (sizeof...(A) != GetNDimensions()) {
@@ -583,6 +603,10 @@ public:
    {
       static_assert(sizeof...(A) >= 1, "need at least one argument to Fill");
       if constexpr (sizeof...(A) >= 1) {
+         while (fSnapshotInProgress.load(std::memory_order_relaxed)) {
+            // Spin while a snapshot is running
+         }
+
          auto t = std::forward_as_tuple(args...);
          if constexpr (std::is_same_v<typename Internal::LastType<A...>::type, RWeight>) {
             static_assert(SupportsWeightedFilling, "weighted filling is not supported for integral bin content types");
@@ -846,6 +870,12 @@ public:
       static_assert(std::is_trivially_copyable_v<BinContentType>,
                     "snapshotting requires a trivially copyable bin content type");
 
+      do {
+         while (fSnapshotInProgress.load(std::memory_order_relaxed)) {
+            // Spin while another snapshot is running
+         }
+      } while (fSnapshotInProgress.exchange(true, std::memory_order_relaxed));
+
       RHistEngine snapshot(fAxes.Get());
       // Do a first collect.
       for (std::size_t i = 0; i < fBinContents.size(); i++) {
@@ -871,6 +901,8 @@ public:
          }
       } while (changed);
 
+      fSnapshotInProgress.store(false, std::memory_order_relaxed);
+
       return snapshot;
    }
 

hist/histv7/test/hist_engine_atomic.cxx

@@ -310,7 +310,37 @@ TEST(RHistEngine, SnapshotAtomic)
    EXPECT_EQ(engineB.GetBinContent(RBinIndex::Overflow()), 1);
 }
 
+// Stress calling SnapshotAtomic from multiple threads.
 TEST(RHistEngine, StressSnapshotAtomic)
+{
+   static constexpr std::size_t Bins = 20;
+   static constexpr std::size_t NThreads = 4;
+   static constexpr std::size_t NSnapshotsPerThread = 10000;
+   static constexpr int ExpectedBinContent0 = 1;
+
+   // Create a histogram with some bins that takes a bit of time to snapshot.
+   RHistEngine<int> engine(Bins, {0, Bins});
+   engine.Fill(0.5);
+
+   std::atomic<int> binContent0 = ExpectedBinContent0;
+
+   StressInParallel(NThreads, [&] {
+      for (std::size_t i = 0; i < NSnapshotsPerThread; i++) {
+         auto snapshot = engine.SnapshotAtomic();
+         // compare_exchange wants a non-const reference...
+         int expected = ExpectedBinContent0;
+         int actual = snapshot.GetBinContent(0);
+         if (actual != expected) {
+            binContent0.compare_exchange_strong(expected, actual);
+         }
+      }
+   });
+
+   EXPECT_EQ(binContent0, ExpectedBinContent0);
+}
+
+// Stress calling SnapshotAtomic while other threads call FillAtomic.
+TEST(RHistEngine, StressFillSnapshotAtomic)
 {
    static constexpr std::size_t Bins = 20;
    static constexpr std::size_t NThreads = 4;
@@ -485,7 +515,7 @@ TEST(RHistEngine_RBinWithError, SnapshotAtomic)
    EXPECT_EQ(engineB.GetBinContent(RBinIndex::Overflow()).fSum, 1);
 }
 
-TEST(RHistEngine_RBinWithError, StressSnapshotAtomic)
+TEST(RHistEngine_RBinWithError, StressFillSnapshotAtomic)
 {
    static constexpr std::size_t Bins = 20;
    static constexpr std::size_t NThreads = 4;

hist/histv7/test/hist_io.cxx

@@ -10,6 +10,8 @@ static void ExpectThrowOnWriteObject(const T &obj)
 {
    ROOT::TestSupport::CheckDiagsRAII diagRAII;
    diagRAII.optionalDiag(kWarning, "TKey::TKey", "no public constructor", /*matchFullMessage=*/false);
+   diagRAII.optionalDiag(kWarning, "TStreamerInfo::Build", "data member \"fSnapshotInProgress\" will not be saved",
+                         /*matchFullMessage=*/false);
 
    TMemFile f("mem.root", "RECREATE");
    EXPECT_THROW(f.WriteObject(&obj, "o"), std::runtime_error);