GCC Code Coverage Report

include/simdb/
Coverage:
low: ≥ 0%
medium: ≥ 75.0%
high: ≥ 90.0%
Lines:
47 / 51
92.2%
Branches:
9 / 14
64.3%
Functions:
19 / 22
86.4%
apps/argos/PipelineDataTypes.hpp
Line Branch Exec Source
1 // <PipelineDataTypes.hpp> -*- C++ -*-
2
3 #pragma once
4
5 #include "simdb/Exceptions.hpp"
6 #include "simdb/apps/argos/CollectedData.hpp"
7 #include "simdb/apps/argos/Timestamps.hpp"
8 #include "simdb/utils/Demangle.hpp"
9 #include "simdb/utils/ValidValue.hpp"
10
11 #include <map>
12 #include <memory>
13 #include <string>
14 #include <unordered_map>
15 #include <unordered_set>
16 #include <utility>
17 #include <vector>
18
19 //! This file contains data types that are used to pass information
20 //! down ArgosCollector's pipeline to the database.
21
22 namespace simdb::argos {
23
24 //! \brief Holds onto all byte buffers containing collected data and open/close state changes.
25 //! All records of everything that happened at a given simulation time T are processed together
26 //! and end up in the DB in the same blob/rowid. Clocks are tracked so we can support multi-clock
27 //! simulations. The clock_ids are a set (not a scalar) since two clocks can collect at the same
28 //! time (example: one clock might be exactly twice as fast as another).
29 //!
30 //! Used as input to the zlib pipeline stage.
31 struct CollectionEntries
32 {
33 ValidValue<uint64_t> sim_time;
34 std::vector<std::unique_ptr<CollectedData>> entries;
35 std::unordered_set<uint32_t> clock_ids;
36 };
37
38 //! \brief Output data structure from the zlib pipeline stage.
39 struct CompressedCollectionEntries
40 {
41 ValidValue<uint64_t> sim_time;
42 std::vector<char> compressed_entries;
43 std::unordered_set<uint32_t> clock_ids;
44 };
45
46 //! \brief Enum used for warning/error/message filtering in Argos.
47 enum class NotifType { WARNING, ERROR, MESSAGE, __INVALID__ };
48
49 //! \brief Notification captured during collection. These appear in a dedicated tab in Argos.
50 struct Notification
51 {
52 ValidValue<uint16_t> cid;
53 std::string notif;
54 NotifType type = NotifType::__INVALID__;
55 ValidValue<uint64_t> sim_time;
56
57 Notification(uint16_t cid, const std::string& notif, const NotifType type, uint64_t sim_time) :
58 cid(cid),
59 notif(notif),
60 type(type),
61 sim_time(sim_time)
62 {
63 }
64
65 Notification(uint16_t cid, const std::string& notif, const NotifType type) :
66 cid(cid),
67 notif(notif),
68 type(type)
69 {
70 }
71
72 // Default ctor needed for ConcurrentQueue::try_pop
73 Notification() = default;
74 };
75
76 //! \brief Single byte buffer with CID for scalar collectables (including scalar structs).
77 struct ScalarEntry
78 {
79 uint16_t cid = 0;
80 std::vector<char> scalar_bytes;
81
82 13008 ScalarEntry(uint16_t cid, std::vector<char>&& scalar_bytes) :
83 13008 cid(cid),
84 13008 scalar_bytes(std::move(scalar_bytes))
85 {
86 13008 }
87
88 ScalarEntry() = default;
89 ✗ ScalarEntry(ScalarEntry&&) = default;
90 ScalarEntry& operator=(ScalarEntry&&) = default;
91 };
92
93 //! \brief Vector of byte buffers with CID for contig container collectables.
94 struct ContigEntry
95 {
96 uint16_t cid = 0;
97 std::vector<std::vector<char>> contig_bin_bytes;
98
99 12015 ContigEntry(uint16_t cid, std::vector<std::vector<char>>&& contig_bin_bytes) :
100 12015 cid(cid),
101 12015 contig_bin_bytes(std::move(contig_bin_bytes))
102 {
103 12015 }
104
105 ContigEntry() = default;
106 ✗ ContigEntry(ContigEntry&&) = default;
107 ContigEntry& operator=(ContigEntry&&) = default;
108 };
109
110 //! \brief Bin-mapped byte buffers with CID for sparse container collectables.
111 struct SparseEntry
112 {
113 uint16_t cid = 0;
114 std::map<uint16_t, std::vector<char>> sparse_bin_bytes;
115
116 12438 SparseEntry(uint16_t cid, std::map<uint16_t, std::vector<char>>&& sparse_bin_bytes) :
117 12438 cid(cid),
118 12438 sparse_bin_bytes(std::move(sparse_bin_bytes))
119 {
120 12438 }
121
122 SparseEntry() = default;
123 ✗ SparseEntry(SparseEntry&&) = default;
124 SparseEntry& operator=(SparseEntry&&) = default;
125 };
126
127 //! \brief Timestamped notification.
128 struct NotifEntry
129 {
130 ValidValue<uint64_t> sim_time;
131 std::string notif;
132 NotifType type = NotifType::__INVALID__;
133
134 1 NotifEntry(const Timestamp* timestamp, const std::string& notif, NotifType type) :
135 1 notif(notif),
136 1 type(type)
137 {
138
1/2
✗ Branch 4 → 5 not taken.
✓ Branch 4 → 8 taken 1 time.
 1 if (timestamp)
139 {
140
0/1
✗ Branch 5 → 6 not taken.
 ✗ sim_time = timestamp->getTime();
141 }
142 1 }
143
144 3361 NotifEntry() = default;
145 2 NotifEntry(NotifEntry&&) = default;
146 2 NotifEntry& operator=(NotifEntry&&) = default;
147 };
148
149 //! \brief This class is used to create all the *Entries above. Tracks everything that happened
150 //! across all collectables at a specific simulation time point. A ledger is filled at time T
151 //! until simulation has moved forward, then the ledger is moved to the 1st async stage of
152 //! the DB pipeline. ArgosCollector will create a new Ledger and add to it until time advances
153 //! again.
154 class Ledger
155 {
156 public:
157 3005 Ledger(uint64_t sim_time, uint64_t window_id, uint64_t reserve_num_scalars = 0, uint64_t reserve_num_contigs = 0,
158 3005 uint64_t reserve_num_sparses = 0) :
159 3005 sim_time_(sim_time),
160 3005 window_id_(window_id)
161 {
162
1/1
✓ Branch 6 → 7 taken 3005 times.
 3005 scalar_records_.reserve(reserve_num_scalars);
163
1/1
✓ Branch 7 → 8 taken 3005 times.
 3005 contig_records_.reserve(reserve_num_contigs);
164
1/1
✓ Branch 8 → 9 taken 3005 times.
 3005 sparse_records_.reserve(reserve_num_sparses);
165 3005 }
166
167 Ledger(Ledger&&) = default;
168 Ledger(const Ledger&) = delete;
169
170 13008 void recordScalar(uint16_t cid, std::vector<char>&& scalar_bytes)
171 {
172 13008 scalar_records_.emplace_back(cid, std::move(scalar_bytes));
173 13008 }
174
175 12015 void recordContig(uint16_t cid, std::vector<std::vector<char>>&& contig_bytes)
176 {
177 12015 contig_records_.emplace_back(cid, std::move(contig_bytes));
178 12015 }
179
180 12438 void recordSparse(uint16_t cid, std::map<uint16_t, std::vector<char>>&& sparse_bin_bytes)
181 {
182 12438 sparse_records_.emplace_back(cid, std::move(sparse_bin_bytes));
183 12438 }
184
185 4412 void closeRecord(uint16_t cid) { closed_cids_.insert(cid); }
186
187 3005 uint64_t getSimTime() const { return sim_time_; }
188
189 3005 uint64_t getWindowId() const { return window_id_; }
190
191 4 bool hasEntries() const
192 {
193
4/6
✓ Branch 3 → 4 taken 1 time.
✓ Branch 3 → 10 taken 3 times.
✓ Branch 5 → 6 taken 1 time.
✗ Branch 5 → 10 not taken.
✓ Branch 7 → 8 taken 1 time.
✗ Branch 7 → 10 not taken.
 5 return !scalar_records_.empty() || !contig_records_.empty() || !sparse_records_.empty() ||
194
1/2
✓ Branch 9 → 10 taken 1 time.
✗ Branch 9 → 11 not taken.
 5 !closed_cids_.empty();
195 }
196
197 3005 std::vector<ScalarEntry> releaseScalarEntries() { return std::move(scalar_records_); }
198
199 3005 std::vector<ContigEntry> releaseContigEntries() { return std::move(contig_records_); }
200
201 3005 std::vector<SparseEntry> releaseSparseEntries() { return std::move(sparse_records_); }
202
203 3005 const std::unordered_set<uint16_t>& getClosedCIDs() const { return closed_cids_; }
204
205 private:
206 uint64_t sim_time_ = 0;
207 uint64_t window_id_ = 0;
208 std::vector<ScalarEntry> scalar_records_;
209 std::vector<ContigEntry> contig_records_;
210 std::vector<SparseEntry> sparse_records_;
211 std::unordered_set<uint16_t> closed_cids_;
212 };
213
214 using LedgerPtr = std::unique_ptr<Ledger>;
215
216 } // namespace simdb::argos
217