GCC Code Coverage Report

include/simdb/
Coverage:
low: ≥ 0%
medium: ≥ 75.0%
high: ≥ 90.0%
Lines:
148 / 150
98.7%
Branches:
73 / 508
14.4%
Functions:
23 / 24
95.8%
pipeline/Pipeline.hpp
Line Branch Exec Source
1 // <Pipeline.hpp> -*- C++ -*-
2
3 #pragma once
4
5 #include "simdb/Assert.hpp"
6 #include "simdb/pipeline/Flusher.hpp"
7 #include "simdb/pipeline/Stage.hpp"
8 #include <map>
9
10 namespace simdb {
11 class App;
12 class DatabaseManager;
13 } // namespace simdb
14
15 namespace simdb::pipeline {
16
17 /*!
18 * \class Pipeline
19 *
20 * \brief Multi-stage data processor with typed input/output ports between
21 * stages. Add stages, declare port bindings, then finalize; stages
22 * run on PollingThreads. Move-only semantics; no restriction on
23 * I/O type changes between stages.
24 *
25 * \note Create via PipelineManager::createPipeline().
26 */
27 class Pipeline
28 {
29 public:
30 /// \brief Construct a pipeline (typically via PipelineManager::createPipeline()).
31 /// \param db_mgr DatabaseManager for database stages.
32 /// \param name Pipeline name.
33 /// \param app The App that owns this pipeline.
34 13 Pipeline(DatabaseManager* db_mgr, const std::string& name, const App* app) :
35 13 db_mgr_(db_mgr),
36 13 pipeline_name_(name),
37 13 app_(app)
38 {
39 13 }
40
41 /// \brief Return the App that owns this pipeline.
42 ✗ const App* getOwningApp() const { return app_; }
43
44 /// \brief Return the DatabaseManager used by database stages.
45 DatabaseManager* getDatabaseManager() const { return db_mgr_; }
46
47 /// \brief Return the pipeline name.
48 13 std::string getName() const { return pipeline_name_; }
49
50 /// \brief Add a stage; only valid before noMoreStages(). Returns pointer to the new stage.
51 /// \tparam StageType Stage class (must derive from Stage).
52 /// \tparam StageCtorArgs Constructor argument types.
53 /// \param name Unique stage name within this pipeline.
54 /// \param args Arguments forwarded to the stage constructor.
55 /// \throws DBException if not accepting stages or stage name already exists.
56 template <typename StageType, typename... StageCtorArgs>
57 32 StageType* addStage(const std::string& name, StageCtorArgs&&... args)
58 {
59
10/160
SimplePipeline::DatabaseStage* simdb::pipeline::Pipeline::addStage<SimplePipeline::DatabaseStage, unsigned long>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unsigned long&&):
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 29 taken 6 times.
✗ Branch 5 → 6 not taken.
✗ Branch 6 → 7 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 22 → 23 not taken.
✗ Branch 23 → 24 not taken.
✗ Branch 24 → 25 not taken.
✗ Branch 25 → 26 not taken.
✗ Branch 27 → 28 not taken.
SimplePipeline::CompressionStage* simdb::pipeline::Pipeline::addStage<SimplePipeline::CompressionStage>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 29 taken 6 times.
✗ Branch 5 → 6 not taken.
✗ Branch 6 → 7 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 22 → 23 not taken.
✗ Branch 23 → 24 not taken.
✗ Branch 24 → 25 not taken.
✗ Branch 25 → 26 not taken.
✗ Branch 27 → 28 not taken.
PipelineSnooper::BoostStage* simdb::pipeline::Pipeline::addStage<PipelineSnooper::BoostStage, unsigned int&>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unsigned int&):
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 29 taken 1 time.
✗ Branch 5 → 6 not taken.
✗ Branch 6 → 7 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 22 → 23 not taken.
✗ Branch 23 → 24 not taken.
✗ Branch 24 → 25 not taken.
✗ Branch 25 → 26 not taken.
✗ Branch 27 → 28 not taken.
PipelineSnooper::BufferStage* simdb::pipeline::Pipeline::addStage<PipelineSnooper::BufferStage, unsigned int&, unsigned int&>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unsigned int&, unsigned int&):
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 29 taken 1 time.
✗ Branch 5 → 6 not taken.
✗ Branch 6 → 7 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 22 → 23 not taken.
✗ Branch 23 → 24 not taken.
✗ Branch 24 → 25 not taken.
✗ Branch 25 → 26 not taken.
✗ Branch 27 → 28 not taken.
PipelineSnooper::DatabaseStage* simdb::pipeline::Pipeline::addStage<PipelineSnooper::DatabaseStage, unsigned int&, unsigned int&>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unsigned int&, unsigned int&):
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 29 taken 1 time.
✗ Branch 5 → 6 not taken.
✗ Branch 6 → 7 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 22 → 23 not taken.
✗ Branch 23 → 24 not taken.
✗ Branch 24 → 25 not taken.
✗ Branch 25 → 26 not taken.
✗ Branch 27 → 28 not taken.
PipelineSnooper::ZlibStage* simdb::pipeline::Pipeline::addStage<PipelineSnooper::ZlibStage, unsigned int&>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unsigned int&):
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 29 taken 1 time.
✗ Branch 5 → 6 not taken.
✗ Branch 6 → 7 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 22 → 23 not taken.
✗ Branch 23 → 24 not taken.
✗ Branch 24 → 25 not taken.
✗ Branch 25 → 26 not taken.
✗ Branch 27 → 28 not taken.
DatabaseWatchdog::Watchdog* simdb::pipeline::Pipeline::addStage<DatabaseWatchdog::Watchdog, DatabaseWatchdog*>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, DatabaseWatchdog*&&):
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 29 taken 1 time.
✗ Branch 5 → 6 not taken.
✗ Branch 6 → 7 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 22 → 23 not taken.
✗ Branch 23 → 24 not taken.
✗ Branch 24 → 25 not taken.
✗ Branch 25 → 26 not taken.
✗ Branch 27 → 28 not taken.
simdb::argos::ArgosCollector::Compressor* simdb::pipeline::Pipeline::addStage<simdb::argos::ArgosCollector::Compressor>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 29 taken 5 times.
✗ Branch 5 → 6 not taken.
✗ Branch 6 → 7 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 22 → 23 not taken.
✗ Branch 23 → 24 not taken.
✗ Branch 24 → 25 not taken.
✗ Branch 25 → 26 not taken.
✗ Branch 27 → 28 not taken.
simdb::argos::ArgosCollector::PipelineStager* simdb::pipeline::Pipeline::addStage<simdb::argos::ArgosCollector::PipelineStager, unsigned long&>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unsigned long&):
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 29 taken 5 times.
✗ Branch 5 → 6 not taken.
✗ Branch 6 → 7 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 22 → 23 not taken.
✗ Branch 23 → 24 not taken.
✗ Branch 24 → 25 not taken.
✗ Branch 25 → 26 not taken.
✗ Branch 27 → 28 not taken.
simdb::argos::ArgosCollector::Writer* simdb::pipeline::Pipeline::addStage<simdb::argos::ArgosCollector::Writer>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 29 taken 5 times.
✗ Branch 5 → 6 not taken.
✗ Branch 6 → 7 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 22 → 23 not taken.
✗ Branch 23 → 24 not taken.
✗ Branch 24 → 25 not taken.
✗ Branch 25 → 26 not taken.
✗ Branch 27 → 28 not taken.
 32 simdb_assert(state_ == State::ACCEPTING_STAGES,
60 "Cannot add stage '" + name + "' to pipeline '" + pipeline_name_ + "'; not accepting stages.");
61
62 32 auto& stage = stages_[name];
63
10/160
SimplePipeline::DatabaseStage* simdb::pipeline::Pipeline::addStage<SimplePipeline::DatabaseStage, unsigned long>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unsigned long&&):
✗ Branch 31 → 32 not taken.
✓ Branch 31 → 58 taken 6 times.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 37 → 38 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 44 → 45 not taken.
✗ Branch 45 → 46 not taken.
✗ Branch 50 → 51 not taken.
✗ Branch 51 → 52 not taken.
✗ Branch 52 → 53 not taken.
✗ Branch 53 → 54 not taken.
✗ Branch 54 → 55 not taken.
✗ Branch 56 → 57 not taken.
SimplePipeline::CompressionStage* simdb::pipeline::Pipeline::addStage<SimplePipeline::CompressionStage>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 31 → 32 not taken.
✓ Branch 31 → 58 taken 6 times.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 37 → 38 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 44 → 45 not taken.
✗ Branch 45 → 46 not taken.
✗ Branch 50 → 51 not taken.
✗ Branch 51 → 52 not taken.
✗ Branch 52 → 53 not taken.
✗ Branch 53 → 54 not taken.
✗ Branch 54 → 55 not taken.
✗ Branch 56 → 57 not taken.
PipelineSnooper::BoostStage* simdb::pipeline::Pipeline::addStage<PipelineSnooper::BoostStage, unsigned int&>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unsigned int&):
✗ Branch 31 → 32 not taken.
✓ Branch 31 → 58 taken 1 time.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 37 → 38 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 44 → 45 not taken.
✗ Branch 45 → 46 not taken.
✗ Branch 50 → 51 not taken.
✗ Branch 51 → 52 not taken.
✗ Branch 52 → 53 not taken.
✗ Branch 53 → 54 not taken.
✗ Branch 54 → 55 not taken.
✗ Branch 56 → 57 not taken.
PipelineSnooper::BufferStage* simdb::pipeline::Pipeline::addStage<PipelineSnooper::BufferStage, unsigned int&, unsigned int&>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unsigned int&, unsigned int&):
✗ Branch 31 → 32 not taken.
✓ Branch 31 → 58 taken 1 time.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 37 → 38 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 44 → 45 not taken.
✗ Branch 45 → 46 not taken.
✗ Branch 50 → 51 not taken.
✗ Branch 51 → 52 not taken.
✗ Branch 52 → 53 not taken.
✗ Branch 53 → 54 not taken.
✗ Branch 54 → 55 not taken.
✗ Branch 56 → 57 not taken.
PipelineSnooper::DatabaseStage* simdb::pipeline::Pipeline::addStage<PipelineSnooper::DatabaseStage, unsigned int&, unsigned int&>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unsigned int&, unsigned int&):
✗ Branch 31 → 32 not taken.
✓ Branch 31 → 58 taken 1 time.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 37 → 38 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 44 → 45 not taken.
✗ Branch 45 → 46 not taken.
✗ Branch 50 → 51 not taken.
✗ Branch 51 → 52 not taken.
✗ Branch 52 → 53 not taken.
✗ Branch 53 → 54 not taken.
✗ Branch 54 → 55 not taken.
✗ Branch 56 → 57 not taken.
PipelineSnooper::ZlibStage* simdb::pipeline::Pipeline::addStage<PipelineSnooper::ZlibStage, unsigned int&>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unsigned int&):
✗ Branch 31 → 32 not taken.
✓ Branch 31 → 58 taken 1 time.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 37 → 38 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 44 → 45 not taken.
✗ Branch 45 → 46 not taken.
✗ Branch 50 → 51 not taken.
✗ Branch 51 → 52 not taken.
✗ Branch 52 → 53 not taken.
✗ Branch 53 → 54 not taken.
✗ Branch 54 → 55 not taken.
✗ Branch 56 → 57 not taken.
DatabaseWatchdog::Watchdog* simdb::pipeline::Pipeline::addStage<DatabaseWatchdog::Watchdog, DatabaseWatchdog*>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, DatabaseWatchdog*&&):
✗ Branch 31 → 32 not taken.
✓ Branch 31 → 58 taken 1 time.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 37 → 38 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 44 → 45 not taken.
✗ Branch 45 → 46 not taken.
✗ Branch 50 → 51 not taken.
✗ Branch 51 → 52 not taken.
✗ Branch 52 → 53 not taken.
✗ Branch 53 → 54 not taken.
✗ Branch 54 → 55 not taken.
✗ Branch 56 → 57 not taken.
simdb::argos::ArgosCollector::Compressor* simdb::pipeline::Pipeline::addStage<simdb::argos::ArgosCollector::Compressor>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 31 → 32 not taken.
✓ Branch 31 → 58 taken 5 times.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 37 → 38 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 44 → 45 not taken.
✗ Branch 45 → 46 not taken.
✗ Branch 50 → 51 not taken.
✗ Branch 51 → 52 not taken.
✗ Branch 52 → 53 not taken.
✗ Branch 53 → 54 not taken.
✗ Branch 54 → 55 not taken.
✗ Branch 56 → 57 not taken.
simdb::argos::ArgosCollector::PipelineStager* simdb::pipeline::Pipeline::addStage<simdb::argos::ArgosCollector::PipelineStager, unsigned long&>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unsigned long&):
✗ Branch 31 → 32 not taken.
✓ Branch 31 → 58 taken 5 times.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 37 → 38 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 44 → 45 not taken.
✗ Branch 45 → 46 not taken.
✗ Branch 50 → 51 not taken.
✗ Branch 51 → 52 not taken.
✗ Branch 52 → 53 not taken.
✗ Branch 53 → 54 not taken.
✗ Branch 54 → 55 not taken.
✗ Branch 56 → 57 not taken.
simdb::argos::ArgosCollector::Writer* simdb::pipeline::Pipeline::addStage<simdb::argos::ArgosCollector::Writer>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 31 → 32 not taken.
✓ Branch 31 → 58 taken 5 times.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 37 → 38 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 44 → 45 not taken.
✗ Branch 45 → 46 not taken.
✗ Branch 50 → 51 not taken.
✗ Branch 51 → 52 not taken.
✗ Branch 52 → 53 not taken.
✗ Branch 53 → 54 not taken.
✗ Branch 54 → 55 not taken.
✗ Branch 56 → 57 not taken.
 32 simdb_assert(!stage, "Stage '" + name + "' already exists in pipeline '" + pipeline_name_ + "'.");
