libraries/JANA/JEventSource.cc

0001 #include <JANA/JEventSource.h>
0002
0003 void JEventSource::DoOpen(bool with_lock) {
0004     if (with_lock) {
0005         std::lock_guard<std::mutex> lock(m_mutex);
0006         if (!m_is_initialized) {
0007             throw JException("Attempted to open a JEventSource that hasn't been initialized!");
0008         }
0009         CallWithJExceptionWrapper("JEventSource::Open", [&](){ Open();});
0010         if (GetResourceName().empty()) {
0011             LOG_INFO(GetLogger()) << "Opened JEventSource '" << GetTypeName() << "'" << LOG_END;
0012         }
0013         else {
0014             LOG_INFO(GetLogger()) << "Opened JEventSource '" << GetTypeName() << "' ('" << GetResourceName() << "')" << LOG_END;
0015         }
0016         m_status = Status::Opened;
0017     }
0018     else {
0019         if (!m_is_initialized) {
0020             throw JException("Attempted to open a JEventSource that hasn't been initialized!");
0021         }
0022         CallWithJExceptionWrapper("JEventSource::Open", [&](){ Open();});
0023         if (GetResourceName().empty()) {
0024             LOG_INFO(GetLogger()) << "Opened JEventSource '" << GetTypeName() << "'" << LOG_END;
0025         }
0026         else {
0027             LOG_INFO(GetLogger()) << "Opened JEventSource '" << GetTypeName() << "' ('" << GetResourceName() << "')" << LOG_END;
0028         }
0029         m_status = Status::Opened;
0030     }
0031 }
0032
0033 void JEventSource::DoClose(bool with_lock) {
0034     if (with_lock) {
0035         std::lock_guard<std::mutex> lock(m_mutex);
0036
0037         if (m_status != JEventSource::Status::Opened) return;
0038
0039         CallWithJExceptionWrapper("JEventSource::Close", [&](){ Close();});
0040         if (GetResourceName().empty()) {
0041             LOG_INFO(GetLogger()) << "Closed JEventSource '" << GetTypeName() << "'" << LOG_END;
0042         }
0043         else {
0044             LOG_INFO(GetLogger()) << "Closed JEventSource '" << GetTypeName() << "' ('" << GetResourceName() << "')" << LOG_END;
0045         }
0046         m_status = Status::Closed;
0047     }
0048     else {
0049         if (m_status != JEventSource::Status::Opened) return;
0050
0051         CallWithJExceptionWrapper("JEventSource::Close", [&](){ Close();});
0052         if (GetResourceName().empty()) {
0053             LOG_INFO(GetLogger()) << "Closed JEventSource '" << GetTypeName() << "'" << LOG_END;
0054         }
0055         else {
0056             LOG_INFO(GetLogger()) << "Closed JEventSource '" << GetTypeName() << "' ('" << GetResourceName() << "')" << LOG_END;
0057         }
0058         m_status = Status::Closed;
0059     }
0060 }
0061
0062 JEventSource::Result JEventSource::DoNext(std::shared_ptr<JEvent> event) {
0063
0064     std::lock_guard<std::mutex> lock(m_mutex); // In general, DoNext must be synchronized.
0065
0066     if (!m_is_initialized) {
0067         throw JException("JEventSource has not been initialized!");
0068     }
0069     if (m_status == Status::Unopened) {
0070         DoOpen(false);
0071     }
0072     if (m_status == Status::Opened) {
0073
0074         // First we check whether there are events to skip. If so, we skip as many as possible
0075         if (m_nskip > 0) {
0076             auto [result, remaining_events] = Skip(*event.get(), m_nskip);
0077             m_events_skipped += (m_nskip - remaining_events);
0078             m_nskip = remaining_events;
0079
0080             LOG_DEBUG(GetLogger()) << "Finished with Skip: " << m_events_skipped << " events skipped, " << m_nskip << " events to skip remain";
0081
0082             // If we encountered a problem, exit and let the arrow figure out when and whether to resume.
0083             // Note that Skip() will call Close() on our behalf.
0084             if (result != Result::Success) return result;
0085         }
0086
0087         // Next we check whether we are limited by jana:nevents
0088         if (m_nevents != 0 && m_events_emitted >= m_nevents) {
0089             LOG_DEBUG(GetLogger()) << "Closing EventSource due to reaching nevent limit";
0090             DoClose(false);
0091             return Result::FailureFinished;
0092         }
0093
0094         // By this point we know that we are done skipping events and are ready to emit them
0095
0096         // We configure the event
0097         event->SetEventNumber(m_events_emitted); // Default event number to event count
0098         event->SetJEventSource(this);
0099         event->SetSequential(false);
0100         event->GetJCallGraphRecorder()->Reset();
0101
0102
0103         JEventSource::Result result = Result::Success;
0104         try {
0105             auto previous_origin = event->GetJCallGraphRecorder()->SetInsertDataOrigin( JCallGraphRecorder::ORIGIN_FROM_SOURCE);  // (see note at top of JCallGraphRecorder.h)
0106             if (m_callback_style == CallbackStyle::LegacyMode) {
0107                 GetEvent(event);
0108             }
0109             else {
0110                 result = Emit(*event);
0111             }
0112             event->GetJCallGraphRecorder()->SetInsertDataOrigin( previous_origin );
0113         }
0114         catch (RETURN_STATUS rs) {
0115
0116             if (rs == RETURN_STATUS::kNO_MORE_EVENTS) {
0117                 DoClose(false);
0118                 result = Result::FailureFinished;
0119             }
0120             else if (rs == RETURN_STATUS::kTRY_AGAIN || rs == RETURN_STATUS::kBUSY) {
0121                 result = Result::FailureTryAgain;
0122             }
0123             else if (rs == RETURN_STATUS::kERROR || rs == RETURN_STATUS::kUNKNOWN) {
0124                 JException ex ("JEventSource threw RETURN_STATUS::kERROR or kUNKNOWN");
0125                 ex.plugin_name = m_plugin_name;
0126                 ex.type_name = m_type_name;
0127                 ex.function_name = "JEventSource::GetEvent";
0128                 ex.instance_name = m_resource_name;
0129                 throw ex;
0130             }
0131         }
0132         catch (JException& ex) {
0133             if (ex.function_name.empty()) ex.function_name = "JEventSource::GetEvent";
0134             if (ex.type_name.empty()) ex.type_name = m_type_name;
0135             if (ex.instance_name.empty()) ex.instance_name = m_prefix;
0136             if (ex.plugin_name.empty()) ex.plugin_name = m_plugin_name;
0137             throw ex;
0138         }
0139         catch (std::exception& e){
0140             auto ex = JException(e.what());
0141             ex.exception_type = JTypeInfo::demangle_current_exception_type();
0142             ex.nested_exception = std::current_exception();
0143             ex.function_name = "JEventSource::GetEvent";
0144             ex.type_name = m_type_name;
0145             ex.instance_name = m_prefix;
0146             ex.plugin_name = m_plugin_name;
0147             throw ex;
0148         }
0149         catch (...) {
0150             auto ex = JException("Unknown exception");
0151             ex.exception_type = JTypeInfo::demangle_current_exception_type();
0152             ex.nested_exception = std::current_exception();
0153             ex.function_name = "JEventSource::GetEvent";
0154             ex.type_name = m_type_name;
0155             ex.instance_name = m_prefix;
0156             ex.plugin_name = m_plugin_name;
0157             throw ex;
0158         }
0159
0160         if (result == Result::Success) {
0161             m_events_emitted += 1;
0162             // We end up here if we read an entry in our file or retrieved a message from our socket,
0163             // and believe we could obtain another one immediately if we wanted to
0164             for (auto* output : GetOutputs()) {
0165                 output->EulerianStore(*event->GetFactorySet());
0166             }
0167             for (auto* output : GetVariadicOutputs()) {
0168                 output->EulerianStore(*event->GetFactorySet());
0169             }
0170             return Result::Success;
0171         }
0172         else if (result == Result::FailureFinished) {
0173             // We end up here if we tried to read an entry in a file, but found EOF
0174             // or if we received a message from a socket that contained no data and indicated no more data will be coming
0175             DoClose(false);
0176             return Result::FailureFinished;
0177         }
0178         else {
0179             return result;
0180         }
0181     }
0182     else { // status == Closed
0183         return Result::FailureFinished;
0184     }
0185 }
0186
0187
0188 void JEventSource::DoFinishEvent(JEvent& event) {
0189
0190     m_events_processed.fetch_add(1);
0191     if (m_enable_finish_event) {
0192         std::lock_guard<std::mutex> lock(m_mutex);
0193         CallWithJExceptionWrapper("JEventSource::FinishEvent", [&](){
0194             FinishEvent(event);
0195         });
0196     }
0197 }
0198
0199 void JEventSource::Summarize(JComponentSummary& summary) const {
0200
0201     auto* result = new JComponentSummary::Component(
0202         "Source", GetPrefix(), GetTypeName(), GetLevel(), GetPluginName());
0203
0204     SummarizeOutputs(*result);
0205     summary.Add(result);
0206 }
0207
0208
0209 std::pair<JEventSource::Result, size_t> JEventSource::Skip(JEvent& event, size_t events_to_skip) {
0210
0211     // Return values
0212     Result result = Result::Success;
0213
0214     while (events_to_skip > 0 && result == Result::Success) {
0215         try {
0216             auto previous_origin = event.GetJCallGraphRecorder()->SetInsertDataOrigin( JCallGraphRecorder::ORIGIN_FROM_SOURCE);  // (see note at top of JCallGraphRecorder.h)
0217             if (m_callback_style == CallbackStyle::LegacyMode) {
0218                 GetEvent(event.shared_from_this());
0219             }
0220             else {
0221                 result = Emit(event);
0222             }
0223             event.GetJCallGraphRecorder()->SetInsertDataOrigin( previous_origin );
0224             // We need to call FinishEvent, but we don't want to call it from inside JEvent::Clear() because we are already inside the lock.
0225             // So instead, we call it ourselves out here. This has the added benefit of letting us avoid updating m_events_processed.
0226             if (m_enable_finish_event) {
0227                 CallWithJExceptionWrapper("JEventSource::FinishEvent", [&](){ FinishEvent(event); });
0228             }
0229             event.Clear(false);
0230             events_to_skip -= 1;
0231         }
0232         catch (RETURN_STATUS rs) {
0233
0234             if (rs == RETURN_STATUS::kNO_MORE_EVENTS) {
0235                 DoClose(false);
0236                 result = Result::FailureFinished;
0237             }
0238             else if (rs == RETURN_STATUS::kTRY_AGAIN || rs == RETURN_STATUS::kBUSY) {
0239                 result = Result::FailureTryAgain;
0240             }
0241             else if (rs == RETURN_STATUS::kERROR || rs == RETURN_STATUS::kUNKNOWN) {
0242                 JException ex ("JEventSource threw RETURN_STATUS::kERROR or kUNKNOWN");
0243                 ex.plugin_name = m_plugin_name;
0244                 ex.type_name = m_type_name;
0245                 ex.function_name = (m_callback_style == CallbackStyle::LegacyMode) ? "JEventSource::GetEvent" : "JEventSource::Emit";
0246                 ex.instance_name = m_resource_name;
0247                 throw ex;
0248             }
0249         }
0250         catch (JException& ex) {
0251             if (ex.function_name.empty()) ex.function_name = (m_callback_style == CallbackStyle::LegacyMode) ? "JEventSource::GetEvent" : "JEventSource::Emit";
0252             if (ex.type_name.empty()) ex.type_name = m_type_name;
0253             if (ex.instance_name.empty()) ex.instance_name = m_prefix;
0254             if (ex.plugin_name.empty()) ex.plugin_name = m_plugin_name;
0255             throw ex;
0256         }
0257         catch (std::exception& e){
0258             auto ex = JException(e.what());
0259             ex.exception_type = JTypeInfo::demangle_current_exception_type();
0260             ex.nested_exception = std::current_exception();
0261             ex.function_name = (m_callback_style == CallbackStyle::LegacyMode) ? "JEventSource::GetEvent" : "JEventSource::Emit";
0262             ex.type_name = m_type_name;
0263             ex.instance_name = m_prefix;
0264             ex.plugin_name = m_plugin_name;
0265             throw ex;
0266         }
0267         catch (...) {
0268             auto ex = JException("Unknown exception");
0269             ex.exception_type = JTypeInfo::demangle_current_exception_type();
0270             ex.nested_exception = std::current_exception();
0271             ex.function_name = (m_callback_style == CallbackStyle::LegacyMode) ? "JEventSource::GetEvent" : "JEventSource::Emit";
0272             ex.type_name = m_type_name;
0273             ex.instance_name = m_prefix;
0274             ex.plugin_name = m_plugin_name;
0275             throw ex;
0276         }
0277     }
0278
0279     return {result, events_to_skip};
0280 }
0281
0282
0283
0284