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->SetSequential(false);
0099         event->GetJCallGraphRecorder()->Reset();
0100
0101
0102         JEventSource::Result result = Result::Success;
0103         try {
0104             auto previous_origin = event->GetJCallGraphRecorder()->SetInsertDataOrigin( JCallGraphRecorder::ORIGIN_FROM_SOURCE);  // (see note at top of JCallGraphRecorder.h)
0105             if (m_callback_style == CallbackStyle::LegacyMode) {
0106                 GetEvent(event);
0107             }
0108             else {
0109                 result = Emit(*event);
0110             }
0111             event->GetJCallGraphRecorder()->SetInsertDataOrigin( previous_origin );
0112         }
0113         catch (RETURN_STATUS rs) {
0114
0115             if (rs == RETURN_STATUS::kNO_MORE_EVENTS) {
0116                 DoClose(false);
0117                 result = Result::FailureFinished;
0118             }
0119             else if (rs == RETURN_STATUS::kTRY_AGAIN || rs == RETURN_STATUS::kBUSY) {
0120                 result = Result::FailureTryAgain;
0121             }
0122             else if (rs == RETURN_STATUS::kERROR || rs == RETURN_STATUS::kUNKNOWN) {
0123                 JException ex ("JEventSource threw RETURN_STATUS::kERROR or kUNKNOWN");
0124                 ex.plugin_name = m_plugin_name;
0125                 ex.type_name = m_type_name;
0126                 ex.function_name = "JEventSource::GetEvent";
0127                 ex.instance_name = m_resource_name;
0128                 throw ex;
0129             }
0130         }
0131         catch (JException& ex) {
0132             if (ex.function_name.empty()) ex.function_name = "JEventSource::GetEvent";
0133             if (ex.type_name.empty()) ex.type_name = m_type_name;
0134             if (ex.instance_name.empty()) ex.instance_name = m_prefix;
0135             if (ex.plugin_name.empty()) ex.plugin_name = m_plugin_name;
0136             throw ex;
0137         }
0138         catch (std::exception& e){
0139             auto ex = JException(e.what());
0140             ex.exception_type = JTypeInfo::demangle_current_exception_type();
0141             ex.nested_exception = std::current_exception();
0142             ex.function_name = "JEventSource::GetEvent";
0143             ex.type_name = m_type_name;
0144             ex.instance_name = m_prefix;
0145             ex.plugin_name = m_plugin_name;
0146             throw ex;
0147         }
0148         catch (...) {
0149             auto ex = JException("Unknown exception");
0150             ex.exception_type = JTypeInfo::demangle_current_exception_type();
0151             ex.nested_exception = std::current_exception();
0152             ex.function_name = "JEventSource::GetEvent";
0153             ex.type_name = m_type_name;
0154             ex.instance_name = m_prefix;
0155             ex.plugin_name = m_plugin_name;
0156             throw ex;
0157         }
0158
0159         if (result == Result::Success) {
0160             m_events_emitted += 1;
0161             // We only set the JEventSource once Emit/GetEvent succeeded because it also controls
0162             // m_events_processed and EventSource::FinishEvent
0163             event->SetJEventSource(this);
0164
0165             // We end up here if we read an entry in our file or retrieved a message from our socket,
0166             // and believe we could obtain another one immediately if we wanted to
0167             for (auto* output : GetOutputs()) {
0168                 output->EulerianStore(*event->GetFactorySet());
0169             }
0170             for (auto* output : GetVariadicOutputs()) {
0171                 output->EulerianStore(*event->GetFactorySet());
0172             }
0173             return Result::Success;
0174         }
0175         else if (result == Result::FailureFinished) {
0176             // We end up here if we tried to read an entry in a file, but found EOF
0177             // or if we received a message from a socket that contained no data and indicated no more data will be coming
0178             DoClose(false);
0179             return Result::FailureFinished;
0180         }
0181         else {
0182             return result;
0183         }
0184     }
0185     else { // status == Closed
0186         return Result::FailureFinished;
0187     }
0188 }
0189
0190
0191 void JEventSource::DoFinishEvent(JEvent& event) {
0192
0193     m_events_processed.fetch_add(1);
0194     if (m_enable_finish_event) {
0195         std::lock_guard<std::mutex> lock(m_mutex);
0196         CallWithJExceptionWrapper("JEventSource::FinishEvent", [&](){
0197             FinishEvent(event);
0198         });
0199     }
0200 }
0201
0202 void JEventSource::Summarize(JComponentSummary& summary) const {
0203
0204     auto* result = new JComponentSummary::Component(
0205         "Source", GetPrefix(), GetTypeName(), GetLevel(), GetPluginName());
0206
0207     SummarizeOutputs(*result);
0208     summary.Add(result);
0209 }
0210
0211
0212 std::pair<JEventSource::Result, size_t> JEventSource::Skip(JEvent& event, size_t events_to_skip) {
0213
0214     // Return values
0215     Result result = Result::Success;
0216
0217     while (events_to_skip > 0 && result == Result::Success) {
0218         try {
0219             event.SetJEventSource(this);
0220             auto previous_origin = event.GetJCallGraphRecorder()->SetInsertDataOrigin( JCallGraphRecorder::ORIGIN_FROM_SOURCE);  // (see note at top of JCallGraphRecorder.h)
0221             if (m_callback_style == CallbackStyle::LegacyMode) {
0222                 GetEvent(event.shared_from_this());
0223             }
0224             else {
0225                 result = Emit(event);
0226             }
0227             event.GetJCallGraphRecorder()->SetInsertDataOrigin( previous_origin );
0228             // We need to call FinishEvent, but we don't want to call it from inside JEvent::Clear() because we are already inside the lock.
0229             // So instead, we call it ourselves out here. This has the added benefit of letting us avoid updating m_events_processed.
0230             if (m_enable_finish_event) {
0231                 CallWithJExceptionWrapper("JEventSource::FinishEvent", [&](){ FinishEvent(event); });
0232             }
0233             event.SetJEventSource(nullptr); // This tells event::Clear() _not_ to call FinishEvent()
0234             event.Clear();
0235             events_to_skip -= 1;
0236         }
0237         catch (RETURN_STATUS rs) {
0238
0239             if (rs == RETURN_STATUS::kNO_MORE_EVENTS) {
0240                 DoClose(false);
0241                 result = Result::FailureFinished;
0242             }
0243             else if (rs == RETURN_STATUS::kTRY_AGAIN || rs == RETURN_STATUS::kBUSY) {
0244                 result = Result::FailureTryAgain;
0245             }
0246             else if (rs == RETURN_STATUS::kERROR || rs == RETURN_STATUS::kUNKNOWN) {
0247                 JException ex ("JEventSource threw RETURN_STATUS::kERROR or kUNKNOWN");
0248                 ex.plugin_name = m_plugin_name;
0249                 ex.type_name = m_type_name;
0250                 ex.function_name = (m_callback_style == CallbackStyle::LegacyMode) ? "JEventSource::GetEvent" : "JEventSource::Emit";
0251                 ex.instance_name = m_resource_name;
0252                 throw ex;
0253             }
0254         }
0255         catch (JException& ex) {
0256             if (ex.function_name.empty()) ex.function_name = (m_callback_style == CallbackStyle::LegacyMode) ? "JEventSource::GetEvent" : "JEventSource::Emit";
0257             if (ex.type_name.empty()) ex.type_name = m_type_name;
0258             if (ex.instance_name.empty()) ex.instance_name = m_prefix;
0259             if (ex.plugin_name.empty()) ex.plugin_name = m_plugin_name;
0260             throw ex;
0261         }
0262         catch (std::exception& e){
0263             auto ex = JException(e.what());
0264             ex.exception_type = JTypeInfo::demangle_current_exception_type();
0265             ex.nested_exception = std::current_exception();
0266             ex.function_name = (m_callback_style == CallbackStyle::LegacyMode) ? "JEventSource::GetEvent" : "JEventSource::Emit";
0267             ex.type_name = m_type_name;
0268             ex.instance_name = m_prefix;
0269             ex.plugin_name = m_plugin_name;
0270             throw ex;
0271         }
0272         catch (...) {
0273             auto ex = JException("Unknown exception");
0274             ex.exception_type = JTypeInfo::demangle_current_exception_type();
0275             ex.nested_exception = std::current_exception();
0276             ex.function_name = (m_callback_style == CallbackStyle::LegacyMode) ? "JEventSource::GetEvent" : "JEventSource::Emit";
0277             ex.type_name = m_type_name;
0278             ex.instance_name = m_prefix;
0279             ex.plugin_name = m_plugin_name;
0280             throw ex;
0281         }
0282     }
0283
0284     return {result, events_to_skip};
0285 }
0286
0287
0288
0289