64
10/10
SimplePipeline::DatabaseStage* simdb::pipeline::Pipeline::addStage<SimplePipeline::DatabaseStage, unsigned long>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unsigned long&&):
✓ Branch 59 → 60 taken 6 times.
SimplePipeline::CompressionStage* simdb::pipeline::Pipeline::addStage<SimplePipeline::CompressionStage>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✓ Branch 58 → 59 taken 6 times.
PipelineSnooper::BoostStage* simdb::pipeline::Pipeline::addStage<PipelineSnooper::BoostStage, unsigned int&>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unsigned int&):
✓ Branch 59 → 60 taken 1 time.
PipelineSnooper::BufferStage* simdb::pipeline::Pipeline::addStage<PipelineSnooper::BufferStage, unsigned int&, unsigned int&>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unsigned int&, unsigned int&):
✓ Branch 60 → 61 taken 1 time.
PipelineSnooper::DatabaseStage* simdb::pipeline::Pipeline::addStage<PipelineSnooper::DatabaseStage, unsigned int&, unsigned int&>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unsigned int&, unsigned int&):
✓ Branch 60 → 61 taken 1 time.
PipelineSnooper::ZlibStage* simdb::pipeline::Pipeline::addStage<PipelineSnooper::ZlibStage, unsigned int&>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unsigned int&):
✓ Branch 59 → 60 taken 1 time.
DatabaseWatchdog::Watchdog* simdb::pipeline::Pipeline::addStage<DatabaseWatchdog::Watchdog, DatabaseWatchdog*>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, DatabaseWatchdog*&&):
✓ Branch 59 → 60 taken 1 time.
simdb::argos::ArgosCollector::Compressor* simdb::pipeline::Pipeline::addStage<simdb::argos::ArgosCollector::Compressor>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✓ Branch 58 → 59 taken 5 times.
simdb::argos::ArgosCollector::PipelineStager* simdb::pipeline::Pipeline::addStage<simdb::argos::ArgosCollector::PipelineStager, unsigned long&>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unsigned long&):
✓ Branch 59 → 60 taken 5 times.
simdb::argos::ArgosCollector::Writer* simdb::pipeline::Pipeline::addStage<simdb::argos::ArgosCollector::Writer>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✓ Branch 58 → 59 taken 5 times.
 32 stage = std::make_unique<StageType>(std::forward<StageCtorArgs>(args)...);
