Merge pull request #9135 from mark-grimes/guessCollectionSizes

Backport RunningAverage and use for PFRecHitProducer reserve

FWCore/Utilities/interface/RunningAverage.h

@@ -0,0 +1,41 @@
+#ifndef FWCore_Utilities_RunningAverage_H
+#define FWCore_Utilities_RunningAverage_H
+#include <atomic>
+#include <algorithm>
+#include <array>
+
+namespace edm {
+// keeps the running average of the last N entries
+// thread safe, fast: does not garantee precise update in case of collision
+  class RunningAverage {
+  public:
+    static constexpr int N = 16;  // better be a power of 2
+    explicit RunningAverage(unsigned int k=4) : m_mean(N*k), m_curr(0) {
+      for (auto & i : m_buffer) i=k; 
+    }
+
+    int mean() const { return m_mean/N;}
+
+    int upper() const { auto lm = mean(); return lm+=(std::abs(m_buffer[0]-lm)+std::abs(m_buffer[N/2]-lm)); }  // about 2 sigma
+
+   void update(unsigned int q) {
+      int e=m_curr;
+      while(!m_curr.compare_exchange_weak(e,e+1)); 
+      int k = (N-1)&e;
+      int old = m_buffer[k];
+      if (!m_buffer[k].compare_exchange_strong(old,q)) return;
+      m_mean+= (q-old); 
+    }
+
+
+private:
+    std::array<std::atomic<int>,N> m_buffer;
+    std::atomic<int> m_mean;
+    std::atomic<int> m_curr;
+
+  };
+}
+
+#endif
+
+

FWCore/Utilities/test/BuildFile.xml

@@ -42,3 +42,7 @@
 </bin>
 <bin   file="EDGetToken_t.cpp">
 </bin>
+
+<bin   file="RunningAverage_t.cpp">
+  <use name="tbb"/>
+</bin>

FWCore/Utilities/test/RunningAverage_t.cpp

@@ -0,0 +1,70 @@
+#define private public
+#include "FWCore/Utilities/interface/RunningAverage.h"
+#undef private
+
+namespace {
+
+  edm::RunningAverage localRA;
+
+}
+
+#include "tbb/parallel_for.h"
+#include "tbb/task_scheduler_init.h"
+#include <iostream>
+#include <vector>
+#include <atomic>
+#include <random>
+#include <algorithm>
+#include <type_traits>
+
+int main() {
+
+
+  //tbb::task_scheduler_init init;  // Automatic number of threads
+  tbb::task_scheduler_init init(tbb::task_scheduler_init::default_num_threads());  // Explicit number of threads
+
+  // std::random_device rd;
+  std::mt19937 e2; // (rd());
+  std::normal_distribution<> normal_dist(1000., 200.);
+
+
+
+  int kk=0;
+
+
+  int n=2000;
+
+  int res[n];
+  int qq[n]; for (auto & y:qq) y = std::max(0.,normal_dist(e2));
+
+  auto theLoop = [&](int i, std::vector<float>& v ) {
+    kk++;
+    v.reserve(res[i]=localRA.upper());
+    v.resize(qq[i]);
+    localRA.update(v.size());
+    std::vector<float> t;
+    swap(v,t);
+  };
+
+
+  tbb::parallel_for(
+		    tbb::blocked_range<size_t>(0,n),
+		    [&](const tbb::blocked_range<size_t>& r) {
+		      std::vector<float> v; // Had to move this declaration here because there's no thread_local in this version of gcc (in 7_5_X it's declared above)
+		      for (size_t i=r.begin();i<r.end();++i) theLoop(i,v);
+    }
+  );
+
+
+  auto mm = std::max_element(res,res+n);
+  std::cout << kk << ' ' << localRA.m_curr << ' ' << localRA.mean() << std::endl;
+  for (auto & i : localRA.m_buffer) std::cout << i << ' ';
+  std::cout << std::endl;
+  std::cout << std::accumulate(res,res+n,0)/n 
+	    << ' ' << *std::min_element(res+16,res+n) << ',' << *mm << std::endl;
+
+  return 0;
+
+
+}
+

RecoParticleFlow/PFClusterProducer/plugins/PFRecHitProducer.cc

@@ -47,6 +47,9 @@ void
    std::auto_ptr<reco::PFRecHitCollection> out(new reco::PFRecHitCollection );
    std::auto_ptr<reco::PFRecHitCollection> cleaned(new reco::PFRecHitCollection );
 
+   out->reserve( outputSizeGuess_.upper() );
+   cleaned->reserve( cleanedOutputSizeGuess_.upper() );
+
    navigator_->beginEvent(iSetup);
 
    for (unsigned int i=0;i<creators_.size();++i) {
@@ -75,6 +78,9 @@ void
      }
    }
 
+   outputSizeGuess_.update( out->size() );
+   cleanedOutputSizeGuess_.update( cleaned->size() );
+
    iEvent.put(out,"");
    iEvent.put(cleaned,"Cleaned");
    hitmap.clear();

RecoParticleFlow/PFClusterProducer/plugins/PFRecHitProducer.h

@@ -9,6 +9,7 @@
 #include "FWCore/Framework/interface/Event.h"
 #include "FWCore/Framework/interface/MakerMacros.h"
 
+#include "FWCore/Utilities/interface/RunningAverage.h"
 #include "FWCore/ParameterSet/interface/ParameterSet.h"
 
 #include "RecoParticleFlow/PFClusterProducer/interface/PFRecHitCreatorBase.h"
@@ -30,6 +31,8 @@ class PFRecHitProducer : public edm::EDProducer {
       virtual void produce(edm::Event&, const edm::EventSetup&) override;
       std::vector<std::unique_ptr<PFRecHitCreatorBase> > creators_;
       std::unique_ptr<PFRecHitNavigatorBase> navigator_;
+      edm::RunningAverage outputSizeGuess_; ///< Use this to try and accurately reserve space for the output
+      edm::RunningAverage cleanedOutputSizeGuess_; ///< Use this to try and accurately reserve space for the output
 };
 
 #include "FWCore/Framework/interface/MakerMacros.h"