JANA/Topology/JArrow.cc

0001
0002 #include <JANA/Topology/JArrow.h>
0003
0004
0005 JArrow::Port& JArrow::AddPort(std::string name, JEventLevel level, PortDirection direction) {
0006     if (m_port_lookup.find(name) != m_port_lookup.end()) {
0007         throw JException("Port with name '%s' already exists", name.c_str());
0008     }
0009     auto port = std::make_unique<Port>(name, level);
0010     auto port_raw_ptr = port.get();
0011     m_ports.push_back(std::move(port));
0012     m_port_lookup[name] = m_ports.size()-1;
0013
0014     auto it = m_auto_port_lookup.find({level, direction});
0015     if (it != m_auto_port_lookup.end()) {
0016         // There's a conflict! Multiple ports with the same level, direction
0017         // Handle this case by disabling the lookup
0018         it->second = -1;
0019     }
0020     else {
0021         m_auto_port_lookup[{level, direction}] = m_ports.size()-1;
0022     }
0023     return *port_raw_ptr;
0024 }
0025
0026 JEvent* JArrow::Pull(size_t port_index, size_t location_id) {
0027     JEvent* event = nullptr;
0028     auto& port = m_ports.at(port_index);
0029     if (port->GetQueue() != nullptr) {
0030         event = port->GetQueue()->Pop(location_id);
0031     }
0032     else if (port->GetPool() != nullptr){
0033         event = port->GetPool()->Pop( location_id);
0034     }
0035     else {
0036         throw JException("Arrow %s: Port %d not wired!", m_name.c_str(), port_index);
0037     }
0038     // If pop() failed, the returned event is nullptr
0039     return event;
0040 }
0041
0042
0043 void JArrow::Push(OutputData& outputs, size_t output_count, size_t location_id) {
0044     for (size_t output = 0; output < output_count; ++output) {
0045         JEvent* event = outputs[output].first;
0046         int port_index = outputs[output].second;
0047         Port& port = GetPort(port_index);
0048         if (port.GetQueue() != nullptr) {
0049             port.GetQueue()->Push(event, location_id);
0050         }
0051         else if (port.GetPool() != nullptr) {
0052             port.GetPool()->Ingest(event, location_id);
0053         }
0054         else {
0055             throw JException("Arrow %s: Port %s not wired!", m_name.c_str(), port.GetName().c_str());
0056         }
0057     }
0058 }
0059
0060 JArrow::FireResult JArrow::Execute(size_t location_id) {
0061
0062     auto start_total_time = std::chrono::steady_clock::now();
0063     if (m_next_visit_time > start_total_time) {
0064         // If we haven't reached the next visit time, exit immediately
0065         return FireResult::ComeBackLater;
0066     }
0067
0068     JEvent* input = nullptr;
0069     if (m_next_input_port != -1) {
0070         input = Pull(m_next_input_port, location_id);
0071     }
0072
0073     if (input == nullptr && m_next_input_port != -1) {
0074         // Failed to obtain the input we needed; arrow is NOT ready to fire
0075         return FireResult::NotRunYet;
0076     }
0077
0078     // Obtained the input we needed; arrow is ready to fire
0079     // Remember that `input` might be nullptr, in case arrow doesn't need any input event
0080
0081     OutputData outputs;
0082     size_t output_count = 0;
0083     JArrow::FireResult result = JArrow::FireResult::KeepGoing;
0084
0085     Fire(input, outputs, output_count, result);
0086
0087     Push(outputs, output_count, location_id);
0088
0089     return result;
0090 }
0091
0092
0093 std::string ToString(JArrow::FireResult r) {
0094     switch (r) {
0095         case JArrow::FireResult::NotRunYet:     return "NotRunYet";
0096         case JArrow::FireResult::KeepGoing:     return "KeepGoing";
0097         case JArrow::FireResult::ComeBackLater: return "ComeBackLater";
0098         case JArrow::FireResult::Finished:      return "Finished";
0099         default:                                return "Error";
0100     }
0101 }
0102
0103 std::string ToString(JArrow::PortDirection d) {
0104     switch (d) {
0105         case JArrow::PortDirection::In:  return "In";
0106         case JArrow::PortDirection::Out: return "Out";
0107         default:                         return "Unknown";
0108     }
0109 }
0110
0111 int JArrow::GetPortIndex(JEventLevel level, PortDirection direction) {
0112     auto it = m_auto_port_lookup.find({level, direction});
0113     if (it == m_auto_port_lookup.end()) {
0114         throw JException("Unable to find port with (level=%s, direction=%s) on arrow '%s'",
0115                          toString(level).c_str(),
0116                          ToString(direction).c_str(),
0117                          GetName().c_str());
0118     }
0119     else if (it->second == -1) {
0120         throw JException("Ambiguous port with (level=%s, direction=%s) on arrow '%s'",
0121                          toString(level).c_str(),
0122                          ToString(direction).c_str(),
0123                          GetName().c_str());
0124     }
0125     return it->second;
0126 }
0127
0128 int JArrow::GetPortIndex(const std::string& port_name) {
0129     auto it = m_port_lookup.find(port_name);
0130     if (it == m_port_lookup.end()) {
0131         LOG_FATAL(GetLogger()) << "Unable to find port_name '" << port_name << "' on arrow '" << GetName() << "'. Valid port names are:";
0132         for (auto& port : m_ports) {
0133             LOG_FATAL(GetLogger()) << "    " << port->GetName();
0134         }
0135         throw JException("Unable to find port_name '%s' on arrow '%s'", port_name.c_str(), GetName().c_str());
0136     }
0137     return it->second;
0138 }
0139
0140
0141