Merge pull request #367 from JeffersonLab/nbrei_eventproc_beginrun_fix

Bugfix: JEventProcessor::Process() called before BeginRun()

src/libraries/JANA/Components/JHasOutputs.h

@@ -31,6 +31,7 @@ struct JHasOutputs {
  template <typename T>
  class Output : public OutputBase {
  std::vector<T*> m_data;
+ bool is_not_owner = false;
 
  public:
  Output(JHasOutputs* owner, std::string default_tag_name="") {
@@ -43,9 +44,12 @@ struct JHasOutputs {
 
  protected:
  void InsertCollection(JEvent& event) override {
- event.Insert(m_data, this->collection_names[0]);
+ auto fac = event.Insert(m_data, this->collection_names[0]);
+ fac->SetIsNotOwnerFlag(is_not_owner);
  }
  void Reset() override { }
+
+ void SetIsNotOwnerFlag(bool not_owner=true) { is_not_owner = not_owner; }
  };
 
 

src/libraries/JANA/JEventProcessor.h

@@ -8,6 +8,7 @@
 #include <JANA/Components/JHasInputs.h>
 #include <JANA/Components/JHasRunCallbacks.h>
 #include <JANA/JEvent.h>
+#include <mutex>
 
 class JApplication;
 
@@ -30,10 +31,9 @@ class JEventProcessor : public jana::components::JComponent,
 
  uint64_t GetEventCount() const { return m_event_count; };
 
- bool AreEventsOrdered() const { return m_receive_events_in_order; }
-
 
  virtual void DoInitialize() {
+ std::lock_guard<std::mutex> lock(m_mutex);
  for (auto* parameter : m_parameters) {
  parameter->Configure(*(m_app->GetJParameterManager()), m_prefix);
  }
@@ -109,8 +109,12 @@ class JEventProcessor : public jana::components::JComponent,
 
  virtual void DoLegacyProcess(const std::shared_ptr<const JEvent>& event) {
 
- // DoLegacyProcess doesn't hold any locks, as it requires the user to hold a lock for it.
- // Because of this, 
+ // DoLegacyProcess holds a lock to make sure that {Begin,Change,End}Run() are always called before Process(). 
+ // Note that in LegacyMode, Process() requires the user to manage a _separate_ lock for its critical section.
+ // This arrangement means that {Begin,Change,End}Run() will definitely be called at least once before `Process`, but there
+ // may be races when there are multiple run numbers present in the stream. This isn't a problem in practice for now, 
+ // but future work should use ExpertMode or DeclarativeMode for this reason (but also for the usability improvements!)
+
  if (m_callback_style != CallbackStyle::LegacyMode) {
  throw JException("Called DoLegacyProcess() on a non-legacy-mode JEventProcessor");
  }
@@ -123,15 +127,19 @@ class JEventProcessor : public jana::components::JComponent,
  else if (m_status == Status::Finalized) {
  throw JException("JEventProcessor: Attempted to call DoMap() after Finalize()");
  }
- if (m_last_run_number != run_number) {
- if (m_last_run_number != -1) {
- CallWithJExceptionWrapper("JEventProcessor::EndRun", [&](){ EndRun(); });
+ {
+ // Protect the call to BeginRun(), etc, to prevent some threads from running Process() before BeginRun().
+ std::lock_guard<std::mutex> lock(m_mutex);
+ if (m_last_run_number != run_number) {
+ if (m_last_run_number != -1) {
+ CallWithJExceptionWrapper("JEventProcessor::EndRun", [&](){ EndRun(); });
+ }
+ for (auto* resource : m_resources) {
+ resource->ChangeRun(event->GetRunNumber(), m_app);
+ }
+ m_last_run_number = run_number;
+ CallWithJExceptionWrapper("JEventProcessor::BeginRun", [&](){ BeginRun(event); });
  }
- for (auto* resource : m_resources) {
- resource->ChangeRun(event->GetRunNumber(), m_app);
- }
- m_last_run_number = run_number;
- CallWithJExceptionWrapper("JEventProcessor::BeginRun", [&](){ BeginRun(event); });
  }
  CallWithJExceptionWrapper("JEventProcessor::Process", [&](){ Process(event); });
  m_event_count += 1;
@@ -170,33 +178,29 @@ class JEventProcessor : public jana::components::JComponent,
  // LegacyMode-specific callbacks
 
  virtual void Process(const std::shared_ptr<const JEvent>& /*event*/) {
- throw JException("Not implemented yet!");
  }
- 
+
  // ExpertMode-specific callbacks
 
  virtual void ProcessParallel(const JEvent& /*event*/) {
  }
 
  virtual void Process(const JEvent& /*event*/) {
- throw JException("Not implemented yet!");
  }
 
  // DeclarativeMode-specific callbacks
 
  virtual void ProcessParallel(int64_t /*run_nr*/, uint64_t /*event_nr*/, uint64_t /*event_idx*/) {
- throw JException("Not implemented yet!");
  }
 
  virtual void Process(int64_t /*run_nr*/, uint64_t /*event_nr*/, uint64_t /*event_idx*/) {
- throw JException("Not implemented yet!");
  }
 
 
  virtual void Finish() {}
 
 
- // TODO: Deprecate
+ [[deprecated]]
  virtual std::string GetType() const {
  return m_type_name;
  }
@@ -217,18 +221,10 @@ class JEventProcessor : public jana::components::JComponent,
 
  // void SetResourceName(std::string resource_name) { m_resource_name = std::move(resource_name); }
 
- /// SetEventsOrdered allows the user to tell the parallelization engine that it needs to see
- /// the event stream ordered by increasing event IDs. (Note that this requires all EventSources
- /// emit event IDs which are consecutive.) Ordering by event ID makes for cleaner output, but comes
- /// with a performance penalty, so it is best if this is enabled during debugging, and disabled otherwise.
-
- // void SetEventsOrdered(bool receive_events_in_order) { m_receive_events_in_order = receive_events_in_order; }
-
 
 private:
  std::string m_resource_name;
  std::atomic_ullong m_event_count {0};
- bool m_receive_events_in_order = false;
 
 };