GCC Code Coverage Report

include/simdb/
Coverage:
low: ≥ 0%
medium: ≥ 75.0%
high: ≥ 90.0%
Lines:
325 / 406
80.0%
Branches:
152 / 1127
13.5%
Functions:
72 / 86
83.7%
pipeline/QueueRepo.hpp
Line Branch Exec Source
1 // <QueueRepo.hpp> -*- C++ -*-
2
3 #pragma once
4
5 #include "simdb/Assert.hpp"
6 #include "simdb/pipeline/Queue.hpp"
7 #include <set>
8
9 namespace simdb::pipeline {
10
11 /*!
12 * \class QueuePlaceholder
13 *
14 * \brief Abstract placeholder for a pipeline port queue. Used by StageQueueRepo
15 * and PipelineQueueRepo to create and bind ConcurrentQueues when stages
16 * are finalized and bindings are applied.
17 */
18 class QueuePlaceholder
19 {
20 public:
21 55 virtual ~QueuePlaceholder() = default;
22 /// \brief Create a new Queue and return ownership.
23 virtual std::unique_ptr<QueueBase> createQueue() = 0;
24 /// \brief Assign an existing queue to this placeholder (e.g. after binding).
25 virtual void assignQueue(QueueBase* queue) = 0;
26 /// \brief Return true if a queue has been created or assigned.
27 virtual bool hasQueue() const = 0;
28 };
29
30 /*!
31 * \class InputQueuePlaceholder
32 *
33 * \brief Placeholder for an input port queue; holds a reference to the stage's
34 * ConcurrentQueue<T>* so the queue can be assigned when bound.
35 * \tparam T Element type of the queue.
36 */
37 template <typename T> class InputQueuePlaceholder : public QueuePlaceholder
38 {
39 public:
40 /// \brief Construct with a reference to the stage's queue pointer (must be null initially).
41 /// \param queue Reference to the stage's ConcurrentQueue<T>*; set when
42 /// \throws DBException if \p queue is non-null.
43 36 InputQueuePlaceholder(ConcurrentQueue<T>*& queue) :
44 36 queue_(queue)
45 {
46
9/108
simdb::pipeline::InputQueuePlaceholder<DummyDataWindowBytes>::InputQueuePlaceholder(simdb::ConcurrentQueue<DummyDataWindowBytes>*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 2 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
simdb::pipeline::InputQueuePlaceholder<DummyData>::InputQueuePlaceholder(simdb::ConcurrentQueue<DummyData>*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 1 time.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
simdb::pipeline::InputQueuePlaceholder<simdb::argos::NotifEntry>::InputQueuePlaceholder(simdb::ConcurrentQueue<simdb::argos::NotifEntry>*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 5 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
simdb::pipeline::InputQueuePlaceholder<simdb::argos::CollectionEntries>::InputQueuePlaceholder(simdb::ConcurrentQueue<simdb::argos::CollectionEntries>*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 5 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
simdb::pipeline::InputQueuePlaceholder<simdb::argos::CompressedCollectionEntries>::InputQueuePlaceholder(simdb::ConcurrentQueue<simdb::argos::CompressedCollectionEntries>*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 5 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
simdb::pipeline::InputQueuePlaceholder<std::unique_ptr<simdb::argos::Ledger, std::default_delete<simdb::argos::Ledger> > >::InputQueuePlaceholder(simdb::ConcurrentQueue<std::unique_ptr<simdb::argos::Ledger, std::default_delete<simdb::argos::Ledger> > >*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 5 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
simdb::pipeline::InputQueuePlaceholder<std::vector<DummyData, std::allocator<DummyData> > >::InputQueuePlaceholder(simdb::ConcurrentQueue<std::vector<DummyData, std::allocator<DummyData> > >*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 1 time.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
simdb::pipeline::InputQueuePlaceholder<std::vector<char, std::allocator<char> > >::InputQueuePlaceholder(simdb::ConcurrentQueue<std::vector<char, std::allocator<char> > >*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 6 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
simdb::pipeline::InputQueuePlaceholder<std::vector<double, std::allocator<double> > >::InputQueuePlaceholder(simdb::ConcurrentQueue<std::vector<double, std::allocator<double> > >*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 6 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
 36 simdb_assert(!queue_, "Input queue placeholder cannot be initialized "
47 "with non-null queue pointer.");
48 36 }
49
50 17 std::unique_ptr<QueueBase> createQueue() override
51 {
52
4/9
simdb::pipeline::InputQueuePlaceholder<DummyDataWindowBytes>::createQueue():
✗ Branch 2 → 3 not taken.
simdb::pipeline::InputQueuePlaceholder<DummyData>::createQueue():
✓ Branch 2 → 3 taken 1 time.
simdb::pipeline::InputQueuePlaceholder<simdb::argos::NotifEntry>::createQueue():
✓ Branch 2 → 3 taken 5 times.
simdb::pipeline::InputQueuePlaceholder<simdb::argos::CollectionEntries>::createQueue():
✗ Branch 2 → 3 not taken.
simdb::pipeline::InputQueuePlaceholder<simdb::argos::CompressedCollectionEntries>::createQueue():
✗ Branch 2 → 3 not taken.
simdb::pipeline::InputQueuePlaceholder<std::unique_ptr<simdb::argos::Ledger, std::default_delete<simdb::argos::Ledger> > >::createQueue():
✓ Branch 2 → 3 taken 5 times.
simdb::pipeline::InputQueuePlaceholder<std::vector<DummyData, std::allocator<DummyData> > >::createQueue():
✗ Branch 2 → 3 not taken.
simdb::pipeline::InputQueuePlaceholder<std::vector<char, std::allocator<char> > >::createQueue():
✗ Branch 2 → 3 not taken.
simdb::pipeline::InputQueuePlaceholder<std::vector<double, std::allocator<double> > >::createQueue():
✓ Branch 2 → 3 taken 6 times.
 17 auto queue = std::make_unique<Queue<T>>();
53 17 queue_ = &queue->get();
54 34 return queue;
55 17 }
56
57 19 void assignQueue(QueueBase* queue) override
58 {
59
5/18
simdb::pipeline::InputQueuePlaceholder<DummyDataWindowBytes>::assignQueue(simdb::pipeline::QueueBase*):
✓ Branch 2 → 3 taken 2 times.
✗ Branch 2 → 4 not taken.
simdb::pipeline::InputQueuePlaceholder<DummyData>::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 2 → 3 not taken.
✗ Branch 2 → 4 not taken.
simdb::pipeline::InputQueuePlaceholder<simdb::argos::NotifEntry>::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 2 → 3 not taken.
✗ Branch 2 → 4 not taken.
simdb::pipeline::InputQueuePlaceholder<simdb::argos::CollectionEntries>::assignQueue(simdb::pipeline::QueueBase*):
✓ Branch 2 → 3 taken 5 times.
✗ Branch 2 → 4 not taken.
simdb::pipeline::InputQueuePlaceholder<simdb::argos::CompressedCollectionEntries>::assignQueue(simdb::pipeline::QueueBase*):
✓ Branch 2 → 3 taken 5 times.
✗ Branch 2 → 4 not taken.
simdb::pipeline::InputQueuePlaceholder<std::unique_ptr<simdb::argos::Ledger, std::default_delete<simdb::argos::Ledger> > >::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 2 → 3 not taken.
✗ Branch 2 → 4 not taken.
simdb::pipeline::InputQueuePlaceholder<std::vector<DummyData, std::allocator<DummyData> > >::assignQueue(simdb::pipeline::QueueBase*):
✓ Branch 2 → 3 taken 1 time.
✗ Branch 2 → 4 not taken.
simdb::pipeline::InputQueuePlaceholder<std::vector<char, std::allocator<char> > >::assignQueue(simdb::pipeline::QueueBase*):
✓ Branch 2 → 3 taken 6 times.
✗ Branch 2 → 4 not taken.
simdb::pipeline::InputQueuePlaceholder<std::vector<double, std::allocator<double> > >::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 2 → 3 not taken.
✗ Branch 2 → 4 not taken.
 19 auto typed_queue = dynamic_cast<Queue<T>*>(queue);
60
5/108
simdb::pipeline::InputQueuePlaceholder<DummyDataWindowBytes>::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 5 → 6 not taken.
✓ Branch 5 → 24 taken 2 times.
✗ Branch 8 → 9 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 11 → 12 not taken.
✗ Branch 15 → 16 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 22 → 23 not taken.
simdb::pipeline::InputQueuePlaceholder<DummyData>::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 5 → 6 not taken.
✗ Branch 5 → 24 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 11 → 12 not taken.
✗ Branch 15 → 16 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 22 → 23 not taken.
simdb::pipeline::InputQueuePlaceholder<simdb::argos::NotifEntry>::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 5 → 6 not taken.
✗ Branch 5 → 24 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 11 → 12 not taken.
✗ Branch 15 → 16 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 22 → 23 not taken.
simdb::pipeline::InputQueuePlaceholder<simdb::argos::CollectionEntries>::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 5 → 6 not taken.
✓ Branch 5 → 24 taken 5 times.
✗ Branch 8 → 9 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 11 → 12 not taken.
✗ Branch 15 → 16 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 22 → 23 not taken.
simdb::pipeline::InputQueuePlaceholder<simdb::argos::CompressedCollectionEntries>::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 5 → 6 not taken.
✓ Branch 5 → 24 taken 5 times.
✗ Branch 8 → 9 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 11 → 12 not taken.
✗ Branch 15 → 16 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 22 → 23 not taken.
simdb::pipeline::InputQueuePlaceholder<std::unique_ptr<simdb::argos::Ledger, std::default_delete<simdb::argos::Ledger> > >::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 5 → 6 not taken.
✗ Branch 5 → 24 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 11 → 12 not taken.
✗ Branch 15 → 16 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 22 → 23 not taken.
simdb::pipeline::InputQueuePlaceholder<std::vector<DummyData, std::allocator<DummyData> > >::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 5 → 6 not taken.
✓ Branch 5 → 24 taken 1 time.
✗ Branch 8 → 9 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 11 → 12 not taken.
✗ Branch 15 → 16 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 22 → 23 not taken.
simdb::pipeline::InputQueuePlaceholder<std::vector<char, std::allocator<char> > >::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 5 → 6 not taken.
✓ Branch 5 → 24 taken 6 times.
✗ Branch 8 → 9 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 11 → 12 not taken.
✗ Branch 15 → 16 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 22 → 23 not taken.
simdb::pipeline::InputQueuePlaceholder<std::vector<double, std::allocator<double> > >::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 5 → 6 not taken.
✗ Branch 5 → 24 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 11 → 12 not taken.
✗ Branch 15 → 16 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 22 → 23 not taken.
 19 simdb_assert(typed_queue, "Incompatible queue types in binding");
61 19 queue_ = &typed_queue->get();
62 19 }
63
64 /// \brief Return true if a queue has been created or assigned.
65 36 bool hasQueue() const override { return queue_ != nullptr; }
66
67 /// \brief Return the assigned queue pointer (null until createQueue or assignQueue).
68 17 ConcurrentQueue<T>* getQueue() { return queue_; }
69
70 private:
71 ConcurrentQueue<T>*& queue_;
72 };
73
74 /*!
75 * \class OutputQueuePlaceholder
76 *
77 * \brief Placeholder for an output port queue; holds a reference to the stage's
78 * ConcurrentQueue<T>* so the queue can be assigned when bound.
79 * \tparam T Element type of the queue.
80 */
81 template <typename T> class OutputQueuePlaceholder : public QueuePlaceholder
82 {
83 public:
84 /// \brief Construct with a reference to the stage's queue pointer (must be null initially).
85 /// \param queue Reference to the stage's ConcurrentQueue<T>*; set when noMoreStages() is called.
86 /// \throws DBException if \p queue is non-null.
87 19 OutputQueuePlaceholder(ConcurrentQueue<T>*& queue) :
88 19 queue_(queue)
89 {
90
5/60
simdb::pipeline::OutputQueuePlaceholder<DummyDataWindowBytes>::OutputQueuePlaceholder(simdb::ConcurrentQueue<DummyDataWindowBytes>*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 2 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
simdb::pipeline::OutputQueuePlaceholder<simdb::argos::CollectionEntries>::OutputQueuePlaceholder(simdb::ConcurrentQueue<simdb::argos::CollectionEntries>*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 5 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
simdb::pipeline::OutputQueuePlaceholder<simdb::argos::CompressedCollectionEntries>::OutputQueuePlaceholder(simdb::ConcurrentQueue<simdb::argos::CompressedCollectionEntries>*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 5 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
simdb::pipeline::OutputQueuePlaceholder<std::vector<DummyData, std::allocator<DummyData> > >::OutputQueuePlaceholder(simdb::ConcurrentQueue<std::vector<DummyData, std::allocator<DummyData> > >*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 1 time.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
simdb::pipeline::OutputQueuePlaceholder<std::vector<char, std::allocator<char> > >::OutputQueuePlaceholder(simdb::ConcurrentQueue<std::vector<char, std::allocator<char> > >*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 6 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
 19 simdb_assert(!queue_, "Output queue placeholder cannot be initialized "
91 "with non-null queue pointer.");
92 19 }
93
94 /// \brief Create a new Queue and return ownership.
95 19 std::unique_ptr<QueueBase> createQueue() override
96 {
97
5/5
simdb::pipeline::OutputQueuePlaceholder<DummyDataWindowBytes>::createQueue():
✓ Branch 2 → 3 taken 2 times.
simdb::pipeline::OutputQueuePlaceholder<simdb::argos::CollectionEntries>::createQueue():
✓ Branch 2 → 3 taken 5 times.
simdb::pipeline::OutputQueuePlaceholder<simdb::argos::CompressedCollectionEntries>::createQueue():
✓ Branch 2 → 3 taken 5 times.
simdb::pipeline::OutputQueuePlaceholder<std::vector<DummyData, std::allocator<DummyData> > >::createQueue():
✓ Branch 2 → 3 taken 1 time.
simdb::pipeline::OutputQueuePlaceholder<std::vector<char, std::allocator<char> > >::createQueue():
✓ Branch 2 → 3 taken 6 times.
 19 auto queue = std::make_unique<Queue<T>>();
98 19 queue_ = &queue->get();
99 38 return queue;
100 19 }
101
102 /// \brief Assign an existing queue to this placeholder (e.g. after binding).
103 ✗ void assignQueue(QueueBase* queue) override
104 {
105
0/10
simdb::pipeline::OutputQueuePlaceholder<DummyDataWindowBytes>::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 2 → 3 not taken.
✗ Branch 2 → 4 not taken.
simdb::pipeline::OutputQueuePlaceholder<simdb::argos::CollectionEntries>::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 2 → 3 not taken.
✗ Branch 2 → 4 not taken.
simdb::pipeline::OutputQueuePlaceholder<simdb::argos::CompressedCollectionEntries>::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 2 → 3 not taken.
✗ Branch 2 → 4 not taken.
simdb::pipeline::OutputQueuePlaceholder<std::vector<DummyData, std::allocator<DummyData> > >::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 2 → 3 not taken.
✗ Branch 2 → 4 not taken.
simdb::pipeline::OutputQueuePlaceholder<std::vector<char, std::allocator<char> > >::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 2 → 3 not taken.
✗ Branch 2 → 4 not taken.
 ✗ auto typed_queue = dynamic_cast<Queue<T>*>(queue);
106
0/60
simdb::pipeline::OutputQueuePlaceholder<DummyDataWindowBytes>::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 5 → 6 not taken.
✗ Branch 5 → 24 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 11 → 12 not taken.
✗ Branch 15 → 16 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 22 → 23 not taken.
simdb::pipeline::OutputQueuePlaceholder<simdb::argos::CollectionEntries>::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 5 → 6 not taken.
✗ Branch 5 → 24 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 11 → 12 not taken.
✗ Branch 15 → 16 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 22 → 23 not taken.
simdb::pipeline::OutputQueuePlaceholder<simdb::argos::CompressedCollectionEntries>::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 5 → 6 not taken.
✗ Branch 5 → 24 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 11 → 12 not taken.
✗ Branch 15 → 16 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 22 → 23 not taken.
simdb::pipeline::OutputQueuePlaceholder<std::vector<DummyData, std::allocator<DummyData> > >::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 5 → 6 not taken.
✗ Branch 5 → 24 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 11 → 12 not taken.
✗ Branch 15 → 16 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 22 → 23 not taken.
simdb::pipeline::OutputQueuePlaceholder<std::vector<char, std::allocator<char> > >::assignQueue(simdb::pipeline::QueueBase*):
✗ Branch 5 → 6 not taken.
✗ Branch 5 → 24 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 11 → 12 not taken.
✗ Branch 15 → 16 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 22 → 23 not taken.
 ✗ simdb_assert(typed_queue, "Incompatible queue types in binding");
107 ✗ queue_ = &typed_queue->get();
108 ✗ }
109
110 19 bool hasQueue() const override { return queue_ != nullptr; }
111
112 /// \brief Return the assigned queue pointer (null until createQueue or assignQueue).
113 ConcurrentQueue<T>* getQueue() { return queue_; }
114
115 private:
116 ConcurrentQueue<T>*& queue_;
117 };
118
119 /*!
120 * \class StageQueueRepo
121 *
122 * \brief Per-stage registry of input and output port placeholders. Used in
123 * Stage subclasses (addInPort_/addOutPort_) before the stage name is set.
124 * Keys become "stage_name.port_name" after setStageName().
125 */
126 class StageQueueRepo
127 {
128 public:
129 /// \brief Register an input port placeholder; only in Stage subclass constructors (before name is set).
130 /// \throws DBException if stage name already set, or port name already exists.
131 36 template <typename T> void addInPortPlaceholder(const std::string& port_name, ConcurrentQueue<T>*& queue)
132 {
133
9/108
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<DummyDataWindowBytes>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<DummyDataWindowBytes>*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 2 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<DummyData>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<DummyData>*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 1 time.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<simdb::argos::NotifEntry>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<simdb::argos::NotifEntry>*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 5 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<simdb::argos::CollectionEntries>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<simdb::argos::CollectionEntries>*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 5 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<simdb::argos::CompressedCollectionEntries>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<simdb::argos::CompressedCollectionEntries>*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 5 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<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&, simdb::ConcurrentQueue<std::unique_ptr<simdb::argos::Ledger, std::default_delete<simdb::argos::Ledger> > >*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 5 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<std::vector<DummyData, std::allocator<DummyData> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<std::vector<DummyData, std::allocator<DummyData> > >*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 1 time.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<std::vector<char, std::allocator<char> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<std::vector<char, std::allocator<char> > >*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 6 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<std::vector<double, std::allocator<double> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<std::vector<double, std::allocator<double> > >*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 6 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
 36 simdb_assert(stage_name_.empty(), "You may only add in/out ports in Stage subclass constructors");
134
135 36 auto& placeholder = input_placeholders_[port_name];
136
9/126
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<DummyDataWindowBytes>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<DummyDataWindowBytes>*&):
✗ Branch 24 → 25 not taken.
✓ Branch 24 → 47 taken 2 times.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 36 → 37 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 45 → 46 not taken.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<DummyData>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<DummyData>*&):
✗ Branch 24 → 25 not taken.
✓ Branch 24 → 47 taken 1 time.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 36 → 37 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 45 → 46 not taken.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<simdb::argos::NotifEntry>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<simdb::argos::NotifEntry>*&):
✗ Branch 24 → 25 not taken.
✓ Branch 24 → 47 taken 5 times.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 36 → 37 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 45 → 46 not taken.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<simdb::argos::CollectionEntries>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<simdb::argos::CollectionEntries>*&):
✗ Branch 24 → 25 not taken.
✓ Branch 24 → 47 taken 5 times.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 36 → 37 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 45 → 46 not taken.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<simdb::argos::CompressedCollectionEntries>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<simdb::argos::CompressedCollectionEntries>*&):
✗ Branch 24 → 25 not taken.
✓ Branch 24 → 47 taken 5 times.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 36 → 37 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 45 → 46 not taken.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<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&, simdb::ConcurrentQueue<std::unique_ptr<simdb::argos::Ledger, std::default_delete<simdb::argos::Ledger> > >*&):
✗ Branch 24 → 25 not taken.
✓ Branch 24 → 47 taken 5 times.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 36 → 37 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 45 → 46 not taken.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<std::vector<DummyData, std::allocator<DummyData> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<std::vector<DummyData, std::allocator<DummyData> > >*&):
✗ Branch 24 → 25 not taken.
✓ Branch 24 → 47 taken 1 time.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 36 → 37 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 45 → 46 not taken.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<std::vector<char, std::allocator<char> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<std::vector<char, std::allocator<char> > >*&):
✗ Branch 24 → 25 not taken.
✓ Branch 24 → 47 taken 6 times.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 36 → 37 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 45 → 46 not taken.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<std::vector<double, std::allocator<double> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<std::vector<double, std::allocator<double> > >*&):
✗ Branch 24 → 25 not taken.
✓ Branch 24 → 47 taken 6 times.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 36 → 37 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 45 → 46 not taken.
 36 simdb_assert(!placeholder, "Input port placeholder '" + port_name + "' already exists in QueueRepo");
137
9/9
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<DummyDataWindowBytes>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<DummyDataWindowBytes>*&):
✓ Branch 47 → 48 taken 2 times.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<DummyData>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<DummyData>*&):
✓ Branch 47 → 48 taken 1 time.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<simdb::argos::NotifEntry>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<simdb::argos::NotifEntry>*&):
✓ Branch 47 → 48 taken 5 times.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<simdb::argos::CollectionEntries>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<simdb::argos::CollectionEntries>*&):
✓ Branch 47 → 48 taken 5 times.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<simdb::argos::CompressedCollectionEntries>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<simdb::argos::CompressedCollectionEntries>*&):
✓ Branch 47 → 48 taken 5 times.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<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&, simdb::ConcurrentQueue<std::unique_ptr<simdb::argos::Ledger, std::default_delete<simdb::argos::Ledger> > >*&):
✓ Branch 47 → 48 taken 5 times.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<std::vector<DummyData, std::allocator<DummyData> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<std::vector<DummyData, std::allocator<DummyData> > >*&):
✓ Branch 47 → 48 taken 1 time.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<std::vector<char, std::allocator<char> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<std::vector<char, std::allocator<char> > >*&):
✓ Branch 47 → 48 taken 6 times.
void simdb::pipeline::StageQueueRepo::addInPortPlaceholder<std::vector<double, std::allocator<double> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<std::vector<double, std::allocator<double> > >*&):
✓ Branch 47 → 48 taken 6 times.
 36 placeholder = std::make_unique<InputQueuePlaceholder<T>>(queue);
