apps/argos/ArgosCollector.hpp - GCC Code Coverage Report

apps/argos/ArgosCollector.hpp

Line	Branch	Exec	Source
1			// <ArgosCollector.hpp> -- C++ -*
2
3			#pragma once
4
5			#include "simdb/Assert.hpp"
6			#include "simdb/apps/App.hpp"
7			#include "simdb/apps/argos/Checkpointer.hpp"
8			#include "simdb/apps/argos/EntryPoint.hpp"
9			#include "simdb/apps/argos/EnumInspector.hpp"
10			#include "simdb/apps/argos/PipelineStagerInterface.hpp"
11			#include "simdb/apps/argos/Timestamps.hpp"
12			#include "simdb/pipeline/PipelineManager.hpp"
13			#include "simdb/sqlite/Dump.hpp"
14			#include "simdb/utils/Compress.hpp"
15			#include "simdb/utils/ConcurrentQueue.hpp"
16			#include "simdb/utils/TinyStrings.hpp"
17
18			namespace simdb::argos {
19
20			// TODO cnyce: Fix byte alignment issue surfaced on Mac. While Linux doesn't complain,
21			// we have a potential wrong answer bug. Disabling the checkpointers results in a DB size
22			// roughly 30% larger than using the default heartbeat of 10. But even the larger DB is
23			// still nearly 70% smaller than legacy collection.
24			// See GitHub issue #198.
25			inline constexpr size_t DEFAULT_HEARTBEAT = 1;
26
27			//! \class ArgosCollector
28			//! \brief Main entry point into the Argos collection system.
29			class ArgosCollector : public App, public PipelineStagerInterface
30			{
31			public:
32			//! Required by all SimDB apps
33			static constexpr auto NAME = "argos-collector";
34
35		5	ArgosCollector(DatabaseManager* db_mgr) :
36	2/2 ✓ Branch 4 → 5 taken 5 times. ✓ Branch 11 → 12 taken 5 times.	5	db_mgr_(db_mgr)
37			{
38		5	}
39
40		9	static void defineSchema(Schema& schema)
41			{
42			using dt = SqlDataType;
43
44	2/2 ✓ Branch 4 → 5 taken 9 times. ✓ Branch 5 → 6 taken 9 times.	18	auto& globals_tbl = schema.addTable("CollectionGlobals");
45	2/2 ✓ Branch 10 → 11 taken 9 times. ✓ Branch 11 → 12 taken 9 times.	18	globals_tbl.addColumn("Heartbeat", dt::int32_t);
46
47	2/2 ✓ Branch 16 → 17 taken 9 times. ✓ Branch 17 → 18 taken 9 times.	18	auto& clks_tbl = schema.addTable("Clocks");
48	2/2 ✓ Branch 22 → 23 taken 9 times. ✓ Branch 23 → 24 taken 9 times.	18	clks_tbl.addColumn("Name", dt::string_t);
49	2/2 ✓ Branch 28 → 29 taken 9 times. ✓ Branch 29 → 30 taken 9 times.	18	clks_tbl.addColumn("Period", dt::uint32_t);
50	2/2 ✓ Branch 34 → 35 taken 9 times. ✓ Branch 35 → 36 taken 9 times.	18	clks_tbl.addColumn("Numer", dt::uint32_t);
51	2/2 ✓ Branch 40 → 41 taken 9 times. ✓ Branch 41 → 42 taken 9 times.	18	clks_tbl.addColumn("Denom", dt::uint32_t);
52	2/2 ✓ Branch 46 → 47 taken 9 times. ✓ Branch 47 → 48 taken 9 times.	18	clks_tbl.setColumnDefaultValue("Numer", 0);
53	2/2 ✓ Branch 52 → 53 taken 9 times. ✓ Branch 53 → 54 taken 9 times.	18	clks_tbl.setColumnDefaultValue("Denom", 0);
54
55	2/2 ✓ Branch 58 → 59 taken 9 times. ✓ Branch 59 → 60 taken 9 times.	18	auto& collectable_tns_tbl = schema.addTable("CollectableTreeNodes");
56	2/2 ✓ Branch 64 → 65 taken 9 times. ✓ Branch 65 → 66 taken 9 times.	18	collectable_tns_tbl.addColumn("CID", dt::int32_t);
57	2/2 ✓ Branch 70 → 71 taken 9 times. ✓ Branch 71 → 72 taken 9 times.	18	collectable_tns_tbl.addColumn("FullPath", dt::string_t);
58	2/2 ✓ Branch 76 → 77 taken 9 times. ✓ Branch 77 → 78 taken 9 times.	18	collectable_tns_tbl.addColumn("ClockID", dt::int32_t);
59	2/2 ✓ Branch 82 → 83 taken 9 times. ✓ Branch 83 → 84 taken 9 times.	18	collectable_tns_tbl.addColumn("TypeName", dt::string_t);
60	2/2 ✓ Branch 88 → 89 taken 9 times. ✓ Branch 89 → 90 taken 9 times.	18	collectable_tns_tbl.addColumn("ShowInUI", dt::int32_t);
61	2/2 ✓ Branch 94 → 95 taken 9 times. ✓ Branch 95 → 96 taken 9 times.	18	collectable_tns_tbl.setColumnDefaultValue("ShowInUI", 0);
62	2/2 ✓ Branch 100 → 101 taken 9 times. ✓ Branch 101 → 102 taken 9 times.	18	collectable_tns_tbl.ensureUnique("CID");
63	2/2 ✓ Branch 106 → 107 taken 9 times. ✓ Branch 107 → 108 taken 9 times.	9	collectable_tns_tbl.createIndexOn("CID");
64		9	collectable_tns_tbl.unsetPrimaryKey();
65
66	2/2 ✓ Branch 113 → 114 taken 9 times. ✓ Branch 114 → 115 taken 9 times.	18	auto& enum_itypes_tbl = schema.addTable("CollectedEnums");
67	2/2 ✓ Branch 119 → 120 taken 9 times. ✓ Branch 120 → 121 taken 9 times.	18	enum_itypes_tbl.addColumn("EnumName", dt::string_t);
68	2/2 ✓ Branch 125 → 126 taken 9 times. ✓ Branch 126 → 127 taken 9 times.	18	enum_itypes_tbl.addColumn("EnumIntTypeName", dt::string_t);
69
70	2/2 ✓ Branch 131 → 132 taken 9 times. ✓ Branch 132 → 133 taken 9 times.	18	auto& enum_members_tbl = schema.addTable("EnumMembers");
71	2/2 ✓ Branch 137 → 138 taken 9 times. ✓ Branch 138 → 139 taken 9 times.	18	enum_members_tbl.addColumn("EnumID", dt::int32_t);
72	2/2 ✓ Branch 143 → 144 taken 9 times. ✓ Branch 144 → 145 taken 9 times.	18	enum_members_tbl.addColumn("MemberName", dt::string_t);
73	2/2 ✓ Branch 149 → 150 taken 9 times. ✓ Branch 150 → 151 taken 9 times.	18	enum_members_tbl.addColumn("MemberValueStr", dt::string_t);
74
75	2/2 ✓ Branch 155 → 156 taken 9 times. ✓ Branch 156 → 157 taken 9 times.	18	auto& timestamps_tbl = schema.addTable("Timestamps");
76	2/2 ✓ Branch 161 → 162 taken 9 times. ✓ Branch 162 → 163 taken 9 times.	18	timestamps_tbl.addColumn("Timestamp", dt::uint64_t);
77	2/2 ✓ Branch 167 → 168 taken 9 times. ✓ Branch 168 → 169 taken 9 times.	18	timestamps_tbl.ensureUnique("Timestamp");
78
79	2/2 ✓ Branch 173 → 174 taken 9 times. ✓ Branch 174 → 175 taken 9 times.	18	auto& collection_records_tbl = schema.addTable("CollectionRecords");
80	2/2 ✓ Branch 179 → 180 taken 9 times. ✓ Branch 180 → 181 taken 9 times.	18	collection_records_tbl.addColumn("TimestampID", dt::int32_t);
81	2/2 ✓ Branch 185 → 186 taken 9 times. ✓ Branch 186 → 187 taken 9 times.	18	collection_records_tbl.addColumn("Records", dt::blob_t);
82	2/2 ✓ Branch 191 → 192 taken 9 times. ✓ Branch 192 → 193 taken 9 times.	9	collection_records_tbl.ensureUnique("TimestampID");
83		9	collection_records_tbl.unsetPrimaryKey();
84
85	2/2 ✓ Branch 198 → 199 taken 9 times. ✓ Branch 199 → 200 taken 9 times.	18	auto& timestamp_clocks_tbl = schema.addTable("TimestampClocks");
86	2/2 ✓ Branch 204 → 205 taken 9 times. ✓ Branch 205 → 206 taken 9 times.	18	timestamp_clocks_tbl.addColumn("TimestampID", dt::int32_t);
87	2/2 ✓ Branch 210 → 211 taken 9 times. ✓ Branch 211 → 212 taken 9 times.	9	timestamp_clocks_tbl.addColumn("ClockID", dt::int32_t);
88	6/12 ✓ Branch 217 → 218 taken 18 times. ✓ Branch 220 → 215 taken 18 times. ✓ Branch 220 → 221 taken 9 times. ✓ Branch 221 → 222 taken 9 times. ✓ Branch 223 → 224 taken 18 times. ✓ Branch 223 → 225 taken 9 times. ✗ Branch 532 → 533 not taken. ✗ Branch 532 → 534 not taken. ✗ Branch 535 → 536 not taken. ✗ Branch 535 → 539 not taken. ✗ Branch 537 → 538 not taken. ✗ Branch 537 → 539 not taken.	45	timestamp_clocks_tbl.createCompoundIndexOn({"TimestampID", "ClockID"});
89		9	timestamp_clocks_tbl.unsetPrimaryKey();
90
91	2/2 ✓ Branch 228 → 229 taken 9 times. ✓ Branch 229 → 230 taken 9 times.	18	auto& queue_max_sizes_tbl = schema.addTable("QueueMaxSizes");
92	2/2 ✓ Branch 234 → 235 taken 9 times. ✓ Branch 235 → 236 taken 9 times.	18	queue_max_sizes_tbl.addColumn("CID", dt::int32_t);
93	2/2 ✓ Branch 240 → 241 taken 9 times. ✓ Branch 241 → 242 taken 9 times.	18	queue_max_sizes_tbl.addColumn("MaxSize", dt::int32_t);
94	2/2 ✓ Branch 246 → 247 taken 9 times. ✓ Branch 247 → 248 taken 9 times.	9	queue_max_sizes_tbl.ensureUnique("CID");
95		9	queue_max_sizes_tbl.unsetPrimaryKey();
96
97	2/2 ✓ Branch 253 → 254 taken 9 times. ✓ Branch 254 → 255 taken 9 times.	18	auto& notif_tbl = schema.addTable("Notifications");
98	2/2 ✓ Branch 259 → 260 taken 9 times. ✓ Branch 260 → 261 taken 9 times.	18	notif_tbl.addColumn("Timestamp", dt::uint64_t);
99	2/2 ✓ Branch 265 → 266 taken 9 times. ✓ Branch 266 → 267 taken 9 times.	18	notif_tbl.addColumn("NotifType", dt::int32_t);
100	2/2 ✓ Branch 271 → 272 taken 9 times. ✓ Branch 272 → 273 taken 9 times.	18	notif_tbl.addColumn("NotifStr", dt::string_t);
101
102			// TODO cnyce: populate this table in SimDB (Sparta will handle it for now)
103	2/2 ✓ Branch 277 → 278 taken 9 times. ✓ Branch 278 → 279 taken 9 times.	18	auto& dtype_schemas_tbl = schema.addTable("DataTypeSchemas");
104	2/2 ✓ Branch 283 → 284 taken 9 times. ✓ Branch 284 → 285 taken 9 times.	18	dtype_schemas_tbl.addColumn("RootTypeName", dt::string_t);
105
106			// TODO cnyce: populate this table in SimDB (Sparta will handle it for now)
107	2/2 ✓ Branch 289 → 290 taken 9 times. ✓ Branch 290 → 291 taken 9 times.	18	auto& dtype_nodes_tbl = schema.addTable("DataTypeNodes");
108	2/2 ✓ Branch 295 → 296 taken 9 times. ✓ Branch 296 → 297 taken 9 times.	18	dtype_nodes_tbl.addColumn("SchemaId", dt::int32_t);
109	2/2 ✓ Branch 301 → 302 taken 9 times. ✓ Branch 302 → 303 taken 9 times.	18	dtype_nodes_tbl.addColumn("Name", dt::string_t);
110	2/2 ✓ Branch 307 → 308 taken 9 times. ✓ Branch 308 → 309 taken 9 times.	18	dtype_nodes_tbl.addColumn("TypeName", dt::string_t);
111	2/2 ✓ Branch 313 → 314 taken 9 times. ✓ Branch 314 → 315 taken 9 times.	9	dtype_nodes_tbl.addColumn("FormatStr", dt::string_t);
112		9	}
113
114		4	void setHeartbeat(size_t heartbeat)
115			{
116	1/12 ✗ Branch 2 → 3 not taken. ✓ Branch 2 → 21 taken 4 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.	4	simdb_assert(heartbeat != 0, "Cannot use 0 for Argos collector heartbeat");
117
118	2/2 ✓ Branch 21 → 22 taken 3 times. ✓ Branch 21 → 26 taken 1 time.	4	if (heartbeat > 1)
119			{
120			// TODO cnyce: Fix byte alignment issue surfaced on Mac. While Linux doesn't complain,
121			// we have a potential wrong answer bug. Disabling the checkpointers results in a DB size
122			// roughly 30% larger than using the default heartbeat of 10. But even the larger DB is
123			// still nearly 70% smaller than legacy collection.
124			// See GitHub issue #198.
125		3	std::cout << "Ignoring Argos collection heartbeat (" << heartbeat
126		3	<< "). Defaulting to 1. This is a temporary workaround while a bug fix is implemented.";
127		3	std::cout << std::endl;
128			}
129		4	}
130
131		5	void addClock(const std::string& clk_name, size_t period) { addClock(clk_name, period, 0, 0); }
132
133		5	void addClock(const std::string& clk_name, size_t period, size_t numer, size_t denom)
134			{
135	1/2 ✗ Branch 41 → 4 not taken. ✓ Branch 41 → 42 taken 5 times.	5	for (const auto& [_clk_name, _period, _numer, _denom] : clocks_)
136			{
137	0/2 ✗ Branch 10 → 11 not taken. ✗ Branch 10 → 39 not taken.	✗	if (clk_name == _clk_name)
138			{
139	0/21 ✗ Branch 11 → 12 not taken. ✗ Branch 11 → 13 not taken. ✗ Branch 12 → 13 not taken. ✗ Branch 12 → 14 not taken. ✗ Branch 15 → 16 not taken. ✗ Branch 15 → 17 not taken. ✗ Branch 16 → 17 not taken. ✗ Branch 16 → 18 not taken. ✗ Branch 19 → 20 not taken. ✗ Branch 19 → 39 not taken. ✗ Branch 22 → 23 not taken. ✗ Branch 23 → 24 not taken. ✗ Branch 25 → 26 not taken. ✗ Branch 29 → 30 not taken. ✗ Branch 30 → 31 not taken. ✗ Branch 31 → 32 not taken. ✗ Branch 32 → 33 not taken. ✗ Branch 33 → 34 not taken. ✗ Branch 34 → 35 not taken. ✗ Branch 35 → 36 not taken. ✗ Branch 37 → 38 not taken.	✗	simdb_assert(period == _period && numer == _numer && denom == _denom,
140			"Clock mismatch - already registered with different params: " << clk_name);
141			}
142			}
143
144	1/1 ✓ Branch 42 → 43 taken 5 times.	5	auto clk_desc = std::make_tuple(clk_name, period, numer, denom);
145	1/1 ✓ Branch 44 → 45 taken 5 times.	5	clocks_.emplace_back(std::move(clk_desc));
146		5	}
147
148		5	void timestampWith(const uint64_t* backpointer)
149			{
150	1/12 ✗ 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.	5	simdb_assert(timestamp_ == nullptr, "Cannot change timestamp object once created!");
151	1/1 ✓ Branch 22 → 23 taken 5 times.	5	timestamp_ = std::make_unique<Timestamp>(backpointer);
152		5	}
153
154			void timestampWith(uint64_t (*fn)())
155			{
156			simdb_assert(timestamp_ == nullptr, "Cannot change timestamp object once created!");
157			timestamp_ = std::make_unique<Timestamp>(fn);
158			}
159
160			void timestampWith(std::function<uint64_t()> fn)
161			{
162			simdb_assert(timestamp_ == nullptr, "Cannot change timestamp object once created!");
163			timestamp_ = std::make_unique<Timestamp>(fn);
164			}
165
166			//! TODO cnyce: Once the collection code from Sparta is moved to SimDB, change this
167			//! to a template method so we can figure out the encoded data type name ourselves.
168			//! Scalar types are encoded as follows:
169			//!
170			//! For scalar PODs:
171			//! typeid(T).name()
172			//! "bool"
173			//! "unsigned long"
174			//! ...
175			//!
176			//! For scalar enums with operator<< (defn held in separate table for string-int map):
177			//! typeid(T).name()
178			//! "IssueType"
179			//! "MMUState"
180			//! ...
181			//! Enum values in collection blobs are serialized as std::underlying_type_t<T>.
182			//!
183			//! For scalar enums without operator<< (treated just like int PODs):
184			//! typeid(std::underlying_type_t<T>).name()
185			//! "int"
186			//! "unsigned int"
187			//! ...
188			//!
189			//! For scalar struct-like types:
190			//! typeid(T).name()
191			//! "Packet"
192			//! "Inst"
193			//! ...
194			//!
195			//! For scalar string-like types (std::string, const char*):
196			//! "string"
197		31	EntryPoint* createScalarCollector(const std::string& path, const std::string& clk_name,
198			const std::string& encoded_scalar_type)
199			{
200	1/1 ✓ Branch 2 → 3 taken 31 times.	31	auto entry_point = std::make_unique<EntryPoint>(this, &tiny_strings_);
201	2/2 ✓ Branch 3 → 4 taken 31 times. ✓ Branch 6 → 7 taken 31 times.	31	meta_by_cid_[entry_point->getID()] = std::make_tuple(path, clk_name, encoded_scalar_type);
202	1/1 ✓ Branch 10 → 11 taken 31 times.	31	entry_points_.emplace_back(std::move(entry_point));
203		62	return entry_points_.back().get();
204		31	}
205
206			//! TODO cnyce: Once the collection code from Sparta is moved to SimDB, change this
207			//! to a template method so we can figure out the encoded data type name ourselves.
208			//! Container types are encoded as follows:
209			//!
210			//! <encoded_scalar_type>_<sparse/contig>_capacity<N>
211			//! "Inst_sparse_capacity32"
212			//! "bool_contig_capacity4"
213			//! ...
214		60	EntryPoint* createContainerCollector(const std::string& path, const std::string& clk_name,
215			const std::string& encoded_container_type)
216			{
217	1/1 ✓ Branch 2 → 3 taken 60 times.	60	auto entry_point = std::make_unique<EntryPoint>(this, &tiny_strings_);
218	2/2 ✓ Branch 3 → 4 taken 60 times. ✓ Branch 6 → 7 taken 60 times.	60	meta_by_cid_[entry_point->getID()] = std::make_tuple(path, clk_name, encoded_container_type);
219	1/1 ✓ Branch 10 → 11 taken 60 times.	60	entry_points_.emplace_back(std::move(entry_point));
220		120	return entry_points_.back().get();
221		60	}
222
223		5	TinyStrings<>* getTinyStrings() { return &tiny_strings_; }
224
225		1	EnumInspector* getEnumInspector() { return &enum_inspector_; }
226
227		5	void createPipeline(pipeline::PipelineManager* pipeline_mgr) override
228			{
229	2/2 ✓ Branch 4 → 5 taken 5 times. ✓ Branch 5 → 6 taken 5 times.	10	auto pipeline = pipeline_mgr->createPipeline(NAME, this);
230
231	2/2 ✓ Branch 10 → 11 taken 5 times. ✓ Branch 11 → 12 taken 5 times.	10	pipeline_stager_ = pipeline->addStage<PipelineStager>("stager", heartbeat_);
232	2/2 ✓ Branch 16 → 17 taken 5 times. ✓ Branch 17 → 18 taken 5 times.	10	pipeline->addStage<Compressor>("compressor");
233	2/2 ✓ Branch 22 → 23 taken 5 times. ✓ Branch 23 → 24 taken 5 times.	10	pipeline->addStage<Writer>("writer");
234	1/1 ✓ Branch 26 → 27 taken 5 times.	5	pipeline->noMoreStages();
235
236	3/3 ✓ Branch 29 → 30 taken 5 times. ✓ Branch 32 → 33 taken 5 times. ✓ Branch 33 → 34 taken 5 times.	20	pipeline->bind("stager.data_output_queue", "compressor.data_input_queue");
237	3/3 ✓ Branch 40 → 41 taken 5 times. ✓ Branch 43 → 44 taken 5 times. ✓ Branch 44 → 45 taken 5 times.	20	pipeline->bind("compressor.data_output_queue", "writer.data_input_queue");
238	1/1 ✓ Branch 49 → 50 taken 5 times.	5	pipeline->noMoreBindings();
239
240	2/2 ✓ Branch 52 → 53 taken 5 times. ✓ Branch 53 → 54 taken 5 times.	10	pipeline_head_ = pipeline->getInPortQueue<LedgerPtr>("stager.main_input_queue");
241	2/2 ✓ Branch 58 → 59 taken 5 times. ✓ Branch 59 → 60 taken 5 times.	10	notif_head_ = pipeline->getInPortQueue<NotifEntry>("writer.notif_input_queue");
242
243	2/2 ✓ Branch 76 → 64 taken 91 times. ✓ Branch 76 → 77 taken 5 times.	96	for (const auto& [cid, meta] : meta_by_cid_)
244			{
245		91	const auto& encoded_dtype = std::get<2>(meta);
246
247		91	ContainerMeta container_meta;
248	3/3 ✓ Branch 68 → 69 taken 91 times. ✓ Branch 69 → 70 taken 60 times. ✓ Branch 69 → 73 taken 31 times.	91	if (extractContainerMeta_(encoded_dtype, container_meta))
249			{
250	3/3 ✓ Branch 70 → 71 taken 60 times. ✓ Branch 71 → 72 taken 60 times. ✓ Branch 72 → 74 taken 60 times.	60	pipeline_stager_->setContainerDataType(cid, container_meta.sparse, container_meta.capacity);
251			} else
252			{
253	1/1 ✓ Branch 73 → 74 taken 31 times.	31	pipeline_stager_->setScalarDataType(cid);
254			}
255			}
256
257		5	NotifEntry notif_entry;
258	2/3 ✓ Branch 81 → 82 taken 5 times. ✗ Branch 82 → 79 not taken. ✓ Branch 82 → 83 taken 5 times.	5	while (pending_notif_entries_.try_pop(notif_entry))
259			{
260	0/1 ✗ Branch 80 → 81 not taken.	✗	notif_head_->emplace(std::move(notif_entry));
261			}
262
263	2/2 ✓ Branch 94 → 85 taken 91 times. ✓ Branch 94 → 95 taken 5 times.	96	for (const auto& collector : entry_points_)
264			{
265		91	auto cid = collector->getID();
266	1/1 ✓ Branch 88 → 89 taken 91 times.	91	const auto& clk_name = std::get<1>(meta_by_cid_.at(cid));
267	2/2 ✓ Branch 90 → 91 taken 91 times. ✓ Branch 91 → 92 taken 91 times.	91	pipeline_stager_->setCollectableClock(cid, getClockId_(clk_name));
268			}
269
270		5	is_live_ = true;
271		5	}
272
273		5	void postInit(int, char**) override
274			{
275	3/3 ✓ Branch 2 → 3 taken 5 times. ✓ Branch 4 → 5 taken 5 times. ✓ Branch 5 → 6 taken 5 times.	5	db_mgr_->INSERT(SQL_TABLE("CollectionGlobals"), SQL_VALUES(heartbeat_));
276
277		5	std::map<std::string, int> clk_ids;
278	2/2 ✓ Branch 11 → 12 taken 5 times. ✓ Branch 12 → 13 taken 5 times.	5	auto clk_inserter = db_mgr_->prepareINSERT(SQL_TABLE("Clocks"));
279	2/2 ✓ Branch 26 → 16 taken 4 times. ✓ Branch 26 → 27 taken 5 times.	9	for (const auto& [_clk_name, _period, _numer, _denom] : clocks_)
280			{
281	1/1 ✓ Branch 22 → 23 taken 4 times.	4	auto id = clk_inserter->createRecordWithColValues(_clk_name, _period, _numer, _denom);
282	1/1 ✓ Branch 23 → 24 taken 4 times.	4	clk_ids[_clk_name] = id;
283			}
284
285	2/2 ✓ Branch 28 → 29 taken 5 times. ✓ Branch 29 → 30 taken 5 times.	5	auto ctn_inserter = db_mgr_->prepareINSERT(SQL_TABLE("CollectableTreeNodes"));
286	2/2 ✓ Branch 47 → 33 taken 91 times. ✓ Branch 47 → 48 taken 5 times.	96	for (const auto& collector : entry_points_)
287			{
288		91	auto cid = (int)collector->getID();
289	1/1 ✓ Branch 36 → 37 taken 91 times.	91	const auto& full_path = std::get<0>(meta_by_cid_.at(cid));
290	1/1 ✓ Branch 38 → 39 taken 91 times.	91	const auto& clk_name = std::get<1>(meta_by_cid_.at(cid));
291	1/1 ✓ Branch 40 → 41 taken 91 times.	91	const auto& dtype_name = std::get<2>(meta_by_cid_.at(cid));
292	1/1 ✓ Branch 42 → 43 taken 91 times.	91	const auto clk_id = clk_ids.at(clk_name);
293	1/1 ✓ Branch 44 → 45 taken 91 times.	91	ctn_inserter->createRecordWithColValues(cid, full_path, clk_id, dtype_name);
294			}
295		5	}
296
297		13008	void stage(uint16_t cid, std::vector<char>&& scalar_bytes) override
298			{
299		13008	assertLive_();
300		13008	checkTimeAdvanced_();
301		13008	ledger_->recordScalar(cid, std::move(scalar_bytes));
302		13008	}
303
304		12015	void stage(uint16_t cid, std::vector<std::vector<char>>&& contig_bin_bytes) override
305			{
306		12015	assertLive_();
307		12015	checkTimeAdvanced_();
308		12015	ledger_->recordContig(cid, std::move(contig_bin_bytes));
309		12015	}
310
311		12438	void stage(uint16_t cid, std::map<uint16_t, std::vector<char>>&& sparse_bin_bytes) override
312			{
313		12438	assertLive_();
314		12438	checkTimeAdvanced_();
315		12438	ledger_->recordSparse(cid, std::move(sparse_bin_bytes));
316		12438	}
317
318		4412	void closeRecord(uint16_t cid) override
319			{
320		4412	assertLive_();
321		4412	checkTimeAdvanced_();
322		4412	ledger_->closeRecord(cid);
323		4412	}
324
325		✗	void postNotif(const std::string& notif, NotifType type) override
326			{
327	0/1 ✗ Branch 3 → 4 not taken.	✗	NotifEntry entry(timestamp_.get(), notif, type);
328	0/2 ✗ Branch 4 → 5 not taken. ✗ Branch 4 → 6 not taken.	✗	auto notif_entries = is_live_ ? notif_head_ : &pending_notif_entries_;
329	0/1 ✗ Branch 8 → 9 not taken.	✗	notif_entries->emplace(std::move(entry));
330		✗	}
331
332		5	void preTeardown() override
333			{
334	5/6 ✓ Branch 3 → 4 taken 4 times. ✓ Branch 3 → 8 taken 1 time. ✓ Branch 6 → 7 taken 4 times. ✗ Branch 6 → 8 not taken. ✓ Branch 9 → 10 taken 4 times. ✓ Branch 9 → 12 taken 1 time.	5	if (ledger_ && ledger_->hasEntries())
335			{
336		4	pipeline_head_->emplace(std::move(ledger_));
337			}
338		5	}
339
340		5	void postTeardown() override
341			{
342	1/2 ✓ Branch 2 → 3 taken 5 times. ✗ Branch 2 → 4 not taken.	5	if (pipeline_stager_)
343			{
344		5	pipeline_stager_->writeMetaOnPostTeardown(db_mgr_);
345			}
346		5	tiny_strings_.serialize(db_mgr_);
347		5	enum_inspector_.serializeEnumMaps(db_mgr_);
348
349	2/2 ✓ Branch 7 → 8 taken 1 time. ✓ Branch 7 → 64 taken 4 times.	5	if (verbose())
350			{
351		1	std::cout << "[simdb] Collection tables at the end of simulation (except timestamps/blobs):\n\n";
352	2/2 ✓ Branch 11 → 12 taken 1 time. ✓ Branch 12 → 13 taken 1 time.	2	dumpTable(db_mgr_, "CollectionGlobals");
353	2/2 ✓ Branch 17 → 18 taken 1 time. ✓ Branch 18 → 19 taken 1 time.	2	dumpTable(db_mgr_, "Clocks");
354	2/2 ✓ Branch 23 → 24 taken 1 time. ✓ Branch 24 → 25 taken 1 time.	2	dumpTable(db_mgr_, "CollectedEnums");
355	2/2 ✓ Branch 29 → 30 taken 1 time. ✓ Branch 30 → 31 taken 1 time.	2	dumpTable(db_mgr_, "EnumMembers");
356	2/2 ✓ Branch 35 → 36 taken 1 time. ✓ Branch 36 → 37 taken 1 time.	2	dumpTable(db_mgr_, "CollectableTreeNodes");
357	2/2 ✓ Branch 41 → 42 taken 1 time. ✓ Branch 42 → 43 taken 1 time.	2	dumpTable(db_mgr_, "DataTypeSchemas");
358	2/2 ✓ Branch 47 → 48 taken 1 time. ✓ Branch 48 → 49 taken 1 time.	2	dumpTable(db_mgr_, "DataTypeNodes");
359	2/2 ✓ Branch 53 → 54 taken 1 time. ✓ Branch 54 → 55 taken 1 time.	2	dumpTable(db_mgr_, "QueueMaxSizes");
360	2/2 ✓ Branch 59 → 60 taken 1 time. ✓ Branch 60 → 61 taken 1 time.	2	dumpTable(db_mgr_, "Notifications");
361			}
362		5	}
363
364			private:
365			/// Entry to the DB pipeline (1st async stage input)
366			ConcurrentQueue<LedgerPtr>* pipeline_head_ = nullptr;
367
368			/// Entry to the DB writer stage's notifications
369			ConcurrentQueue<NotifEntry>* notif_head_ = nullptr;
370
371			/// Queued notifications prior to createPipeline()
372			ConcurrentQueue<NotifEntry> pending_notif_entries_;
373
374			/// Ledger of all collection activity at one simulation time point
375			LedgerPtr ledger_;
376
377			/// \class PipelineStager
378			/// \brief 1st async stage. Takes collected data from the Ledger and appends checkpoints to the
379			/// CID's checkpoint chains for snapshot-delta compression.
380			class PipelineStager : public pipeline::Stage
381			{
382			public:
383		5	PipelineStager(size_t heartbeat) :
384		5	heartbeat_(heartbeat)
385			{
386	2/2 ✓ Branch 10 → 11 taken 5 times. ✓ Branch 11 → 12 taken 5 times.	10	addInPort_<LedgerPtr>("main_input_queue", main_input_queue_);
387	2/2 ✓ Branch 16 → 17 taken 5 times. ✓ Branch 17 → 18 taken 5 times.	10	addOutPort_<CollectionEntries>("data_output_queue", data_output_queue_);
388		5	}
389
390			/// Must be called from main thread
391		91	void setCollectableClock(uint16_t cid, uint32_t clock_id)
392			{
393		91	collectable_clock_ids_[cid] = clock_id;
394		91	clock_ids_.insert(clock_id);
395		91	}
396
397			/// Must be called from main thread
398		31	void setScalarDataType(uint16_t cid)
399			{
400	2/2 ✓ Branch 2 → 3 taken 31 times. ✓ Branch 3 → 4 taken 31 times.	31	checkpointers_[cid] = std::make_unique<ScalarCheckpointer>(heartbeat_);
401		31	++num_scalars_;
402		31	}
403
404			/// Must be called from main thread
405		60	void setContainerDataType(uint16_t cid, bool sparse, size_t capacity)
406			{
407	2/2 ✓ Branch 2 → 3 taken 30 times. ✓ Branch 2 → 8 taken 30 times.	60	if (sparse)
408			{
409	2/2 ✓ Branch 3 → 4 taken 30 times. ✓ Branch 4 → 5 taken 30 times.	30	checkpointers_[cid] = std::make_unique<SparseContainerCheckpointer>(heartbeat_, capacity);
410		30	++num_sparses_;
411			} else
412			{
413	2/2 ✓ Branch 8 → 9 taken 30 times. ✓ Branch 9 → 10 taken 30 times.	30	checkpointers_[cid] = std::make_unique<ContigContainerCheckpointer>(heartbeat_, capacity);
414		30	++num_contigs_;
415			}
416		60	}
417
418		3005	size_t getNumScalars() const { return num_scalars_; }
419
420		3005	size_t getNumContigs() const { return num_contigs_; }
421
422		3005	size_t getNumSparses() const { return num_sparses_; }
423
424			/// Must be called from main thread
425		5	void writeMetaOnPostTeardown(DatabaseManager* db_mgr)
426			{
427		5	writeShowInUI_(db_mgr);
428		5	writeQueueMaxSizes_(db_mgr);
429		5	}
430
431			private:
432		27	pipeline::PipelineAction run_(bool) override
433			{
434		27	auto action = pipeline::PipelineAction::SLEEP;
435
436		27	LedgerPtr ledger;
437	3/3 ✓ Branch 121 → 122 taken 3032 times. ✓ Branch 122 → 3 taken 3005 times. ✓ Branch 122 → 123 taken 27 times.	3032	while (main_input_queue_->try_pop(ledger))
438			{
439		3005	auto window_id = ledger->getWindowId();
440
441			// Keep track of the CIDs that are used to call createCheckpoint(),
442			// and skip those CIDs in the calls to closeRecord(). If we allow
443			// the closeRecord() API call, the collected data does not show
444			// up in the UI.
445		3005	std::unordered_set<uint16_t> chkpt_cids;
446
447		3005	auto scalars = ledger->releaseScalarEntries();
448	2/2 ✓ Branch 21 → 10 taken 13008 times. ✓ Branch 21 → 22 taken 3005 times.	16013	for (auto& scalar : scalars)
449			{
450		13008	auto cid = scalar.cid;
451		13008	auto data = std::move(scalar.scalar_bytes);
452	2/2 ✓ Branch 13 → 14 taken 13008 times. ✓ Branch 16 → 17 taken 13008 times.	13008	checkpointers_.at(cid)->createCheckpoint(window_id, std::move(data));
453	1/1 ✓ Branch 17 → 18 taken 13008 times.	13008	chkpt_cids.insert(cid);
454		13008	}
455
456		3005	auto contigs = ledger->releaseContigEntries();
457	2/2 ✓ Branch 39 → 26 taken 12015 times. ✓ Branch 39 → 40 taken 3005 times.	15020	for (auto& contig : contigs)
458			{
459		12015	auto cid = contig.cid;
460		12015	auto data = std::move(contig.contig_bin_bytes);
461	2/2 ✓ Branch 29 → 30 taken 12015 times. ✓ Branch 30 → 31 taken 12015 times.	12015	updateContainerMaxSize_(cid, detail::getContainerSize(data));
462	2/2 ✓ Branch 31 → 32 taken 12015 times. ✓ Branch 34 → 35 taken 12015 times.	12015	checkpointers_.at(cid)->createCheckpoint(window_id, std::move(data));
463	1/1 ✓ Branch 35 → 36 taken 12015 times.	12015	chkpt_cids.insert(cid);
464		12015	}
465
466		3005	auto sparses = ledger->releaseSparseEntries();
467	2/2 ✓ Branch 57 → 44 taken 12438 times. ✓ Branch 57 → 58 taken 3005 times.	15443	for (auto& sparse : sparses)
468			{
469		12438	auto cid = sparse.cid;
470		12438	auto data = std::move(sparse.sparse_bin_bytes);
471	2/2 ✓ Branch 47 → 48 taken 12438 times. ✓ Branch 48 → 49 taken 12438 times.	12438	updateContainerMaxSize_(cid, detail::getContainerSize(data));
472	2/2 ✓ Branch 49 → 50 taken 12438 times. ✓ Branch 52 → 53 taken 12438 times.	12438	checkpointers_.at(cid)->createCheckpoint(window_id, std::move(data));
473	1/1 ✓ Branch 53 → 54 taken 12438 times.	12438	chkpt_cids.insert(cid);
474		12438	}
475
476	2/2 ✓ Branch 70 → 62 taken 4412 times. ✓ Branch 70 → 71 taken 3005 times.	7417	for (const auto cid : ledger->getClosedCIDs())
477			{
478	2/3 ✓ Branch 63 → 64 taken 4412 times. ✓ Branch 64 → 65 taken 4412 times. ✗ Branch 64 → 68 not taken.	4412	if (chkpt_cids.count(cid) == 0)
479			{
480	2/2 ✓ Branch 65 → 66 taken 4412 times. ✓ Branch 67 → 68 taken 4412 times.	4412	checkpointers_.at(cid)->closeRecord(window_id);
481			}
482			}
483
484		3005	CollectionEntries to_send;
485		3005	auto sim_time = ledger->getSimTime();
486		3005	to_send.sim_time = sim_time;
487
488		3005	const auto multi_clock = clock_ids_.size() > 1;
489	2/2 ✓ Branch 110 → 78 taken 90005 times. ✓ Branch 110 → 111 taken 3005 times.	93010	for (auto& [cid, checkpointer] : checkpointers_)
490			{
491	2/7 ✗ Branch 81 → 82 not taken. ✓ Branch 81 → 86 taken 90005 times. ✗ Branch 83 → 84 not taken. ✗ Branch 84 → 85 not taken. ✗ Branch 84 → 86 not taken. ✗ Branch 87 → 88 not taken. ✓ Branch 87 → 89 taken 90005 times.	90005	if (multi_clock && !checkpointer->participatedInWindow(window_id))
492			{
493		✗	continue;
494			}
495
496	1/1 ✓ Branch 90 → 91 taken 90005 times.	90005	auto entries = checkpointer->encodeForPipeline(window_id, sim_time, cid);
497	2/2 ✓ Branch 105 → 93 taken 128308 times. ✓ Branch 105 → 106 taken 90005 times.	218313	for (auto& entry : entries)
498			{
499	1/1 ✓ Branch 94 → 95 taken 128308 times.	128308	cids_with_data_.insert(cid);
500	1/1 ✓ Branch 96 → 97 taken 128308 times.	128308	to_send.entries.emplace_back(std::move(entry));
501	1/1 ✓ Branch 97 → 98 taken 128308 times.	128308	if (const auto clk_it = collectable_clock_ids_.find(cid);
502	1/2 ✓ Branch 100 → 101 taken 128308 times. ✗ Branch 100 → 103 not taken.	128308	clk_it != collectable_clock_ids_.end())
503			{
504	1/1 ✓ Branch 102 → 103 taken 128308 times.	128308	to_send.clock_ids.insert(clk_it->second);
505			}
506			}
507		90005	}
508
509	1/2 ✓ Branch 112 → 113 taken 3005 times. ✗ Branch 112 → 115 not taken.	3005	if (!to_send.entries.empty())
510			{
511	1/1 ✓ Branch 114 → 115 taken 3005 times.	3005	data_output_queue_->emplace(std::move(to_send));
512			}
513
514		3005	action = pipeline::PipelineAction::PROCEED;
515		3005	}
516
517		27	return action;
518		27	}
519
520		24453	void updateContainerMaxSize_(uint16_t cid, uint16_t size)
521			{
522	1/1 ✓ Branch 2 → 3 taken 24453 times.	24453	auto it = container_max_sizes_.find(cid);
523	2/2 ✓ Branch 5 → 6 taken 60 times. ✓ Branch 5 → 7 taken 24393 times.	24453	if (it == container_max_sizes_.end())
524			{
525	1/1 ✓ Branch 6 → 11 taken 60 times.	60	container_max_sizes_.emplace(cid, size);
526			} else
527			{
528		24393	it->second = std::max(it->second, size);
529			}
530		24453	}
531
532			// Set ShowInUI=1 for all the collectables that actually collected data
533		5	void writeShowInUI_(DatabaseManager* db_mgr) const
534			{
535		5	std::vector<int> valid_cids;
536	2/2 ✓ Branch 18 → 5 taken 91 times. ✓ Branch 18 → 19 taken 5 times.	96	for (auto cid : cids_with_data_)
537			{
538	3/3 ✓ Branch 6 → 7 taken 91 times. ✓ Branch 9 → 10 taken 60 times. ✓ Branch 9 → 13 taken 31 times.	91	if (auto it = container_max_sizes_.find(cid); it != container_max_sizes_.end())
539			{
540	1/2 ✗ Branch 11 → 12 not taken. ✓ Branch 11 → 13 taken 60 times.	60	if (it->second == 0)
541			{
542		✗	continue;
543			}
544			}
545	1/1 ✓ Branch 14 → 15 taken 91 times.	91	valid_cids.push_back(cid);
546			}
547
548	2/2 ✓ Branch 20 → 21 taken 4 times. ✓ Branch 20 → 38 taken 1 time.	5	if (!valid_cids.empty())
549			{
550	1/1 ✓ Branch 21 → 22 taken 4 times.	4	std::ostringstream oss;
551	1/1 ✓ Branch 22 → 23 taken 4 times.	4	oss << "UPDATE CollectableTreeNodes SET ShowInUI=1 WHERE CID IN (";
552
553		4	bool comma = false;
554	2/2 ✓ Branch 31 → 25 taken 91 times. ✓ Branch 31 → 32 taken 4 times.	95	for (const auto cid : valid_cids)
555			{
556	2/2 ✓ Branch 26 → 27 taken 87 times. ✓ Branch 26 → 28 taken 4 times.	91	if (comma)
557			{
558	1/1 ✓ Branch 27 → 28 taken 87 times.	87	oss << ",";
559			}
560	1/1 ✓ Branch 28 → 29 taken 91 times.	91	oss << cid;
561		91	comma = true;
562			}
563	1/1 ✓ Branch 32 → 33 taken 4 times.	4	oss << ")";
564	2/2 ✓ Branch 33 → 34 taken 4 times. ✓ Branch 34 → 35 taken 4 times.	4	db_mgr->EXECUTE(oss.str());
565		4	}
566
567	1/1 ✓ Branch 38 → 39 taken 5 times.	5	auto query = db_mgr->createQuery("CollectableTreeNodes");
568	1/1 ✓ Branch 40 → 41 taken 5 times.	5	query->addConstraintForInt("CID", SetConstraints::NOT_IN_SET, valid_cids);
569
570			struct CID_Info
571			{
572			std::string path;
573			std::string type;
574
575		✗	CID_Info(const std::string& path, const std::string& type) :
576		✗	path(path),
577	0/1 ✗ Branch 3 → 4 not taken.	✗	type(type)
578			{
579		✗	}
580			};
581
582		5	std::string path;
583	1/1 ✓ Branch 43 → 44 taken 5 times.	5	query->select("FullPath", path);
584
585		5	std::string type;
586	1/1 ✓ Branch 46 → 47 taken 5 times.	5	query->select("TypeName", type);
587
588		5	std::vector<CID_Info> cid_infos;
589	1/1 ✓ Branch 49 → 50 taken 5 times.	5	auto results = query->getResultSet();
590	2/3 ✓ Branch 53 → 54 taken 5 times. ✗ Branch 54 → 51 not taken. ✓ Branch 54 → 55 taken 5 times.	5	while (results.getNextRecord())
591			{
592	0/1 ✗ Branch 51 → 52 not taken.	✗	cid_infos.emplace_back(path, type);
593			}
594
595	1/2 ✗ Branch 56 → 57 not taken. ✓ Branch 56 → 113 taken 5 times.	5	if (!cid_infos.empty())
596			{
597	0/1 ✗ Branch 57 → 58 not taken.	✗	std::ostringstream oss;
598	0/1 ✗ Branch 58 → 59 not taken.	✗	oss << "No data was ever collected for the following collectables, and will not be shown in Argos:\n";
599		✗	size_t leftcol_w = 0;
600	0/2 ✗ Branch 66 → 61 not taken. ✗ Branch 66 → 67 not taken.	✗	for (const auto& info : cid_infos)
601			{
602		✗	leftcol_w = std::max(leftcol_w, info.path.size());
603			}
604
605		✗	leftcol_w += 12;
606	0/2 ✗ Branch 96 → 69 not taken. ✗ Branch 96 → 97 not taken.	✗	for (const auto& info : cid_infos)
607			{
608	0/2 ✗ Branch 70 → 71 not taken. ✗ Branch 73 → 74 not taken.	✗	oss << std::left << std::setw(leftcol_w) << info.path;
609	0/2 ✗ Branch 75 → 76 not taken. ✗ Branch 75 → 82 not taken.	✗	if (auto idx = info.type.find("_sparse"); idx != std::string::npos)
610			{
611	0/1 ✗ Branch 76 → 77 not taken.	✗	auto base_type = info.type.substr(0, idx);
612	0/3 ✗ Branch 77 → 78 not taken. ✗ Branch 78 → 79 not taken. ✗ Branch 79 → 80 not taken.	✗	oss << "(Sparse container of '" << base_type << "')";
613	0/2 ✗ Branch 83 → 84 not taken. ✗ Branch 83 → 90 not taken.	✗	} else if (auto idx = info.type.find("_contig"); idx != std::string::npos)
614			{
615	0/1 ✗ Branch 84 → 85 not taken.	✗	auto base_type = info.type.substr(0, idx);
616	0/3 ✗ Branch 85 → 86 not taken. ✗ Branch 86 → 87 not taken. ✗ Branch 87 → 88 not taken.	✗	oss << "(Contig container of '" << base_type << "')";
617		✗	} else
618			{
619	0/3 ✗ Branch 90 → 91 not taken. ✗ Branch 91 → 92 not taken. ✗ Branch 92 → 93 not taken.	✗	oss << "(" << info.type << ")";
620			}
621	0/1 ✗ Branch 93 → 94 not taken.	✗	oss << "\n";
622			}
623
624	0/1 ✗ Branch 97 → 98 not taken.	✗	auto warning = oss.str();
625		✗	warning.pop_back();
626
627	0/3 ✗ Branch 101 → 102 not taken. ✗ Branch 103 → 104 not taken. ✗ Branch 104 → 105 not taken.	✗	db_mgr->INSERT(SQL_TABLE("Notifications"), SQL_COLUMNS("NotifStr", "NotifType"),
628	0/2 ✗ Branch 99 → 100 not taken. ✗ Branch 100 → 101 not taken.	✗	SQL_VALUES(warning, (int)NotifType::WARNING));
629		✗	}
630		5	}
631
632		5	void writeQueueMaxSizes_(DatabaseManager* db_mgr) const
633			{
634	2/2 ✓ Branch 3 → 4 taken 5 times. ✓ Branch 4 → 5 taken 5 times.	5	auto inserter = db_mgr->prepareINSERT(SQL_TABLE("QueueMaxSizes"));
635	2/2 ✓ Branch 15 → 8 taken 60 times. ✓ Branch 15 → 16 taken 5 times.	65	for (const auto& [cid, max_size] : container_max_sizes_)
636			{
637	1/1 ✓ Branch 12 → 13 taken 60 times.	60	inserter->createRecordWithColValues((int)cid, (int)max_size);
638			}
639		5	}
640
641			const size_t heartbeat_;
642
643			ConcurrentQueue<LedgerPtr>* main_input_queue_ = nullptr;
644			ConcurrentQueue<CollectionEntries>* data_output_queue_ = nullptr;
645
646			std::unordered_map<uint16_t, uint32_t> collectable_clock_ids_;
647			std::unordered_set<uint32_t> clock_ids_;
648			std::unordered_set<uint16_t> cids_with_data_;
649
650			std::unordered_map<uint16_t, std::unique_ptr<CollectableCheckpointer>> checkpointers_;
651			std::unordered_map<uint16_t, uint16_t> container_max_sizes_;
652
653			size_t num_scalars_ = 0;
654			size_t num_contigs_ = 0;
655			size_t num_sparses_ = 0;
656			};
657
658			/// \class Compressor
659			/// \brief 2nd async stage. Performs zlib compression.
660			class Compressor : public pipeline::CompressionStage
661			{
662			public:
663		5	Compressor()
664		5	{
665	2/2 ✓ Branch 5 → 6 taken 5 times. ✓ Branch 6 → 7 taken 5 times.	10	addInPort_<CollectionEntries>("data_input_queue", input_queue_);
666	2/2 ✓ Branch 11 → 12 taken 5 times. ✓ Branch 12 → 13 taken 5 times.	10	addOutPort_<CompressedCollectionEntries>("data_output_queue", output_queue_);
667		5	}
668
669			private:
670		3072	pipeline::PipelineAction run_(bool) override
671			{
672		3072	CollectionEntries collection_at_time;
673	3/3 ✓ Branch 3 → 4 taken 3072 times. ✓ Branch 4 → 5 taken 3005 times. ✓ Branch 4 → 29 taken 67 times.	3072	if (input_queue_->try_pop(collection_at_time))
674			{
675		3005	std::vector<char> uncompressed;
676	2/2 ✓ Branch 18 → 8 taken 128308 times. ✓ Branch 18 → 19 taken 3005 times.	131313	for (const auto& src : collection_at_time.entries)
677			{
678		128308	const auto& src_data = src->getData();
679	1/1 ✓ Branch 15 → 16 taken 128308 times.	128308	uncompressed.insert(uncompressed.end(), src_data.begin(), src_data.end());
680			}
681
682		3005	CompressedCollectionEntries compressed;
683	2/2 ✓ Branch 20 → 21 taken 3005 times. ✓ Branch 21 → 22 taken 3005 times.	3005	compressData(uncompressed, compressed.compressed_entries, getBestCompressionLevel_());
684		3005	compressed.sim_time = collection_at_time.sim_time;
685		3005	compressed.clock_ids = std::move(collection_at_time.clock_ids);
686	1/1 ✓ Branch 25 → 26 taken 3005 times.	3005	output_queue_->emplace(std::move(compressed));
687		3005	return pipeline::PipelineAction::PROCEED;
688		3005	}
689
690		67	return pipeline::PipelineAction::SLEEP;
691		3072	}
692
693			ConcurrentQueue<CollectionEntries>* input_queue_ = nullptr;
694			ConcurrentQueue<CompressedCollectionEntries>* output_queue_ = nullptr;
695			};
696
697			/// \class Writer
698			/// \brief Final async stage. Writes collected/checkpointed/compressed data and notifications
699			/// to the database.
700			class Writer : public pipeline::DatabaseStage<ArgosCollector>
701			{
702			public:
703		5	Writer()
704		5	{
705	2/2 ✓ Branch 5 → 6 taken 5 times. ✓ Branch 6 → 7 taken 5 times.	10	addInPort_<CompressedCollectionEntries>("data_input_queue", data_input_queue_);
706	2/2 ✓ Branch 11 → 12 taken 5 times. ✓ Branch 12 → 13 taken 5 times.	10	addInPort_<NotifEntry>("notif_input_queue", notif_input_queue_);
707		5	}
708
709			private:
710		3356	pipeline::PipelineAction run_(bool) override
711			{
712		3356	auto action = pipeline::PipelineAction::SLEEP;
713
714		3356	CompressedCollectionEntries collection_at_time;
715	3/3 ✓ Branch 3 → 4 taken 3356 times. ✓ Branch 4 → 5 taken 3005 times. ✓ Branch 4 → 32 taken 351 times.	3356	if (data_input_queue_->try_pop(collection_at_time))
716			{
717	1/1 ✓ Branch 5 → 6 taken 3005 times.	3005	auto db_mgr = getDatabaseManager_();
718	2/2 ✓ Branch 6 → 7 taken 3005 times. ✓ Branch 7 → 8 taken 3005 times.	3005	auto id = Timestamp::createTimestampInDatabase(db_mgr, collection_at_time.sim_time);
719
720	2/2 ✓ Branch 10 → 11 taken 3005 times. ✓ Branch 11 → 12 taken 3005 times.	6010	auto inserter = getTableInserter_("CollectionRecords");
721		3005	const auto& bytes = collection_at_time.compressed_entries;
722	1/1 ✓ Branch 14 → 15 taken 3005 times.	3005	inserter->createRecordWithColValues(id, bytes);
723
724	1/2 ✓ Branch 16 → 17 taken 3005 times. ✗ Branch 16 → 31 not taken.	3005	if (!collection_at_time.clock_ids.empty())
725			{
726	2/2 ✓ Branch 19 → 20 taken 3005 times. ✓ Branch 20 → 21 taken 3005 times.	6010	auto clk_inserter = getTableInserter_("TimestampClocks");
727	2/2 ✓ Branch 29 → 25 taken 3005 times. ✓ Branch 29 → 30 taken 3005 times.	6010	for (const auto clock_id : collection_at_time.clock_ids)
728			{
729	1/1 ✓ Branch 26 → 27 taken 3005 times.	3005	clk_inserter->createRecordWithColValues(id, static_cast<int>(clock_id));
730			}
731			}
732
733		3005	action = pipeline::PipelineAction::PROCEED;
734			}
735
736			// Notifications are expected to be "sporadic" / one-off. Flush all
737			// of the notifs now, and do not consider "notification-only" to mean
738			// "keep greedily processing / keep threads running". This is the
739			// reason why we use WHILE here instead of IF like above, and the
740			// returned action is not set to PROCEED based on the availability
741			// of new notifications.
742		3356	NotifEntry notif;
743	2/3 ✓ Branch 47 → 48 taken 3356 times. ✗ Branch 48 → 34 not taken. ✓ Branch 48 → 49 taken 3356 times.	3356	while (notif_input_queue_->try_pop(notif))
744			{
745	0/2 ✗ Branch 36 → 37 not taken. ✗ Branch 37 → 38 not taken.	✗	auto inserter = getTableInserter_("Notifications");
746
747		✗	const auto& notif_str = notif.notif;
748		✗	const auto notif_type = notif.type;
749
750	0/2 ✗ Branch 41 → 42 not taken. ✗ Branch 41 → 44 not taken.	✗	if (notif.sim_time.isValid())
751			{
752	0/2 ✗ Branch 42 → 43 not taken. ✗ Branch 43 → 44 not taken.	✗	inserter->setColumnValue(0, notif.sim_time.getValue());
753			}
754	0/1 ✗ Branch 44 → 45 not taken.	✗	inserter->setColumnValue(1, (int)notif_type);
755	0/1 ✗ Branch 45 → 46 not taken.	✗	inserter->setColumnValue(2, notif_str);
756	0/1 ✗ Branch 46 → 47 not taken.	✗	inserter->createRecord();
757			}
758
759		3356	return action;
760		3356	}
761
762			ConcurrentQueue<CompressedCollectionEntries>* data_input_queue_ = nullptr;
763			ConcurrentQueue<NotifEntry>* notif_input_queue_ = nullptr;
764			};
765
766		83746	void assertLive_() const
767			{
768	3/16 ✓ Branch 2 → 3 taken 83746 times. ✗ Branch 2 → 5 not taken. ✗ Branch 4 → 5 not taken. ✓ Branch 4 → 6 taken 83746 times. ✗ Branch 7 → 8 not taken. ✓ Branch 7 → 26 taken 83746 times. ✗ Branch 10 → 11 not taken. ✗ Branch 11 → 12 not taken. ✗ Branch 13 → 14 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 22 → 23 not taken. ✗ Branch 24 → 25 not taken.	83746	simdb_assert(is_live_ && timestamp_, "API call cannot be made until pipeline is open and "
769			"timestampWith() was called");
770		83746	}
771
772		41873	void checkTimeAdvanced_()
773			{
774	1/1 ✓ Branch 2 → 3 taken 41873 times.	41873	assertLive_();
775
776	1/1 ✓ Branch 4 → 5 taken 41873 times.	41873	auto current_time = timestamp_->getTime();
777	2/2 ✓ Branch 6 → 7 taken 4 times. ✓ Branch 6 → 15 taken 41869 times.	41873	if (!current_stage_time_.isValid())
778			{
779		4	current_stage_time_ = current_time;
780	1/1 ✓ Branch 11 → 12 taken 4 times.	8	ledger_ = std::make_unique<Ledger>(current_time, current_window_id_++, pipeline_stager_->getNumScalars(),
781		12	pipeline_stager_->getNumContigs(), pipeline_stager_->getNumSparses());
782	2/3 ✓ Branch 15 → 16 taken 41869 times. ✗ Branch 16 → 17 not taken. ✓ Branch 16 → 37 taken 41869 times.	41869	} else if (current_time < current_stage_time_.getValue())
783			{
784	0/13 ✗ Branch 17 → 18 not taken. ✗ Branch 18 → 19 not taken. ✗ Branch 18 → 51 not taken. ✗ Branch 21 → 22 not taken. ✗ Branch 22 → 23 not taken. ✗ Branch 24 → 25 not taken. ✗ Branch 28 → 29 not taken. ✗ Branch 29 → 30 not taken. ✗ Branch 30 → 31 not taken. ✗ Branch 31 → 32 not taken. ✗ Branch 32 → 33 not taken. ✗ Branch 33 → 34 not taken. ✗ Branch 35 → 36 not taken.	✗	simdb_assert(current_time >= current_stage_time_.getValue(), "Time must be monotonically increasing");
785	3/3 ✓ Branch 37 → 38 taken 41869 times. ✓ Branch 38 → 39 taken 3001 times. ✓ Branch 38 → 51 taken 38868 times.	41869	} else if (current_time > current_stage_time_.getValue())
786			{
787	1/2 ✓ Branch 40 → 41 taken 3001 times. ✗ Branch 40 → 50 not taken.	3001	if (ledger_)
788			{
789			// Send current ledger to the pipeline and create a new one.
790	1/1 ✓ Branch 42 → 43 taken 3001 times.	3001	pipeline_head_->emplace(std::move(ledger_));
791
792			ledger_ =
793	1/1 ✓ Branch 46 → 47 taken 3001 times.	6002	std::make_unique<Ledger>(current_time, current_window_id_++, pipeline_stager_->getNumScalars(),
794		9003	pipeline_stager_->getNumContigs(), pipeline_stager_->getNumSparses());
795			}
796
797		3001	current_stage_time_ = current_time;
798			}
799		41873	}
800
801			struct ContainerMeta
802			{
803			ValidValue<bool> sparse;
804			ValidValue<size_t> capacity;
805			};
806
807		91	static bool extractContainerMeta_(const std::string& encoded_dtype, ContainerMeta& meta)
808			{
809		152	auto extract_capacity = [&](const bool contig) {
810	3/3 ✓ Branch 4 → 5 taken 91 times. ✓ Branch 4 → 6 taken 61 times. ✓ Branch 7 → 8 taken 152 times.	152	const std::string lookfor = contig ? "_contig_capacity" : "_sparse_capacity";
811	2/2 ✓ Branch 10 → 11 taken 60 times. ✓ Branch 10 → 18 taken 92 times.	152	if (auto idx = encoded_dtype.find(lookfor); idx != std::string::npos)
812			{
813	2/2 ✓ Branch 12 → 13 taken 60 times. ✓ Branch 13 → 14 taken 60 times.	60	auto capacity = std::stoi(encoded_dtype.substr(idx + lookfor.size()));
814		60	meta.sparse = !contig;
815		60	meta.capacity = capacity;
816		60	return true;
817			}
818		92	return false;
819		152	};
820
821	6/6 ✓ Branch 2 → 3 taken 91 times. ✓ Branch 3 → 4 taken 61 times. ✓ Branch 3 → 6 taken 30 times. ✓ Branch 4 → 5 taken 61 times. ✓ Branch 5 → 6 taken 30 times. ✓ Branch 5 → 7 taken 31 times.	182	return extract_capacity(true) \|\| extract_capacity(false);
822			}
823
824		91	uint32_t getClockId_(const std::string& clk_name) const
825			{
826	1/2 ✓ Branch 10 → 3 taken 91 times. ✗ Branch 10 → 11 not taken.	91	for (size_t i = 0; i < clocks_.size(); ++i)
827			{
828	1/2 ✓ Branch 6 → 7 taken 91 times. ✗ Branch 6 → 8 not taken.	91	if (std::get<0>(clocks_[i]) == clk_name)
829			{
830		91	return static_cast<uint32_t>(i + 1);
831			}
832			}
833	0/3 ✗ Branch 13 → 14 not taken. ✗ Branch 14 → 15 not taken. ✗ Branch 15 → 16 not taken.	✗	throw DBException("Unknown clock: ") << clk_name;
834			}
835
836			DatabaseManager* const db_mgr_;
837			size_t heartbeat_ = DEFAULT_HEARTBEAT;
838
839			using ClockDescriptor = std::tuple<std::string, // clk name
840			uint32_t, // period
841			uint32_t, // numerator
842			uint32_t // denominator
843			>;
844			std::vector<ClockDescriptor> clocks_;
845
846			using CollectableMeta = std::tuple<std::string, // dot-delimited path
847			std::string, // clk name
848			std::string // encoded data type
849			>;
850			std::map<uint16_t, CollectableMeta> meta_by_cid_;
851
852			std::unique_ptr<Timestamp> timestamp_;
853			std::vector<std::unique_ptr<EntryPoint>> entry_points_;
854			PipelineStager* pipeline_stager_ = nullptr;
855			ValidValue<uint64_t> current_stage_time_;
856			uint64_t current_window_id_ = 1;
857			TinyStrings<> tiny_strings_;
858			EnumInspector enum_inspector_;
859			bool is_live_ = false;
860			};
861
862			} // namespace simdb::argos
863