Back to home page

EIC code displayed by LXR
 
 

    

Warning, file /jana2/src/libraries/JANA/Topology/JTopologyBuilder.cc was not indexed or was modified since last indexation (in which case cross-reference links may be missing, inaccurate or erroneous).

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     LOG_INFO(GetLogger()) << "Creating event pool with level=" << current_level << " and size=" << m_max_inflight_events;
0264     JEventPool* pool_at_level = new JEventPool(m_components, m_max_inflight_events, m_location_count, current_level);
0265     pools.push_back(pool_at_level); // Hand over ownership of the pool to the topology
0266     LOG_INFO(GetLogger()) << "Created event pool with level=" << current_level << " and size=" << m_max_inflight_events;
0267 
0268     // --------------------------
0269     // 1. Source
0270     // --------------------------
0271     JEventSourceArrow* src_arrow = nullptr;
0272     bool need_source = !sources_at_level.empty();
0273     if (need_source) {
0274         src_arrow = new JEventSourceArrow(level_str+"Source", sources_at_level);
0275         src_arrow->attach(pool_at_level, JEventSourceArrow::EVENT_IN);
0276         src_arrow->attach(pool_at_level, JEventSourceArrow::EVENT_OUT);
0277         arrows.push_back(src_arrow);
0278     }
0279 
0280     // --------------------------
0281     // 2. Map1
0282     // --------------------------
0283     bool have_parallel_sources = false;
0284     for (JEventSource* source: sources_at_level) {
0285         have_parallel_sources |= source->IsProcessParallelEnabled();
0286     }
0287     bool have_unfolder = !unfolders_at_level.empty();
0288     JEventMapArrow* map1_arrow = nullptr;
0289     bool need_map1 = (have_parallel_sources || have_unfolder);
0290 
0291     if (need_map1) {
0292         map1_arrow = new JEventMapArrow(level_str+"Map1");
0293         for (JEventSource* source: sources_at_level) {
0294             if (source->IsProcessParallelEnabled()) {
0295                 map1_arrow->add_source(source);
0296             }
0297         }
0298         for (JEventUnfolder* unf: unfolders_at_level) {
0299             map1_arrow->add_unfolder(unf);
0300         }
0301         map1_arrow->attach(pool_at_level, JEventMapArrow::EVENT_IN);
0302         map1_arrow->attach(pool_at_level, JEventMapArrow::EVENT_OUT);
0303         arrows.push_back(map1_arrow);
0304     }
0305 
0306     // --------------------------
0307     // 3. Unfold
0308     // --------------------------
0309     JUnfoldArrow* unfold_arrow = nullptr;
0310     bool need_unfold = have_unfolder;
0311     if (need_unfold) {
0312         unfold_arrow = new JUnfoldArrow(level_str+"Unfold", unfolders_at_level[0]);
0313         unfold_arrow->attach(pool_at_level, JUnfoldArrow::REJECTED_PARENT_OUT); 
0314         arrows.push_back(unfold_arrow);
0315     }
0316 
0317     // --------------------------
0318     // 4. Fold
0319     // --------------------------
0320     JFoldArrow* fold_arrow = nullptr;
0321     bool need_fold = have_unfolder;
0322     if(need_fold) {
0323         fold_arrow = new JFoldArrow(level_str+"Fold", current_level, unfolders_at_level[0]->GetChildLevel());
0324         arrows.push_back(fold_arrow);
0325         fold_arrow->attach(pool_at_level, JFoldArrow::PARENT_OUT); 
0326     }
0327 
0328     // --------------------------
0329     // 5. Map2
0330     // --------------------------
0331     JEventMapArrow* map2_arrow = nullptr;
0332     bool need_map2 = !mappable_procs_at_level.empty();
0333     if (need_map2) {
0334         map2_arrow = new JEventMapArrow(level_str+"Map2");
0335         for (JEventProcessor* proc : mappable_procs_at_level) {
0336             map2_arrow->add_processor(proc);
0337             map2_arrow->attach(pool_at_level, JEventMapArrow::EVENT_IN);
0338             map2_arrow->attach(pool_at_level, JEventMapArrow::EVENT_OUT);
0339         }
0340         arrows.push_back(map2_arrow);
0341     }
0342 
0343     // --------------------------
0344     // 6. Tap
0345     // --------------------------
0346     JEventTapArrow* first_tap_arrow = nullptr;
0347     JEventTapArrow* last_tap_arrow = nullptr;
0348     bool need_tap = !tappable_procs_at_level.empty();
0349     if (need_tap) {
0350         std::tie(first_tap_arrow, last_tap_arrow) = create_tap_chain(tappable_procs_at_level, level_str);
0351         first_tap_arrow->attach(pool_at_level, JEventTapArrow::EVENT_IN);
0352         last_tap_arrow->attach(pool_at_level, JEventTapArrow::EVENT_OUT);
0353     }
0354 
0355 
0356     // Now that we've set up our component grid, we can do wiring!
0357     // --------------------------
0358     // 1. Source
0359     // --------------------------
0360     if (parent_unfolder != nullptr) {
0361         parent_unfolder->attach(pool_at_level, JUnfoldArrow::CHILD_IN);
0362         connect_to_first_available(parent_unfolder, JUnfoldArrow::CHILD_OUT,
0363                                    {{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}});
0364     }
0365     if (src_arrow != nullptr) {
0366         connect_to_first_available(src_arrow, JEventSourceArrow::EVENT_OUT,
0367                                    {{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}});
0368     }
0369     if (map1_arrow != nullptr) {
0370         connect_to_first_available(map1_arrow, JEventMapArrow::EVENT_OUT,
0371                                    {{unfold_arrow, JUnfoldArrow::PARENT_IN}, {map2_arrow, JEventMapArrow::EVENT_IN}, {first_tap_arrow, JEventTapArrow::EVENT_IN}, {parent_folder, JFoldArrow::CHILD_IN}});
0372     }
0373     if (unfold_arrow != nullptr) {
0374         connect_to_first_available(unfold_arrow, JUnfoldArrow::REJECTED_PARENT_OUT,
0375                                    {{map2_arrow, JEventMapArrow::EVENT_IN}, {first_tap_arrow, JEventTapArrow::EVENT_IN}, {parent_folder, JFoldArrow::CHILD_IN}});
0376     }
0377     if (fold_arrow != nullptr) {
0378         connect_to_first_available(fold_arrow, JFoldArrow::CHILD_OUT,
0379                                    {{map2_arrow, JEventMapArrow::EVENT_IN}, {first_tap_arrow, JEventTapArrow::EVENT_IN}, {parent_folder, JFoldArrow::CHILD_IN}});
0380     }
0381     if (map2_arrow != nullptr) {
0382         connect_to_first_available(map2_arrow, JEventMapArrow::EVENT_OUT,
0383                                    {{first_tap_arrow, JEventTapArrow::EVENT_IN}, {parent_folder, JFoldArrow::CHILD_IN}});
0384     }
0385     if (last_tap_arrow != nullptr) {
0386         connect_to_first_available(last_tap_arrow, JEventTapArrow::EVENT_OUT,
0387                                    {{parent_folder, JFoldArrow::CHILD_IN}});
0388     }
0389     if (parent_folder != nullptr) {
0390         parent_folder->attach(pool_at_level, JFoldArrow::CHILD_OUT);
0391     }
0392 
0393     // Finally, we recur over lower levels!
0394     if (need_unfold) {
0395         auto next_level = unfolders_at_level[0]->GetChildLevel();
0396         attach_level(next_level, unfold_arrow, fold_arrow);
0397     }
0398     else {
0399         // This is the lowest level
0400         // TODO: Improve logic for determining event counts for multilevel topologies
0401         if (last_tap_arrow != nullptr) {
0402             last_tap_arrow->set_is_sink(true);
0403         }
0404         else if (map2_arrow != nullptr) {
0405             map2_arrow->set_is_sink(true);
0406         }
0407         else if (map1_arrow != nullptr) {
0408             map1_arrow->set_is_sink(true);
0409         }
0410         else if (src_arrow != nullptr) {
0411             src_arrow->set_is_sink(true);
0412         }
0413     }
0414 }
0415 
0416 
0417
[ Source navigation ] [ Diff markup ] [ Identifier search ] [ general search ]

This page was automatically generated by the 2.3.7 LXR engine.

The LXR team
Valid CSS 2.1! Valid HTML 4.01!