138 36 }
139
140 /// \brief Register an output port placeholder; only in Stage subclass constructors (before name is set).
141 /// \throws DBException if stage name already set, or port name already exists.
142 19 template <typename T> void addOutPortPlaceholder(const std::string& port_name, ConcurrentQueue<T>*& queue)
143 {
144
5/60
void simdb::pipeline::StageQueueRepo::addOutPortPlaceholder<DummyDataWindowBytes>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<DummyDataWindowBytes>*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 2 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
void simdb::pipeline::StageQueueRepo::addOutPortPlaceholder<simdb::argos::CollectionEntries>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<simdb::argos::CollectionEntries>*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 5 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
void simdb::pipeline::StageQueueRepo::addOutPortPlaceholder<simdb::argos::CompressedCollectionEntries>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<simdb::argos::CompressedCollectionEntries>*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 5 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
void simdb::pipeline::StageQueueRepo::addOutPortPlaceholder<std::vector<DummyData, std::allocator<DummyData> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<std::vector<DummyData, std::allocator<DummyData> > >*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 1 time.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
void simdb::pipeline::StageQueueRepo::addOutPortPlaceholder<std::vector<char, std::allocator<char> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<std::vector<char, std::allocator<char> > >*&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 6 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
 19 simdb_assert(stage_name_.empty(), "You may only add in/out ports in Stage subclass constructors");
145
146 19 auto& placeholder = output_placeholders_[port_name];
147
5/70
void simdb::pipeline::StageQueueRepo::addOutPortPlaceholder<DummyDataWindowBytes>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<DummyDataWindowBytes>*&):
✗ Branch 24 → 25 not taken.
✓ Branch 24 → 47 taken 2 times.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 36 → 37 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 45 → 46 not taken.
void simdb::pipeline::StageQueueRepo::addOutPortPlaceholder<simdb::argos::CollectionEntries>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<simdb::argos::CollectionEntries>*&):
✗ Branch 24 → 25 not taken.
✓ Branch 24 → 47 taken 5 times.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 36 → 37 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 45 → 46 not taken.
void simdb::pipeline::StageQueueRepo::addOutPortPlaceholder<simdb::argos::CompressedCollectionEntries>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<simdb::argos::CompressedCollectionEntries>*&):
✗ Branch 24 → 25 not taken.
✓ Branch 24 → 47 taken 5 times.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 36 → 37 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 45 → 46 not taken.
void simdb::pipeline::StageQueueRepo::addOutPortPlaceholder<std::vector<DummyData, std::allocator<DummyData> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<std::vector<DummyData, std::allocator<DummyData> > >*&):
✗ Branch 24 → 25 not taken.
✓ Branch 24 → 47 taken 1 time.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 36 → 37 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 45 → 46 not taken.
void simdb::pipeline::StageQueueRepo::addOutPortPlaceholder<std::vector<char, std::allocator<char> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<std::vector<char, std::allocator<char> > >*&):
✗ Branch 24 → 25 not taken.
✓ Branch 24 → 47 taken 6 times.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 36 → 37 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 45 → 46 not taken.
 19 simdb_assert(!placeholder, "Output port placeholder '" + port_name + "' already exists in QueueRepo");