65 32 stage->setName_(name);
66 32 stages_in_order_.push_back(name);
67 32 return static_cast<StageType*>(stage.get());
68 }
69
70 /// \brief Finalize stages and start accepting bindings.
71 /// \throws DBException if not accepting stages.
72 13 void noMoreStages()
73 {
74
1/14
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 25 taken 13 times.
✗ Branch 5 → 6 not taken.
✗ Branch 6 → 7 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 19 → 20 not taken.
✗ Branch 20 → 21 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 23 → 24 not taken.
 13 simdb_assert(state_ == State::ACCEPTING_STAGES,
75 "Cannot finalize stages for pipeline '" + pipeline_name_ + "'; stage changes already finalized.");
76
2/2
✓ Branch 34 → 27 taken 32 times.
✓ Branch 34 → 35 taken 13 times.
 45 for (auto& [stage_name, stage] : stages_)
77 {
78
1/1
✓ Branch 31 → 32 taken 32 times.
 32 stage->mergeQueueRepo_(queue_repo_);
79 }
80 13 queue_repo_.noMoreStages();
81 13 state_ = State::ACCEPTING_BINDINGS;
82 13 }
83
84 /// \brief Bind an output port to an input port (full names: "StageName.port").
85 /// \throws DBException if not accepting bindings.
86 19 void bind(const std::string& output_port_full_name, const std::string& input_port_full_name)
87 {
88
1/14
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 25 taken 19 times.
✗ Branch 5 → 6 not taken.
✗ Branch 6 → 7 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 19 → 20 not taken.
✗ Branch 20 → 21 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 23 → 24 not taken.
 19 simdb_assert(state_ == State::ACCEPTING_BINDINGS,
89 "Cannot bind ports for pipeline '" + pipeline_name_ + "'; not accepting bindings.");
90 19 queue_repo_.bind(output_port_full_name, input_port_full_name);
91 19 }
92
93 /// \brief Finalize bindings and create queues; required before getInPortQueue/getOutPortQueue.
94 /// \throws DBException if not accepting bindings.
95 13 void noMoreBindings()
96 {
97
1/14
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 25 taken 13 times.
✗ Branch 5 → 6 not taken.
✗ Branch 6 → 7 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 19 → 20 not taken.
✗ Branch 20 → 21 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 23 → 24 not taken.
 13 simdb_assert(state_ == State::ACCEPTING_BINDINGS, "Cannot finalize bindings for pipeline '" + pipeline_name_ +
98 "'; binding changes already finalized.");
99 13 queue_repo_.finalizeBindings();
100 13 state_ = State::BINDINGS_COMPLETE;
101 13 }
102
103 /// \brief Return the input port queue; only valid after noMoreBindings().
104 /// \tparam T Element type of the queue.
105 17 template <typename T> simdb::ConcurrentQueue<T>* getInPortQueue(const std::string& port_full_name)
106 {
107
8/64
simdb::ConcurrentQueue<DummyData>* simdb::pipeline::Pipeline::getInPortQueue<DummyData>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 5 taken 1 time.
✗ Branch 3 → 4 not taken.
✗ Branch 3 → 5 not taken.
✗ Branch 6 → 7 not taken.
✓ Branch 6 → 25 taken 1 time.
✗ Branch 9 → 10 not taken.
✗ Branch 10 → 11 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 19 → 20 not taken.
✗ Branch 20 → 21 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 23 → 24 not taken.
simdb::ConcurrentQueue<simdb::argos::NotifEntry>* simdb::pipeline::Pipeline::getInPortQueue<simdb::argos::NotifEntry>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 5 taken 5 times.
✗ Branch 3 → 4 not taken.
✗ Branch 3 → 5 not taken.
✗ Branch 6 → 7 not taken.
✓ Branch 6 → 25 taken 5 times.
✗ Branch 9 → 10 not taken.
✗ Branch 10 → 11 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 19 → 20 not taken.
✗ Branch 20 → 21 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 23 → 24 not taken.
simdb::ConcurrentQueue<std::unique_ptr<simdb::argos::Ledger, std::default_delete<simdb::argos::Ledger> > >* simdb::pipeline::Pipeline::getInPortQueue<std::unique_ptr<simdb::argos::Ledger, std::default_delete<simdb::argos::Ledger> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 5 taken 5 times.
✗ Branch 3 → 4 not taken.
✗ Branch 3 → 5 not taken.
✗ Branch 6 → 7 not taken.
✓ Branch 6 → 25 taken 5 times.
✗ Branch 9 → 10 not taken.
✗ Branch 10 → 11 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 19 → 20 not taken.
✗ Branch 20 → 21 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 23 → 24 not taken.
simdb::ConcurrentQueue<std::vector<double, std::allocator<double> > >* simdb::pipeline::Pipeline::getInPortQueue<std::vector<double, std::allocator<double> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 5 taken 6 times.
✗ Branch 3 → 4 not taken.
✗ Branch 3 → 5 not taken.
✗ Branch 6 → 7 not taken.
✓ Branch 6 → 25 taken 6 times.
✗ Branch 9 → 10 not taken.
✗ Branch 10 → 11 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 19 → 20 not taken.
✗ Branch 20 → 21 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 23 → 24 not taken.
 17 simdb_assert(state_ == State::BINDINGS_COMPLETE || state_ == State::FINALIZED,
108 "Cannot access port queues until noMoreBindings() is called.");
109 17 return queue_repo_.getInPortQueue<T>(port_full_name);
110 }
111
112 /// \brief Return the output port queue; only valid after noMoreBindings().
113 /// \tparam T Element type of the queue.
114 template <typename T> simdb::ConcurrentQueue<T>* getOutPortQueue(const std::string& port_full_name)
115 {
116 simdb_assert(state_ == State::BINDINGS_COMPLETE || state_ == State::FINALIZED,
117 "Cannot access port queues until noMoreBindings() is called.");
118 return queue_repo_.getOutPortQueue<T>(port_full_name);
119 }
120
121 /// \brief Assign each stage to a PollingThread (or the shared DatabaseThread); call after noMoreBindings().
122 /// \param threads Vector to which new PollingThreads may be appended.
123 /// \param database_thread Single shared DatabaseThread for all DatabaseStages (created if null).
124 13 void assignStageThreads(std::vector<std::unique_ptr<PollingThread>>& threads,
125 std::unique_ptr<DatabaseThread>& database_thread)
126 {
127
1/14
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 25 taken 13 times.
✗ Branch 5 → 6 not taken.
✗ Branch 6 → 7 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 19 → 20 not taken.
✗ Branch 20 → 21 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 23 → 24 not taken.
 13 simdb_assert(state_ == State::BINDINGS_COMPLETE,
128 "Cannot assign stage threads for pipeline '" + pipeline_name_ + "; noMoreBindings() never called");
129
130 13 queue_repo_.validateQueues();
131
132
2/2
✓ Branch 35 → 28 taken 32 times.
✓ Branch 35 → 36 taken 13 times.
 45 for (auto& [stage_name, stage] : stages_)
133 {
134
1/1
✓ Branch 32 → 33 taken 32 times.
 32 stage->assignThread_(db_mgr_, threads, database_thread);
135 }
136
137 13 state_ = State::FINALIZED;
138 13 }
139
140 /// \brief Create a Flusher that runs the given stages (or all in order); wraps in transaction
141 /// if any stage is a DatabaseStage.
142 /// \param stage_names Stage names in flush order; if empty, use add order.
143 /// \return Flusher or FlusherWithTransaction; caller owns.
144 /// \throws DBException if a stage name does not exist.
145 7 std::unique_ptr<Flusher> createFlusher(const std::vector<std::string>& stage_names = {})
146 {
147 7 std::vector<Stage*> stages;
148
2/2
✓ Branch 4 → 5 taken 1 time.
✓ Branch 4 → 6 taken 6 times.
 7 const auto& ordered_stage_names = !stage_names.empty() ? stage_names : stages_in_order_;
149
2/2
✓ Branch 38 → 9 taken 16 times.
✓ Branch 38 → 39 taken 7 times.
 23 for (const auto& name : ordered_stage_names)
150 {
151
1/1
✓ Branch 10 → 11 taken 16 times.
 16 auto it = stages_.find(name);
152
1/13
✗ Branch 13 → 14 not taken.
✓ Branch 13 → 33 taken 16 times.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 19 → 20 not taken.
✗ Branch 23 → 24 not taken.
✗ Branch 24 → 25 not taken.
✗ Branch 25 → 26 not taken.
✗ Branch 26 → 27 not taken.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 29 → 30 not taken.
✗ Branch 31 → 32 not taken.
 16 simdb_assert(it != stages_.end(), "Stage does not exist: " << name);
153
1/1
✓ Branch 35 → 36 taken 16 times.
 16 stages.emplace_back(it->second.get());
154 }
155
156 7 bool has_db_stage = false;
157
2/2
✓ Branch 49 → 41 taken 16 times.
✓ Branch 49 → 50 taken 7 times.
 23 for (auto stage : stages)
158 {
159
3/4
✓ Branch 42 → 43 taken 16 times.
✗ Branch 42 → 44 not taken.
✓ Branch 45 → 46 taken 7 times.
✓ Branch 45 → 47 taken 9 times.
 16 if (dynamic_cast<DatabaseStageBase*>(stage))
160 {
161 7 has_db_stage = true;
162 }
163 }
164
165
3/10
✓ Branch 51 → 52 taken 7 times.
✓ Branch 52 → 53 taken 7 times.
✓ Branch 59 → 60 taken 7 times.
✗ Branch 59 → 61 not taken.
✗ Branch 86 → 87 not taken.
✗ Branch 86 → 88 not taken.
✗ Branch 90 → 91 not taken.
✗ Branch 90 → 93 not taken.
✗ Branch 91 → 92 not taken.
✗ Branch 91 → 93 not taken.
 14 return has_db_stage ? std::unique_ptr<FlusherWithTransaction>(new FlusherWithTransaction(stages, db_mgr_))
166
1/8
✓ Branch 50 → 51 taken 7 times.
✗ Branch 50 → 55 not taken.
✗ Branch 55 → 56 not taken.
✗ Branch 56 → 57 not taken.
✗ Branch 82 → 83 not taken.
✗ Branch 82 → 85 not taken.
✗ Branch 83 → 84 not taken.
✗ Branch 83 → 85 not taken.
 21 : std::unique_ptr<Flusher>(new Flusher(stages));
167 7 }
168
169 /// \brief Return the AsyncDatabaseAccessor for this pipeline (set by PipelineManager after openPipelines()).
170 AsyncDatabaseAccessor* getAsyncDatabaseAccessor() const { return async_db_accessor_; }
171
172 /// \brief Return a map of stage name to Stage* (all stages in this pipeline).
173 26 std::map<std::string, Stage*> getStages()
174 {
175 26 std::map<std::string, Stage*> stages;
176
2/2
✓ Branch 12 → 5 taken 64 times.
✓ Branch 12 → 13 taken 26 times.
 90 for (auto& [name, stage] : stages_)
177 {
178
1/1
✓ Branch 9 → 10 taken 64 times.
 64 stages[name] = stage.get();
179 }
180 26 return stages;
181 ✗ }
182
183 /// \brief Return stages in the order they were added (name, Stage* pairs).
184 26 std::vector<std::pair<std::string, Stage*>> getOrderedStages()
185 {
186 26 std::vector<std::pair<std::string, Stage*>> ordered_stages;
187
1/1
✓ Branch 3 → 4 taken 26 times.
 26 auto unordered_stages = getStages();
188
2/2
✓ Branch 13 → 6 taken 64 times.
✓ Branch 13 → 14 taken 26 times.
 90 for (auto& name : stages_in_order_)
189 {
190
1/1
✓ Branch 7 → 8 taken 64 times.
 64 auto stage = unordered_stages.at(name);
191
2/2
✓ Branch 8 → 9 taken 64 times.
✓ Branch 9 → 10 taken 64 times.
 64 ordered_stages.emplace_back(std::make_pair(name, stage));
192 }
193 26 return ordered_stages;
194 26 }
195
196 private:
197 DatabaseManager* db_mgr_ = nullptr;
198 std::string pipeline_name_;
199 const App* app_ = nullptr;
200 std::unordered_map<std::string, std::unique_ptr<Stage>> stages_;
201 std::vector<std::string> stages_in_order_;
202 PipelineQueueRepo queue_repo_;
203 AsyncDatabaseAccessor* async_db_accessor_ = nullptr;
204
205 enum class State { ACCEPTING_STAGES, ACCEPTING_BINDINGS, BINDINGS_COMPLETE, FINALIZED };
206
207 State state_ = State::ACCEPTING_STAGES;
208 };
209
210 } // namespace simdb::pipeline
211