JANA/Topology/JEventPool.h

0001 // Copyright 2024, Jefferson Science Associates, LLC.
0002 // Subject to the terms in the LICENSE file found in the top-level directory.
0003 // Author: Nathan Brei
0004
0005 #pragma once
0006
0007 #include <JANA/Topology/JEventQueue.h>
0008 #include <JANA/Services/JComponentManager.h>
0009 #include <vector>
0010
0011
0012 class JEventPool : public JEventQueue {
0013 private:
0014     std::vector<std::shared_ptr<JEvent>> m_owned_events;
0015     std::shared_ptr<JComponentManager> m_component_manager;
0016     JEventLevel m_level;
0017     JLogger m_logger;
0018
0019 public:
0020     inline JEventPool(std::shared_ptr<JComponentManager> component_manager,
0021                       size_t max_inflight_events,
0022                       size_t location_count,
0023                       JEventLevel level = JEventLevel::PhysicsEvent)
0024
0025             : JEventQueue(max_inflight_events, location_count)
0026             , m_component_manager(component_manager)
0027             , m_level(level) {
0028
0029         // Create JEvents and distribute them among the pools.
0030         m_owned_events.reserve(max_inflight_events);
0031         for (size_t evt_idx=0; evt_idx<max_inflight_events; evt_idx++) {
0032
0033             m_owned_events.push_back(std::make_shared<JEvent>());
0034             auto evt = &m_owned_events.back();
0035             m_component_manager->configure_event(**evt);
0036             (*evt)->SetLevel(m_level);
0037             Push(evt->get(), evt_idx % location_count);
0038         }
0039     }
0040
0041     void Scale(size_t capacity) override {
0042         auto old_capacity = m_capacity;
0043         if (capacity < old_capacity) {
0044             // Downscaling a JEventPool is tricky because even draining the topology
0045             // doesn't guarantee that all JEvents have been returned to the JEventPools by
0046             // the time the topology pauses. Consider JEventUnfolder, which may very well hold
0047             // on to a child event that it won't need because the parent event source has
0048             // paused or finished.
0049             // For now we don't reduce the size of the pools or queues because there's little
0050             // penalty to keeping them around (the main penalty comes from event creation time
0051             // and memory footprint, but if we are downscaling we already paid that)
0052             return;
0053         }
0054
0055         // Resize queues to fit new capacity
0056         m_capacity = capacity;
0057         for (auto& local_queue: m_local_queues) {
0058             local_queue->ringbuffer.resize(capacity, nullptr);
0059             local_queue->capacity = capacity;
0060         }
0061
0062         // Create new JEvents, add to owned_events, and distribute to queues
0063         m_owned_events.reserve(capacity);
0064
0065         for (size_t evt_idx=old_capacity; evt_idx<capacity; evt_idx++) {
0066             m_owned_events.push_back(std::make_shared<JEvent>());
0067             auto evt = &m_owned_events.back();
0068             m_component_manager->configure_event(**evt);
0069             (*evt)->SetLevel(m_level);
0070             Push(evt->get(), evt_idx % GetLocationCount());
0071         }
0072     }
0073
0074     void Finalize() {
0075         for (auto& evt : m_owned_events) {
0076             evt->Finish();
0077         }
0078     }
0079
0080 };
0081
0082