148
5/5
void simdb::pipeline::StageQueueRepo::addOutPortPlaceholder<DummyDataWindowBytes>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<DummyDataWindowBytes>*&):
✓ Branch 47 → 48 taken 2 times.
void simdb::pipeline::StageQueueRepo::addOutPortPlaceholder<simdb::argos::CollectionEntries>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<simdb::argos::CollectionEntries>*&):
✓ Branch 47 → 48 taken 5 times.
void simdb::pipeline::StageQueueRepo::addOutPortPlaceholder<simdb::argos::CompressedCollectionEntries>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<simdb::argos::CompressedCollectionEntries>*&):
✓ Branch 47 → 48 taken 5 times.
void simdb::pipeline::StageQueueRepo::addOutPortPlaceholder<std::vector<DummyData, std::allocator<DummyData> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<std::vector<DummyData, std::allocator<DummyData> > >*&):
✓ Branch 47 → 48 taken 1 time.
void simdb::pipeline::StageQueueRepo::addOutPortPlaceholder<std::vector<char, std::allocator<char> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, simdb::ConcurrentQueue<std::vector<char, std::allocator<char> > >*&):
✓ Branch 47 → 48 taken 6 times.
 19 placeholder = std::make_unique<OutputQueuePlaceholder<T>>(queue);
