Merge pull request #1575 from fwyzard/add_StreamContext_to_Pre_PostSo…

…urce Multithreading fixes -- Add a StreamContext to the Pre/PostSource signals
cms-sw · Nov 27, 2013 · 4953851 · 4953851
2 parents 0c16833 + 9f94d72
commit 4953851
Show file tree

Hide file tree

Showing 27 changed files with 115 additions and 92 deletions.
diff --git a/DQMServices/Core/src/DQMService.cc b/DQMServices/Core/src/DQMService.cc
@@ -34,6 +34,10 @@ static void
 restrictDQMAccessM(const edm::ModuleDescription &)
 { restrictDQMAccess(); }
 
+static void
+restrictDQMAccessS(edm::StreamID)
+{ restrictDQMAccess(); }
+
 /// Release access to the DQM core.
 static void
 releaseDQMAccess(void)
@@ -43,6 +47,10 @@ static void
 releaseDQMAccessM(const edm::ModuleDescription &)
 { releaseDQMAccess(); }
 
+static void
+releaseDQMAccessS(edm::StreamID)
+{ releaseDQMAccess(); }
+
 // -------------------------------------------------------------------
 DQMService::DQMService(const edm::ParameterSet &pset, edm::ActivityRegistry &ar)
   : store_(&*edm::Service<DQMStore>()),
