Merge pull request #270 from JeffersonLab/nbrei_arrow_refactoring

Arrow refactoring
JeffersonLab · Dec 28, 2023 · 5f2826b · 5f2826b
2 parents a14ba26 + 560727c
commit 5f2826b
Show file tree

Hide file tree

Showing 45 changed files with 1,334 additions and 1,471 deletions.
diff --git a/src/examples/BlockExample/BlockExampleSource.h b/src/examples/BlockExample/BlockExampleSource.h
@@ -34,10 +34,10 @@ class BlockExampleSource : public JBlockedEventSource<MyBlock> {
 		return Status::Success;
 	}
 
-	virtual std::vector<std::shared_ptr<JEvent>> DisentangleBlock(MyBlock& block, JEventPool& pool) {
+	virtual std::vector<std::shared_ptr<JEvent>*> DisentangleBlock(MyBlock& block, JEventPool& pool) {
 
 		LOG_DEBUG(m_logger) <<  "BlockDisentangler: Disentangling block " << block.block_number << LOG_END;
-		std::vector<std::shared_ptr<JEvent>> events;
+		std::vector<std::shared_ptr<JEvent>*> events;
 		bool result = pool.get_many(events, block.data.size());
 
 		if (result == false) {
@@ -51,7 +51,7 @@ class BlockExampleSource : public JBlockedEventSource<MyBlock> {
 		size_t i = 0;
 		for (auto datum : block.data) {
 			LOG_DEBUG(m_logger) << "BlockDisentangler: extracted event containing " << datum << LOG_END;
-			events[i++]->Insert(new MyObject(datum));
+			(*(events[i++]))->Insert(new MyObject(datum));
 		}
 		return events;
 	}

diff --git a/src/examples/BlockExample/main.cc b/src/examples/BlockExample/main.cc
@@ -17,13 +17,15 @@ std::shared_ptr<JArrowTopology> configure_block_topology(std::shared_ptr<JArrowT
     auto processor = new BlockExampleProcessor;
 
     auto block_queue = new JMailbox<MyBlock*>;
-    auto event_queue = new JMailbox<std::shared_ptr<JEvent>>;
+    auto event_queue = new JMailbox<std::shared_ptr<JEvent>*>;
+    auto block_pool = new JPool<MyBlock>(0, 1, false);
+    block_pool->init();
 
-    // topology->queues.push_back(block_queue);
-    // FIXME: block_queue is a (very minor) memory leak
     topology->queues.push_back(event_queue);
+    topology->queues.push_back(block_queue);
+    topology->pools.push_back(block_pool);
 
-    auto block_source_arrow = new JBlockSourceArrow<MyBlock>("block_source", source, block_queue);
+    auto block_source_arrow = new JBlockSourceArrow<MyBlock>("block_source", source, block_pool, block_queue);
     auto block_disentangler_arrow = new JBlockDisentanglerArrow<MyBlock>("block_disentangler", source, block_queue, event_queue, topology->event_pool);
     auto processor_arrow = new JEventProcessorArrow("processors", event_queue, nullptr, topology->event_pool);
 
@@ -33,9 +35,6 @@ std::shared_ptr<JArrowTopology> configure_block_topology(std::shared_ptr<JArrowT
     topology->arrows.push_back(block_disentangler_arrow);
     topology->arrows.push_back(processor_arrow);
 
-    topology->sources.push_back(block_source_arrow);
-    topology->sinks.push_back(processor_arrow);
-
     block_source_arrow->attach(block_disentangler_arrow);
     block_disentangler_arrow->attach(processor_arrow);
 

diff --git a/src/examples/SubeventExample/SubeventExample.cc b/src/examples/SubeventExample/SubeventExample.cc
@@ -4,7 +4,6 @@
 
 #include <JANA/JApplication.h>
 #include <JANA/JObject.h>
-#include <JANA/Engine/JSubeventMailbox.h>
 #include <JANA/Engine/JSubeventArrow.h>
 #include <JANA/JEventSource.h>
 #include <JANA/JEventProcessor.h>
@@ -80,8 +79,8 @@ struct SimpleProcessor : public JEventProcessor {
 int main() {
 
     MyProcessor processor;
-    JMailbox<std::shared_ptr<JEvent>> events_in;
-    JMailbox<std::shared_ptr<JEvent>> events_out;
+    JMailbox<std::shared_ptr<JEvent>*> events_in;
+    JMailbox<std::shared_ptr<JEvent>*> events_out;
     JMailbox<SubeventWrapper<MyInput>> subevents_in;
     JMailbox<SubeventWrapper<MyOutput>> subevents_out;
 
@@ -109,12 +108,10 @@ int main() {
     proc_arrow->add_processor(new SimpleProcessor);
 
     topology->arrows.push_back(source_arrow);
-    topology->sources.push_back(source_arrow);
     topology->arrows.push_back(split_arrow);
     topology->arrows.push_back(subprocess_arrow);
     topology->arrows.push_back(merge_arrow);
     topology->arrows.push_back(proc_arrow);
-    topology->sinks.push_back(proc_arrow);
 
     source_arrow->attach(split_arrow);
     split_arrow->attach(subprocess_arrow);

diff --git a/src/libraries/JANA/CMakeLists.txt b/src/libraries/JANA/CMakeLists.txt
@@ -41,6 +41,7 @@ set(JANA2_SOURCES
     Engine/JEventSourceArrow.h
     Engine/JBlockSourceArrow.h
     Engine/JBlockDisentanglerArrow.h
+    Engine/JPool.h
 
     Engine/JMailbox.h
     Engine/JScheduler.cc