149 19 }
150
151 /// \brief Set the stage name; rekeys all placeholders to "stage_name.port_name". Call at most once.
152 /// \throws DBException if renaming or if name already set differently.
153 32 void setStageName(const std::string& stage_name)
154 {
155
3/20
✓ Branch 3 → 4 taken 32 times.
✗ Branch 3 → 7 not taken.
✗ Branch 5 → 6 not taken.
✓ Branch 5 → 7 taken 32 times.
✗ Branch 8 → 9 not taken.
✓ Branch 8 → 35 taken 32 times.
✗ Branch 11 → 12 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 18 → 19 not taken.
✗ Branch 19 → 20 not taken.
✗ Branch 20 → 21 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 22 → 23 not taken.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 29 → 30 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 31 → 32 not taken.
✗ Branch 33 → 34 not taken.
 32 simdb_assert(stage_name_ == stage_name || stage_name_.empty(),
156 "Cannot rename StageQueueRepo stage name from '" + stage_name_ + "' to '" + stage_name +
157 "'. Can only set the stage name once.");
158 32 stage_name_ = stage_name;
159
160 32 rekeyPlaceholders_(input_placeholders_);
161 32 rekeyPlaceholders_(output_placeholders_);
162 32 }
163
164 private:
165 /// \brief Rekey every placeholder to "stage_name.port_name" without invalidating the
166 /// map while iterating. We extract each node, mutate its key in place, and
167 /// reinsert it, which moves the node (and its unique_ptr) rather than copying.
168 64 void rekeyPlaceholders_(std::unordered_map<std::string, std::unique_ptr<QueuePlaceholder>>& placeholders)
169 {
170 64 std::vector<std::string> old_keys;
171
1/1
✓ Branch 4 → 5 taken 64 times.
 64 old_keys.reserve(placeholders.size());
172
2/2
✓ Branch 13 → 7 taken 55 times.
✓ Branch 13 → 14 taken 64 times.
 119 for (const auto& [key, placeholder] : placeholders)
173 {
174 (void)placeholder;
175
1/1
✓ Branch 10 → 11 taken 55 times.
 55 old_keys.push_back(key);
176 }
177
178
2/2
✓ Branch 30 → 16 taken 55 times.
✓ Branch 30 → 31 taken 64 times.
 119 for (const auto& old_key : old_keys)
179 {
180
1/1
✓ Branch 17 → 18 taken 55 times.
 55 auto node = placeholders.extract(old_key);
181
2/2
✓ Branch 18 → 19 taken 55 times.
✓ Branch 19 → 20 taken 55 times.
 55 node.key() = stage_name_ + "." + old_key;
182
1/1
✓ Branch 25 → 26 taken 55 times.
 55 placeholders.insert(std::move(node));
183 55 }
184 64 }
185
186 std::unordered_map<std::string, std::unique_ptr<QueuePlaceholder>> input_placeholders_;
187 std::unordered_map<std::string, std::unique_ptr<QueuePlaceholder>> output_placeholders_;
188 std::string stage_name_;
189 friend class PipelineQueueRepo;
190 };
191
192 /*!
193 * \class PipelineQueueRepo
194 *
195 * \brief Aggregates StageQueueRepos, manages port bindings, and creates/assigns
196 * queues in finalizeBindings(). Lifecycle: add stages via merge(), then
197 * noMoreStages(), then bind() pairs, then finalizeBindings(); after that
198 * getInPortQueue/getOutPortQueue and validateQueues() are valid.
199 */
200 class PipelineQueueRepo
201 {
202 public:
203 /// \brief Merge another stage's placeholders into this repo; only while accepting stages.
204 /// \throws DBException if not accepting stages or if a port name collides.
205 32 void merge(StageQueueRepo& other)
206 {
207
1/12
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 21 taken 32 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 17 → 18 not taken.
✗ Branch 19 → 20 not taken.
 32 simdb_assert(state_ == RepoState::ACCEPTING_STAGES, "Cannot merge StageQueueRepo; PipelineQueueRepo "
208 "not accepting stages.");
209
210
2/2
✓ Branch 57 → 23 taken 36 times.
✓ Branch 57 → 58 taken 32 times.
 68 for (auto& [key, placeholder] : other.input_placeholders_)
211 {
212
2/15
✓ Branch 27 → 28 taken 36 times.
✗ Branch 29 → 30 not taken.
✓ Branch 29 → 52 taken 36 times.
✗ Branch 32 → 33 not taken.
✗ Branch 33 → 34 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 44 → 45 not taken.
✗ Branch 45 → 46 not taken.
✗ Branch 46 → 47 not taken.
✗ Branch 47 → 48 not taken.
✗ Branch 48 → 49 not taken.
✗ Branch 50 → 51 not taken.
 36 simdb_assert(input_placeholders_.find(key) == input_placeholders_.end(),
213 "Input port placeholder '" + key + "' already exists in QueueRepo");
214
1/1
✓ Branch 53 → 54 taken 36 times.
 36 input_placeholders_[key] = std::move(placeholder);
215 }
216
217
2/2
✓ Branch 94 → 60 taken 19 times.
✓ Branch 94 → 95 taken 32 times.
 51 for (auto& [key, placeholder] : other.output_placeholders_)
218 {
219
2/15
✓ Branch 64 → 65 taken 19 times.
✗ Branch 66 → 67 not taken.
✓ Branch 66 → 89 taken 19 times.
✗ Branch 69 → 70 not taken.
✗ Branch 70 → 71 not taken.
✗ Branch 72 → 73 not taken.
✗ Branch 76 → 77 not taken.
✗ Branch 77 → 78 not taken.
✗ Branch 78 → 79 not taken.
✗ Branch 81 → 82 not taken.
✗ Branch 82 → 83 not taken.
✗ Branch 83 → 84 not taken.
✗ Branch 84 → 85 not taken.
✗ Branch 85 → 86 not taken.
✗ Branch 87 → 88 not taken.
 19 simdb_assert(output_placeholders_.find(key) == output_placeholders_.end(),
220 "Output port placeholder '" + key + "' already exists in QueueRepo");
221
1/1
✓ Branch 90 → 91 taken 19 times.
 19 output_placeholders_[key] = std::move(placeholder);
222 }
223
224 32 other.input_placeholders_.clear();
225 32 other.output_placeholders_.clear();
226 32 }
227
228 /// \brief Signal that no more stages will be merged; switch to accepting bindings.
229 /// \throws DBException if not accepting stages.
230 13 void noMoreStages()
231 {
232
1/12
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 21 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 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 19 → 20 not taken.
 13 simdb_assert(state_ == RepoState::ACCEPTING_STAGES, "Cannot finalize stages; PipelineQueueRepo not "
233 "accepting stages.");
234 13 state_ = RepoState::ACCEPTING_BINDINGS;
235 13 }
236
237 /// \brief Bind an output port to an input port (both full names, e.g. "StageA.out" -> "StageB.in").
238 /// \throws DBException if not accepting bindings.
239 19 void bind(const std::string& output_port_full_name, const std::string& input_port_full_name)
240 {
241
1/12
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 21 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 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 19 → 20 not taken.
 19 simdb_assert(state_ == RepoState::ACCEPTING_BINDINGS,
242 "Cannot bind ports; PipelineQueueRepo not accepting bindings.");
243 19 port_bindings_[output_port_full_name] = input_port_full_name;
244 19 }
245
246 /// \brief Create queues from bindings and unbound ports; switch to bindings complete.
247 /// \throws DBException if not accepting bindings or if a port is missing.
248 13 void finalizeBindings()
249 {
250
1/12
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 21 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 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 19 → 20 not taken.
 13 simdb_assert(state_ == RepoState::ACCEPTING_BINDINGS, "Cannot finalize bindings; PipelineQueueRepo not "
251 "accepting bindings.");
252
253
2/2
✓ Branch 91 → 23 taken 19 times.
✓ Branch 91 → 92 taken 13 times.
 32 for (const auto& [out_port, in_port] : port_bindings_)
254 {
255
1/1
✓ Branch 26 → 27 taken 19 times.
 19 auto out_it = output_placeholders_.find(out_port);
256
1/14
✗ Branch 29 → 30 not taken.
✓ Branch 29 → 52 taken 19 times.
✗ Branch 32 → 33 not taken.
✗ Branch 33 → 34 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 44 → 45 not taken.
✗ Branch 45 → 46 not taken.
✗ Branch 46 → 47 not taken.
✗ Branch 47 → 48 not taken.
✗ Branch 48 → 49 not taken.
✗ Branch 50 → 51 not taken.
 19 simdb_assert(out_it != output_placeholders_.end(),
257 "Output port placeholder '" + out_port + "' not found in QueueRepo");
258
1/1
✓ Branch 52 → 53 taken 19 times.
 19 auto in_it = input_placeholders_.find(in_port);
259
1/14
✗ Branch 55 → 56 not taken.
✓ Branch 55 → 78 taken 19 times.
✗ Branch 58 → 59 not taken.
✗ Branch 59 → 60 not taken.
✗ Branch 61 → 62 not taken.
✗ Branch 65 → 66 not taken.
✗ Branch 66 → 67 not taken.
✗ Branch 67 → 68 not taken.
✗ Branch 70 → 71 not taken.
✗ Branch 71 → 72 not taken.
✗ Branch 72 → 73 not taken.
✗ Branch 73 → 74 not taken.
✗ Branch 74 → 75 not taken.
✗ Branch 76 → 77 not taken.
 19 simdb_assert(in_it != input_placeholders_.end(),
260 "Input port placeholder '" + in_port + "' not found in QueueRepo");
261
262
1/1
✓ Branch 80 → 81 taken 19 times.
 19 auto queue = out_it->second->createQueue();
263
1/1
✓ Branch 82 → 83 taken 19 times.
 19 queues_.emplace_back(std::move(queue));
264
1/1
✓ Branch 87 → 88 taken 19 times.
 19 in_it->second->assignQueue(queues_.back().get());
265 19 }
266
267
2/2
✓ Branch 109 → 94 taken 36 times.
✓ Branch 109 → 110 taken 13 times.
 49 for (const auto& [key, placeholder] : input_placeholders_)
268 {
269
3/3
✓ Branch 98 → 99 taken 36 times.
✓ Branch 99 → 100 taken 17 times.
✓ Branch 99 → 107 taken 19 times.
 36 if (!placeholder->hasQueue())
270 {
271
1/1
✓ Branch 101 → 102 taken 17 times.
 17 auto queue = placeholder->createQueue();
272
1/1
✓ Branch 103 → 104 taken 17 times.
 17 queues_.emplace_back(std::move(queue));
273
1/1
✓ Branch 104 → 105 taken 17 times.
 17 unbound_input_queues_.insert(key);
274 17 }
275 }
276
277
2/2
✓ Branch 127 → 112 taken 19 times.
✓ Branch 127 → 128 taken 13 times.
 32 for (const auto& [key, placeholder] : output_placeholders_)
278 {
279
2/3
✓ Branch 116 → 117 taken 19 times.
✗ Branch 117 → 118 not taken.
✓ Branch 117 → 125 taken 19 times.
 19 if (!placeholder->hasQueue())
280 {
281
0/1
✗ Branch 119 → 120 not taken.
 ✗ auto queue = placeholder->createQueue();
282
0/1
✗ Branch 121 → 122 not taken.
 ✗ queues_.emplace_back(std::move(queue));
283
0/1
✗ Branch 122 → 123 not taken.
 ✗ unbound_output_queues_.insert(key);
284 ✗ }
285 }
286
287 13 state_ = RepoState::BINDINGS_COMPLETE;
288 13 }
289
290 /// \brief Return the input port queue for the given full port name; only after finalizeBindings().
291 /// \tparam T Element type of the queue.
292 /// \throws DBException if bindings not finalized, port not found, or type mismatch.
293 17 template <typename T> ConcurrentQueue<T>* getInPortQueue(const std::string& port_full_name)
294 {
295
4/48
simdb::ConcurrentQueue<DummyData>* simdb::pipeline::PipelineQueueRepo::getInPortQueue<DummyData>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 21 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 17 → 18 not taken.
✗ Branch 19 → 20 not taken.
simdb::ConcurrentQueue<simdb::argos::NotifEntry>* simdb::pipeline::PipelineQueueRepo::getInPortQueue<simdb::argos::NotifEntry>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 21 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 17 → 18 not taken.
✗ Branch 19 → 20 not taken.
simdb::ConcurrentQueue<std::unique_ptr<simdb::argos::Ledger, std::default_delete<simdb::argos::Ledger> > >* simdb::pipeline::PipelineQueueRepo::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 → 21 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 17 → 18 not taken.
✗ Branch 19 → 20 not taken.
simdb::ConcurrentQueue<std::vector<double, std::allocator<double> > >* simdb::pipeline::PipelineQueueRepo::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 → 21 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 17 → 18 not taken.
✗ Branch 19 → 20 not taken.
 17 simdb_assert(state_ == RepoState::BINDINGS_COMPLETE, "Cannot access port queues until "
296 "finalizeBindings() is called.");
297
298
4/4
simdb::ConcurrentQueue<DummyData>* simdb::pipeline::PipelineQueueRepo::getInPortQueue<DummyData>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✓ Branch 21 → 22 taken 1 time.
simdb::ConcurrentQueue<simdb::argos::NotifEntry>* simdb::pipeline::PipelineQueueRepo::getInPortQueue<simdb::argos::NotifEntry>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✓ Branch 21 → 22 taken 5 times.
simdb::ConcurrentQueue<std::unique_ptr<simdb::argos::Ledger, std::default_delete<simdb::argos::Ledger> > >* simdb::pipeline::PipelineQueueRepo::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 21 → 22 taken 5 times.
simdb::ConcurrentQueue<std::vector<double, std::allocator<double> > >* simdb::pipeline::PipelineQueueRepo::getInPortQueue<std::vector<double, std::allocator<double> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✓ Branch 21 → 22 taken 6 times.
 17 auto it = input_placeholders_.find(port_full_name);
299
4/56
simdb::ConcurrentQueue<DummyData>* simdb::pipeline::PipelineQueueRepo::getInPortQueue<DummyData>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 24 → 25 not taken.
✓ Branch 24 → 47 taken 1 time.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 36 → 37 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 45 → 46 not taken.
simdb::ConcurrentQueue<simdb::argos::NotifEntry>* simdb::pipeline::PipelineQueueRepo::getInPortQueue<simdb::argos::NotifEntry>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 24 → 25 not taken.
✓ Branch 24 → 47 taken 5 times.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 36 → 37 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 45 → 46 not taken.
simdb::ConcurrentQueue<std::unique_ptr<simdb::argos::Ledger, std::default_delete<simdb::argos::Ledger> > >* simdb::pipeline::PipelineQueueRepo::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 24 → 25 not taken.
✓ Branch 24 → 47 taken 5 times.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 36 → 37 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 45 → 46 not taken.
simdb::ConcurrentQueue<std::vector<double, std::allocator<double> > >* simdb::pipeline::PipelineQueueRepo::getInPortQueue<std::vector<double, std::allocator<double> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 24 → 25 not taken.
✓ Branch 24 → 47 taken 6 times.
✗ Branch 27 → 28 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 36 → 37 not taken.
✗ Branch 39 → 40 not taken.
✗ Branch 40 → 41 not taken.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 43 → 44 not taken.
✗ Branch 45 → 46 not taken.
 17 simdb_assert(it != input_placeholders_.end(),
300 "Input port placeholder '" + port_full_name + "' not found in PipelineQueueRepo");
301
4/8
simdb::ConcurrentQueue<DummyData>* simdb::pipeline::PipelineQueueRepo::getInPortQueue<DummyData>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✓ Branch 49 → 50 taken 1 time.
✗ Branch 49 → 51 not taken.
simdb::ConcurrentQueue<simdb::argos::NotifEntry>* simdb::pipeline::PipelineQueueRepo::getInPortQueue<simdb::argos::NotifEntry>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✓ Branch 49 → 50 taken 5 times.
✗ Branch 49 → 51 not taken.
simdb::ConcurrentQueue<std::unique_ptr<simdb::argos::Ledger, std::default_delete<simdb::argos::Ledger> > >* simdb::pipeline::PipelineQueueRepo::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 49 → 50 taken 5 times.
✗ Branch 49 → 51 not taken.
simdb::ConcurrentQueue<std::vector<double, std::allocator<double> > >* simdb::pipeline::PipelineQueueRepo::getInPortQueue<std::vector<double, std::allocator<double> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✓ Branch 49 → 50 taken 6 times.
✗ Branch 49 → 51 not taken.
 17 auto typed_placeholder = dynamic_cast<InputQueuePlaceholder<T>*>(it->second.get());
302
4/56
simdb::ConcurrentQueue<DummyData>* simdb::pipeline::PipelineQueueRepo::getInPortQueue<DummyData>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 52 → 53 not taken.
✓ Branch 52 → 75 taken 1 time.
✗ Branch 55 → 56 not taken.
✗ Branch 56 → 57 not taken.
✗ Branch 58 → 59 not taken.
✗ Branch 62 → 63 not taken.
✗ Branch 63 → 64 not taken.
✗ Branch 64 → 65 not taken.
✗ Branch 67 → 68 not taken.
✗ Branch 68 → 69 not taken.
✗ Branch 69 → 70 not taken.
✗ Branch 70 → 71 not taken.
✗ Branch 71 → 72 not taken.
✗ Branch 73 → 74 not taken.
simdb::ConcurrentQueue<simdb::argos::NotifEntry>* simdb::pipeline::PipelineQueueRepo::getInPortQueue<simdb::argos::NotifEntry>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 52 → 53 not taken.
✓ Branch 52 → 75 taken 5 times.
✗ Branch 55 → 56 not taken.
✗ Branch 56 → 57 not taken.
✗ Branch 58 → 59 not taken.
✗ Branch 62 → 63 not taken.
✗ Branch 63 → 64 not taken.
✗ Branch 64 → 65 not taken.
✗ Branch 67 → 68 not taken.
✗ Branch 68 → 69 not taken.
✗ Branch 69 → 70 not taken.
✗ Branch 70 → 71 not taken.
✗ Branch 71 → 72 not taken.
✗ Branch 73 → 74 not taken.
simdb::ConcurrentQueue<std::unique_ptr<simdb::argos::Ledger, std::default_delete<simdb::argos::Ledger> > >* simdb::pipeline::PipelineQueueRepo::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 52 → 53 not taken.
✓ Branch 52 → 75 taken 5 times.
✗ Branch 55 → 56 not taken.
✗ Branch 56 → 57 not taken.
✗ Branch 58 → 59 not taken.
✗ Branch 62 → 63 not taken.
✗ Branch 63 → 64 not taken.
✗ Branch 64 → 65 not taken.
✗ Branch 67 → 68 not taken.
✗ Branch 68 → 69 not taken.
✗ Branch 69 → 70 not taken.
✗ Branch 70 → 71 not taken.
✗ Branch 71 → 72 not taken.
✗ Branch 73 → 74 not taken.
simdb::ConcurrentQueue<std::vector<double, std::allocator<double> > >* simdb::pipeline::PipelineQueueRepo::getInPortQueue<std::vector<double, std::allocator<double> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 52 → 53 not taken.
✓ Branch 52 → 75 taken 6 times.
✗ Branch 55 → 56 not taken.
✗ Branch 56 → 57 not taken.
✗ Branch 58 → 59 not taken.
✗ Branch 62 → 63 not taken.
✗ Branch 63 → 64 not taken.
✗ Branch 64 → 65 not taken.
✗ Branch 67 → 68 not taken.
✗ Branch 68 → 69 not taken.
✗ Branch 69 → 70 not taken.
✗ Branch 70 → 71 not taken.
✗ Branch 71 → 72 not taken.
✗ Branch 73 → 74 not taken.
 17 simdb_assert(typed_placeholder, "Incompatible queue types for input port '" + port_full_name + "'");
303
4/4
simdb::ConcurrentQueue<DummyData>* simdb::pipeline::PipelineQueueRepo::getInPortQueue<DummyData>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✓ Branch 75 → 76 taken 1 time.
simdb::ConcurrentQueue<simdb::argos::NotifEntry>* simdb::pipeline::PipelineQueueRepo::getInPortQueue<simdb::argos::NotifEntry>(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✓ Branch 75 → 76 taken 5 times.
simdb::ConcurrentQueue<std::unique_ptr<simdb::argos::Ledger, std::default_delete<simdb::argos::Ledger> > >* simdb::pipeline::PipelineQueueRepo::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 75 → 76 taken 5 times.
simdb::ConcurrentQueue<std::vector<double, std::allocator<double> > >* simdb::pipeline::PipelineQueueRepo::getInPortQueue<std::vector<double, std::allocator<double> > >(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✓ Branch 75 → 76 taken 6 times.
 17 unbound_input_queues_.erase(port_full_name);
304 34 return typed_placeholder->getQueue();
305 }
306
307 /// \brief Return the output port queue for the given full port name; only after finalizeBindings().
308 /// \tparam T Element type of the queue.
309 /// \throws DBException if bindings not finalized, port not found, or type mismatch.
310 template <typename T> ConcurrentQueue<T>* getOutPortQueue(const std::string& port_full_name)
311 {
312 simdb_assert(state_ == RepoState::BINDINGS_COMPLETE, "Cannot access port queues until "
313 "finalizeBindings() is called.");
314
315 auto it = output_placeholders_.find(port_full_name);
316 simdb_assert(it != output_placeholders_.end(),
317 "Output port placeholder '" + port_full_name + "' not found in PipelineQueueRepo");
318 auto typed_placeholder = dynamic_cast<OutputQueuePlaceholder<T>*>(it->second.get());
319 simdb_assert(typed_placeholder, "Incompatible queue types for output port '" + port_full_name + "'");
320 unbound_output_queues_.erase(port_full_name);
321 return typed_placeholder->getQueue();
322 }
323
324 /// \brief Throw if any input or output port is unbound (not connected and not explicitly unbound).
325 /// \throws DBException with a message listing unbound ports.
326 13 void validateQueues()
327 {
328
1/1
✓ Branch 2 → 3 taken 13 times.
 13 std::ostringstream oss;
329
330
1/2
✗ Branch 4 → 5 not taken.
✓ Branch 4 → 16 taken 13 times.
 13 if (!unbound_input_queues_.empty())
331 {
332
0/1
✗ Branch 5 → 6 not taken.
 ✗ oss << "The following input queues are not attached to anything:\n";
333
0/2
✗ Branch 14 → 8 not taken.
✗ Branch 14 → 15 not taken.
 ✗ for (const auto& name : unbound_input_queues_)
334 {
335
0/3
✗ Branch 9 → 10 not taken.
✗ Branch 10 → 11 not taken.
✗ Branch 11 → 12 not taken.
 ✗ oss << " " << name << "\n";
336 }
337 }
338
339
1/2
✗ Branch 17 → 18 not taken.
✓ Branch 17 → 29 taken 13 times.
 13 if (!unbound_output_queues_.empty())
340 {
341 oss << "The following output queues are not attached to "
342
0/1
✗ Branch 18 → 19 not taken.
 ✗ "anything:\n";
343
0/2
✗ Branch 27 → 21 not taken.
✗ Branch 27 → 28 not taken.
 ✗ for (const auto& name : unbound_output_queues_)
344 {
345
0/3
✗ Branch 22 → 23 not taken.
✗ Branch 23 → 24 not taken.
✗ Branch 24 → 25 not taken.
 ✗ oss << " " << name << "\n";
346 }
347 }
348
349
1/1
✓ Branch 29 → 30 taken 13 times.
 13 auto err = oss.str();
350
1/12
✗ Branch 31 → 32 not taken.
✓ Branch 31 → 50 taken 13 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 46 → 47 not taken.
✗ Branch 48 → 49 not taken.
 13 simdb_assert(err.empty(), err);
351 13 }
352
353 private:
354 std::unordered_map<std::string, std::unique_ptr<QueuePlaceholder>> input_placeholders_;
355 std::unordered_map<std::string, std::unique_ptr<QueuePlaceholder>> output_placeholders_;
356 std::unordered_map<std::string, std::string> port_bindings_;
357 std::set<std::string> unbound_input_queues_;
358 std::set<std::string> unbound_output_queues_;
359 std::vector<std::unique_ptr<QueueBase>> queues_;
360
361 enum class RepoState { ACCEPTING_STAGES, ACCEPTING_BINDINGS, BINDINGS_COMPLETE };
362
363 RepoState state_ = RepoState::ACCEPTING_STAGES;
364 };
365
366 } // namespace simdb::pipeline
367