JANA/Topology/JTopologyBuilder.cc

0001
0002 // Copyright 2020, Jefferson Science Associates, LLC.
0003 // Subject to the terms in the LICENSE file found in the top-level directory.
0004
0005
0006 #include "JTopologyBuilder.h"
0007
0008 #include "JEventSourceArrow.h"
0009 #include "JEventMapArrow.h"
0010 #include "JEventTapArrow.h"
0011 #include "JUnfoldArrow.h"
0012 #include "JFoldArrow.h"
0013 #include <JANA/JEventProcessor.h>
0014 #include <JANA/Utils/JTablePrinter.h>
0015 #include <string>
0016 #include <vector>
0017
0018
0019 JTopologyBuilder::JTopologyBuilder() {
0020     SetPrefix("jana");
0021 }
0022
0023 JTopologyBuilder::~JTopologyBuilder() {
0024     for (auto arrow : arrows) {
0025         delete arrow;
0026     }
0027     for (auto queue : queues) {
0028         delete queue;
0029     }
0030     for (auto pool : pools) {
0031         delete pool;
0032     }
0033 }
0034
0035 std::string JTopologyBuilder::print_topology() {
0036     JTablePrinter t;
0037     t.AddColumn("Arrow", JTablePrinter::Justify::Left, 0);
0038     t.AddColumn("Parallel", JTablePrinter::Justify::Center, 0);
0039     t.AddColumn("Direction", JTablePrinter::Justify::Left, 0);
0040     t.AddColumn("Place", JTablePrinter::Justify::Left, 0);
0041     t.AddColumn("ID", JTablePrinter::Justify::Left, 0);
0042
0043     // Build index lookup for queues
0044     int i = 0;
0045     std::map<void*, int> lookup;
0046     for (JEventQueue* queue : queues) {
0047         lookup[queue] = i;
0048         i += 1;
0049     }
0050     // Build index lookup for pools
0051     for (JEventPool* pool : pools) {
0052         lookup[pool] = i;
0053         i += 1;
0054     }
0055     // Build table
0056
0057     bool show_row = true;
0058
0059     for (JArrow* arrow : arrows) {
0060
0061         show_row = true;
0062         for (JArrow::Port& port : arrow->m_ports) {
0063             if (show_row) {
0064                 t | arrow->get_name();
0065                 t | arrow->is_parallel();
0066                 show_row = false;
0067             }
0068             else {
0069                 t | "" | "" ;
0070             }
0071             auto place_index = lookup[(port.queue!=nullptr) ? (void*) port.queue : (void*) port.pool];
0072
0073             t | ((port.is_input) ? "Input ": "Output");
0074             t | ((port.queue != nullptr) ? "Queue ": "Pool");
0075             t | place_index;
0076         }
0077     }
0078     return t.Render();
0079 }
0080
0081
0082 /// set_cofigure_fn lets the user provide a lambda that sets up a topology after all components have been loaded.
0083 /// It provides an 'empty' JArrowTopology which has been furnished with a pointer to the JComponentManager, the JEventPool,
0084 /// and the JProcessorMapping (in case you care about NUMA details). However, it does not contain any queues or arrows.
0085 /// You have to furnish those yourself.
0086 void JTopologyBuilder::set_configure_fn(std::function<void(JTopologyBuilder&)> configure_fn) {
0087     m_configure_topology = std::move(configure_fn);
0088 }
0089
0090 void JTopologyBuilder::create_topology() {
0091     mapping.initialize(static_cast<JProcessorMapping::AffinityStrategy>(m_affinity),
0092                        static_cast<JProcessorMapping::LocalityStrategy>(m_locality));
0093
0094     if (m_configure_topology) {
0095         m_configure_topology(*this);
0096         LOG_WARN(GetLogger()) << "Found custom topology configurator! Modified arrow topology is: \n" << print_topology() << LOG_END;
0097     }
0098     else {
0099         attach_level(JEventLevel::Run, nullptr, nullptr);
0100         LOG_INFO(GetLogger()) << "Arrow topology is:\n" << print_topology() << LOG_END;
0101     }
0102     for (auto* arrow : arrows) {
0103         arrow->set_logger(GetLogger());
0104     }
0105 }
0106
0107
0108 void JTopologyBuilder::acquire_services(JServiceLocator *sl) {
0109
0110     m_components = sl->get<JComponentManager>();
0111
0112     // We default event pool size to be equal to nthreads
0113     // We parse the 'nthreads' parameter two different ways for backwards compatibility.
0114     if (m_params->Exists("nthreads")) {
0115         if (m_params->GetParameterValue<std::string>("nthreads") == "Ncores") {
0116             m_max_inflight_events = JCpuInfo::GetNumCpus();
0117         } else {
0118             m_max_inflight_events = m_params->GetParameterValue<int>("nthreads");
0119         }
0120     }
0121
0122     m_params->SetDefaultParameter("jana:max_inflight_events", m_max_inflight_events,
0123                                     "The number of events which may be in-flight at once. Should be at least `nthreads` to prevent starvation; more gives better load balancing.")
0124             ->SetIsAdvanced(true);
0125
0126     /*
0127     m_params->SetDefaultParameter("jana:enable_stealing", m_enable_stealing,
0128                                     "Enable work stealing. Improves load balancing when jana:locality != 0; otherwise does nothing.")
0129             ->SetIsAdvanced(true);
0130     */
0131     m_params->SetDefaultParameter("jana:affinity", m_affinity,
0132                                     "Constrain worker thread CPU affinity. 0=Let the OS decide. 1=Avoid extra memory movement at the expense of using hyperthreads. 2=Avoid hyperthreads at the expense of extra memory movement")
0133             ->SetIsAdvanced(true);
0134     m_params->SetDefaultParameter("jana:locality", m_locality,
0135                                     "Constrain memory locality. 0=No constraint. 1=Events stay on the same socket. 2=Events stay on the same NUMA domain. 3=Events stay on same core. 4=Events stay on same cpu/hyperthread.")
0136             ->SetIsAdvanced(true);
0137 };
0138
0139
0140 void JTopologyBuilder::connect(JArrow* upstream, size_t upstream_port_id, JArrow* downstream, size_t downstream_port_id) {
0141
0142     JEventQueue* queue = nullptr;
0143
0144     JArrow::Port& downstream_port = downstream->m_ports.at(downstream_port_id);
0145     if (downstream_port.queue != nullptr) {
0146         // If the queue already exists, use that!
0147         queue = downstream_port.queue;
0148     }
0149     else {
0150         // Create a new queue
0151         queue = new JEventQueue(m_max_inflight_events, mapping.get_loc_count());
0152         downstream_port.queue = queue;
0153         queues.push_back(queue);
0154     }
0155     downstream_port.pool = nullptr;
0156
0157     JArrow::Port& upstream_port = upstream->m_ports.at(upstream_port_id);
0158     upstream_port.queue = queue;
0159     upstream_port.pool = nullptr;
0160 }
0161
0162
0163 void JTopologyBuilder::connect_to_first_available(JArrow* upstream, size_t upstream_port, std::vector<std::pair<JArrow*, size_t>> downstreams) {
0164     for (auto& [downstream, downstream_port_id] : downstreams) {
0165         if (downstream != nullptr) {
0166             connect(upstream, upstream_port, downstream, downstream_port_id);
0167             return;
0168         }
0169     }
0170 }
0171
0172 std::pair<JEventTapArrow*, JEventTapArrow*> JTopologyBuilder::create_tap_chain(std::vector<JEventProcessor*>& procs, std::string level) {
0173
0174     JEventTapArrow* first = nullptr;
0175     JEventTapArrow* last = nullptr;
0176
0177     int i=1;
0178     std::string arrow_name = level + "Tap";
0179     for (JEventProcessor* proc : procs) {
0180         if (procs.size() > 1) {
0181             arrow_name += std::to_string(i++);
0182         }
0183         JEventTapArrow* current = new JEventTapArrow(arrow_name);
0184         current->add_processor(proc);
0185         arrows.push_back(current);
0186         if (first == nullptr) {
0187             first = current;
0188         }
0189         if (last != nullptr) {
0190             connect(last, JEventTapArrow::EVENT_OUT, current, JEventTapArrow::EVENT_IN);
0191         }
0192         last = current;
0193     }
0194     return {first, last};
0195 }
0196
0197
0198 void JTopologyBuilder::attach_level(JEventLevel current_level, JUnfoldArrow* parent_unfolder, JFoldArrow* parent_folder) {
0199     std::stringstream ss;
0200     ss << current_level;
0201     auto level_str = ss.str();
0202
0203     // Find all event sources at this level
0204     std::vector<JEventSource*> sources_at_level;
0205     for (JEventSource* source : m_components->get_evt_srces()) {
0206         if (source->GetLevel() == current_level) {
0207             sources_at_level.push_back(source);
0208         }
0209     }
0210
0211     // Find all unfolders at this level
0212     std::vector<JEventUnfolder*> unfolders_at_level;
0213     for (JEventUnfolder* unfolder : m_components->get_unfolders()) {
0214         if (unfolder->GetLevel() == current_level) {
0215             unfolders_at_level.push_back(unfolder);
0216         }
0217     }
0218
0219     // Find all processors at this level
0220     std::vector<JEventProcessor*> mappable_procs_at_level;
0221     std::vector<JEventProcessor*> tappable_procs_at_level;
0222
0223     for (JEventProcessor* proc : m_components->get_evt_procs()) {
0224         if (proc->GetLevel() == current_level) {
0225             mappable_procs_at_level.push_back(proc);
0226             if (proc->GetCallbackStyle() != JEventProcessor::CallbackStyle::LegacyMode) {
0227                 tappable_procs_at_level.push_back(proc);
0228             }
0229         }
0230     }
0231
0232
0233     bool is_top_level = (parent_unfolder == nullptr);
0234     if (is_top_level && sources_at_level.size() == 0) {
0235         // Skip level entirely when no source is present.
0236         LOG_TRACE(GetLogger()) << "JTopologyBuilder: No sources found at level " << current_level << ", skipping" << LOG_END;
0237         JEventLevel next = next_level(current_level);
0238         if (next == JEventLevel::None) {
0239             LOG_WARN(GetLogger()) << "No sources found: Processing topology will be empty." << LOG_END;
0240             return;
0241         }
0242         return attach_level(next, nullptr, nullptr);
0243     }
0244
0245     // Enforce constraints on what our builder will accept (at least for now)
0246     if (!is_top_level && !sources_at_level.empty()) {
0247         throw JException("Topology forbids event sources at lower event levels in the topology");
0248     }
0249     if ((parent_unfolder == nullptr && parent_folder != nullptr) || (parent_unfolder != nullptr && parent_folder == nullptr)) {
0250         throw JException("Topology requires matching unfolder/folder arrow pairs");
0251     }
0252     if (unfolders_at_level.size() > 1) {
0253         throw JException("Multiple JEventUnfolders provided for level %s", level_str.c_str());
0254     }
0255     // Another constraint is that the highest level of the topology has an event sources, but this is automatically handled by
0256     // the level-skipping logic above
0257
0258
0259     // Fill out arrow grid from components at this event level
0260     // --------------------------
0261     // 0. Pool
0262     // --------------------------
0263     JEventPool* pool_at_level = new JEventPool(m_components, m_max_inflight_events, m_location_count, current_level);
0264     pools.push_back(pool_at_level); // Hand over ownership of the pool to the topology
0265     LOG_INFO(GetLogger()) << "Created event pool with level=" << current_level << " and size=" << m_max_inflight_events;
0266
0267     // --------------------------
0268     // 1. Source
0269     // --------------------------
0270     JEventSourceArrow* src_arrow = nullptr;
0271     bool need_source = !sources_at_level.empty();
0272     if (need_source) {
0273         src_arrow = new JEventSourceArrow(level_str+"Source", sources_at_level);
0274         src_arrow->attach(pool_at_level, JEventSourceArrow::EVENT_IN);
0275         src_arrow->attach(pool_at_level, JEventSourceArrow::EVENT_OUT);
0276         arrows.push_back(src_arrow);
0277     }
0278
0279     // --------------------------
0280     // 2. Map1
0281     // --------------------------
0282     bool have_parallel_sources = false;
0283     for (JEventSource* source: sources_at_level) {
0284         have_parallel_sources |= source->IsPreprocessEnabled();
0285     }
0286     bool have_unfolder = !unfolders_at_level.empty();
0287     JEventMapArrow* map1_arrow = nullptr;
0288     bool need_map1 = (have_parallel_sources || have_unfolder);
0289
0290     if (need_map1) {
0291         map1_arrow = new JEventMapArrow(level_str+"Map1");
0292         for (JEventSource* source: sources_at_level) {
0293             if (source->IsPreprocessEnabled()) {
0294                 map1_arrow->add_source(source);
0295             }
0296         }
0297         for (JEventUnfolder* unf: unfolders_at_level) {
0298             map1_arrow->add_unfolder(unf);
0299         }
0300         map1_arrow->attach(pool_at_level, JEventMapArrow::EVENT_IN);
0301         map1_arrow->attach(pool_at_level, JEventMapArrow::EVENT_OUT);
0302         arrows.push_back(map1_arrow);
0303     }
0304
0305     // --------------------------
0306     // 3. Unfold
0307     // --------------------------
0308     JUnfoldArrow* unfold_arrow = nullptr;
0309     bool need_unfold = have_unfolder;
0310     if (need_unfold) {
0311         unfold_arrow = new JUnfoldArrow(level_str+"Unfold", unfolders_at_level[0]);
0312         unfold_arrow->attach(pool_at_level, JUnfoldArrow::REJECTED_PARENT_OUT);
0313         arrows.push_back(unfold_arrow);
0314     }
0315
0316     // --------------------------
0317     // 4. Fold
0318     // --------------------------
0319     JFoldArrow* fold_arrow = nullptr;
0320     bool need_fold = have_unfolder;
0321     if(need_fold) {
0322         fold_arrow = new JFoldArrow(level_str+"Fold", current_level, unfolders_at_level[0]->GetChildLevel());
0323         arrows.push_back(fold_arrow);
0324         fold_arrow->attach(pool_at_level, JFoldArrow::PARENT_OUT);
0325     }
0326
0327     // --------------------------
0328     // 5. Map2
0329     // --------------------------
0330     JEventMapArrow* map2_arrow = nullptr;
0331     bool need_map2 = !mappable_procs_at_level.empty();
0332     if (need_map2) {
0333         map2_arrow = new JEventMapArrow(level_str+"Map2");
0334         for (JEventProcessor* proc : mappable_procs_at_level) {
0335             map2_arrow->add_processor(proc);
0336             map2_arrow->attach(pool_at_level, JEventMapArrow::EVENT_IN);
0337             map2_arrow->attach(pool_at_level, JEventMapArrow::EVENT_OUT);
0338         }
0339         arrows.push_back(map2_arrow);
0340     }
0341
0342     // --------------------------
0343     // 6. Tap
0344     // --------------------------
0345     JEventTapArrow* first_tap_arrow = nullptr;
0346     JEventTapArrow* last_tap_arrow = nullptr;
0347     bool need_tap = !tappable_procs_at_level.empty();
0348     if (need_tap) {
0349         std::tie(first_tap_arrow, last_tap_arrow) = create_tap_chain(tappable_procs_at_level, level_str);
0350         first_tap_arrow->attach(pool_at_level, JEventTapArrow::EVENT_IN);
0351         last_tap_arrow->attach(pool_at_level, JEventTapArrow::EVENT_OUT);
0352     }
0353
0354
0355     // Now that we've set up our component grid, we can do wiring!
0356     // --------------------------
0357     // 1. Source
0358     // --------------------------
0359     if (parent_unfolder != nullptr) {
0360         parent_unfolder->attach(pool_at_level, JUnfoldArrow::CHILD_IN);
0361         connect_to_first_available(parent_unfolder, JUnfoldArrow::CHILD_OUT,
0362                                    {{map1_arrow, JEventMapArrow::EVENT_IN}, {unfold_arrow, JUnfoldArrow::PARENT_IN}, {map2_arrow, JEventMapArrow::EVENT_IN}, {first_tap_arrow, JEventTapArrow::EVENT_IN}, {parent_folder, JFoldArrow::CHILD_IN}});
0363     }
0364     if (src_arrow != nullptr) {
0365         connect_to_first_available(src_arrow, JEventSourceArrow::EVENT_OUT,
0366                                    {{map1_arrow, JEventMapArrow::EVENT_IN}, {unfold_arrow, JUnfoldArrow::PARENT_IN}, {map2_arrow, JEventMapArrow::EVENT_IN}, {first_tap_arrow, JEventTapArrow::EVENT_IN}, {parent_folder, JFoldArrow::CHILD_IN}});
0367     }
0368     if (map1_arrow != nullptr) {
0369         connect_to_first_available(map1_arrow, JEventMapArrow::EVENT_OUT,
0370                                    {{unfold_arrow, JUnfoldArrow::PARENT_IN}, {map2_arrow, JEventMapArrow::EVENT_IN}, {first_tap_arrow, JEventTapArrow::EVENT_IN}, {parent_folder, JFoldArrow::CHILD_IN}});
0371     }
0372     if (unfold_arrow != nullptr) {
0373         connect_to_first_available(unfold_arrow, JUnfoldArrow::REJECTED_PARENT_OUT,
0374                                    {{map2_arrow, JEventMapArrow::EVENT_IN}, {first_tap_arrow, JEventTapArrow::EVENT_IN}, {parent_folder, JFoldArrow::CHILD_IN}});
0375     }
0376     if (fold_arrow != nullptr) {
0377         connect_to_first_available(fold_arrow, JFoldArrow::CHILD_OUT,
0378                                    {{map2_arrow, JEventMapArrow::EVENT_IN}, {first_tap_arrow, JEventTapArrow::EVENT_IN}, {parent_folder, JFoldArrow::CHILD_IN}});
0379     }
0380     if (map2_arrow != nullptr) {
0381         connect_to_first_available(map2_arrow, JEventMapArrow::EVENT_OUT,
0382                                    {{first_tap_arrow, JEventTapArrow::EVENT_IN}, {parent_folder, JFoldArrow::CHILD_IN}});
0383     }
0384     if (last_tap_arrow != nullptr) {
0385         connect_to_first_available(last_tap_arrow, JEventTapArrow::EVENT_OUT,
0386                                    {{parent_folder, JFoldArrow::CHILD_IN}});
0387     }
0388     if (parent_folder != nullptr) {
0389         parent_folder->attach(pool_at_level, JFoldArrow::CHILD_OUT);
0390     }
0391
0392     // Finally, we recur over lower levels!
0393     if (need_unfold) {
0394         auto next_level = unfolders_at_level[0]->GetChildLevel();
0395         attach_level(next_level, unfold_arrow, fold_arrow);
0396     }
0397     else {
0398         // This is the lowest level
0399         // TODO: Improve logic for determining event counts for multilevel topologies
0400         if (last_tap_arrow != nullptr) {
0401             last_tap_arrow->set_is_sink(true);
0402         }
0403         else if (map2_arrow != nullptr) {
0404             map2_arrow->set_is_sink(true);
0405         }
0406         else if (map1_arrow != nullptr) {
0407             map1_arrow->set_is_sink(true);
0408         }
0409         else if (src_arrow != nullptr) {
0410             src_arrow->set_is_sink(true);
0411         }
0412     }
0413 }
0414
0415
0416