Merge pull request #34567 from Dr15Jones/streamCacheGlobalOutputModule

Allow global::OutputModule to use a StreamCache
cms-sw · Jul 21, 2021 · 0805f82 · 0805f82
2 parents 9d63f03 + 651966b
commit 0805f82
Show file tree

Hide file tree

Showing 3 changed files with 56 additions and 4 deletions.
diff --git a/FWCore/Framework/interface/global/OutputModuleBase.h b/FWCore/Framework/interface/global/OutputModuleBase.h
@@ -128,8 +128,8 @@ namespace edm {
       void doBeginJob();
       void doEndJob();
 
-      void doBeginStream(StreamID id);
-      void doEndStream(StreamID id);
+      void doBeginStream(StreamID id) { doBeginStream_(id); }
+      void doEndStream(StreamID id) { doEndStream_(id); }
 
       bool doEvent(EventTransitionInfo const&, ActivityRegistry*, ModuleCallingContext const*);
       //For now this is a placeholder

diff --git a/FWCore/Framework/interface/global/outputmoduleAbilityToImplementor.h b/FWCore/Framework/interface/global/outputmoduleAbilityToImplementor.h
@@ -48,6 +48,37 @@ namespace edm {
         virtual void respondToCloseInputFile(FileBlock const&) = 0;
       };
 
+      template <typename T, typename C>
+      class StreamCacheHolder : public virtual T {
+      public:
+        explicit StreamCacheHolder(edm::ParameterSet const& iPSet) : OutputModuleBase(iPSet) {}
+        StreamCacheHolder(StreamCacheHolder<T, C> const&) = delete;
+        StreamCacheHolder<T, C>& operator=(StreamCacheHolder<T, C> const&) = delete;
+        ~StreamCacheHolder() override {
+          for (auto c : caches_) {
+            delete c;
+          }
+        }
+
+      protected:
+        C* streamCache(edm::StreamID iID) const { return caches_[iID.value()]; }
+
+      private:
+        void preallocStreams(unsigned int iNStreams) final { caches_.resize(iNStreams, static_cast<C*>(nullptr)); }
+        void doBeginStream_(StreamID id) final { caches_[id.value()] = beginStream(id).release(); }
+        void doEndStream_(StreamID id) final {
+          endStream(id);
+          delete caches_[id.value()];
+          caches_[id.value()] = nullptr;
+        }
+
+        virtual std::unique_ptr<C> beginStream(edm::StreamID) const = 0;
+        virtual void endStream(edm::StreamID) const {}
+
+        //When threaded we will have a container for N items whre N is # of streams
+        std::vector<C*> caches_;
+      };
+
       template <typename T, typename C>
       class RunCacheHolder : public virtual T {
       public:
@@ -117,6 +148,11 @@ namespace edm {
         typedef edm::global::outputmodule::LuminosityBlockCacheHolder<edm::global::OutputModuleBase, C> Type;
       };
 
+      template <typename C>
+      struct AbilityToImplementor<edm::StreamCache<C>> {
+        typedef edm::global::outputmodule::StreamCacheHolder<edm::global::OutputModuleBase, C> Type;
+      };
+
     }  // namespace outputmodule
   }    // namespace global
 }  // namespace edm

diff --git a/FWCore/Framework/src/WorkerT.cc b/FWCore/Framework/src/WorkerT.cc
@@ -76,6 +76,16 @@ namespace edm {
       static bool constexpr value = true;
     };
 
+    template <typename T>
+    struct has_only_stream_transition_functions {
+      static bool constexpr value = false;
+    };
+
+    template <>
+    struct has_only_stream_transition_functions<edm::global::OutputModuleBase> {
+      static bool constexpr value = true;
+    };
+
     struct DoNothing {
       template <typename... T>
       inline void operator()(const T&...) {}
@@ -456,7 +466,10 @@ namespace edm {
 
   template <typename T>
   inline void WorkerT<T>::implBeginStream(StreamID id) {
-    std::conditional_t<workerimpl::has_stream_functions<T>::value, workerimpl::DoBeginStream<T>, workerimpl::DoNothing>
+    std::conditional_t<workerimpl::has_stream_functions<T>::value or
+                           workerimpl::has_only_stream_transition_functions<T>::value,
+                       workerimpl::DoBeginStream<T>,
+                       workerimpl::DoNothing>
         might_call;
     might_call(this, id);
   }
@@ -469,7 +482,10 @@ namespace edm {
 
   template <typename T>
   inline void WorkerT<T>::implEndStream(StreamID id) {
-    std::conditional_t<workerimpl::has_stream_functions<T>::value, workerimpl::DoEndStream<T>, workerimpl::DoNothing>
+    std::conditional_t<workerimpl::has_stream_functions<T>::value or
+                           workerimpl::has_only_stream_transition_functions<T>::value,
+                       workerimpl::DoEndStream<T>,
+                       workerimpl::DoNothing>
         might_call;
     might_call(this, id);
   }