@@ -53,8 +61,8 @@ DQMService::DQMService(const edm::ParameterSet &pset, edm::ActivityRegistry &ar)
 {
   ar.watchPreSourceConstruction(&restrictDQMAccessM);
   ar.watchPostSourceConstruction(&releaseDQMAccessM);
-  ar.watchPreSource(&restrictDQMAccess);
-  ar.watchPostSource(&releaseDQMAccess);
+  ar.watchPreSourceEvent(&restrictDQMAccessS);
+  ar.watchPostSourceEvent(&releaseDQMAccessS);
   ar.watchPreModule(&restrictDQMAccessM);
   ar.watchPostModule(&releaseDQMAccessM);
   ar.watchPostProcessEvent(this, &DQMService::flush);

diff --git a/DataFormats/GeometrySurface/plugins/BlockWipedAllocatorService.cc b/DataFormats/GeometrySurface/plugins/BlockWipedAllocatorService.cc
@@ -50,7 +50,6 @@ class BlockWipedAllocatorService {
 
     blockWipedPool(&pool);
     if (m_useAlloc) BlockWipedPoolAllocated::usePool();
-    iAR.watchPreSource(this,&BlockWipedAllocatorService::preSource);
     iAR.watchPreProcessEvent(this,&BlockWipedAllocatorService::preEventProcessing);
     iAR.watchPostEndJob(this,&BlockWipedAllocatorService::postEndJob);
     iAR.watchPreModule(this,&BlockWipedAllocatorService::preModule);
@@ -60,11 +59,6 @@ class BlockWipedAllocatorService {
   // wipe the workspace before each event
   void preEventProcessing(const edm::EventID&, const edm::Timestamp&) { wiper();}
 
-  // nope event-principal deleted in source
-  void preSource() {
-   // wiper();
-  }
-
   void dump() {
     if (m_silent) return;
     std::cout << "ReferenceCounted stat"<< std::endl;

diff --git a/EventFilter/Utilities/interface/MicroStateServiceClassic.h b/EventFilter/Utilities/interface/MicroStateServiceClassic.h
@@ -37,9 +37,8 @@ namespace evf{
       void preEventProcessing(const edm::EventID&, const edm::Timestamp&);
       void postEventProcessing(const edm::Event&, const edm::EventSetup&);
 
-      void preSource();
-      void postSource();
-
+      void preSourceEvent(edm::StreamID);
+      void postSourceEvent(edm::StreamID);
 
       void preModule(const edm::ModuleDescription&);
       void postModule(const edm::ModuleDescription&);

diff --git a/EventFilter/Utilities/plugins/FastMonitoringService.cc b/EventFilter/Utilities/plugins/FastMonitoringService.cc
@@ -44,8 +44,8 @@ namespace evf{
     reg.watchPreProcessPath(this,&FastMonitoringService::preProcessPath);
     reg.watchPreProcessEvent(this,&FastMonitoringService::preEventProcessing);
     reg.watchPostProcessEvent(this,&FastMonitoringService::postEventProcessing);
-    reg.watchPreSource(this,&FastMonitoringService::preSource);
-    reg.watchPostSource(this,&FastMonitoringService::postSource);
+    reg.watchPreSourceEvent(this,&FastMonitoringService::preSourceEvent);
+    reg.watchPostSourceEvent(this,&FastMonitoringService::postSourceEvent);
 
     reg.watchPreModule(this,&FastMonitoringService::preModule);
     reg.watchPostModule(this,&FastMonitoringService::postModule);
@@ -246,13 +246,14 @@ namespace evf{
     fmt_.m_data.processedJ_.value()++;
     fmt_.monlock_.unlock();
   }
-  void FastMonitoringService::preSource()
+
+  void FastMonitoringService::preSourceEvent(edm::StreamID)
   {
     //    boost::mutex::scoped_lock sl(lock_);
     fmt_.m_data.microstate_ = &reservedMicroStateNames[mIdle];
   }
 
-  void FastMonitoringService::postSource()
+  void FastMonitoringService::postSourceEvent(edm::StreamID)
   {
     //    boost::mutex::scoped_lock sl(lock_);
     fmt_.m_data.microstate_ = &reservedMicroStateNames[mFwkOvh];

diff --git a/EventFilter/Utilities/plugins/FastMonitoringService.h b/EventFilter/Utilities/plugins/FastMonitoringService.h
@@ -118,8 +118,8 @@ namespace evf{
       void preEventProcessing(const edm::EventID&, const edm::Timestamp&);
       void postEventProcessing(const edm::Event&, const edm::EventSetup&);
 
-      void preSource();
-      void postSource();
+      void preSourceEvent(edm::StreamID);
+      void postSourceEvent(edm::StreamID);
 
       void preModule(const edm::ModuleDescription&);
       void postModule(const edm::ModuleDescription&);

diff --git a/EventFilter/Utilities/src/MicroStateServiceClassic.cc b/EventFilter/Utilities/src/MicroStateServiceClassic.cc
@@ -15,8 +15,8 @@ namespace evf{
 
     reg.watchPreProcessEvent(this,&MicroStateServiceClassic::preEventProcessing);
     reg.watchPostProcessEvent(this,&MicroStateServiceClassic::postEventProcessing);
-    reg.watchPreSource(this,&MicroStateServiceClassic::preSource);
-    reg.watchPostSource(this,&MicroStateServiceClassic::postSource);
+    reg.watchPreSourceEvent(this,&MicroStateServiceClassic::preSourceEvent);
+    reg.watchPostSourceEvent(this,&MicroStateServiceClassic::postSourceEvent);
 
     reg.watchPreModule(this,&MicroStateServiceClassic::preModule);
     reg.watchPostModule(this,&MicroStateServiceClassic::postModule);
@@ -53,13 +53,14 @@ namespace evf{
     boost::mutex::scoped_lock sl(lock_);
     microstate2_ = &input;
   }
-  void MicroStateServiceClassic::preSource()
+
+  void MicroStateServiceClassic::preSourceEvent(edm::StreamID)
   {
     boost::mutex::scoped_lock sl(lock_);
     microstate2_ = &input;
   }
 
-  void MicroStateServiceClassic::postSource()
+  void MicroStateServiceClassic::postSourceEvent(edm::StreamID)
   {
     boost::mutex::scoped_lock sl(lock_);
     microstate2_ = &fwkovh;

diff --git a/FWCore/Framework/interface/InputSource.h b/FWCore/Framework/interface/InputSource.h
@@ -109,10 +109,10 @@ namespace edm {
     ItemType nextItemType();
 
     /// Read next event
-    void readEvent(EventPrincipal& ep, StreamContext *);
+    void readEvent(EventPrincipal& ep, StreamContext &);
 
     /// Read a specific event
-    bool readEvent(EventPrincipal& ep, EventID const&, StreamContext *);
+    bool readEvent(EventPrincipal& ep, EventID const&, StreamContext &);
 
     /// Read next luminosity block Auxilary
     boost::shared_ptr<LuminosityBlockAuxiliary> readLuminosityBlockAuxiliary();
@@ -279,9 +279,15 @@ namespace edm {
 
     class EventSourceSentry {
     public:
-      explicit EventSourceSentry(InputSource const& source);
+      EventSourceSentry(InputSource const& source, StreamContext & sc);
+      ~EventSourceSentry();
+
+      EventSourceSentry(EventSourceSentry const&) = delete; // Disallow copying and moving
+      EventSourceSentry& operator=(EventSourceSentry const&) = delete; // Disallow copying and moving
+
     private:
-      SourceSentry sentry_;
+      InputSource const& source_;
+      StreamContext & sc_;
     };
 
     class LumiSourceSentry {
@@ -386,7 +392,7 @@ namespace edm {
     virtual void readRun_(RunPrincipal& runPrincipal);
     virtual void readLuminosityBlock_(LuminosityBlockPrincipal& lumiPrincipal);
     virtual void readEvent_(EventPrincipal& eventPrincipal) = 0;
-    virtual bool readIt(EventID const&, EventPrincipal& eventPrincipal);
+    virtual bool readIt(EventID const& id, EventPrincipal& eventPrincipal, StreamContext& streamContext);
     virtual std::unique_ptr<FileBlock> readFile_();
     virtual void closeFile_() {}
     virtual bool goToEvent_(EventID const& eventID);

diff --git a/FWCore/Framework/src/EventProcessor.cc b/FWCore/Framework/src/EventProcessor.cc
@@ -1806,7 +1806,7 @@ namespace edm {
     //TODO this will have to become per stream
     auto& event = principalCache_.eventPrincipal(iStreamIndex);
     StreamContext streamContext(event.streamID(), &processContext_);
-    input_->readEvent(event, &streamContext);
+    input_->readEvent(event, streamContext);
     FDEBUG(1) << "\treadEvent\n";
   }
   void EventProcessor::processEvent(unsigned int iStreamIndex) {

diff --git a/FWCore/Framework/src/InputSource.cc b/FWCore/Framework/src/InputSource.cc
@@ -19,6 +19,7 @@
 #include "FWCore/ParameterSet/interface/ParameterSet.h"
 #include "FWCore/ServiceRegistry/interface/ActivityRegistry.h"
 #include "FWCore/ServiceRegistry/interface/Service.h"
+#include "FWCore/ServiceRegistry/interface/StreamContext.h"
 #include "FWCore/ParameterSet/interface/ConfigurationDescriptions.h"
 #include "FWCore/ParameterSet/interface/ParameterSetDescription.h"
 #include "FWCore/Utilities/interface/do_nothing_deleter.h"
@@ -329,13 +330,17 @@ namespace edm {
   }
 
   void
-  InputSource::readEvent(EventPrincipal& ep, StreamContext* streamContext) {
+  InputSource::readEvent(EventPrincipal& ep, StreamContext& streamContext) {
     assert(state_ == IsEvent);
     assert(!eventLimitReached());
+    {
+      // block scope, in order to issue the PostSourceEvent signal before calling postRead and issueReports
+      EventSourceSentry sentry(*this, streamContext);
 
-    callWithTryCatchAndPrint<void>( [this,&ep](){ readEvent_(ep); }, "Calling InputSource::readEvent_" );
-    if(receiver_) {
-      --numberOfEventsBeforeBigSkip_;
+      callWithTryCatchAndPrint<void>( [this,&ep](){ readEvent_(ep); }, "Calling InputSource::readEvent_" );
+      if(receiver_) {
+        --numberOfEventsBeforeBigSkip_;
+      }
     }
 
     Event event(ep, moduleDescription(), nullptr);
@@ -347,14 +352,17 @@ namespace edm {
   }
 
   bool
-  InputSource::readEvent(EventPrincipal& ep, EventID const& eventID, StreamContext* streamContext) {
+  InputSource::readEvent(EventPrincipal& ep, EventID const& eventID, StreamContext& streamContext) {
     bool result = false;
 
-    if(!limitReached()) {
-      //result = callWithTryCatchAndPrint<bool>( [this,ep,&eventID](){ return readIt(eventID, ep); }, "Calling InputSource::readIt" );
-      result = readIt(eventID, ep);
-      if(result) {
+    if (not limitReached()) {
+      // the Pre/PostSourceEvent signals should be generated only if the event is actually found.
+      // this should be taken care of by an EventSourceSentry in the implementaion of readIt()
+
+      //result = callWithTryCatchAndPrint<bool>( [this,&eventID,&ep](){ return readIt(eventID, ep); }, "Calling InputSource::readIt" );
+      result = readIt(eventID, ep, streamContext);
 
+      if (result) {
         Event event(ep, moduleDescription(), nullptr);
         postRead(event);
         if(remainingEvents_ > 0) --remainingEvents_;
@@ -409,7 +417,7 @@ namespace edm {
   }
 
   bool
-  InputSource::readIt(EventID const&, EventPrincipal&) {
+  InputSource::readIt(EventID const&, EventPrincipal&, StreamContext&) {
     throw Exception(errors::LogicError)
       << "InputSource::readIt()\n"
       << "Random access is not implemented for this type of Input Source\n"
@@ -608,8 +616,15 @@ namespace edm {
     post_();
   }
 
-  InputSource::EventSourceSentry::EventSourceSentry(InputSource const& source) :
-     sentry_(source.actReg()->preSourceSignal_, source.actReg()->postSourceSignal_) {
+  InputSource::EventSourceSentry::EventSourceSentry(InputSource const& source, StreamContext & sc) :
+    source_(source),
+    sc_(sc)
+  {
+    source.actReg()->preSourceSignal_(sc_.streamID());
+  }
+
+  InputSource::EventSourceSentry::~EventSourceSentry() {
+    source_.actReg()->postSourceSignal_(sc_.streamID());
   }
 
   InputSource::LumiSourceSentry::LumiSourceSentry(InputSource const& source) :

diff --git a/FWCore/Integration/test/ThrowingSource.cc b/FWCore/Integration/test/ThrowingSource.cc
@@ -130,7 +130,6 @@ namespace edm {
   ThrowingSource::readEvent_(EventPrincipal& eventPrincipal) {
     if (whenToThrow_ == kReadEvent) throw cms::Exception("TestThrow") << "ThrowingSource::readEvent_";
     assert(eventCached() || processingMode() != RunsLumisAndEvents);
-    EventSourceSentry sentry(*this);
     EventAuxiliary aux(eventID(), processGUID(), Timestamp(presentTime()), false, EventAuxiliary::Undefined);
     eventPrincipal.fillEventPrincipal(aux, processHistoryRegistry());
   }

diff --git a/FWCore/MessageService/interface/MessageLogger.h b/FWCore/MessageService/interface/MessageLogger.h
@@ -64,12 +64,14 @@ class MessageLogger {
   void  postEndJob();
   void  jobFailure();
 
-  void  preSource  ();
-  void  postSource ();
+  void  preSourceEvent  ( StreamID );
+  void  postSourceEvent ( StreamID );
+  void  preSourceRunLumi  ();
+  void  postSourceRunLumi ();
 
-  void  preFile       ( std::string const &, bool );
+  void  preFile       ( std::string const&, bool );
   void  preFileClose  ( std::string const&, bool );
-  void  postFile      ( std::string const &, bool );
+  void  postFile      ( std::string const&, bool );
 
   void  preModuleConstruction ( ModuleDescription const & );
   void  postModuleConstruction( ModuleDescription const & );

diff --git a/FWCore/MessageService/src/MessageLogger.cc b/FWCore/MessageService/src/MessageLogger.cc
@@ -267,13 +267,13 @@ namespace edm {
       iRegistry.watchPreModuleEvent(this,&MessageLogger::preModuleEvent);
       iRegistry.watchPostModuleEvent(this,&MessageLogger::postModuleEvent);
 
-      iRegistry.watchPreSource(this,&MessageLogger::preSource);
-      iRegistry.watchPostSource(this,&MessageLogger::postSource);
+      iRegistry.watchPreSourceEvent(this,&MessageLogger::preSourceEvent);
+      iRegistry.watchPostSourceEvent(this,&MessageLogger::postSourceEvent);
       // change log 14:
-      iRegistry.watchPreSourceRun(this,&MessageLogger::preSource);
-      iRegistry.watchPostSourceRun(this,&MessageLogger::postSource);
-      iRegistry.watchPreSourceLumi(this,&MessageLogger::preSource);
-      iRegistry.watchPostSourceLumi(this,&MessageLogger::postSource);
+      iRegistry.watchPreSourceRun(this,&MessageLogger::preSourceRunLumi);
+      iRegistry.watchPostSourceRun(this,&MessageLogger::postSourceRunLumi);
+      iRegistry.watchPreSourceLumi(this,&MessageLogger::preSourceRunLumi);
+      iRegistry.watchPostSourceLumi(this,&MessageLogger::postSourceRunLumi);
       iRegistry.watchPreOpenFile(this,&MessageLogger::preFile);
       iRegistry.watchPostOpenFile(this,&MessageLogger::postFile);
       iRegistry.watchPreCloseFile(this,&MessageLogger::preFileClose);
@@ -692,12 +692,13 @@ namespace edm {
       MessageDrop::instance()->setSinglet("AfterBeginJob");     // Change Log 17
     }
 
-    void
-    MessageLogger::preSource()
-    {
-      establish("source");
-    }
-    void MessageLogger::postSource()
+    void MessageLogger::preSourceEvent(StreamID)
+    { establish("source"); }
+    void MessageLogger::postSourceEvent(StreamID)
+    { unEstablish("AfterSource"); }
+    void MessageLogger::preSourceRunLumi()
+    { establish("source"); }
+    void MessageLogger::postSourceRunLumi()
     { unEstablish("AfterSource"); }
 
     void MessageLogger::preFile( std::string const &, bool )

diff --git a/FWCore/ServiceRegistry/interface/ActivityRegistry.h b/FWCore/ServiceRegistry/interface/ActivityRegistry.h
@@ -31,6 +31,7 @@ unscheduled execution. The tests are in FWCore/Integration/test:
 // system include files
 //#include "boost/signal.hpp"
 #include "FWCore/Utilities/interface/Signal.h"
+#include "FWCore/Utilities/interface/StreamID.h"
 #include "boost/bind.hpp"
 #include "boost/mem_fn.hpp"
 #include "boost/utility.hpp"
@@ -102,20 +103,20 @@ namespace edm {
       AR_WATCH_USING_METHOD_0(watchJobFailure)
 
       /// signal is emitted before the source starts creating an Event
-      typedef signalslot::Signal<void()> PreSource;
-      PreSource preSourceSignal_;
-      void watchPreSource(PreSource::slot_type const& iSlot) {
+      typedef signalslot::Signal<void(StreamID)> PreSourceEvent;
+      PreSourceEvent preSourceSignal_;
+      void watchPreSourceEvent(PreSourceEvent::slot_type const& iSlot) {
         preSourceSignal_.connect(iSlot);
       }
-      AR_WATCH_USING_METHOD_0(watchPreSource)
+      AR_WATCH_USING_METHOD_1(watchPreSourceEvent)
 
       /// signal is emitted after the source starts creating an Event
-      typedef signalslot::Signal<void()> PostSource;
-      PostSource postSourceSignal_;
-      void watchPostSource(PostSource::slot_type const& iSlot) {
+      typedef signalslot::Signal<void(StreamID)> PostSourceEvent;
+      PostSourceEvent postSourceSignal_;
+      void watchPostSourceEvent(PostSourceEvent::slot_type const& iSlot) {
          postSourceSignal_.connect_front(iSlot);
       }
-      AR_WATCH_USING_METHOD_0(watchPostSource)
+      AR_WATCH_USING_METHOD_1(watchPostSourceEvent)
 
       /// signal is emitted before the source starts creating a Lumi
       typedef signalslot::Signal<void()> PreSourceLumi;

diff --git a/FWCore/Services/src/Memory.cc b/FWCore/Services/src/Memory.cc
@@ -143,7 +143,7 @@ namespace edm {
       if(!oncePerEventMode_) { // default, prints on increases
         iReg.watchPreSourceConstruction(this, &SimpleMemoryCheck::preSourceConstruction);
         iReg.watchPostSourceConstruction(this, &SimpleMemoryCheck::postSourceConstruction);
-        iReg.watchPostSource(this, &SimpleMemoryCheck::postSource);
+        iReg.watchPostSourceEvent(this, &SimpleMemoryCheck::postSourceEvent);
         iReg.watchPostModuleConstruction(this, &SimpleMemoryCheck::postModuleConstruction);
         iReg.watchPostModuleBeginJob(this, &SimpleMemoryCheck::postModuleBeginJob);
         iReg.watchPostEvent(this, &SimpleMemoryCheck::postEvent);
@@ -267,7 +267,7 @@ namespace edm {
       }
     }
 
-    void SimpleMemoryCheck::postSource() {
+    void SimpleMemoryCheck::postSourceEvent(StreamID sid) {
       bool expected = false;
       if(measurementUnderway_.compare_exchange_strong(expected,true,std::memory_order_acq_rel) ) {
         std::shared_ptr<void> guard(nullptr,[this](void const*) {

diff --git a/FWCore/Services/src/Memory.h b/FWCore/Services/src/Memory.h
@@ -62,7 +62,7 @@ namespace edm {
 
       void preSourceConstruction(const ModuleDescription&);
       void postSourceConstruction(const ModuleDescription&);
-      void postSource();
+      void postSourceEvent(StreamID);
 
       void postBeginJob();