GCC Code Coverage Report

include/simdb/
Coverage:
low: ≥ 0%
medium: ≥ 75.0%
high: ≥ 90.0%
Lines:
110 / 110
100.0%
Branches:
25 / 84
29.8%
Functions:
34 / 34
100.0%
pipeline/Stage.hpp
Line Branch Exec Source
1 // <Stage.hpp> -*- C++ -*-
2
3 #pragma once
4
5 #include "simdb/Assert.hpp"
6 #include "simdb/pipeline/DatabaseAccessor.hpp"
7 #include "simdb/pipeline/DatabaseThread.hpp"
8 #include "simdb/pipeline/PollingThread.hpp"
9 #include "simdb/pipeline/QueueRepo.hpp"
10 #include "simdb/pipeline/Runnable.hpp"
11 #include "simdb/utils/Compress.hpp"
12 #include <algorithm>
13 #include <memory>
14 #include <unordered_map>
15
16 namespace simdb::pipeline {
17
18 /*!
19 * \class Stage
20 *
21 * \brief Base class for pipeline stages (Runnables with input/output ports
22 * and an optional polling interval). Subclass to implement run_(); use
23 * addInPort_/addOutPort_ in derived constructors to define queues.
24 * The interval is the PollingThread sleep time when no work is done.
25 */
26 class Stage : public Runnable
27 {
28 protected:
29 /// \brief Construct with the polling interval (ms) for the thread when no work is done.
30 /// \param interval_milliseconds Sleep time for the PollingThread; non-database stages
31 /// that share a thread must use the same interval.
32 32 Stage(size_t interval_milliseconds = 100) :
33 32 interval_milliseconds_(interval_milliseconds)
34 {
35 32 }
36
37 36 template <typename T> void addInPort_(const std::string& port_name, ConcurrentQueue<T>*& queue)
38 {
39 36 queue_repo_.addInPortPlaceholder<T>(port_name, queue);
40 36 }
41
42 19 template <typename T> void addOutPort_(const std::string& port_name, ConcurrentQueue<T>*& queue)
43 {
44 19 queue_repo_.addOutPortPlaceholder<T>(port_name, queue);
45 19 }
46
47 10 virtual AsyncDatabaseAccessor* getAsyncDatabaseAccessor_() const { return async_db_accessor_; }
48
49 private:
50 32 void setName_(const std::string& name)
51 {
52 32 name_ = name;
53 32 queue_repo_.setStageName(name);
54 32 }
55
56 32 void mergeQueueRepo_(PipelineQueueRepo& master_repo) { master_repo.merge(queue_repo_); }
57
58 20 virtual void assignThread_(DatabaseManager*, std::vector<std::unique_ptr<PollingThread>>& threads,
59 std::unique_ptr<DatabaseThread>&)
60 {
61
2/2
✓ Branch 2 → 3 taken 20 times.
✓ Branch 3 → 4 taken 20 times.
 20 threads.emplace_back(std::make_unique<PollingThread>(interval_milliseconds_));
62 20 threads.back()->addRunnable(this);
63 20 }
64
65 20 void setAsyncDatabaseAccessor_(AsyncDatabaseAccessor* async_db_accessor) { async_db_accessor_ = async_db_accessor; }
66
67 32 std::string getDescription_() const override { return name_; }
68
69 86281 PipelineAction processOne(bool force) override final { return run_(force); }
70
71 56 PipelineAction processAll(bool force) override final
72 {
73 56 PipelineAction outcome = PipelineAction::SLEEP;
74 while (true)
75 {
76 2087 auto result = processOne(force);
77
2/2
✓ Branch 4 → 5 taken 2031 times.
✓ Branch 4 → 6 taken 56 times.
 2087 if (result == PipelineAction::PROCEED)
78 {
79 2031 outcome = PipelineAction::PROCEED;
80
1/2
✓ Branch 6 → 7 taken 56 times.
✗ Branch 6 → 8 not taken.
 56 } else if (result == PipelineAction::SLEEP)
81 {
82 56 break;
83 }
84 2031 }
85 56 return outcome;
86 }
87
88 virtual PipelineAction run_(bool force) = 0;
89
90 std::string name_;
91 StageQueueRepo queue_repo_;
92 const size_t interval_milliseconds_;
93 AsyncDatabaseAccessor* async_db_accessor_ = nullptr;
94
95 friend class Pipeline;
96 friend class PipelineManager;
97 friend class Flusher;
98 };
99
100 /*!
101 * \class CompressionStage
102 *
103 * \brief For stages that perform zlib compression, subclassing from
104 * CompressionStage adds an API to get the best compression level given
105 * the thread's current pct time spent sleeping vs working.
106 */
107 class CompressionStage : public Stage
108 {
109 public:
110 // \brief Construct
111 // \param force_compress_regardless If true, the minimum compression
112 // level will always be 1 (fastest compression) even if the thread
113 // is not keeping up (never sleeping). If false, level=0 can be
114 // returned to denote "thread is too slow; do not compress"
115 11 CompressionStage(bool force_compress_regardless = true) :
116 11 force_compress_regardless_(force_compress_regardless)
117 {
118 11 }
119
120 protected:
121 25688 int getBestCompressionLevel_() const
122 {
123
3/3
✓ Branch 2 → 3 taken 25688 times.
✓ Branch 3 → 4 taken 25688 times.
✓ Branch 4 → 5 taken 25688 times.
 25688 const auto sleep_pct = std::clamp(getThread_()->getSleepPct(), 0.0, 100.0);
124
125 // Idle threads can spend CPU on better compression; busy threads favor speed.
126
1/2
✓ Branch 5 → 6 taken 25688 times.
✗ Branch 5 → 7 not taken.
 25688 const int min_level = force_compress_regardless_ ? static_cast<int>(CompressionLevel::FASTEST)
127 : static_cast<int>(CompressionLevel::DISABLED);
128 25688 const int max_level = static_cast<int>(CompressionLevel::HIGHEST);
129 25688 return min_level + static_cast<int>((max_level - min_level) * sleep_pct / 100.0 + 0.5);
130 }
131
132 private:
133 const bool force_compress_regardless_;
134 };
135
136 /*!
137 * \class DatabaseStageBase
138 *
139 * \brief Base for stages that run on the dedicated DatabaseThread and use
140 * DatabaseAccessor (getDatabaseManager_(), getTableInserter_()). Do not
141 * use getAsyncDatabaseAccessor_() from a DatabaseStage.
142 */
143 class DatabaseStageBase : public Stage
144 {
145 protected:
146 38024 AsyncDatabaseAccessor* getAsyncDatabaseAccessor_() const override final
147 {
148 throw DBException("Cannot access the AsyncDatabaseAccessor from a "
149 "DatabaseStage - use "
150
1/1
✓ Branch 4 → 5 taken 38024 times.
 38024 "getDatabaseManager_()");
151 }
152 };
153
154 /*!
155 * \class DatabaseStage
156 *
157 * \brief Concrete base for app-specific database stages. Provides
158 * getDatabaseManager_() and getTableInserter_() keyed by \p AppT.
159 * Derive from DatabaseStage<YourApp> and implement run_().
160 * \tparam AppT The App type (for schema and table inserters).
161 */
162 template <typename AppT> class DatabaseStage : public DatabaseStageBase
163 {
164 protected:
165 11172 DatabaseManager* getDatabaseManager_()
166 {
167
2/24
simdb::pipeline::DatabaseStage<PipelineSnooper>::getDatabaseManager_():
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 8167 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::DatabaseStage<simdb::argos::ArgosCollector>::getDatabaseManager_():
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 3005 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.
 11172 simdb_assert(db_accessor_, "DatabaseAccessor not initialized");
168 11172 return db_accessor_->getDatabaseManager();
169 }
170
171 33691 PreparedINSERT* getTableInserter_(const std::string& tbl_name)
172 {
173
3/36
simdb::pipeline::DatabaseStage<SimplePipeline>::getTableInserter_(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 22683 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::DatabaseStage<PipelineSnooper>::getTableInserter_(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 4998 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::DatabaseStage<simdb::argos::ArgosCollector>::getTableInserter_(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&):
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 6010 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.
 33691 simdb_assert(db_accessor_, "DatabaseAccessor not initialized");
174 33691 return db_accessor_->template getTableInserter<AppT>(tbl_name);
175 }
176
177 private:
178 12 void assignThread_(DatabaseManager* db_mgr, std::vector<std::unique_ptr<PollingThread>>&,
179 std::unique_ptr<DatabaseThread>& database_thread) override final
180 {
181 // Prepare the DatabaseAccessor
182
3/3
simdb::pipeline::DatabaseStage<SimplePipeline>::assignThread_(simdb::DatabaseManager*, std::vector<std::unique_ptr<simdb::pipeline::PollingThread, std::default_delete<simdb::pipeline::PollingThread> >, std::allocator<std::unique_ptr<simdb::pipeline::PollingThread, std::default_delete<simdb::pipeline::PollingThread> > > >&, std::unique_ptr<simdb::pipeline::DatabaseThread, std::default_delete<simdb::pipeline::DatabaseThread> >&):
✓ Branch 2 → 3 taken 6 times.
simdb::pipeline::DatabaseStage<PipelineSnooper>::assignThread_(simdb::DatabaseManager*, std::vector<std::unique_ptr<simdb::pipeline::PollingThread, std::default_delete<simdb::pipeline::PollingThread> >, std::allocator<std::unique_ptr<simdb::pipeline::PollingThread, std::default_delete<simdb::pipeline::PollingThread> > > >&, std::unique_ptr<simdb::pipeline::DatabaseThread, std::default_delete<simdb::pipeline::DatabaseThread> >&):
✓ Branch 2 → 3 taken 1 time.
simdb::pipeline::DatabaseStage<simdb::argos::ArgosCollector>::assignThread_(simdb::DatabaseManager*, std::vector<std::unique_ptr<simdb::pipeline::PollingThread, std::default_delete<simdb::pipeline::PollingThread> >, std::allocator<std::unique_ptr<simdb::pipeline::PollingThread, std::default_delete<simdb::pipeline::PollingThread> > > >&, std::unique_ptr<simdb::pipeline::DatabaseThread, std::default_delete<simdb::pipeline::DatabaseThread> >&):
✓ Branch 2 → 3 taken 5 times.
 12 db_accessor_ = std::make_unique<DatabaseAccessor>(db_mgr);
183
184 // Use dedicated database thread
185
4/6
simdb::pipeline::DatabaseStage<SimplePipeline>::assignThread_(simdb::DatabaseManager*, std::vector<std::unique_ptr<simdb::pipeline::PollingThread, std::default_delete<simdb::pipeline::PollingThread> >, std::allocator<std::unique_ptr<simdb::pipeline::PollingThread, std::default_delete<simdb::pipeline::PollingThread> > > >&, std::unique_ptr<simdb::pipeline::DatabaseThread, std::default_delete<simdb::pipeline::DatabaseThread> >&):
✓ Branch 6 → 7 taken 3 times.
✓ Branch 6 → 11 taken 3 times.
simdb::pipeline::DatabaseStage<PipelineSnooper>::assignThread_(simdb::DatabaseManager*, std::vector<std::unique_ptr<simdb::pipeline::PollingThread, std::default_delete<simdb::pipeline::PollingThread> >, std::allocator<std::unique_ptr<simdb::pipeline::PollingThread, std::default_delete<simdb::pipeline::PollingThread> > > >&, std::unique_ptr<simdb::pipeline::DatabaseThread, std::default_delete<simdb::pipeline::DatabaseThread> >&):
✓ Branch 6 → 7 taken 1 time.
✗ Branch 6 → 11 not taken.
simdb::pipeline::DatabaseStage<simdb::argos::ArgosCollector>::assignThread_(simdb::DatabaseManager*, std::vector<std::unique_ptr<simdb::pipeline::PollingThread, std::default_delete<simdb::pipeline::PollingThread> >, std::allocator<std::unique_ptr<simdb::pipeline::PollingThread, std::default_delete<simdb::pipeline::PollingThread> > > >&, std::unique_ptr<simdb::pipeline::DatabaseThread, std::default_delete<simdb::pipeline::DatabaseThread> >&):
✓ Branch 6 → 7 taken 5 times.
✗ Branch 6 → 11 not taken.
 12 if (!database_thread)
186 {
187
3/3
simdb::pipeline::DatabaseStage<SimplePipeline>::assignThread_(simdb::DatabaseManager*, std::vector<std::unique_ptr<simdb::pipeline::PollingThread, std::default_delete<simdb::pipeline::PollingThread> >, std::allocator<std::unique_ptr<simdb::pipeline::PollingThread, std::default_delete<simdb::pipeline::PollingThread> > > >&, std::unique_ptr<simdb::pipeline::DatabaseThread, std::default_delete<simdb::pipeline::DatabaseThread> >&):
✓ Branch 7 → 8 taken 3 times.
simdb::pipeline::DatabaseStage<PipelineSnooper>::assignThread_(simdb::DatabaseManager*, std::vector<std::unique_ptr<simdb::pipeline::PollingThread, std::default_delete<simdb::pipeline::PollingThread> >, std::allocator<std::unique_ptr<simdb::pipeline::PollingThread, std::default_delete<simdb::pipeline::PollingThread> > > >&, std::unique_ptr<simdb::pipeline::DatabaseThread, std::default_delete<simdb::pipeline::DatabaseThread> >&):
✓ Branch 7 → 8 taken 1 time.
simdb::pipeline::DatabaseStage<simdb::argos::ArgosCollector>::assignThread_(simdb::DatabaseManager*, std::vector<std::unique_ptr<simdb::pipeline::PollingThread, std::default_delete<simdb::pipeline::PollingThread> >, std::allocator<std::unique_ptr<simdb::pipeline::PollingThread, std::default_delete<simdb::pipeline::PollingThread> > > >&, std::unique_ptr<simdb::pipeline::DatabaseThread, std::default_delete<simdb::pipeline::DatabaseThread> >&):
✓ Branch 7 → 8 taken 5 times.
 9 database_thread = std::make_unique<DatabaseThread>(db_mgr);
188 }
189
190 12 database_thread->addRunnable(this);
191 12 }
192
193 std::unique_ptr<DatabaseAccessor> db_accessor_;
194 };
195
196 } // namespace simdb::pipeline
197