Austin Schuh | c243b42 | 2020-10-11 15:35:08 -0700 | [diff] [blame] | 1 | #include "aos/events/logging/logfile_utils.h" |
| 2 | |
Austin Schuh | e243aaf | 2020-10-11 15:46:02 -0700 | [diff] [blame] | 3 | #include <chrono> |
| 4 | #include <string> |
Austin Schuh | c243b42 | 2020-10-11 15:35:08 -0700 | [diff] [blame] | 5 | |
Austin Schuh | c41603c | 2020-10-11 16:17:37 -0700 | [diff] [blame] | 6 | #include "aos/events/logging/logfile_sorting.h" |
Austin Schuh | c243b42 | 2020-10-11 15:35:08 -0700 | [diff] [blame] | 7 | #include "aos/events/logging/test_message_generated.h" |
Austin Schuh | 4b5c22a | 2020-11-30 22:58:43 -0800 | [diff] [blame] | 8 | #include "aos/flatbuffer_merge.h" |
Austin Schuh | e243aaf | 2020-10-11 15:46:02 -0700 | [diff] [blame] | 9 | #include "aos/flatbuffers.h" |
Austin Schuh | c243b42 | 2020-10-11 15:35:08 -0700 | [diff] [blame] | 10 | #include "aos/json_to_flatbuffer.h" |
| 11 | #include "aos/testing/tmpdir.h" |
Austin Schuh | e243aaf | 2020-10-11 15:46:02 -0700 | [diff] [blame] | 12 | #include "gtest/gtest.h" |
Austin Schuh | c243b42 | 2020-10-11 15:35:08 -0700 | [diff] [blame] | 13 | |
| 14 | namespace aos { |
| 15 | namespace logger { |
| 16 | namespace testing { |
Austin Schuh | e243aaf | 2020-10-11 15:46:02 -0700 | [diff] [blame] | 17 | namespace chrono = std::chrono; |
Austin Schuh | c243b42 | 2020-10-11 15:35:08 -0700 | [diff] [blame] | 18 | |
Austin Schuh | e243aaf | 2020-10-11 15:46:02 -0700 | [diff] [blame] | 19 | // Creates a size prefixed flatbuffer from json. |
Austin Schuh | c243b42 | 2020-10-11 15:35:08 -0700 | [diff] [blame] | 20 | template <typename T> |
| 21 | SizePrefixedFlatbufferDetachedBuffer<T> JsonToSizedFlatbuffer( |
| 22 | const std::string_view data) { |
| 23 | flatbuffers::FlatBufferBuilder fbb; |
| 24 | fbb.ForceDefaults(true); |
| 25 | fbb.FinishSizePrefixed(JsonToFlatbuffer<T>(data, &fbb)); |
| 26 | return fbb.Release(); |
| 27 | } |
| 28 | |
Austin Schuh | e243aaf | 2020-10-11 15:46:02 -0700 | [diff] [blame] | 29 | // Tests that we can write and read 2 flatbuffers to file. |
Austin Schuh | c243b42 | 2020-10-11 15:35:08 -0700 | [diff] [blame] | 30 | TEST(SpanReaderTest, ReadWrite) { |
| 31 | const std::string logfile = aos::testing::TestTmpDir() + "/log.bfbs"; |
| 32 | unlink(logfile.c_str()); |
| 33 | |
| 34 | const aos::SizePrefixedFlatbufferDetachedBuffer<TestMessage> m1 = |
Austin Schuh | 4b5c22a | 2020-11-30 22:58:43 -0800 | [diff] [blame] | 35 | JsonToSizedFlatbuffer<TestMessage>(R"({ "value": 1 })"); |
Austin Schuh | c243b42 | 2020-10-11 15:35:08 -0700 | [diff] [blame] | 36 | const aos::SizePrefixedFlatbufferDetachedBuffer<TestMessage> m2 = |
Austin Schuh | 4b5c22a | 2020-11-30 22:58:43 -0800 | [diff] [blame] | 37 | JsonToSizedFlatbuffer<TestMessage>(R"({ "value": 2 })"); |
Austin Schuh | c243b42 | 2020-10-11 15:35:08 -0700 | [diff] [blame] | 38 | |
| 39 | { |
| 40 | DetachedBufferWriter writer(logfile, std::make_unique<DummyEncoder>()); |
Austin Schuh | add6eb3 | 2020-11-09 21:24:26 -0800 | [diff] [blame] | 41 | writer.QueueSpan(m1.span()); |
| 42 | writer.QueueSpan(m2.span()); |
Austin Schuh | c243b42 | 2020-10-11 15:35:08 -0700 | [diff] [blame] | 43 | } |
| 44 | |
| 45 | SpanReader reader(logfile); |
| 46 | |
| 47 | EXPECT_EQ(reader.filename(), logfile); |
Austin Schuh | add6eb3 | 2020-11-09 21:24:26 -0800 | [diff] [blame] | 48 | EXPECT_EQ(reader.ReadMessage(), m1.span()); |
| 49 | EXPECT_EQ(reader.ReadMessage(), m2.span()); |
Austin Schuh | c243b42 | 2020-10-11 15:35:08 -0700 | [diff] [blame] | 50 | EXPECT_EQ(reader.ReadMessage(), absl::Span<const uint8_t>()); |
| 51 | } |
| 52 | |
Austin Schuh | e243aaf | 2020-10-11 15:46:02 -0700 | [diff] [blame] | 53 | // Tests that we can actually parse the resulting messages at a basic level |
| 54 | // through MessageReader. |
| 55 | TEST(MessageReaderTest, ReadWrite) { |
| 56 | const std::string logfile = aos::testing::TestTmpDir() + "/log.bfbs"; |
| 57 | unlink(logfile.c_str()); |
| 58 | |
| 59 | const aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> config = |
| 60 | JsonToSizedFlatbuffer<LogFileHeader>( |
| 61 | R"({ "max_out_of_order_duration": 100000000 })"); |
| 62 | const aos::SizePrefixedFlatbufferDetachedBuffer<MessageHeader> m1 = |
| 63 | JsonToSizedFlatbuffer<MessageHeader>( |
| 64 | R"({ "channel_index": 0, "monotonic_sent_time": 1 })"); |
| 65 | const aos::SizePrefixedFlatbufferDetachedBuffer<MessageHeader> m2 = |
| 66 | JsonToSizedFlatbuffer<MessageHeader>( |
| 67 | R"({ "channel_index": 0, "monotonic_sent_time": 2 })"); |
| 68 | |
| 69 | { |
| 70 | DetachedBufferWriter writer(logfile, std::make_unique<DummyEncoder>()); |
Austin Schuh | add6eb3 | 2020-11-09 21:24:26 -0800 | [diff] [blame] | 71 | writer.QueueSpan(config.span()); |
| 72 | writer.QueueSpan(m1.span()); |
| 73 | writer.QueueSpan(m2.span()); |
Austin Schuh | e243aaf | 2020-10-11 15:46:02 -0700 | [diff] [blame] | 74 | } |
| 75 | |
| 76 | MessageReader reader(logfile); |
| 77 | |
| 78 | EXPECT_EQ(reader.filename(), logfile); |
| 79 | |
| 80 | EXPECT_EQ( |
| 81 | reader.max_out_of_order_duration(), |
| 82 | std::chrono::nanoseconds(config.message().max_out_of_order_duration())); |
| 83 | EXPECT_EQ(reader.newest_timestamp(), monotonic_clock::min_time); |
| 84 | EXPECT_TRUE(reader.ReadMessage()); |
| 85 | EXPECT_EQ(reader.newest_timestamp(), |
| 86 | monotonic_clock::time_point(chrono::nanoseconds(1))); |
| 87 | EXPECT_TRUE(reader.ReadMessage()); |
| 88 | EXPECT_EQ(reader.newest_timestamp(), |
| 89 | monotonic_clock::time_point(chrono::nanoseconds(2))); |
| 90 | EXPECT_FALSE(reader.ReadMessage()); |
| 91 | } |
| 92 | |
Austin Schuh | 32f6849 | 2020-11-08 21:45:51 -0800 | [diff] [blame] | 93 | // Tests that we explode when messages are too far out of order. |
| 94 | TEST(PartsMessageReaderDeathTest, TooFarOutOfOrder) { |
| 95 | const std::string logfile0 = aos::testing::TestTmpDir() + "/log0.bfbs"; |
| 96 | unlink(logfile0.c_str()); |
| 97 | |
| 98 | const aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> config0 = |
| 99 | JsonToSizedFlatbuffer<LogFileHeader>( |
| 100 | R"({ |
| 101 | "max_out_of_order_duration": 100000000, |
| 102 | "log_event_uuid": "30ef1283-81d7-4004-8c36-1c162dbcb2b2", |
| 103 | "parts_uuid": "2a05d725-5d5c-4c0b-af42-88de2f3c3876", |
| 104 | "parts_index": 0 |
| 105 | })"); |
| 106 | |
| 107 | const aos::SizePrefixedFlatbufferDetachedBuffer<MessageHeader> m1 = |
| 108 | JsonToSizedFlatbuffer<MessageHeader>( |
| 109 | R"({ "channel_index": 0, "monotonic_sent_time": 100000000 })"); |
| 110 | const aos::SizePrefixedFlatbufferDetachedBuffer<MessageHeader> m2 = |
| 111 | JsonToSizedFlatbuffer<MessageHeader>( |
| 112 | R"({ "channel_index": 0, "monotonic_sent_time": 0 })"); |
| 113 | const aos::SizePrefixedFlatbufferDetachedBuffer<MessageHeader> m3 = |
| 114 | JsonToSizedFlatbuffer<MessageHeader>( |
| 115 | R"({ "channel_index": 0, "monotonic_sent_time": -1 })"); |
| 116 | |
| 117 | { |
| 118 | DetachedBufferWriter writer(logfile0, std::make_unique<DummyEncoder>()); |
Austin Schuh | add6eb3 | 2020-11-09 21:24:26 -0800 | [diff] [blame] | 119 | writer.QueueSpan(config0.span()); |
| 120 | writer.QueueSpan(m1.span()); |
| 121 | writer.QueueSpan(m2.span()); |
| 122 | writer.QueueSpan(m3.span()); |
Austin Schuh | 32f6849 | 2020-11-08 21:45:51 -0800 | [diff] [blame] | 123 | } |
| 124 | |
| 125 | const std::vector<LogFile> parts = SortParts({logfile0}); |
| 126 | |
| 127 | PartsMessageReader reader(parts[0].parts[0]); |
| 128 | |
| 129 | EXPECT_TRUE(reader.ReadMessage()); |
| 130 | EXPECT_TRUE(reader.ReadMessage()); |
| 131 | EXPECT_DEATH({ reader.ReadMessage(); }, "-0.000000001sec vs. 0.000000000sec"); |
| 132 | } |
| 133 | |
Austin Schuh | c41603c | 2020-10-11 16:17:37 -0700 | [diff] [blame] | 134 | // Tests that we can transparently re-assemble part files with a |
| 135 | // PartsMessageReader. |
| 136 | TEST(PartsMessageReaderTest, ReadWrite) { |
| 137 | const std::string logfile0 = aos::testing::TestTmpDir() + "/log0.bfbs"; |
| 138 | const std::string logfile1 = aos::testing::TestTmpDir() + "/log1.bfbs"; |
| 139 | unlink(logfile0.c_str()); |
| 140 | unlink(logfile1.c_str()); |
| 141 | |
| 142 | const aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> config0 = |
| 143 | JsonToSizedFlatbuffer<LogFileHeader>( |
| 144 | R"({ |
| 145 | "max_out_of_order_duration": 100000000, |
| 146 | "log_event_uuid": "30ef1283-81d7-4004-8c36-1c162dbcb2b2", |
| 147 | "parts_uuid": "2a05d725-5d5c-4c0b-af42-88de2f3c3876", |
| 148 | "parts_index": 0 |
| 149 | })"); |
| 150 | const aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> config1 = |
| 151 | JsonToSizedFlatbuffer<LogFileHeader>( |
| 152 | R"({ |
| 153 | "max_out_of_order_duration": 200000000, |
| 154 | "monotonic_start_time": 0, |
| 155 | "realtime_start_time": 0, |
| 156 | "log_event_uuid": "30ef1283-81d7-4004-8c36-1c162dbcb2b2", |
| 157 | "parts_uuid": "2a05d725-5d5c-4c0b-af42-88de2f3c3876", |
| 158 | "parts_index": 1 |
| 159 | })"); |
| 160 | |
| 161 | const aos::SizePrefixedFlatbufferDetachedBuffer<MessageHeader> m1 = |
| 162 | JsonToSizedFlatbuffer<MessageHeader>( |
| 163 | R"({ "channel_index": 0, "monotonic_sent_time": 1 })"); |
| 164 | const aos::SizePrefixedFlatbufferDetachedBuffer<MessageHeader> m2 = |
| 165 | JsonToSizedFlatbuffer<MessageHeader>( |
| 166 | R"({ "channel_index": 0, "monotonic_sent_time": 2 })"); |
| 167 | |
| 168 | { |
| 169 | DetachedBufferWriter writer(logfile0, std::make_unique<DummyEncoder>()); |
Austin Schuh | add6eb3 | 2020-11-09 21:24:26 -0800 | [diff] [blame] | 170 | writer.QueueSpan(config0.span()); |
| 171 | writer.QueueSpan(m1.span()); |
Austin Schuh | c41603c | 2020-10-11 16:17:37 -0700 | [diff] [blame] | 172 | } |
| 173 | { |
| 174 | DetachedBufferWriter writer(logfile1, std::make_unique<DummyEncoder>()); |
Austin Schuh | add6eb3 | 2020-11-09 21:24:26 -0800 | [diff] [blame] | 175 | writer.QueueSpan(config1.span()); |
| 176 | writer.QueueSpan(m2.span()); |
Austin Schuh | c41603c | 2020-10-11 16:17:37 -0700 | [diff] [blame] | 177 | } |
| 178 | |
| 179 | const std::vector<LogFile> parts = SortParts({logfile0, logfile1}); |
| 180 | |
| 181 | PartsMessageReader reader(parts[0].parts[0]); |
| 182 | |
| 183 | EXPECT_EQ(reader.filename(), logfile0); |
| 184 | |
| 185 | // Confirm that the timestamps track, and the filename also updates. |
| 186 | // Read the first message. |
| 187 | EXPECT_EQ(reader.newest_timestamp(), monotonic_clock::min_time); |
| 188 | EXPECT_EQ( |
| 189 | reader.max_out_of_order_duration(), |
| 190 | std::chrono::nanoseconds(config0.message().max_out_of_order_duration())); |
| 191 | EXPECT_TRUE(reader.ReadMessage()); |
| 192 | EXPECT_EQ(reader.filename(), logfile0); |
| 193 | EXPECT_EQ(reader.newest_timestamp(), |
| 194 | monotonic_clock::time_point(chrono::nanoseconds(1))); |
| 195 | EXPECT_EQ( |
| 196 | reader.max_out_of_order_duration(), |
| 197 | std::chrono::nanoseconds(config0.message().max_out_of_order_duration())); |
| 198 | |
| 199 | // Read the second message. |
| 200 | EXPECT_TRUE(reader.ReadMessage()); |
| 201 | EXPECT_EQ(reader.filename(), logfile1); |
| 202 | EXPECT_EQ(reader.newest_timestamp(), |
| 203 | monotonic_clock::time_point(chrono::nanoseconds(2))); |
| 204 | EXPECT_EQ( |
| 205 | reader.max_out_of_order_duration(), |
| 206 | std::chrono::nanoseconds(config1.message().max_out_of_order_duration())); |
| 207 | |
| 208 | // And then confirm that reading again returns no message. |
| 209 | EXPECT_FALSE(reader.ReadMessage()); |
| 210 | EXPECT_EQ(reader.filename(), logfile1); |
| 211 | EXPECT_EQ( |
| 212 | reader.max_out_of_order_duration(), |
| 213 | std::chrono::nanoseconds(config1.message().max_out_of_order_duration())); |
Austin Schuh | 32f6849 | 2020-11-08 21:45:51 -0800 | [diff] [blame] | 214 | EXPECT_EQ(reader.newest_timestamp(), monotonic_clock::max_time); |
Austin Schuh | c41603c | 2020-10-11 16:17:37 -0700 | [diff] [blame] | 215 | } |
Austin Schuh | 32f6849 | 2020-11-08 21:45:51 -0800 | [diff] [blame] | 216 | |
Austin Schuh | 1be0ce4 | 2020-11-29 22:43:26 -0800 | [diff] [blame] | 217 | // Tests that Message's operator < works as expected. |
| 218 | TEST(MessageTest, Sorting) { |
| 219 | const aos::monotonic_clock::time_point e = monotonic_clock::epoch(); |
| 220 | |
| 221 | Message m1{.channel_index = 0, |
| 222 | .queue_index = 0, |
| 223 | .timestamp = e + chrono::milliseconds(1), |
| 224 | .data = SizePrefixedFlatbufferVector<MessageHeader>::Empty()}; |
| 225 | Message m2{.channel_index = 0, |
| 226 | .queue_index = 0, |
| 227 | .timestamp = e + chrono::milliseconds(2), |
| 228 | .data = SizePrefixedFlatbufferVector<MessageHeader>::Empty()}; |
| 229 | |
| 230 | EXPECT_LT(m1, m2); |
| 231 | EXPECT_GE(m2, m1); |
| 232 | |
| 233 | m1.timestamp = e; |
| 234 | m2.timestamp = e; |
| 235 | |
| 236 | m1.channel_index = 1; |
| 237 | m2.channel_index = 2; |
| 238 | |
| 239 | EXPECT_LT(m1, m2); |
| 240 | EXPECT_GE(m2, m1); |
| 241 | |
| 242 | m1.channel_index = 0; |
| 243 | m2.channel_index = 0; |
| 244 | m1.queue_index = 0; |
| 245 | m2.queue_index = 1; |
| 246 | |
| 247 | EXPECT_LT(m1, m2); |
| 248 | EXPECT_GE(m2, m1); |
| 249 | } |
| 250 | |
Austin Schuh | 4b5c22a | 2020-11-30 22:58:43 -0800 | [diff] [blame] | 251 | aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> MakeHeader( |
| 252 | const aos::FlatbufferDetachedBuffer<Configuration> &config, |
| 253 | const std::string_view json) { |
| 254 | flatbuffers::FlatBufferBuilder fbb; |
| 255 | flatbuffers::Offset<Configuration> config_offset = |
| 256 | aos::CopyFlatBuffer(config, &fbb); |
| 257 | LogFileHeader::Builder header_builder(fbb); |
| 258 | header_builder.add_configuration(config_offset); |
| 259 | fbb.Finish(header_builder.Finish()); |
| 260 | aos::FlatbufferDetachedBuffer<LogFileHeader> config_header(fbb.Release()); |
| 261 | |
| 262 | aos::FlatbufferDetachedBuffer<LogFileHeader> header_updates( |
| 263 | JsonToFlatbuffer<LogFileHeader>(json)); |
| 264 | CHECK(header_updates.Verify()); |
| 265 | flatbuffers::FlatBufferBuilder fbb2; |
| 266 | fbb2.FinishSizePrefixed( |
| 267 | aos::MergeFlatBuffers(config_header, header_updates, &fbb2)); |
| 268 | return fbb2.Release(); |
| 269 | } |
| 270 | |
| 271 | class SortingElementTest : public ::testing::Test { |
| 272 | public: |
| 273 | SortingElementTest() |
| 274 | : config_(JsonToFlatbuffer<Configuration>( |
| 275 | R"({ |
| 276 | "channels": [ |
| 277 | { |
| 278 | "name": "/a", |
| 279 | "type": "aos.logger.testing.TestMessage", |
| 280 | "source_node": "pi1", |
| 281 | "destination_nodes": [ |
| 282 | { |
| 283 | "name": "pi2" |
| 284 | }, |
| 285 | { |
| 286 | "name": "pi3" |
| 287 | } |
| 288 | ] |
| 289 | }, |
| 290 | { |
| 291 | "name": "/b", |
| 292 | "type": "aos.logger.testing.TestMessage", |
| 293 | "source_node": "pi1" |
| 294 | }, |
| 295 | { |
| 296 | "name": "/c", |
| 297 | "type": "aos.logger.testing.TestMessage", |
| 298 | "source_node": "pi1" |
| 299 | } |
| 300 | ], |
| 301 | "nodes": [ |
| 302 | { |
| 303 | "name": "pi1" |
| 304 | }, |
| 305 | { |
| 306 | "name": "pi2" |
| 307 | }, |
| 308 | { |
| 309 | "name": "pi3" |
| 310 | } |
| 311 | ] |
| 312 | } |
| 313 | )")), |
| 314 | config0_(MakeHeader(config_, R"({ |
| 315 | /* 100ms */ |
| 316 | "max_out_of_order_duration": 100000000, |
| 317 | "node": { |
| 318 | "name": "pi1" |
| 319 | }, |
| 320 | "logger_node": { |
| 321 | "name": "pi1" |
| 322 | }, |
| 323 | "monotonic_start_time": 0, |
| 324 | "realtime_start_time": 1000000000000, |
| 325 | "log_event_uuid": "30ef1283-81d7-4004-8c36-1c162dbcb2b2", |
| 326 | "parts_uuid": "2a05d725-5d5c-4c0b-af42-88de2f3c3876", |
| 327 | "parts_index": 0 |
| 328 | })")), |
| 329 | config1_(MakeHeader(config_, |
| 330 | R"({ |
| 331 | /* 100ms */ |
| 332 | "max_out_of_order_duration": 100000000, |
| 333 | "node": { |
| 334 | "name": "pi1" |
| 335 | }, |
| 336 | "logger_node": { |
| 337 | "name": "pi1" |
| 338 | }, |
| 339 | "monotonic_start_time": 0, |
| 340 | "realtime_start_time": 1000000000000, |
| 341 | "log_event_uuid": "d4724d35-a6c6-4a30-8a94-d192f4c18260", |
| 342 | "parts_uuid": "bafe9f8e-7dea-4bd9-95f5-3d8390e49208", |
| 343 | "parts_index": 0 |
| 344 | })")) { |
| 345 | unlink(logfile0_.c_str()); |
| 346 | unlink(logfile1_.c_str()); |
| 347 | queue_index_.resize(kChannels); |
| 348 | } |
| 349 | |
| 350 | protected: |
| 351 | static constexpr size_t kChannels = 3u; |
| 352 | |
| 353 | flatbuffers::DetachedBuffer MakeLogMessage( |
| 354 | const aos::monotonic_clock::time_point monotonic_now, int channel_index, |
| 355 | int value) { |
| 356 | flatbuffers::FlatBufferBuilder message_fbb; |
| 357 | message_fbb.ForceDefaults(true); |
| 358 | TestMessage::Builder test_message_builder(message_fbb); |
| 359 | test_message_builder.add_value(value); |
| 360 | message_fbb.Finish(test_message_builder.Finish()); |
| 361 | |
| 362 | aos::Context context; |
| 363 | context.monotonic_event_time = monotonic_now; |
| 364 | context.realtime_event_time = aos::realtime_clock::epoch() + |
| 365 | chrono::seconds(1000) + |
| 366 | monotonic_now.time_since_epoch(); |
| 367 | context.queue_index = queue_index_[channel_index]; |
| 368 | context.size = message_fbb.GetSize(); |
| 369 | context.data = message_fbb.GetBufferPointer(); |
| 370 | |
| 371 | ++queue_index_[channel_index]; |
| 372 | |
| 373 | flatbuffers::FlatBufferBuilder fbb; |
| 374 | fbb.FinishSizePrefixed( |
| 375 | PackMessage(&fbb, context, channel_index, LogType::kLogMessage)); |
| 376 | |
| 377 | return fbb.Release(); |
| 378 | } |
| 379 | |
| 380 | flatbuffers::DetachedBuffer MakeTimestampMessage( |
| 381 | const aos::monotonic_clock::time_point sender_monotonic_now, |
| 382 | int channel_index, chrono::nanoseconds receiver_monotonic_offset) { |
| 383 | aos::Context context; |
| 384 | context.monotonic_remote_time = sender_monotonic_now; |
| 385 | context.realtime_remote_time = aos::realtime_clock::epoch() + |
| 386 | chrono::seconds(1000) + |
| 387 | sender_monotonic_now.time_since_epoch(); |
| 388 | context.remote_queue_index = queue_index_[channel_index] - 1; |
| 389 | context.monotonic_event_time = |
| 390 | sender_monotonic_now + receiver_monotonic_offset; |
| 391 | context.realtime_event_time = |
| 392 | aos::realtime_clock::epoch() + chrono::seconds(1000) + |
| 393 | context.monotonic_event_time.time_since_epoch(); |
| 394 | context.queue_index = queue_index_[channel_index] - 1 + 100; |
| 395 | context.size = 0; |
| 396 | context.data = nullptr; |
| 397 | |
| 398 | flatbuffers::FlatBufferBuilder fbb; |
| 399 | fbb.FinishSizePrefixed(PackMessage(&fbb, context, channel_index, |
| 400 | LogType::kLogDeliveryTimeOnly)); |
| 401 | LOG(INFO) << aos::FlatbufferToJson( |
| 402 | aos::SizePrefixedFlatbufferSpan<MessageHeader>( |
| 403 | absl::Span<uint8_t>(fbb.GetBufferPointer(), fbb.GetSize()))); |
| 404 | |
| 405 | return fbb.Release(); |
| 406 | } |
| 407 | |
| 408 | const std::string logfile0_ = aos::testing::TestTmpDir() + "/log0.bfbs"; |
| 409 | const std::string logfile1_ = aos::testing::TestTmpDir() + "/log1.bfbs"; |
| 410 | |
| 411 | const aos::FlatbufferDetachedBuffer<Configuration> config_; |
| 412 | const aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> config0_; |
| 413 | const aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> config1_; |
| 414 | |
| 415 | std::vector<uint32_t> queue_index_; |
| 416 | }; |
| 417 | |
| 418 | using LogPartsSorterTest = SortingElementTest; |
| 419 | using LogPartsSorterDeathTest = LogPartsSorterTest; |
Austin Schuh | 8f52ed5 | 2020-11-30 23:12:39 -0800 | [diff] [blame] | 420 | using NodeMergerTest = SortingElementTest; |
Austin Schuh | 4b5c22a | 2020-11-30 22:58:43 -0800 | [diff] [blame] | 421 | |
| 422 | // Tests that we can pull messages out of a log sorted in order. |
| 423 | TEST_F(LogPartsSorterTest, Pull) { |
| 424 | const aos::monotonic_clock::time_point e = monotonic_clock::epoch(); |
| 425 | { |
| 426 | DetachedBufferWriter writer(logfile0_, std::make_unique<DummyEncoder>()); |
| 427 | writer.QueueSpan(config0_.span()); |
| 428 | writer.QueueSizedFlatbuffer( |
| 429 | MakeLogMessage(e + chrono::milliseconds(1000), 0, 0x005)); |
| 430 | writer.QueueSizedFlatbuffer( |
| 431 | MakeLogMessage(e + chrono::milliseconds(1000), 1, 0x105)); |
| 432 | writer.QueueSizedFlatbuffer( |
| 433 | MakeLogMessage(e + chrono::milliseconds(2000), 0, 0x006)); |
| 434 | writer.QueueSizedFlatbuffer( |
| 435 | MakeLogMessage(e + chrono::milliseconds(1901), 1, 0x107)); |
| 436 | } |
| 437 | |
| 438 | const std::vector<LogFile> parts = SortParts({logfile0_}); |
| 439 | |
| 440 | LogPartsSorter parts_sorter(parts[0].parts[0]); |
| 441 | |
| 442 | // Confirm we aren't sorted until any time until the message is popped. |
| 443 | // Peeking shouldn't change the sorted until time. |
| 444 | EXPECT_EQ(parts_sorter.sorted_until(), monotonic_clock::min_time); |
| 445 | |
| 446 | std::deque<Message> output; |
| 447 | |
| 448 | ASSERT_TRUE(parts_sorter.Front() != nullptr); |
| 449 | output.emplace_back(std::move(*parts_sorter.Front())); |
| 450 | parts_sorter.PopFront(); |
| 451 | EXPECT_EQ(parts_sorter.sorted_until(), e + chrono::milliseconds(1900)); |
| 452 | |
| 453 | ASSERT_TRUE(parts_sorter.Front() != nullptr); |
| 454 | output.emplace_back(std::move(*parts_sorter.Front())); |
| 455 | parts_sorter.PopFront(); |
| 456 | EXPECT_EQ(parts_sorter.sorted_until(), e + chrono::milliseconds(1900)); |
| 457 | |
| 458 | ASSERT_TRUE(parts_sorter.Front() != nullptr); |
| 459 | output.emplace_back(std::move(*parts_sorter.Front())); |
| 460 | parts_sorter.PopFront(); |
| 461 | EXPECT_EQ(parts_sorter.sorted_until(), monotonic_clock::max_time); |
| 462 | |
| 463 | ASSERT_TRUE(parts_sorter.Front() != nullptr); |
| 464 | output.emplace_back(std::move(*parts_sorter.Front())); |
| 465 | parts_sorter.PopFront(); |
| 466 | EXPECT_EQ(parts_sorter.sorted_until(), monotonic_clock::max_time); |
| 467 | |
| 468 | ASSERT_TRUE(parts_sorter.Front() == nullptr); |
| 469 | |
| 470 | EXPECT_EQ(output[0].timestamp, e + chrono::milliseconds(1000)); |
| 471 | EXPECT_EQ(output[1].timestamp, e + chrono::milliseconds(1000)); |
| 472 | EXPECT_EQ(output[2].timestamp, e + chrono::milliseconds(1901)); |
| 473 | EXPECT_EQ(output[3].timestamp, e + chrono::milliseconds(2000)); |
| 474 | } |
| 475 | |
| 476 | // Tests that messages too far out of order trigger death. |
| 477 | TEST_F(LogPartsSorterDeathTest, Pull) { |
| 478 | const aos::monotonic_clock::time_point e = monotonic_clock::epoch(); |
| 479 | { |
| 480 | DetachedBufferWriter writer(logfile0_, std::make_unique<DummyEncoder>()); |
| 481 | writer.QueueSpan(config0_.span()); |
| 482 | writer.QueueSizedFlatbuffer( |
| 483 | MakeLogMessage(e + chrono::milliseconds(1000), 0, 0x005)); |
| 484 | writer.QueueSizedFlatbuffer( |
| 485 | MakeLogMessage(e + chrono::milliseconds(1000), 1, 0x105)); |
| 486 | writer.QueueSizedFlatbuffer( |
| 487 | MakeLogMessage(e + chrono::milliseconds(2001), 0, 0x006)); |
| 488 | // The following message is too far out of order and will trigger the CHECK. |
| 489 | writer.QueueSizedFlatbuffer( |
| 490 | MakeLogMessage(e + chrono::milliseconds(1900), 1, 0x107)); |
| 491 | } |
| 492 | |
| 493 | const std::vector<LogFile> parts = SortParts({logfile0_}); |
| 494 | |
| 495 | LogPartsSorter parts_sorter(parts[0].parts[0]); |
| 496 | |
| 497 | // Confirm we aren't sorted until any time until the message is popped. |
| 498 | // Peeking shouldn't change the sorted until time. |
| 499 | EXPECT_EQ(parts_sorter.sorted_until(), monotonic_clock::min_time); |
| 500 | std::deque<Message> output; |
| 501 | |
| 502 | ASSERT_TRUE(parts_sorter.Front() != nullptr); |
| 503 | parts_sorter.PopFront(); |
| 504 | ASSERT_TRUE(parts_sorter.Front() != nullptr); |
| 505 | ASSERT_TRUE(parts_sorter.Front() != nullptr); |
| 506 | parts_sorter.PopFront(); |
| 507 | |
| 508 | EXPECT_DEATH({ parts_sorter.Front(); }, "Max out of order exceeded."); |
| 509 | } |
| 510 | |
Austin Schuh | 8f52ed5 | 2020-11-30 23:12:39 -0800 | [diff] [blame] | 511 | // Tests that we can merge data from 2 separate files, including duplicate data. |
| 512 | TEST_F(NodeMergerTest, TwoFileMerger) { |
| 513 | const aos::monotonic_clock::time_point e = monotonic_clock::epoch(); |
| 514 | { |
| 515 | DetachedBufferWriter writer0(logfile0_, std::make_unique<DummyEncoder>()); |
| 516 | writer0.QueueSpan(config0_.span()); |
| 517 | DetachedBufferWriter writer1(logfile1_, std::make_unique<DummyEncoder>()); |
| 518 | writer1.QueueSpan(config1_.span()); |
| 519 | |
| 520 | writer0.QueueSizedFlatbuffer( |
| 521 | MakeLogMessage(e + chrono::milliseconds(1000), 0, 0x005)); |
| 522 | writer1.QueueSizedFlatbuffer( |
| 523 | MakeLogMessage(e + chrono::milliseconds(1001), 1, 0x105)); |
| 524 | |
| 525 | writer0.QueueSizedFlatbuffer( |
| 526 | MakeLogMessage(e + chrono::milliseconds(2000), 0, 0x006)); |
| 527 | writer1.QueueSizedFlatbuffer( |
| 528 | MakeLogMessage(e + chrono::milliseconds(1002), 1, 0x106)); |
| 529 | |
| 530 | // Make a duplicate! |
| 531 | SizePrefixedFlatbufferDetachedBuffer<MessageHeader> msg( |
| 532 | MakeLogMessage(e + chrono::milliseconds(3000), 0, 0x007)); |
| 533 | writer0.QueueSpan(msg.span()); |
| 534 | writer1.QueueSpan(msg.span()); |
| 535 | |
| 536 | writer1.QueueSizedFlatbuffer( |
| 537 | MakeLogMessage(e + chrono::milliseconds(3002), 1, 0x107)); |
| 538 | } |
| 539 | |
| 540 | const std::vector<LogFile> parts = SortParts({logfile0_, logfile1_}); |
| 541 | ASSERT_EQ(parts.size(), 2u); |
| 542 | |
| 543 | std::vector<std::unique_ptr<LogPartsSorter>> parts_sorters; |
| 544 | parts_sorters.emplace_back( |
| 545 | std::make_unique<LogPartsSorter>(parts[0].parts[0])); |
| 546 | parts_sorters.emplace_back( |
| 547 | std::make_unique<LogPartsSorter>(parts[1].parts[0])); |
| 548 | |
| 549 | NodeMerger merger(std::move(parts_sorters)); |
| 550 | |
| 551 | EXPECT_EQ(merger.sorted_until(), monotonic_clock::min_time); |
| 552 | |
| 553 | std::deque<Message> output; |
| 554 | |
| 555 | EXPECT_EQ(merger.sorted_until(), monotonic_clock::min_time); |
| 556 | ASSERT_TRUE(merger.Front() != nullptr); |
| 557 | EXPECT_EQ(merger.sorted_until(), e + chrono::milliseconds(1900)); |
| 558 | |
| 559 | output.emplace_back(std::move(*merger.Front())); |
| 560 | merger.PopFront(); |
| 561 | EXPECT_EQ(merger.sorted_until(), e + chrono::milliseconds(1900)); |
| 562 | |
| 563 | ASSERT_TRUE(merger.Front() != nullptr); |
| 564 | output.emplace_back(std::move(*merger.Front())); |
| 565 | merger.PopFront(); |
| 566 | EXPECT_EQ(merger.sorted_until(), e + chrono::milliseconds(2900)); |
| 567 | |
| 568 | ASSERT_TRUE(merger.Front() != nullptr); |
| 569 | output.emplace_back(std::move(*merger.Front())); |
| 570 | merger.PopFront(); |
| 571 | EXPECT_EQ(merger.sorted_until(), e + chrono::milliseconds(2900)); |
| 572 | |
| 573 | ASSERT_TRUE(merger.Front() != nullptr); |
| 574 | output.emplace_back(std::move(*merger.Front())); |
| 575 | merger.PopFront(); |
| 576 | EXPECT_EQ(merger.sorted_until(), e + chrono::milliseconds(2900)); |
| 577 | |
| 578 | ASSERT_TRUE(merger.Front() != nullptr); |
| 579 | output.emplace_back(std::move(*merger.Front())); |
| 580 | merger.PopFront(); |
| 581 | EXPECT_EQ(merger.sorted_until(), monotonic_clock::max_time); |
| 582 | |
| 583 | ASSERT_TRUE(merger.Front() != nullptr); |
| 584 | output.emplace_back(std::move(*merger.Front())); |
| 585 | merger.PopFront(); |
| 586 | EXPECT_EQ(merger.sorted_until(), monotonic_clock::max_time); |
| 587 | |
| 588 | ASSERT_TRUE(merger.Front() == nullptr); |
| 589 | |
| 590 | EXPECT_EQ(output[0].timestamp, e + chrono::milliseconds(1000)); |
| 591 | EXPECT_EQ(output[1].timestamp, e + chrono::milliseconds(1001)); |
| 592 | EXPECT_EQ(output[2].timestamp, e + chrono::milliseconds(1002)); |
| 593 | EXPECT_EQ(output[3].timestamp, e + chrono::milliseconds(2000)); |
| 594 | EXPECT_EQ(output[4].timestamp, e + chrono::milliseconds(3000)); |
| 595 | EXPECT_EQ(output[5].timestamp, e + chrono::milliseconds(3002)); |
| 596 | } |
| 597 | |
Austin Schuh | c243b42 | 2020-10-11 15:35:08 -0700 | [diff] [blame] | 598 | } // namespace testing |
| 599 | } // namespace logger |
| 600 | } // namespace aos |