blob: 43833fd6787ca4c026cd04f242e6ca9c220004a7 [file] [log] [blame]
Alex Perrycb7da4b2019-08-28 19:35:56 -07001#include "aos/events/simulated_event_loop.h"
2
Austin Schuh5f1cc5c2019-12-01 18:01:11 -08003#include <string_view>
4
Alex Perrycb7da4b2019-08-28 19:35:56 -07005#include "aos/events/event_loop_param_test.h"
Austin Schuh2f8fd752020-09-01 22:38:28 -07006#include "aos/events/logging/logger_generated.h"
Austin Schuh01b4c352020-09-21 23:09:39 -07007#include "aos/events/message_counter.h"
Austin Schuh898f4972020-01-11 17:21:25 -08008#include "aos/events/ping_lib.h"
9#include "aos/events/pong_lib.h"
Austin Schuh7d87b672019-12-01 20:23:49 -080010#include "aos/events/test_message_generated.h"
Austin Schuh4c3b9702020-08-30 11:34:55 -070011#include "aos/network/message_bridge_client_generated.h"
12#include "aos/network/message_bridge_server_generated.h"
Austin Schuh0de30f32020-12-06 12:44:28 -080013#include "aos/network/remote_message_generated.h"
Austin Schuh87dd3832021-01-01 23:07:31 -080014#include "aos/network/testing_time_converter.h"
Austin Schuh4c3b9702020-08-30 11:34:55 -070015#include "aos/network/timestamp_generated.h"
Neil Balchc8f41ed2018-01-20 22:06:53 -080016#include "gtest/gtest.h"
17
18namespace aos {
19namespace testing {
Brian Silverman28d14302020-09-18 15:26:17 -070020namespace {
21
22std::string ConfigPrefix() { return "aos/"; }
23
Austin Schuh0de30f32020-12-06 12:44:28 -080024using message_bridge::RemoteMessage;
Austin Schuh7267c532019-05-19 19:55:53 -070025namespace chrono = ::std::chrono;
26
Austin Schuh0de30f32020-12-06 12:44:28 -080027} // namespace
28
Neil Balchc8f41ed2018-01-20 22:06:53 -080029class SimulatedEventLoopTestFactory : public EventLoopTestFactory {
30 public:
Austin Schuh5f1cc5c2019-12-01 18:01:11 -080031 ::std::unique_ptr<EventLoop> Make(std::string_view name) override {
Austin Schuh217a9782019-12-21 23:02:50 -080032 MaybeMake();
Austin Schuhac0771c2020-01-07 18:36:30 -080033 return event_loop_factory_->MakeEventLoop(name, my_node());
Neil Balchc8f41ed2018-01-20 22:06:53 -080034 }
Austin Schuh5f1cc5c2019-12-01 18:01:11 -080035 ::std::unique_ptr<EventLoop> MakePrimary(std::string_view name) override {
Austin Schuh217a9782019-12-21 23:02:50 -080036 MaybeMake();
Austin Schuhac0771c2020-01-07 18:36:30 -080037 return event_loop_factory_->MakeEventLoop(name, my_node());
Austin Schuh44019f92019-05-19 19:58:27 -070038 }
39
Austin Schuh217a9782019-12-21 23:02:50 -080040 void Run() override { event_loop_factory_->Run(); }
41 void Exit() override { event_loop_factory_->Exit(); }
Austin Schuh44019f92019-05-19 19:58:27 -070042
Austin Schuh52d325c2019-06-23 18:59:06 -070043 // TODO(austin): Implement this. It's used currently for a phased loop test.
44 // I'm not sure how much that matters.
45 void SleepFor(::std::chrono::nanoseconds /*duration*/) override {}
46
Austin Schuh7d87b672019-12-01 20:23:49 -080047 void set_send_delay(std::chrono::nanoseconds send_delay) {
Austin Schuh217a9782019-12-21 23:02:50 -080048 MaybeMake();
49 event_loop_factory_->set_send_delay(send_delay);
Austin Schuh7d87b672019-12-01 20:23:49 -080050 }
51
Neil Balchc8f41ed2018-01-20 22:06:53 -080052 private:
Austin Schuh217a9782019-12-21 23:02:50 -080053 void MaybeMake() {
54 if (!event_loop_factory_) {
55 if (configuration()->has_nodes()) {
Austin Schuhac0771c2020-01-07 18:36:30 -080056 event_loop_factory_ =
57 std::make_unique<SimulatedEventLoopFactory>(configuration());
Austin Schuh217a9782019-12-21 23:02:50 -080058 } else {
59 event_loop_factory_ =
60 std::make_unique<SimulatedEventLoopFactory>(configuration());
61 }
62 }
63 }
64 std::unique_ptr<SimulatedEventLoopFactory> event_loop_factory_;
Neil Balchc8f41ed2018-01-20 22:06:53 -080065};
66
Austin Schuh6bae8252021-02-07 22:01:49 -080067auto CommonParameters() {
68 return ::testing::Combine(
69 ::testing::Values([]() { return new SimulatedEventLoopTestFactory(); }),
70 ::testing::Values(ReadMethod::COPY, ReadMethod::PIN),
71 ::testing::Values(DoTimingReports::kYes, DoTimingReports::kNo));
72}
Austin Schuh6b6dfa52019-06-12 20:16:20 -070073
Austin Schuh6bae8252021-02-07 22:01:49 -080074INSTANTIATE_TEST_CASE_P(SimulatedEventLoopCommonTest, AbstractEventLoopTest,
75 CommonParameters());
Brian Silverman77162972020-08-12 19:52:40 -070076
Austin Schuh6bae8252021-02-07 22:01:49 -080077INSTANTIATE_TEST_CASE_P(SimulatedEventLoopCommonDeathTest,
78 AbstractEventLoopDeathTest, CommonParameters());
Neil Balchc8f41ed2018-01-20 22:06:53 -080079
Austin Schuh89c9b812021-02-20 14:42:10 -080080// Parameters to run all the tests with.
81struct Param {
82 // The config file to use.
83 std::string config;
84 // If true, the RemoteMessage channel should be shared between all the remote
85 // channels. If false, there will be 1 RemoteMessage channel per remote
86 // channel.
87 bool shared;
88};
89
90class RemoteMessageSimulatedEventLoopTest
91 : public ::testing::TestWithParam<struct Param> {
92 public:
93 RemoteMessageSimulatedEventLoopTest()
94 : config(aos::configuration::ReadConfig(
95 absl::StrCat(ConfigPrefix(), "events/", GetParam().config))) {
96 LOG(INFO) << "Config " << GetParam().config;
97 }
98
99 bool shared() const { return GetParam().shared; }
100
101 std::vector<std::unique_ptr<MessageCounter<RemoteMessage>>>
102 MakePi2OnPi1MessageCounters(aos::EventLoop *event_loop) {
103 std::vector<std::unique_ptr<MessageCounter<RemoteMessage>>> counters;
104 if (shared()) {
105 counters.emplace_back(std::make_unique<MessageCounter<RemoteMessage>>(
106 event_loop, "/aos/remote_timestamps/pi2"));
107 } else {
108 counters.emplace_back(std::make_unique<MessageCounter<RemoteMessage>>(
109 event_loop,
110 "/aos/remote_timestamps/pi2/pi1/aos/aos-message_bridge-Timestamp"));
111 counters.emplace_back(std::make_unique<MessageCounter<RemoteMessage>>(
112 event_loop, "/aos/remote_timestamps/pi2/test/aos-examples-Ping"));
113 counters.emplace_back(std::make_unique<MessageCounter<RemoteMessage>>(
114 event_loop, "/aos/remote_timestamps/pi2/reliable/aos-examples-Ping"));
115 }
116 return counters;
117 }
118
119 std::vector<std::unique_ptr<MessageCounter<RemoteMessage>>>
120 MakePi1OnPi2MessageCounters(aos::EventLoop *event_loop) {
121 std::vector<std::unique_ptr<MessageCounter<RemoteMessage>>> counters;
122 if (shared()) {
123 counters.emplace_back(std::make_unique<MessageCounter<RemoteMessage>>(
124 event_loop, "/aos/remote_timestamps/pi1"));
125 } else {
126 counters.emplace_back(std::make_unique<MessageCounter<RemoteMessage>>(
127 event_loop, "/aos/remote_timestamps/pi1/test/aos-examples-Pong"));
128 counters.emplace_back(std::make_unique<MessageCounter<RemoteMessage>>(
129 event_loop,
130 "/aos/remote_timestamps/pi1/pi2/aos/aos-message_bridge-Timestamp"));
131 }
132 return counters;
133 }
134
135 aos::FlatbufferDetachedBuffer<aos::Configuration> config;
136};
137
Neil Balchc8f41ed2018-01-20 22:06:53 -0800138// Test that creating an event and running the scheduler runs the event.
139TEST(EventSchedulerTest, ScheduleEvent) {
140 int counter = 0;
Austin Schuh8bd96322020-02-13 21:18:22 -0800141 EventSchedulerScheduler scheduler_scheduler;
Neil Balchc8f41ed2018-01-20 22:06:53 -0800142 EventScheduler scheduler;
Austin Schuh8bd96322020-02-13 21:18:22 -0800143 scheduler_scheduler.AddEventScheduler(&scheduler);
Neil Balchc8f41ed2018-01-20 22:06:53 -0800144
Austin Schuh8bd96322020-02-13 21:18:22 -0800145 scheduler.Schedule(monotonic_clock::epoch() + chrono::seconds(1),
Austin Schuhac0771c2020-01-07 18:36:30 -0800146 [&counter]() { counter += 1; });
Austin Schuh8bd96322020-02-13 21:18:22 -0800147 scheduler_scheduler.Run();
Neil Balchc8f41ed2018-01-20 22:06:53 -0800148 EXPECT_EQ(counter, 1);
Ravago Jonescf453ab2020-05-06 21:14:53 -0700149 auto token = scheduler.Schedule(monotonic_clock::epoch() + chrono::seconds(2),
150 [&counter]() { counter += 1; });
Neil Balchc8f41ed2018-01-20 22:06:53 -0800151 scheduler.Deschedule(token);
Austin Schuh8bd96322020-02-13 21:18:22 -0800152 scheduler_scheduler.Run();
Neil Balchc8f41ed2018-01-20 22:06:53 -0800153 EXPECT_EQ(counter, 1);
154}
155
156// Test that descheduling an already scheduled event doesn't run the event.
157TEST(EventSchedulerTest, DescheduleEvent) {
158 int counter = 0;
Austin Schuh8bd96322020-02-13 21:18:22 -0800159 EventSchedulerScheduler scheduler_scheduler;
Neil Balchc8f41ed2018-01-20 22:06:53 -0800160 EventScheduler scheduler;
Austin Schuh8bd96322020-02-13 21:18:22 -0800161 scheduler_scheduler.AddEventScheduler(&scheduler);
Neil Balchc8f41ed2018-01-20 22:06:53 -0800162
Austin Schuh8bd96322020-02-13 21:18:22 -0800163 auto token = scheduler.Schedule(monotonic_clock::epoch() + chrono::seconds(1),
164 [&counter]() { counter += 1; });
Neil Balchc8f41ed2018-01-20 22:06:53 -0800165 scheduler.Deschedule(token);
Austin Schuh8bd96322020-02-13 21:18:22 -0800166 scheduler_scheduler.Run();
Neil Balchc8f41ed2018-01-20 22:06:53 -0800167 EXPECT_EQ(counter, 0);
168}
Austin Schuh44019f92019-05-19 19:58:27 -0700169
Austin Schuh8fb315a2020-11-19 22:33:58 -0800170void SendTestMessage(aos::Sender<TestMessage> *sender, int value) {
171 aos::Sender<TestMessage>::Builder builder = sender->MakeBuilder();
172 TestMessage::Builder test_message_builder =
173 builder.MakeBuilder<TestMessage>();
174 test_message_builder.add_value(value);
175 builder.Send(test_message_builder.Finish());
176}
177
178// Test that sending a message after running gets properly notified.
179TEST(SimulatedEventLoopTest, SendAfterRunFor) {
180 SimulatedEventLoopTestFactory factory;
181
182 SimulatedEventLoopFactory simulated_event_loop_factory(
183 factory.configuration());
184
185 ::std::unique_ptr<EventLoop> ping_event_loop =
186 simulated_event_loop_factory.MakeEventLoop("ping");
187 aos::Sender<TestMessage> test_message_sender =
188 ping_event_loop->MakeSender<TestMessage>("/test");
189 SendTestMessage(&test_message_sender, 1);
190
191 std::unique_ptr<EventLoop> pong1_event_loop =
192 simulated_event_loop_factory.MakeEventLoop("pong");
193 MessageCounter<TestMessage> test_message_counter1(pong1_event_loop.get(),
194 "/test");
195
196 EXPECT_FALSE(ping_event_loop->is_running());
197
198 // Watchers start when you start running, so there should be nothing counted.
199 simulated_event_loop_factory.RunFor(chrono::seconds(1));
200 EXPECT_EQ(test_message_counter1.count(), 0u);
201
202 std::unique_ptr<EventLoop> pong2_event_loop =
203 simulated_event_loop_factory.MakeEventLoop("pong");
204 MessageCounter<TestMessage> test_message_counter2(pong2_event_loop.get(),
205 "/test");
206
207 // Pauses in the middle don't count though, so this should be counted.
208 // But, the fresh watcher shouldn't pick it up yet.
209 SendTestMessage(&test_message_sender, 2);
210
211 EXPECT_EQ(test_message_counter1.count(), 0u);
212 EXPECT_EQ(test_message_counter2.count(), 0u);
213 simulated_event_loop_factory.RunFor(chrono::seconds(1));
214
215 EXPECT_EQ(test_message_counter1.count(), 1u);
216 EXPECT_EQ(test_message_counter2.count(), 0u);
217}
218
219// Test that creating an event loop while running dies.
220TEST(SimulatedEventLoopDeathTest, MakeEventLoopWhileRunning) {
221 SimulatedEventLoopTestFactory factory;
222
223 SimulatedEventLoopFactory simulated_event_loop_factory(
224 factory.configuration());
225
226 ::std::unique_ptr<EventLoop> event_loop =
227 simulated_event_loop_factory.MakeEventLoop("ping");
228
229 auto timer = event_loop->AddTimer([&]() {
230 EXPECT_DEATH(
231 {
232 ::std::unique_ptr<EventLoop> event_loop2 =
233 simulated_event_loop_factory.MakeEventLoop("ping");
234 },
235 "event loop while running");
236 simulated_event_loop_factory.Exit();
237 });
238
239 event_loop->OnRun([&event_loop, &timer] {
240 timer->Setup(event_loop->monotonic_now() + chrono::milliseconds(50));
241 });
242
243 simulated_event_loop_factory.Run();
244}
245
246// Test that creating a watcher after running dies.
247TEST(SimulatedEventLoopDeathTest, MakeWatcherAfterRunning) {
248 SimulatedEventLoopTestFactory factory;
249
250 SimulatedEventLoopFactory simulated_event_loop_factory(
251 factory.configuration());
252
253 ::std::unique_ptr<EventLoop> event_loop =
254 simulated_event_loop_factory.MakeEventLoop("ping");
255
256 simulated_event_loop_factory.RunFor(chrono::seconds(1));
257
258 EXPECT_DEATH(
259 { MessageCounter<TestMessage> counter(event_loop.get(), "/test"); },
260 "Can't add a watcher after running");
261
262 ::std::unique_ptr<EventLoop> event_loop2 =
263 simulated_event_loop_factory.MakeEventLoop("ping");
264
265 simulated_event_loop_factory.RunFor(chrono::seconds(1));
266
267 EXPECT_DEATH(
268 { MessageCounter<TestMessage> counter(event_loop2.get(), "/test"); },
269 "Can't add a watcher after running");
270}
271
Austin Schuh44019f92019-05-19 19:58:27 -0700272// Test that running for a time period with no handlers causes time to progress
273// correctly.
274TEST(SimulatedEventLoopTest, RunForNoHandlers) {
Austin Schuh39788ff2019-12-01 18:22:57 -0800275 SimulatedEventLoopTestFactory factory;
276
277 SimulatedEventLoopFactory simulated_event_loop_factory(
278 factory.configuration());
Austin Schuh44019f92019-05-19 19:58:27 -0700279 ::std::unique_ptr<EventLoop> event_loop =
Austin Schuh5f1cc5c2019-12-01 18:01:11 -0800280 simulated_event_loop_factory.MakeEventLoop("loop");
Austin Schuh44019f92019-05-19 19:58:27 -0700281
282 simulated_event_loop_factory.RunFor(chrono::seconds(1));
283
284 EXPECT_EQ(::aos::monotonic_clock::epoch() + chrono::seconds(1),
Austin Schuh44019f92019-05-19 19:58:27 -0700285 event_loop->monotonic_now());
286}
287
288// Test that running for a time with a periodic handler causes time to end
289// correctly.
290TEST(SimulatedEventLoopTest, RunForTimerHandler) {
Austin Schuh39788ff2019-12-01 18:22:57 -0800291 SimulatedEventLoopTestFactory factory;
292
293 SimulatedEventLoopFactory simulated_event_loop_factory(
294 factory.configuration());
Austin Schuh44019f92019-05-19 19:58:27 -0700295 ::std::unique_ptr<EventLoop> event_loop =
Austin Schuh5f1cc5c2019-12-01 18:01:11 -0800296 simulated_event_loop_factory.MakeEventLoop("loop");
Austin Schuh44019f92019-05-19 19:58:27 -0700297
298 int counter = 0;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700299 auto timer = event_loop->AddTimer([&counter]() { ++counter; });
Austin Schuh44019f92019-05-19 19:58:27 -0700300 event_loop->OnRun([&event_loop, &timer] {
301 timer->Setup(event_loop->monotonic_now() + chrono::milliseconds(50),
302 chrono::milliseconds(100));
303 });
304
305 simulated_event_loop_factory.RunFor(chrono::seconds(1));
306
307 EXPECT_EQ(::aos::monotonic_clock::epoch() + chrono::seconds(1),
Austin Schuh44019f92019-05-19 19:58:27 -0700308 event_loop->monotonic_now());
309 EXPECT_EQ(counter, 10);
310}
311
Austin Schuh7d87b672019-12-01 20:23:49 -0800312// Tests that watchers have latency in simulation.
313TEST(SimulatedEventLoopTest, WatcherTimingReport) {
314 SimulatedEventLoopTestFactory factory;
315 factory.set_send_delay(std::chrono::microseconds(50));
316
317 FLAGS_timing_report_ms = 1000;
318 auto loop1 = factory.MakePrimary("primary");
319 loop1->MakeWatcher("/test", [](const TestMessage &) {});
320
321 auto loop2 = factory.Make("sender_loop");
322
323 auto loop3 = factory.Make("report_fetcher");
324
325 Fetcher<timing::Report> report_fetcher =
326 loop3->MakeFetcher<timing::Report>("/aos");
327 EXPECT_FALSE(report_fetcher.Fetch());
328
329 auto sender = loop2->MakeSender<TestMessage>("/test");
330
331 // Send 10 messages in the middle of a timing report period so we get
332 // something interesting back.
333 auto test_timer = loop2->AddTimer([&sender]() {
334 for (int i = 0; i < 10; ++i) {
335 aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
336 TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
337 builder.add_value(200 + i);
338 ASSERT_TRUE(msg.Send(builder.Finish()));
339 }
340 });
341
342 // Quit after 1 timing report, mid way through the next cycle.
343 {
344 auto end_timer = loop1->AddTimer([&factory]() { factory.Exit(); });
345 end_timer->Setup(loop1->monotonic_now() + chrono::milliseconds(2500));
346 end_timer->set_name("end");
347 }
348
349 loop1->OnRun([&test_timer, &loop1]() {
350 test_timer->Setup(loop1->monotonic_now() + chrono::milliseconds(1500));
351 });
352
353 factory.Run();
354
355 // And, since we are here, check that the timing report makes sense.
356 // Start by looking for our event loop's timing.
357 FlatbufferDetachedBuffer<timing::Report> primary_report =
358 FlatbufferDetachedBuffer<timing::Report>::Empty();
359 while (report_fetcher.FetchNext()) {
360 LOG(INFO) << "Report " << FlatbufferToJson(report_fetcher.get());
361 if (report_fetcher->name()->string_view() == "primary") {
362 primary_report = CopyFlatBuffer(report_fetcher.get());
363 }
364 }
365
366 // Check the watcher report.
Ravago Jonescf453ab2020-05-06 21:14:53 -0700367 VLOG(1) << FlatbufferToJson(primary_report, {.multi_line = true});
Austin Schuh7d87b672019-12-01 20:23:49 -0800368
369 EXPECT_EQ(primary_report.message().name()->string_view(), "primary");
370
371 // Just the timing report timer.
372 ASSERT_NE(primary_report.message().timers(), nullptr);
373 EXPECT_EQ(primary_report.message().timers()->size(), 2);
374
375 // No phased loops
376 ASSERT_EQ(primary_report.message().phased_loops(), nullptr);
377
378 // And now confirm that the watcher received all 10 messages, and has latency.
379 ASSERT_NE(primary_report.message().watchers(), nullptr);
380 ASSERT_EQ(primary_report.message().watchers()->size(), 1);
381 EXPECT_EQ(primary_report.message().watchers()->Get(0)->count(), 10);
382 EXPECT_NEAR(
383 primary_report.message().watchers()->Get(0)->wakeup_latency()->average(),
384 0.00005, 1e-9);
385 EXPECT_NEAR(
386 primary_report.message().watchers()->Get(0)->wakeup_latency()->min(),
387 0.00005, 1e-9);
388 EXPECT_NEAR(
389 primary_report.message().watchers()->Get(0)->wakeup_latency()->max(),
390 0.00005, 1e-9);
391 EXPECT_EQ(primary_report.message()
392 .watchers()
393 ->Get(0)
394 ->wakeup_latency()
395 ->standard_deviation(),
396 0.0);
397
398 EXPECT_EQ(
399 primary_report.message().watchers()->Get(0)->handler_time()->average(),
400 0.0);
401 EXPECT_EQ(primary_report.message().watchers()->Get(0)->handler_time()->min(),
402 0.0);
403 EXPECT_EQ(primary_report.message().watchers()->Get(0)->handler_time()->max(),
404 0.0);
405 EXPECT_EQ(primary_report.message()
406 .watchers()
407 ->Get(0)
408 ->handler_time()
409 ->standard_deviation(),
410 0.0);
411}
412
Austin Schuh89c9b812021-02-20 14:42:10 -0800413size_t CountAll(
414 const std::vector<std::unique_ptr<MessageCounter<RemoteMessage>>>
415 &counters) {
416 size_t count = 0u;
417 for (const std::unique_ptr<MessageCounter<RemoteMessage>> &counter :
418 counters) {
419 count += counter->count();
420 }
421 return count;
422}
423
Austin Schuh4c3b9702020-08-30 11:34:55 -0700424// Tests that ping and pong work when on 2 different nodes, and the message
425// gateway messages are sent out as expected.
Austin Schuh89c9b812021-02-20 14:42:10 -0800426TEST_P(RemoteMessageSimulatedEventLoopTest, MultinodePingPong) {
Austin Schuh898f4972020-01-11 17:21:25 -0800427 const Node *pi1 = configuration::GetNode(&config.message(), "pi1");
428 const Node *pi2 = configuration::GetNode(&config.message(), "pi2");
Austin Schuh4c3b9702020-08-30 11:34:55 -0700429 const Node *pi3 = configuration::GetNode(&config.message(), "pi3");
Austin Schuh898f4972020-01-11 17:21:25 -0800430
431 SimulatedEventLoopFactory simulated_event_loop_factory(&config.message());
432
433 std::unique_ptr<EventLoop> ping_event_loop =
434 simulated_event_loop_factory.MakeEventLoop("ping", pi1);
435 Ping ping(ping_event_loop.get());
436
437 std::unique_ptr<EventLoop> pong_event_loop =
438 simulated_event_loop_factory.MakeEventLoop("pong", pi2);
439 Pong pong(pong_event_loop.get());
440
441 std::unique_ptr<EventLoop> pi2_pong_counter_event_loop =
442 simulated_event_loop_factory.MakeEventLoop("pi2_pong_counter", pi2);
Austin Schuh4c3b9702020-08-30 11:34:55 -0700443 MessageCounter<examples::Pong> pi2_pong_counter(
444 pi2_pong_counter_event_loop.get(), "/test");
Austin Schuh2f8fd752020-09-01 22:38:28 -0700445 aos::Fetcher<message_bridge::Timestamp> pi1_on_pi2_timestamp_fetcher =
446 pi2_pong_counter_event_loop->MakeFetcher<message_bridge::Timestamp>(
447 "/pi1/aos");
448 aos::Fetcher<examples::Ping> ping_on_pi2_fetcher =
449 pi2_pong_counter_event_loop->MakeFetcher<examples::Ping>("/test");
Austin Schuh898f4972020-01-11 17:21:25 -0800450
Austin Schuh4c3b9702020-08-30 11:34:55 -0700451 std::unique_ptr<EventLoop> pi3_pong_counter_event_loop =
452 simulated_event_loop_factory.MakeEventLoop("pi3_pong_counter", pi3);
Austin Schuh898f4972020-01-11 17:21:25 -0800453
454 std::unique_ptr<EventLoop> pi1_pong_counter_event_loop =
455 simulated_event_loop_factory.MakeEventLoop("pi1_pong_counter", pi1);
Austin Schuh4c3b9702020-08-30 11:34:55 -0700456 MessageCounter<examples::Pong> pi1_pong_counter(
457 pi1_pong_counter_event_loop.get(), "/test");
Austin Schuh2f8fd752020-09-01 22:38:28 -0700458 aos::Fetcher<examples::Ping> ping_on_pi1_fetcher =
459 pi1_pong_counter_event_loop->MakeFetcher<examples::Ping>("/test");
460 aos::Fetcher<message_bridge::Timestamp> pi1_on_pi1_timestamp_fetcher =
461 pi1_pong_counter_event_loop->MakeFetcher<message_bridge::Timestamp>(
462 "/aos");
463
Austin Schuh4c3b9702020-08-30 11:34:55 -0700464 // Count timestamps.
465 MessageCounter<message_bridge::Timestamp> pi1_on_pi1_timestamp_counter(
466 pi1_pong_counter_event_loop.get(), "/pi1/aos");
467 MessageCounter<message_bridge::Timestamp> pi1_on_pi2_timestamp_counter(
468 pi2_pong_counter_event_loop.get(), "/pi1/aos");
469 MessageCounter<message_bridge::Timestamp> pi1_on_pi3_timestamp_counter(
470 pi3_pong_counter_event_loop.get(), "/pi1/aos");
471 MessageCounter<message_bridge::Timestamp> pi2_on_pi1_timestamp_counter(
472 pi1_pong_counter_event_loop.get(), "/pi2/aos");
473 MessageCounter<message_bridge::Timestamp> pi2_on_pi2_timestamp_counter(
474 pi2_pong_counter_event_loop.get(), "/pi2/aos");
475 MessageCounter<message_bridge::Timestamp> pi3_on_pi1_timestamp_counter(
476 pi1_pong_counter_event_loop.get(), "/pi3/aos");
477 MessageCounter<message_bridge::Timestamp> pi3_on_pi3_timestamp_counter(
478 pi3_pong_counter_event_loop.get(), "/pi3/aos");
479
Austin Schuh2f8fd752020-09-01 22:38:28 -0700480 // Count remote timestamps
Austin Schuh89c9b812021-02-20 14:42:10 -0800481 std::vector<std::unique_ptr<MessageCounter<RemoteMessage>>>
482 remote_timestamps_pi2_on_pi1 =
483 MakePi2OnPi1MessageCounters(pi1_pong_counter_event_loop.get());
484 std::vector<std::unique_ptr<MessageCounter<RemoteMessage>>>
485 remote_timestamps_pi1_on_pi2 =
486 MakePi1OnPi2MessageCounters(pi2_pong_counter_event_loop.get());
Austin Schuh2f8fd752020-09-01 22:38:28 -0700487
Austin Schuh4c3b9702020-08-30 11:34:55 -0700488 // Wait to let timestamp estimation start up before looking for the results.
489 simulated_event_loop_factory.RunFor(chrono::milliseconds(500));
490
Austin Schuh8fb315a2020-11-19 22:33:58 -0800491 std::unique_ptr<EventLoop> pi1_statistics_counter_event_loop =
492 simulated_event_loop_factory.MakeEventLoop("pi1_statistics_counter", pi1);
493 std::unique_ptr<EventLoop> pi2_statistics_counter_event_loop =
494 simulated_event_loop_factory.MakeEventLoop("pi2_statistics_counter", pi2);
495 std::unique_ptr<EventLoop> pi3_statistics_counter_event_loop =
496 simulated_event_loop_factory.MakeEventLoop("pi3_statistics_counter", pi3);
497
Austin Schuh4c3b9702020-08-30 11:34:55 -0700498 int pi1_server_statistics_count = 0;
Austin Schuh8fb315a2020-11-19 22:33:58 -0800499 pi1_statistics_counter_event_loop->MakeWatcher(
Austin Schuh4c3b9702020-08-30 11:34:55 -0700500 "/pi1/aos", [&pi1_server_statistics_count](
501 const message_bridge::ServerStatistics &stats) {
502 VLOG(1) << "pi1 ServerStatistics " << FlatbufferToJson(&stats);
503 EXPECT_EQ(stats.connections()->size(), 2u);
504 for (const message_bridge::ServerConnection *connection :
505 *stats.connections()) {
506 EXPECT_EQ(connection->state(), message_bridge::State::CONNECTED);
Austin Schuh20ac95d2020-12-05 17:24:19 -0800507 EXPECT_TRUE(connection->has_boot_uuid());
Austin Schuh4c3b9702020-08-30 11:34:55 -0700508 if (connection->node()->name()->string_view() == "pi2") {
509 EXPECT_GT(connection->sent_packets(), 50);
510 } else if (connection->node()->name()->string_view() == "pi3") {
511 EXPECT_GE(connection->sent_packets(), 5);
512 } else {
513 LOG(FATAL) << "Unknown connection";
514 }
515
516 EXPECT_TRUE(connection->has_monotonic_offset());
517 EXPECT_EQ(connection->monotonic_offset(), 0);
518 }
519 ++pi1_server_statistics_count;
520 });
521
522 int pi2_server_statistics_count = 0;
Austin Schuh8fb315a2020-11-19 22:33:58 -0800523 pi2_statistics_counter_event_loop->MakeWatcher(
Austin Schuh4c3b9702020-08-30 11:34:55 -0700524 "/pi2/aos", [&pi2_server_statistics_count](
525 const message_bridge::ServerStatistics &stats) {
526 VLOG(1) << "pi2 ServerStatistics " << FlatbufferToJson(&stats);
527 EXPECT_EQ(stats.connections()->size(), 1u);
528
529 const message_bridge::ServerConnection *connection =
530 stats.connections()->Get(0);
Austin Schuh20ac95d2020-12-05 17:24:19 -0800531 EXPECT_TRUE(connection->has_boot_uuid());
Austin Schuh4c3b9702020-08-30 11:34:55 -0700532 EXPECT_EQ(connection->state(), message_bridge::State::CONNECTED);
533 EXPECT_GT(connection->sent_packets(), 50);
534 EXPECT_TRUE(connection->has_monotonic_offset());
535 EXPECT_EQ(connection->monotonic_offset(), 0);
536 ++pi2_server_statistics_count;
537 });
538
539 int pi3_server_statistics_count = 0;
Austin Schuh8fb315a2020-11-19 22:33:58 -0800540 pi3_statistics_counter_event_loop->MakeWatcher(
Austin Schuh4c3b9702020-08-30 11:34:55 -0700541 "/pi3/aos", [&pi3_server_statistics_count](
542 const message_bridge::ServerStatistics &stats) {
543 VLOG(1) << "pi3 ServerStatistics " << FlatbufferToJson(&stats);
544 EXPECT_EQ(stats.connections()->size(), 1u);
545
546 const message_bridge::ServerConnection *connection =
547 stats.connections()->Get(0);
Austin Schuh20ac95d2020-12-05 17:24:19 -0800548 EXPECT_TRUE(connection->has_boot_uuid());
Austin Schuh4c3b9702020-08-30 11:34:55 -0700549 EXPECT_EQ(connection->state(), message_bridge::State::CONNECTED);
550 EXPECT_GE(connection->sent_packets(), 5);
551 EXPECT_TRUE(connection->has_monotonic_offset());
552 EXPECT_EQ(connection->monotonic_offset(), 0);
553 ++pi3_server_statistics_count;
554 });
555
556 int pi1_client_statistics_count = 0;
Austin Schuh8fb315a2020-11-19 22:33:58 -0800557 pi1_statistics_counter_event_loop->MakeWatcher(
Austin Schuh4c3b9702020-08-30 11:34:55 -0700558 "/pi1/aos", [&pi1_client_statistics_count](
559 const message_bridge::ClientStatistics &stats) {
560 VLOG(1) << "pi1 ClientStatistics " << FlatbufferToJson(&stats);
561 EXPECT_EQ(stats.connections()->size(), 2u);
562
563 for (const message_bridge::ClientConnection *connection :
564 *stats.connections()) {
565 EXPECT_EQ(connection->state(), message_bridge::State::CONNECTED);
566 if (connection->node()->name()->string_view() == "pi2") {
567 EXPECT_GT(connection->received_packets(), 50);
568 } else if (connection->node()->name()->string_view() == "pi3") {
569 EXPECT_GE(connection->received_packets(), 5);
570 } else {
571 LOG(FATAL) << "Unknown connection";
572 }
573
574 EXPECT_TRUE(connection->has_monotonic_offset());
575 EXPECT_EQ(connection->monotonic_offset(), 150000);
576 }
577 ++pi1_client_statistics_count;
578 });
579
580 int pi2_client_statistics_count = 0;
Austin Schuh8fb315a2020-11-19 22:33:58 -0800581 pi2_statistics_counter_event_loop->MakeWatcher(
Austin Schuh4c3b9702020-08-30 11:34:55 -0700582 "/pi2/aos", [&pi2_client_statistics_count](
583 const message_bridge::ClientStatistics &stats) {
584 VLOG(1) << "pi2 ClientStatistics " << FlatbufferToJson(&stats);
585 EXPECT_EQ(stats.connections()->size(), 1u);
586
587 const message_bridge::ClientConnection *connection =
588 stats.connections()->Get(0);
589 EXPECT_EQ(connection->state(), message_bridge::State::CONNECTED);
590 EXPECT_GT(connection->received_packets(), 50);
591 EXPECT_TRUE(connection->has_monotonic_offset());
592 EXPECT_EQ(connection->monotonic_offset(), 150000);
593 ++pi2_client_statistics_count;
594 });
595
596 int pi3_client_statistics_count = 0;
Austin Schuh8fb315a2020-11-19 22:33:58 -0800597 pi3_statistics_counter_event_loop->MakeWatcher(
Austin Schuh4c3b9702020-08-30 11:34:55 -0700598 "/pi3/aos", [&pi3_client_statistics_count](
599 const message_bridge::ClientStatistics &stats) {
600 VLOG(1) << "pi3 ClientStatistics " << FlatbufferToJson(&stats);
601 EXPECT_EQ(stats.connections()->size(), 1u);
602
603 const message_bridge::ClientConnection *connection =
604 stats.connections()->Get(0);
605 EXPECT_EQ(connection->state(), message_bridge::State::CONNECTED);
606 EXPECT_GE(connection->received_packets(), 5);
607 EXPECT_TRUE(connection->has_monotonic_offset());
608 EXPECT_EQ(connection->monotonic_offset(), 150000);
609 ++pi3_client_statistics_count;
610 });
611
Austin Schuh2f8fd752020-09-01 22:38:28 -0700612 // Find the channel index for both the /pi1/aos Timestamp channel and Ping
613 // channel.
614 const size_t pi1_timestamp_channel =
615 configuration::ChannelIndex(pi1_pong_counter_event_loop->configuration(),
616 pi1_on_pi2_timestamp_fetcher.channel());
617 const size_t ping_timestamp_channel =
618 configuration::ChannelIndex(pi1_pong_counter_event_loop->configuration(),
619 ping_on_pi2_fetcher.channel());
620
621 for (const Channel *channel :
622 *pi1_pong_counter_event_loop->configuration()->channels()) {
623 VLOG(1) << "Channel "
624 << configuration::ChannelIndex(
625 pi1_pong_counter_event_loop->configuration(), channel)
626 << " " << configuration::CleanedChannelToString(channel);
627 }
628
Austin Schuh8fb315a2020-11-19 22:33:58 -0800629 std::unique_ptr<EventLoop> pi1_remote_timestamp =
630 simulated_event_loop_factory.MakeEventLoop("pi1_remote_timestamp", pi1);
631
Austin Schuh89c9b812021-02-20 14:42:10 -0800632 for (std::pair<int, std::string> channel :
633 shared()
634 ? std::vector<std::pair<
635 int, std::string>>{{-1, "/pi1/aos/remote_timestamps/pi2"}}
636 : std::vector<std::pair<int, std::string>>{
637 {pi1_timestamp_channel,
638 "/pi1/aos/remote_timestamps/pi2/pi1/aos/"
639 "aos-message_bridge-Timestamp"},
640 {ping_timestamp_channel,
641 "/pi1/aos/remote_timestamps/pi2/test/aos-examples-Ping"}}) {
642 // For each remote timestamp we get back, confirm that it is either a ping
643 // message, or a timestamp we sent out. Also confirm that the timestamps
644 // are correct.
645 pi1_remote_timestamp->MakeWatcher(
646 channel.second,
647 [pi1_timestamp_channel, ping_timestamp_channel, &ping_on_pi2_fetcher,
648 &ping_on_pi1_fetcher, &pi1_on_pi2_timestamp_fetcher,
649 &pi1_on_pi1_timestamp_fetcher, &simulated_event_loop_factory, pi2,
650 channel_index = channel.first](const RemoteMessage &header) {
651 VLOG(1) << aos::FlatbufferToJson(&header);
652 EXPECT_TRUE(header.has_boot_uuid());
653 EXPECT_EQ(header.boot_uuid()->string_view(),
654 simulated_event_loop_factory.GetNodeEventLoopFactory(pi2)
655 ->boot_uuid()
656 .string_view());
Austin Schuh2f8fd752020-09-01 22:38:28 -0700657
Austin Schuh89c9b812021-02-20 14:42:10 -0800658 const aos::monotonic_clock::time_point header_monotonic_sent_time(
659 chrono::nanoseconds(header.monotonic_sent_time()));
660 const aos::realtime_clock::time_point header_realtime_sent_time(
661 chrono::nanoseconds(header.realtime_sent_time()));
662 const aos::monotonic_clock::time_point header_monotonic_remote_time(
663 chrono::nanoseconds(header.monotonic_remote_time()));
664 const aos::realtime_clock::time_point header_realtime_remote_time(
665 chrono::nanoseconds(header.realtime_remote_time()));
Austin Schuh2f8fd752020-09-01 22:38:28 -0700666
Austin Schuh89c9b812021-02-20 14:42:10 -0800667 if (channel_index != -1) {
668 ASSERT_EQ(channel_index, header.channel_index());
Austin Schuh2f8fd752020-09-01 22:38:28 -0700669 }
670
Austin Schuh89c9b812021-02-20 14:42:10 -0800671 const Context *pi1_context = nullptr;
672 const Context *pi2_context = nullptr;
673
674 if (header.channel_index() == pi1_timestamp_channel) {
675 // Find the forwarded message.
676 while (pi1_on_pi2_timestamp_fetcher.context().monotonic_event_time <
677 header_monotonic_sent_time) {
678 ASSERT_TRUE(pi1_on_pi2_timestamp_fetcher.FetchNext());
679 }
680
681 // And the source message.
682 while (pi1_on_pi1_timestamp_fetcher.context().monotonic_event_time <
683 header_monotonic_remote_time) {
684 ASSERT_TRUE(pi1_on_pi1_timestamp_fetcher.FetchNext());
685 }
686
687 pi1_context = &pi1_on_pi1_timestamp_fetcher.context();
688 pi2_context = &pi1_on_pi2_timestamp_fetcher.context();
689 } else if (header.channel_index() == ping_timestamp_channel) {
690 // Find the forwarded message.
691 while (ping_on_pi2_fetcher.context().monotonic_event_time <
692 header_monotonic_sent_time) {
693 ASSERT_TRUE(ping_on_pi2_fetcher.FetchNext());
694 }
695
696 // And the source message.
697 while (ping_on_pi1_fetcher.context().monotonic_event_time <
698 header_monotonic_remote_time) {
699 ASSERT_TRUE(ping_on_pi1_fetcher.FetchNext());
700 }
701
702 pi1_context = &ping_on_pi1_fetcher.context();
703 pi2_context = &ping_on_pi2_fetcher.context();
704 } else {
705 LOG(FATAL) << "Unknown channel";
Austin Schuh2f8fd752020-09-01 22:38:28 -0700706 }
707
Austin Schuh89c9b812021-02-20 14:42:10 -0800708 // Confirm the forwarded message has matching timestamps to the
709 // timestamps we got back.
710 EXPECT_EQ(pi2_context->queue_index, header.queue_index());
711 EXPECT_EQ(pi2_context->remote_queue_index,
712 header.remote_queue_index());
713 EXPECT_EQ(pi2_context->monotonic_event_time,
714 header_monotonic_sent_time);
715 EXPECT_EQ(pi2_context->realtime_event_time,
716 header_realtime_sent_time);
717 EXPECT_EQ(pi2_context->realtime_remote_time,
718 header_realtime_remote_time);
719 EXPECT_EQ(pi2_context->monotonic_remote_time,
720 header_monotonic_remote_time);
Austin Schuh2f8fd752020-09-01 22:38:28 -0700721
Austin Schuh89c9b812021-02-20 14:42:10 -0800722 // Confirm the forwarded message also matches the source message.
723 EXPECT_EQ(pi1_context->queue_index, header.remote_queue_index());
724 EXPECT_EQ(pi1_context->monotonic_event_time,
725 header_monotonic_remote_time);
726 EXPECT_EQ(pi1_context->realtime_event_time,
727 header_realtime_remote_time);
728 });
729 }
Austin Schuh2f8fd752020-09-01 22:38:28 -0700730
Austin Schuh4c3b9702020-08-30 11:34:55 -0700731 simulated_event_loop_factory.RunFor(chrono::seconds(10) -
732 chrono::milliseconds(500) +
733 chrono::milliseconds(5));
734
735 EXPECT_EQ(pi1_pong_counter.count(), 1001);
736 EXPECT_EQ(pi2_pong_counter.count(), 1001);
737
738 EXPECT_EQ(pi1_on_pi1_timestamp_counter.count(), 100);
739 EXPECT_EQ(pi1_on_pi2_timestamp_counter.count(), 100);
740 EXPECT_EQ(pi1_on_pi3_timestamp_counter.count(), 100);
741 EXPECT_EQ(pi2_on_pi1_timestamp_counter.count(), 100);
742 EXPECT_EQ(pi2_on_pi2_timestamp_counter.count(), 100);
743 EXPECT_EQ(pi3_on_pi1_timestamp_counter.count(), 100);
744 EXPECT_EQ(pi3_on_pi3_timestamp_counter.count(), 100);
745
Austin Schuh20ac95d2020-12-05 17:24:19 -0800746 EXPECT_EQ(pi1_server_statistics_count, 10);
747 EXPECT_EQ(pi2_server_statistics_count, 10);
748 EXPECT_EQ(pi3_server_statistics_count, 10);
Austin Schuh4c3b9702020-08-30 11:34:55 -0700749
750 EXPECT_EQ(pi1_client_statistics_count, 95);
751 EXPECT_EQ(pi2_client_statistics_count, 95);
752 EXPECT_EQ(pi3_client_statistics_count, 95);
Austin Schuh2f8fd752020-09-01 22:38:28 -0700753
754 // Also confirm that remote timestamps are being forwarded correctly.
Austin Schuh89c9b812021-02-20 14:42:10 -0800755 EXPECT_EQ(CountAll(remote_timestamps_pi2_on_pi1), 1101);
756 EXPECT_EQ(CountAll(remote_timestamps_pi1_on_pi2), 1101);
Austin Schuh4c3b9702020-08-30 11:34:55 -0700757}
758
759// Tests that an offset between nodes can be recovered and shows up in
760// ServerStatistics correctly.
761TEST(SimulatedEventLoopTest, MultinodePingPongWithOffset) {
762 aos::FlatbufferDetachedBuffer<aos::Configuration> config =
Austin Schuh89c9b812021-02-20 14:42:10 -0800763 aos::configuration::ReadConfig(
764 ConfigPrefix() +
765 "events/multinode_pingpong_test_combined_config.json");
Austin Schuh4c3b9702020-08-30 11:34:55 -0700766 const Node *pi1 = configuration::GetNode(&config.message(), "pi1");
Austin Schuh87dd3832021-01-01 23:07:31 -0800767 const size_t pi1_index = configuration::GetNodeIndex(&config.message(), pi1);
768 ASSERT_EQ(pi1_index, 0u);
Austin Schuh4c3b9702020-08-30 11:34:55 -0700769 const Node *pi2 = configuration::GetNode(&config.message(), "pi2");
Austin Schuh87dd3832021-01-01 23:07:31 -0800770 const size_t pi2_index = configuration::GetNodeIndex(&config.message(), pi2);
771 ASSERT_EQ(pi2_index, 1u);
Austin Schuh4c3b9702020-08-30 11:34:55 -0700772 const Node *pi3 = configuration::GetNode(&config.message(), "pi3");
Austin Schuh87dd3832021-01-01 23:07:31 -0800773 const size_t pi3_index = configuration::GetNodeIndex(&config.message(), pi3);
774 ASSERT_EQ(pi3_index, 2u);
Austin Schuh4c3b9702020-08-30 11:34:55 -0700775
Austin Schuh87dd3832021-01-01 23:07:31 -0800776 message_bridge::TestingTimeConverter time(
777 configuration::NodesCount(&config.message()));
Austin Schuh4c3b9702020-08-30 11:34:55 -0700778 SimulatedEventLoopFactory simulated_event_loop_factory(&config.message());
779 NodeEventLoopFactory *pi2_factory =
780 simulated_event_loop_factory.GetNodeEventLoopFactory(pi2);
Austin Schuh87dd3832021-01-01 23:07:31 -0800781 pi2_factory->SetTimeConverter(&time);
Austin Schuh4c3b9702020-08-30 11:34:55 -0700782
783 constexpr chrono::milliseconds kOffset{1501};
Austin Schuh87dd3832021-01-01 23:07:31 -0800784 time.AddNextTimestamp(
785 distributed_clock::epoch(),
786 {monotonic_clock::epoch(), monotonic_clock::epoch() + kOffset,
787 monotonic_clock::epoch()});
Austin Schuh4c3b9702020-08-30 11:34:55 -0700788
789 std::unique_ptr<EventLoop> ping_event_loop =
790 simulated_event_loop_factory.MakeEventLoop("ping", pi1);
791 Ping ping(ping_event_loop.get());
792
793 std::unique_ptr<EventLoop> pong_event_loop =
794 simulated_event_loop_factory.MakeEventLoop("pong", pi2);
795 Pong pong(pong_event_loop.get());
796
Austin Schuh8fb315a2020-11-19 22:33:58 -0800797 // Wait to let timestamp estimation start up before looking for the results.
798 simulated_event_loop_factory.RunFor(chrono::milliseconds(500));
799
Austin Schuh87dd3832021-01-01 23:07:31 -0800800 std::unique_ptr<EventLoop> pi1_pong_counter_event_loop =
801 simulated_event_loop_factory.MakeEventLoop("pi1_pong_counter", pi1);
802
Austin Schuh4c3b9702020-08-30 11:34:55 -0700803 std::unique_ptr<EventLoop> pi2_pong_counter_event_loop =
804 simulated_event_loop_factory.MakeEventLoop("pi2_pong_counter", pi2);
805
806 std::unique_ptr<EventLoop> pi3_pong_counter_event_loop =
807 simulated_event_loop_factory.MakeEventLoop("pi3_pong_counter", pi3);
808
Austin Schuh4c3b9702020-08-30 11:34:55 -0700809 // Confirm the offsets are being recovered correctly.
810 int pi1_server_statistics_count = 0;
811 pi1_pong_counter_event_loop->MakeWatcher(
812 "/pi1/aos", [&pi1_server_statistics_count,
813 kOffset](const message_bridge::ServerStatistics &stats) {
814 VLOG(1) << "pi1 ServerStatistics " << FlatbufferToJson(&stats);
815 EXPECT_EQ(stats.connections()->size(), 2u);
816 for (const message_bridge::ServerConnection *connection :
817 *stats.connections()) {
818 EXPECT_EQ(connection->state(), message_bridge::State::CONNECTED);
Austin Schuh20ac95d2020-12-05 17:24:19 -0800819 EXPECT_TRUE(connection->has_boot_uuid());
Austin Schuh4c3b9702020-08-30 11:34:55 -0700820 if (connection->node()->name()->string_view() == "pi2") {
821 EXPECT_EQ(connection->monotonic_offset(),
822 chrono::nanoseconds(kOffset).count());
823 } else if (connection->node()->name()->string_view() == "pi3") {
824 EXPECT_EQ(connection->monotonic_offset(), 0);
825 } else {
826 LOG(FATAL) << "Unknown connection";
827 }
828
829 EXPECT_TRUE(connection->has_monotonic_offset());
830 }
831 ++pi1_server_statistics_count;
832 });
833
834 int pi2_server_statistics_count = 0;
835 pi2_pong_counter_event_loop->MakeWatcher(
836 "/pi2/aos", [&pi2_server_statistics_count,
837 kOffset](const message_bridge::ServerStatistics &stats) {
838 VLOG(1) << "pi2 ServerStatistics " << FlatbufferToJson(&stats);
839 EXPECT_EQ(stats.connections()->size(), 1u);
840
841 const message_bridge::ServerConnection *connection =
842 stats.connections()->Get(0);
Austin Schuh20ac95d2020-12-05 17:24:19 -0800843 EXPECT_TRUE(connection->has_boot_uuid());
Austin Schuh4c3b9702020-08-30 11:34:55 -0700844 EXPECT_EQ(connection->state(), message_bridge::State::CONNECTED);
845 EXPECT_TRUE(connection->has_monotonic_offset());
846 EXPECT_EQ(connection->monotonic_offset(),
847 -chrono::nanoseconds(kOffset).count());
848 ++pi2_server_statistics_count;
849 });
850
851 int pi3_server_statistics_count = 0;
852 pi3_pong_counter_event_loop->MakeWatcher(
853 "/pi3/aos", [&pi3_server_statistics_count](
854 const message_bridge::ServerStatistics &stats) {
855 VLOG(1) << "pi3 ServerStatistics " << FlatbufferToJson(&stats);
856 EXPECT_EQ(stats.connections()->size(), 1u);
857
858 const message_bridge::ServerConnection *connection =
859 stats.connections()->Get(0);
Austin Schuh20ac95d2020-12-05 17:24:19 -0800860 EXPECT_TRUE(connection->has_boot_uuid());
Austin Schuh4c3b9702020-08-30 11:34:55 -0700861 EXPECT_EQ(connection->state(), message_bridge::State::CONNECTED);
862 EXPECT_TRUE(connection->has_monotonic_offset());
863 EXPECT_EQ(connection->monotonic_offset(), 0);
864 ++pi3_server_statistics_count;
865 });
866
867 simulated_event_loop_factory.RunFor(chrono::seconds(10) -
868 chrono::milliseconds(500) +
869 chrono::milliseconds(5));
870
Austin Schuh20ac95d2020-12-05 17:24:19 -0800871 EXPECT_EQ(pi1_server_statistics_count, 10);
Austin Schuh4c3b9702020-08-30 11:34:55 -0700872 EXPECT_EQ(pi2_server_statistics_count, 9);
Austin Schuh20ac95d2020-12-05 17:24:19 -0800873 EXPECT_EQ(pi3_server_statistics_count, 10);
Austin Schuh4c3b9702020-08-30 11:34:55 -0700874}
875
876// Test that disabling statistics actually disables them.
Austin Schuh89c9b812021-02-20 14:42:10 -0800877TEST_P(RemoteMessageSimulatedEventLoopTest, MultinodeWithoutStatistics) {
Austin Schuh4c3b9702020-08-30 11:34:55 -0700878 const Node *pi1 = configuration::GetNode(&config.message(), "pi1");
879 const Node *pi2 = configuration::GetNode(&config.message(), "pi2");
880 const Node *pi3 = configuration::GetNode(&config.message(), "pi3");
881
882 SimulatedEventLoopFactory simulated_event_loop_factory(&config.message());
883 simulated_event_loop_factory.DisableStatistics();
884
885 std::unique_ptr<EventLoop> ping_event_loop =
886 simulated_event_loop_factory.MakeEventLoop("ping", pi1);
887 Ping ping(ping_event_loop.get());
888
889 std::unique_ptr<EventLoop> pong_event_loop =
890 simulated_event_loop_factory.MakeEventLoop("pong", pi2);
891 Pong pong(pong_event_loop.get());
892
893 std::unique_ptr<EventLoop> pi2_pong_counter_event_loop =
894 simulated_event_loop_factory.MakeEventLoop("pi2_pong_counter", pi2);
895
896 MessageCounter<examples::Pong> pi2_pong_counter(
897 pi2_pong_counter_event_loop.get(), "/test");
898
899 std::unique_ptr<EventLoop> pi3_pong_counter_event_loop =
900 simulated_event_loop_factory.MakeEventLoop("pi3_pong_counter", pi3);
901
902 std::unique_ptr<EventLoop> pi1_pong_counter_event_loop =
903 simulated_event_loop_factory.MakeEventLoop("pi1_pong_counter", pi1);
904
905 MessageCounter<examples::Pong> pi1_pong_counter(
906 pi1_pong_counter_event_loop.get(), "/test");
907
908 // Count timestamps.
909 MessageCounter<message_bridge::Timestamp> pi1_on_pi1_timestamp_counter(
910 pi1_pong_counter_event_loop.get(), "/pi1/aos");
911 MessageCounter<message_bridge::Timestamp> pi1_on_pi2_timestamp_counter(
912 pi2_pong_counter_event_loop.get(), "/pi1/aos");
913 MessageCounter<message_bridge::Timestamp> pi1_on_pi3_timestamp_counter(
914 pi3_pong_counter_event_loop.get(), "/pi1/aos");
915 MessageCounter<message_bridge::Timestamp> pi2_on_pi1_timestamp_counter(
916 pi1_pong_counter_event_loop.get(), "/pi2/aos");
917 MessageCounter<message_bridge::Timestamp> pi2_on_pi2_timestamp_counter(
918 pi2_pong_counter_event_loop.get(), "/pi2/aos");
919 MessageCounter<message_bridge::Timestamp> pi3_on_pi1_timestamp_counter(
920 pi1_pong_counter_event_loop.get(), "/pi3/aos");
921 MessageCounter<message_bridge::Timestamp> pi3_on_pi3_timestamp_counter(
922 pi3_pong_counter_event_loop.get(), "/pi3/aos");
923
Austin Schuh2f8fd752020-09-01 22:38:28 -0700924 // Count remote timestamps
Austin Schuh89c9b812021-02-20 14:42:10 -0800925 std::vector<std::unique_ptr<MessageCounter<RemoteMessage>>>
926 remote_timestamps_pi2_on_pi1 =
927 MakePi2OnPi1MessageCounters(pi1_pong_counter_event_loop.get());
928 std::vector<std::unique_ptr<MessageCounter<RemoteMessage>>>
929 remote_timestamps_pi1_on_pi2 =
930 MakePi1OnPi2MessageCounters(pi2_pong_counter_event_loop.get());
Austin Schuh2f8fd752020-09-01 22:38:28 -0700931
Austin Schuh4c3b9702020-08-30 11:34:55 -0700932 MessageCounter<message_bridge::ServerStatistics>
933 pi1_server_statistics_counter(pi1_pong_counter_event_loop.get(),
934 "/pi1/aos");
935 MessageCounter<message_bridge::ServerStatistics>
936 pi2_server_statistics_counter(pi2_pong_counter_event_loop.get(),
937 "/pi2/aos");
938 MessageCounter<message_bridge::ServerStatistics>
939 pi3_server_statistics_counter(pi3_pong_counter_event_loop.get(),
940 "/pi3/aos");
941
942 MessageCounter<message_bridge::ClientStatistics>
943 pi1_client_statistics_counter(pi1_pong_counter_event_loop.get(),
944 "/pi1/aos");
945 MessageCounter<message_bridge::ClientStatistics>
946 pi2_client_statistics_counter(pi2_pong_counter_event_loop.get(),
947 "/pi2/aos");
948 MessageCounter<message_bridge::ClientStatistics>
949 pi3_client_statistics_counter(pi3_pong_counter_event_loop.get(),
950 "/pi3/aos");
Austin Schuh898f4972020-01-11 17:21:25 -0800951
952 simulated_event_loop_factory.RunFor(chrono::seconds(10) +
953 chrono::milliseconds(5));
954
Austin Schuh4c3b9702020-08-30 11:34:55 -0700955 EXPECT_EQ(pi1_pong_counter.count(), 1001u);
956 EXPECT_EQ(pi2_pong_counter.count(), 1001u);
957
958 EXPECT_EQ(pi1_on_pi1_timestamp_counter.count(), 0u);
959 EXPECT_EQ(pi1_on_pi2_timestamp_counter.count(), 0u);
960 EXPECT_EQ(pi1_on_pi3_timestamp_counter.count(), 0u);
961 EXPECT_EQ(pi2_on_pi1_timestamp_counter.count(), 0u);
962 EXPECT_EQ(pi2_on_pi2_timestamp_counter.count(), 0u);
963 EXPECT_EQ(pi3_on_pi1_timestamp_counter.count(), 0u);
964 EXPECT_EQ(pi3_on_pi3_timestamp_counter.count(), 0u);
965
966 EXPECT_EQ(pi1_server_statistics_counter.count(), 0u);
967 EXPECT_EQ(pi2_server_statistics_counter.count(), 0u);
968 EXPECT_EQ(pi3_server_statistics_counter.count(), 0u);
969
970 EXPECT_EQ(pi1_client_statistics_counter.count(), 0u);
971 EXPECT_EQ(pi2_client_statistics_counter.count(), 0u);
972 EXPECT_EQ(pi3_client_statistics_counter.count(), 0u);
Austin Schuh2f8fd752020-09-01 22:38:28 -0700973
974 // Also confirm that remote timestamps are being forwarded correctly.
Austin Schuh89c9b812021-02-20 14:42:10 -0800975 EXPECT_EQ(CountAll(remote_timestamps_pi2_on_pi1), 1001);
976 EXPECT_EQ(CountAll(remote_timestamps_pi1_on_pi2), 1001);
Austin Schuh898f4972020-01-11 17:21:25 -0800977}
978
Austin Schuhc0b0f722020-12-12 18:36:06 -0800979bool AllConnected(const message_bridge::ServerStatistics *server_statistics) {
980 for (const message_bridge::ServerConnection *connection :
981 *server_statistics->connections()) {
982 if (connection->state() != message_bridge::State::CONNECTED) {
983 return false;
984 }
985 }
986 return true;
987}
988
989bool AllConnectedBut(const message_bridge::ServerStatistics *server_statistics,
990 std::string_view target) {
991 for (const message_bridge::ServerConnection *connection :
992 *server_statistics->connections()) {
993 if (connection->node()->name()->string_view() == target) {
994 if (connection->state() == message_bridge::State::CONNECTED) {
995 return false;
996 }
997 } else {
998 if (connection->state() != message_bridge::State::CONNECTED) {
999 return false;
1000 }
1001 }
1002 }
1003 return true;
1004}
1005
1006bool AllConnected(const message_bridge::ClientStatistics *client_statistics) {
1007 for (const message_bridge::ClientConnection *connection :
1008 *client_statistics->connections()) {
1009 if (connection->state() != message_bridge::State::CONNECTED) {
1010 return false;
1011 }
1012 }
1013 return true;
1014}
1015
1016bool AllConnectedBut(const message_bridge::ClientStatistics *client_statistics,
1017 std::string_view target) {
1018 for (const message_bridge::ClientConnection *connection :
1019 *client_statistics->connections()) {
1020 if (connection->node()->name()->string_view() == target) {
1021 if (connection->state() == message_bridge::State::CONNECTED) {
1022 return false;
1023 }
1024 } else {
1025 if (connection->state() != message_bridge::State::CONNECTED) {
1026 return false;
1027 }
1028 }
1029 }
1030 return true;
1031}
1032
1033// Test that disconnecting nodes actually disconnects them.
Austin Schuh89c9b812021-02-20 14:42:10 -08001034TEST_P(RemoteMessageSimulatedEventLoopTest, MultinodeDisconnect) {
Austin Schuhc0b0f722020-12-12 18:36:06 -08001035 const Node *pi1 = configuration::GetNode(&config.message(), "pi1");
1036 const Node *pi2 = configuration::GetNode(&config.message(), "pi2");
1037 const Node *pi3 = configuration::GetNode(&config.message(), "pi3");
1038
1039 SimulatedEventLoopFactory simulated_event_loop_factory(&config.message());
1040
1041 std::unique_ptr<EventLoop> ping_event_loop =
1042 simulated_event_loop_factory.MakeEventLoop("ping", pi1);
1043 Ping ping(ping_event_loop.get());
1044
1045 std::unique_ptr<EventLoop> pong_event_loop =
1046 simulated_event_loop_factory.MakeEventLoop("pong", pi2);
1047 Pong pong(pong_event_loop.get());
1048
1049 std::unique_ptr<EventLoop> pi2_pong_counter_event_loop =
1050 simulated_event_loop_factory.MakeEventLoop("pi2_pong_counter", pi2);
1051
1052 MessageCounter<examples::Pong> pi2_pong_counter(
1053 pi2_pong_counter_event_loop.get(), "/test");
1054
1055 std::unique_ptr<EventLoop> pi3_pong_counter_event_loop =
1056 simulated_event_loop_factory.MakeEventLoop("pi3_pong_counter", pi3);
1057
1058 std::unique_ptr<EventLoop> pi1_pong_counter_event_loop =
1059 simulated_event_loop_factory.MakeEventLoop("pi1_pong_counter", pi1);
1060
1061 MessageCounter<examples::Pong> pi1_pong_counter(
1062 pi1_pong_counter_event_loop.get(), "/test");
1063
1064 // Count timestamps.
1065 MessageCounter<message_bridge::Timestamp> pi1_on_pi1_timestamp_counter(
1066 pi1_pong_counter_event_loop.get(), "/pi1/aos");
1067 MessageCounter<message_bridge::Timestamp> pi1_on_pi2_timestamp_counter(
1068 pi2_pong_counter_event_loop.get(), "/pi1/aos");
1069 MessageCounter<message_bridge::Timestamp> pi1_on_pi3_timestamp_counter(
1070 pi3_pong_counter_event_loop.get(), "/pi1/aos");
1071 MessageCounter<message_bridge::Timestamp> pi2_on_pi1_timestamp_counter(
1072 pi1_pong_counter_event_loop.get(), "/pi2/aos");
1073 MessageCounter<message_bridge::Timestamp> pi2_on_pi2_timestamp_counter(
1074 pi2_pong_counter_event_loop.get(), "/pi2/aos");
1075 MessageCounter<message_bridge::Timestamp> pi3_on_pi1_timestamp_counter(
1076 pi1_pong_counter_event_loop.get(), "/pi3/aos");
1077 MessageCounter<message_bridge::Timestamp> pi3_on_pi3_timestamp_counter(
1078 pi3_pong_counter_event_loop.get(), "/pi3/aos");
1079
1080 // Count remote timestamps
Austin Schuh89c9b812021-02-20 14:42:10 -08001081 std::vector<std::unique_ptr<MessageCounter<RemoteMessage>>>
1082 remote_timestamps_pi2_on_pi1 =
1083 MakePi2OnPi1MessageCounters(pi1_pong_counter_event_loop.get());
1084 std::vector<std::unique_ptr<MessageCounter<RemoteMessage>>>
1085 remote_timestamps_pi1_on_pi2 =
1086 MakePi1OnPi2MessageCounters(pi2_pong_counter_event_loop.get());
Austin Schuhc0b0f722020-12-12 18:36:06 -08001087
1088 MessageCounter<message_bridge::ServerStatistics>
1089 pi1_server_statistics_counter(pi1_pong_counter_event_loop.get(),
1090 "/pi1/aos");
1091 aos::Fetcher<message_bridge::ServerStatistics> pi1_server_statistics_fetcher =
1092 pi1_pong_counter_event_loop
1093 ->MakeFetcher<message_bridge::ServerStatistics>("/pi1/aos");
1094 aos::Fetcher<message_bridge::ClientStatistics> pi1_client_statistics_fetcher =
1095 pi1_pong_counter_event_loop
1096 ->MakeFetcher<message_bridge::ClientStatistics>("/pi1/aos");
1097
1098 MessageCounter<message_bridge::ServerStatistics>
1099 pi2_server_statistics_counter(pi2_pong_counter_event_loop.get(),
1100 "/pi2/aos");
1101 aos::Fetcher<message_bridge::ServerStatistics> pi2_server_statistics_fetcher =
1102 pi2_pong_counter_event_loop
1103 ->MakeFetcher<message_bridge::ServerStatistics>("/pi2/aos");
1104 aos::Fetcher<message_bridge::ClientStatistics> pi2_client_statistics_fetcher =
1105 pi2_pong_counter_event_loop
1106 ->MakeFetcher<message_bridge::ClientStatistics>("/pi2/aos");
1107
1108 MessageCounter<message_bridge::ServerStatistics>
1109 pi3_server_statistics_counter(pi3_pong_counter_event_loop.get(),
1110 "/pi3/aos");
1111 aos::Fetcher<message_bridge::ServerStatistics> pi3_server_statistics_fetcher =
1112 pi3_pong_counter_event_loop
1113 ->MakeFetcher<message_bridge::ServerStatistics>("/pi3/aos");
1114 aos::Fetcher<message_bridge::ClientStatistics> pi3_client_statistics_fetcher =
1115 pi3_pong_counter_event_loop
1116 ->MakeFetcher<message_bridge::ClientStatistics>("/pi3/aos");
1117
1118 MessageCounter<message_bridge::ClientStatistics>
1119 pi1_client_statistics_counter(pi1_pong_counter_event_loop.get(),
1120 "/pi1/aos");
1121 MessageCounter<message_bridge::ClientStatistics>
1122 pi2_client_statistics_counter(pi2_pong_counter_event_loop.get(),
1123 "/pi2/aos");
1124 MessageCounter<message_bridge::ClientStatistics>
1125 pi3_client_statistics_counter(pi3_pong_counter_event_loop.get(),
1126 "/pi3/aos");
1127
1128 simulated_event_loop_factory.RunFor(chrono::seconds(2) +
1129 chrono::milliseconds(5));
1130
1131 EXPECT_EQ(pi1_pong_counter.count(), 201u);
1132 EXPECT_EQ(pi2_pong_counter.count(), 201u);
1133
1134 EXPECT_EQ(pi1_on_pi1_timestamp_counter.count(), 20u);
1135 EXPECT_EQ(pi1_on_pi2_timestamp_counter.count(), 20u);
1136 EXPECT_EQ(pi1_on_pi3_timestamp_counter.count(), 20u);
1137 EXPECT_EQ(pi2_on_pi1_timestamp_counter.count(), 20u);
1138 EXPECT_EQ(pi2_on_pi2_timestamp_counter.count(), 20u);
1139 EXPECT_EQ(pi3_on_pi1_timestamp_counter.count(), 20u);
1140 EXPECT_EQ(pi3_on_pi3_timestamp_counter.count(), 20u);
1141
1142 EXPECT_EQ(pi1_server_statistics_counter.count(), 2u);
1143 EXPECT_EQ(pi2_server_statistics_counter.count(), 2u);
1144 EXPECT_EQ(pi3_server_statistics_counter.count(), 2u);
1145
1146 EXPECT_EQ(pi1_client_statistics_counter.count(), 20u);
1147 EXPECT_EQ(pi2_client_statistics_counter.count(), 20u);
1148 EXPECT_EQ(pi3_client_statistics_counter.count(), 20u);
1149
1150 // Also confirm that remote timestamps are being forwarded correctly.
Austin Schuh89c9b812021-02-20 14:42:10 -08001151 EXPECT_EQ(CountAll(remote_timestamps_pi2_on_pi1), 221);
1152 EXPECT_EQ(CountAll(remote_timestamps_pi1_on_pi2), 221);
Austin Schuhc0b0f722020-12-12 18:36:06 -08001153
1154 EXPECT_TRUE(pi1_server_statistics_fetcher.Fetch());
1155 EXPECT_TRUE(AllConnected(pi1_server_statistics_fetcher.get()))
1156 << " : " << aos::FlatbufferToJson(pi1_server_statistics_fetcher.get());
1157 EXPECT_TRUE(pi1_client_statistics_fetcher.Fetch());
1158 EXPECT_TRUE(AllConnected(pi1_client_statistics_fetcher.get()))
1159 << " : " << aos::FlatbufferToJson(pi1_client_statistics_fetcher.get());
1160 EXPECT_TRUE(pi2_server_statistics_fetcher.Fetch());
1161 EXPECT_TRUE(AllConnected(pi2_server_statistics_fetcher.get()))
1162 << " : " << aos::FlatbufferToJson(pi2_server_statistics_fetcher.get());
1163 EXPECT_TRUE(pi2_client_statistics_fetcher.Fetch());
1164 EXPECT_TRUE(AllConnected(pi2_client_statistics_fetcher.get()))
1165 << " : " << aos::FlatbufferToJson(pi2_client_statistics_fetcher.get());
1166 EXPECT_TRUE(pi3_server_statistics_fetcher.Fetch());
1167 EXPECT_TRUE(AllConnected(pi3_server_statistics_fetcher.get()))
1168 << " : " << aos::FlatbufferToJson(pi3_server_statistics_fetcher.get());
1169 EXPECT_TRUE(pi3_client_statistics_fetcher.Fetch());
1170 EXPECT_TRUE(AllConnected(pi3_client_statistics_fetcher.get()))
1171 << " : " << aos::FlatbufferToJson(pi3_client_statistics_fetcher.get());
1172
1173 simulated_event_loop_factory.GetNodeEventLoopFactory(pi1)->Disconnect(pi3);
1174
1175 simulated_event_loop_factory.RunFor(chrono::seconds(2));
1176
1177 EXPECT_EQ(pi1_pong_counter.count(), 401u);
1178 EXPECT_EQ(pi2_pong_counter.count(), 401u);
1179
1180 EXPECT_EQ(pi1_on_pi1_timestamp_counter.count(), 40u);
1181 EXPECT_EQ(pi1_on_pi2_timestamp_counter.count(), 40u);
1182 EXPECT_EQ(pi1_on_pi3_timestamp_counter.count(), 20u);
1183 EXPECT_EQ(pi2_on_pi1_timestamp_counter.count(), 40u);
1184 EXPECT_EQ(pi2_on_pi2_timestamp_counter.count(), 40u);
1185 EXPECT_EQ(pi3_on_pi1_timestamp_counter.count(), 40u);
1186 EXPECT_EQ(pi3_on_pi3_timestamp_counter.count(), 40u);
1187
1188 EXPECT_EQ(pi1_server_statistics_counter.count(), 4u);
1189 EXPECT_EQ(pi2_server_statistics_counter.count(), 4u);
1190 EXPECT_EQ(pi3_server_statistics_counter.count(), 4u);
1191
1192 EXPECT_EQ(pi1_client_statistics_counter.count(), 40u);
1193 EXPECT_EQ(pi2_client_statistics_counter.count(), 40u);
1194 EXPECT_EQ(pi3_client_statistics_counter.count(), 40u);
1195
1196 // Also confirm that remote timestamps are being forwarded correctly.
Austin Schuh89c9b812021-02-20 14:42:10 -08001197 EXPECT_EQ(CountAll(remote_timestamps_pi2_on_pi1), 441);
1198 EXPECT_EQ(CountAll(remote_timestamps_pi1_on_pi2), 441);
Austin Schuhc0b0f722020-12-12 18:36:06 -08001199
1200 EXPECT_TRUE(pi1_server_statistics_fetcher.Fetch());
1201 EXPECT_TRUE(AllConnectedBut(pi1_server_statistics_fetcher.get(), "pi3"))
1202 << " : " << aos::FlatbufferToJson(pi1_server_statistics_fetcher.get());
1203 EXPECT_TRUE(pi1_client_statistics_fetcher.Fetch());
1204 EXPECT_TRUE(AllConnected(pi1_client_statistics_fetcher.get()))
1205 << " : " << aos::FlatbufferToJson(pi1_client_statistics_fetcher.get());
1206 EXPECT_TRUE(pi2_server_statistics_fetcher.Fetch());
1207 EXPECT_TRUE(AllConnected(pi2_server_statistics_fetcher.get()))
1208 << " : " << aos::FlatbufferToJson(pi2_server_statistics_fetcher.get());
1209 EXPECT_TRUE(pi2_client_statistics_fetcher.Fetch());
1210 EXPECT_TRUE(AllConnected(pi2_client_statistics_fetcher.get()))
1211 << " : " << aos::FlatbufferToJson(pi2_client_statistics_fetcher.get());
1212 EXPECT_TRUE(pi3_server_statistics_fetcher.Fetch());
1213 EXPECT_TRUE(AllConnected(pi3_server_statistics_fetcher.get()))
1214 << " : " << aos::FlatbufferToJson(pi3_server_statistics_fetcher.get());
1215 EXPECT_TRUE(pi3_client_statistics_fetcher.Fetch());
1216 EXPECT_TRUE(AllConnectedBut(pi3_client_statistics_fetcher.get(), "pi1"))
1217 << " : " << aos::FlatbufferToJson(pi3_client_statistics_fetcher.get());
1218
1219 simulated_event_loop_factory.GetNodeEventLoopFactory(pi1)->Connect(pi3);
1220
1221 simulated_event_loop_factory.RunFor(chrono::seconds(2));
1222
1223 EXPECT_EQ(pi1_pong_counter.count(), 601u);
1224 EXPECT_EQ(pi2_pong_counter.count(), 601u);
1225
1226 EXPECT_EQ(pi1_on_pi1_timestamp_counter.count(), 60u);
1227 EXPECT_EQ(pi1_on_pi2_timestamp_counter.count(), 60u);
1228 EXPECT_EQ(pi1_on_pi3_timestamp_counter.count(), 40u);
1229 EXPECT_EQ(pi2_on_pi1_timestamp_counter.count(), 60u);
1230 EXPECT_EQ(pi2_on_pi2_timestamp_counter.count(), 60u);
1231 EXPECT_EQ(pi3_on_pi1_timestamp_counter.count(), 60u);
1232 EXPECT_EQ(pi3_on_pi3_timestamp_counter.count(), 60u);
1233
1234 EXPECT_EQ(pi1_server_statistics_counter.count(), 6u);
1235 EXPECT_EQ(pi2_server_statistics_counter.count(), 6u);
1236 EXPECT_EQ(pi3_server_statistics_counter.count(), 6u);
1237
1238 EXPECT_EQ(pi1_client_statistics_counter.count(), 60u);
1239 EXPECT_EQ(pi2_client_statistics_counter.count(), 60u);
1240 EXPECT_EQ(pi3_client_statistics_counter.count(), 60u);
1241
1242 // Also confirm that remote timestamps are being forwarded correctly.
Austin Schuh89c9b812021-02-20 14:42:10 -08001243 EXPECT_EQ(CountAll(remote_timestamps_pi2_on_pi1), 661);
1244 EXPECT_EQ(CountAll(remote_timestamps_pi1_on_pi2), 661);
Austin Schuhc0b0f722020-12-12 18:36:06 -08001245
1246 EXPECT_TRUE(pi1_server_statistics_fetcher.Fetch());
1247 EXPECT_TRUE(AllConnected(pi1_server_statistics_fetcher.get()))
1248 << " : " << aos::FlatbufferToJson(pi1_server_statistics_fetcher.get());
1249 EXPECT_TRUE(pi1_client_statistics_fetcher.Fetch());
1250 EXPECT_TRUE(AllConnected(pi1_client_statistics_fetcher.get()))
1251 << " : " << aos::FlatbufferToJson(pi1_client_statistics_fetcher.get());
1252 EXPECT_TRUE(pi2_server_statistics_fetcher.Fetch());
1253 EXPECT_TRUE(AllConnected(pi2_server_statistics_fetcher.get()))
1254 << " : " << aos::FlatbufferToJson(pi2_server_statistics_fetcher.get());
1255 EXPECT_TRUE(pi2_client_statistics_fetcher.Fetch());
1256 EXPECT_TRUE(AllConnected(pi2_client_statistics_fetcher.get()))
1257 << " : " << aos::FlatbufferToJson(pi2_client_statistics_fetcher.get());
1258 EXPECT_TRUE(pi3_server_statistics_fetcher.Fetch());
1259 EXPECT_TRUE(AllConnected(pi3_server_statistics_fetcher.get()))
1260 << " : " << aos::FlatbufferToJson(pi3_server_statistics_fetcher.get());
1261 EXPECT_TRUE(pi3_client_statistics_fetcher.Fetch());
1262 EXPECT_TRUE(AllConnected(pi3_client_statistics_fetcher.get()))
1263 << " : " << aos::FlatbufferToJson(pi3_client_statistics_fetcher.get());
1264}
1265
Austin Schuh2febf0d2020-09-21 22:24:30 -07001266// Tests that the time offset having a slope doesn't break the world.
1267// SimulatedMessageBridge has enough self consistency CHECK statements to
1268// confirm, and we can can also check a message in each direction to make sure
1269// it gets delivered as expected.
1270TEST(SimulatedEventLoopTest, MultinodePingPongWithOffsetAndSlope) {
1271 aos::FlatbufferDetachedBuffer<aos::Configuration> config =
Austin Schuh89c9b812021-02-20 14:42:10 -08001272 aos::configuration::ReadConfig(
1273 ConfigPrefix() +
1274 "events/multinode_pingpong_test_combined_config.json");
Austin Schuh2febf0d2020-09-21 22:24:30 -07001275 const Node *pi1 = configuration::GetNode(&config.message(), "pi1");
Austin Schuh87dd3832021-01-01 23:07:31 -08001276 const size_t pi1_index = configuration::GetNodeIndex(&config.message(), pi1);
1277 ASSERT_EQ(pi1_index, 0u);
Austin Schuh2febf0d2020-09-21 22:24:30 -07001278 const Node *pi2 = configuration::GetNode(&config.message(), "pi2");
Austin Schuh87dd3832021-01-01 23:07:31 -08001279 const size_t pi2_index = configuration::GetNodeIndex(&config.message(), pi2);
1280 ASSERT_EQ(pi2_index, 1u);
1281 const Node *pi3 = configuration::GetNode(&config.message(), "pi3");
1282 const size_t pi3_index = configuration::GetNodeIndex(&config.message(), pi3);
1283 ASSERT_EQ(pi3_index, 2u);
Austin Schuh2febf0d2020-09-21 22:24:30 -07001284
Austin Schuh87dd3832021-01-01 23:07:31 -08001285 message_bridge::TestingTimeConverter time(
1286 configuration::NodesCount(&config.message()));
Austin Schuh2febf0d2020-09-21 22:24:30 -07001287 SimulatedEventLoopFactory simulated_event_loop_factory(&config.message());
1288 NodeEventLoopFactory *pi2_factory =
1289 simulated_event_loop_factory.GetNodeEventLoopFactory(pi2);
Austin Schuh87dd3832021-01-01 23:07:31 -08001290 pi2_factory->SetTimeConverter(&time);
Austin Schuh2febf0d2020-09-21 22:24:30 -07001291
Austin Schuh2febf0d2020-09-21 22:24:30 -07001292 constexpr chrono::milliseconds kOffset{150100};
Austin Schuh87dd3832021-01-01 23:07:31 -08001293 time.AddNextTimestamp(
1294 distributed_clock::epoch(),
1295 {monotonic_clock::epoch(), monotonic_clock::epoch() + kOffset,
1296 monotonic_clock::epoch()});
1297 time.AddNextTimestamp(
1298 distributed_clock::epoch() + chrono::seconds(10),
1299 {monotonic_clock::epoch() + chrono::milliseconds(9999),
1300 monotonic_clock::epoch() + kOffset + chrono::seconds(10),
1301 monotonic_clock::epoch() + chrono::milliseconds(9999)});
Austin Schuh2febf0d2020-09-21 22:24:30 -07001302
1303 std::unique_ptr<EventLoop> ping_event_loop =
1304 simulated_event_loop_factory.MakeEventLoop("ping", pi1);
1305 Ping ping(ping_event_loop.get());
1306
1307 std::unique_ptr<EventLoop> pong_event_loop =
1308 simulated_event_loop_factory.MakeEventLoop("pong", pi2);
1309 Pong pong(pong_event_loop.get());
1310
1311 std::unique_ptr<EventLoop> pi1_counter_event_loop =
1312 simulated_event_loop_factory.MakeEventLoop("pi1_counter", pi1);
1313 std::unique_ptr<EventLoop> pi2_counter_event_loop =
1314 simulated_event_loop_factory.MakeEventLoop("pi2_counter", pi2);
1315
1316 aos::Fetcher<examples::Ping> ping_on_pi1_fetcher =
1317 pi1_counter_event_loop->MakeFetcher<examples::Ping>("/test");
1318 aos::Fetcher<examples::Ping> ping_on_pi2_fetcher =
1319 pi2_counter_event_loop->MakeFetcher<examples::Ping>("/test");
1320
1321 aos::Fetcher<examples::Pong> pong_on_pi2_fetcher =
1322 pi2_counter_event_loop->MakeFetcher<examples::Pong>("/test");
1323 aos::Fetcher<examples::Pong> pong_on_pi1_fetcher =
1324 pi1_counter_event_loop->MakeFetcher<examples::Pong>("/test");
1325
1326 // End after a pong message comes back. This will leave the latest messages
1327 // on all channels so we can look at timestamps easily and check they make
1328 // sense.
1329 std::unique_ptr<EventLoop> pi1_pong_ender =
1330 simulated_event_loop_factory.MakeEventLoop("pi2_counter", pi1);
1331 int count = 0;
1332 pi1_pong_ender->MakeWatcher(
1333 "/test", [&simulated_event_loop_factory, &count](const examples::Pong &) {
1334 if (++count == 100) {
1335 simulated_event_loop_factory.Exit();
1336 }
1337 });
1338
1339 // Run enough that messages should be delivered.
1340 simulated_event_loop_factory.Run();
1341
1342 // Grab the latest messages.
1343 EXPECT_TRUE(ping_on_pi1_fetcher.Fetch());
1344 EXPECT_TRUE(ping_on_pi2_fetcher.Fetch());
1345 EXPECT_TRUE(pong_on_pi1_fetcher.Fetch());
1346 EXPECT_TRUE(pong_on_pi2_fetcher.Fetch());
1347
1348 // Compute their time on the global distributed clock so we can compute
1349 // distance betwen them.
1350 const distributed_clock::time_point pi1_ping_time =
1351 simulated_event_loop_factory.GetNodeEventLoopFactory(pi1)
1352 ->ToDistributedClock(
1353 ping_on_pi1_fetcher.context().monotonic_event_time);
1354 const distributed_clock::time_point pi2_ping_time =
1355 simulated_event_loop_factory.GetNodeEventLoopFactory(pi2)
1356 ->ToDistributedClock(
1357 ping_on_pi2_fetcher.context().monotonic_event_time);
1358 const distributed_clock::time_point pi1_pong_time =
1359 simulated_event_loop_factory.GetNodeEventLoopFactory(pi1)
1360 ->ToDistributedClock(
1361 pong_on_pi1_fetcher.context().monotonic_event_time);
1362 const distributed_clock::time_point pi2_pong_time =
1363 simulated_event_loop_factory.GetNodeEventLoopFactory(pi2)
1364 ->ToDistributedClock(
1365 pong_on_pi2_fetcher.context().monotonic_event_time);
1366
1367 // And confirm the delivery delay is just about exactly 150 uS for both
1368 // directions like expected. There will be a couple ns of rounding errors in
1369 // the conversion functions that aren't worth accounting for right now. This
1370 // will either be really close, or really far.
1371 EXPECT_GE(pi2_ping_time, chrono::microseconds(150) - chrono::nanoseconds(10) +
1372 pi1_ping_time);
1373 EXPECT_LE(pi2_ping_time, chrono::microseconds(150) + chrono::nanoseconds(10) +
1374 pi1_ping_time);
1375
1376 EXPECT_GE(pi1_pong_time, chrono::microseconds(150) - chrono::nanoseconds(10) +
1377 pi2_pong_time);
1378 EXPECT_LE(pi1_pong_time, chrono::microseconds(150) + chrono::nanoseconds(10) +
1379 pi2_pong_time);
1380}
1381
Austin Schuh4c570ea2020-11-19 23:13:24 -08001382void SendPing(aos::Sender<examples::Ping> *sender, int value) {
1383 aos::Sender<examples::Ping>::Builder builder = sender->MakeBuilder();
1384 examples::Ping::Builder ping_builder = builder.MakeBuilder<examples::Ping>();
1385 ping_builder.add_value(value);
1386 builder.Send(ping_builder.Finish());
1387}
1388
1389// Tests that reliable (and unreliable) ping messages get forwarded as expected.
Austin Schuh89c9b812021-02-20 14:42:10 -08001390TEST_P(RemoteMessageSimulatedEventLoopTest, MultinodeStartupTesting) {
Austin Schuh4c570ea2020-11-19 23:13:24 -08001391 const Node *pi1 = configuration::GetNode(&config.message(), "pi1");
1392 const Node *pi2 = configuration::GetNode(&config.message(), "pi2");
1393
1394 SimulatedEventLoopFactory simulated_event_loop_factory(&config.message());
1395
1396 std::unique_ptr<EventLoop> ping_event_loop =
1397 simulated_event_loop_factory.MakeEventLoop("ping", pi1);
1398 aos::Sender<examples::Ping> pi1_reliable_sender =
1399 ping_event_loop->MakeSender<examples::Ping>("/reliable");
1400 aos::Sender<examples::Ping> pi1_unreliable_sender =
1401 ping_event_loop->MakeSender<examples::Ping>("/unreliable");
1402 SendPing(&pi1_reliable_sender, 1);
1403 SendPing(&pi1_unreliable_sender, 1);
1404
1405 std::unique_ptr<EventLoop> pi2_pong_event_loop =
1406 simulated_event_loop_factory.MakeEventLoop("pong", pi2);
1407 MessageCounter<examples::Ping> pi2_reliable_counter(pi2_pong_event_loop.get(),
1408 "/reliable");
1409 MessageCounter<examples::Ping> pi2_unreliable_counter(
1410 pi2_pong_event_loop.get(), "/unreliable");
1411 aos::Fetcher<examples::Ping> reliable_on_pi2_fetcher =
1412 pi2_pong_event_loop->MakeFetcher<examples::Ping>("/reliable");
1413 aos::Fetcher<examples::Ping> unreliable_on_pi2_fetcher =
1414 pi2_pong_event_loop->MakeFetcher<examples::Ping>("/unreliable");
1415
1416 const size_t reliable_channel_index = configuration::ChannelIndex(
1417 pi2_pong_event_loop->configuration(), reliable_on_pi2_fetcher.channel());
1418
1419 std::unique_ptr<EventLoop> pi1_remote_timestamp =
1420 simulated_event_loop_factory.MakeEventLoop("pi1_remote_timestamp", pi1);
1421
Austin Schuheeaa2022021-01-02 21:52:03 -08001422 const chrono::nanoseconds network_delay =
1423 simulated_event_loop_factory.network_delay();
1424
Austin Schuh4c570ea2020-11-19 23:13:24 -08001425 int reliable_timestamp_count = 0;
1426 pi1_remote_timestamp->MakeWatcher(
Austin Schuh89c9b812021-02-20 14:42:10 -08001427 shared() ? "/pi1/aos/remote_timestamps/pi2"
1428 : "/pi1/aos/remote_timestamps/pi2/reliable/aos-examples-Ping",
Austin Schuh20ac95d2020-12-05 17:24:19 -08001429 [reliable_channel_index, &reliable_timestamp_count,
Austin Schuheeaa2022021-01-02 21:52:03 -08001430 &simulated_event_loop_factory, pi2, network_delay, &pi2_pong_event_loop,
1431 &pi1_remote_timestamp](const RemoteMessage &header) {
Austin Schuh20ac95d2020-12-05 17:24:19 -08001432 EXPECT_TRUE(header.has_boot_uuid());
1433 EXPECT_EQ(header.boot_uuid()->string_view(),
1434 simulated_event_loop_factory.GetNodeEventLoopFactory(pi2)
1435 ->boot_uuid()
1436 .string_view());
Austin Schuh4c570ea2020-11-19 23:13:24 -08001437 VLOG(1) << aos::FlatbufferToJson(&header);
1438 if (header.channel_index() == reliable_channel_index) {
1439 ++reliable_timestamp_count;
1440 }
Austin Schuheeaa2022021-01-02 21:52:03 -08001441
1442 const aos::monotonic_clock::time_point header_monotonic_sent_time(
1443 chrono::nanoseconds(header.monotonic_sent_time()));
1444
1445 EXPECT_EQ(pi1_remote_timestamp->context().monotonic_event_time,
1446 header_monotonic_sent_time + network_delay +
1447 (pi1_remote_timestamp->monotonic_now() -
1448 pi2_pong_event_loop->monotonic_now()));
Austin Schuh4c570ea2020-11-19 23:13:24 -08001449 });
1450
1451 // Wait to let timestamp estimation start up before looking for the results.
1452 simulated_event_loop_factory.RunFor(chrono::milliseconds(500));
1453
1454 EXPECT_EQ(pi2_reliable_counter.count(), 1u);
1455 // This one isn't reliable, but was sent before the start. It should *not* be
1456 // delivered.
1457 EXPECT_EQ(pi2_unreliable_counter.count(), 0u);
1458 // Confirm we got a timestamp logged for the message that was forwarded.
1459 EXPECT_EQ(reliable_timestamp_count, 1u);
1460
1461 SendPing(&pi1_reliable_sender, 2);
1462 SendPing(&pi1_unreliable_sender, 2);
1463 simulated_event_loop_factory.RunFor(chrono::milliseconds(500));
1464 EXPECT_EQ(pi2_reliable_counter.count(), 2u);
1465 EXPECT_EQ(pi2_unreliable_counter.count(), 1u);
1466
1467 EXPECT_EQ(reliable_timestamp_count, 2u);
1468}
1469
Austin Schuh20ac95d2020-12-05 17:24:19 -08001470// Tests that rebooting a node changes the ServerStatistics message and the
1471// RemoteTimestamp message.
Austin Schuh89c9b812021-02-20 14:42:10 -08001472TEST_P(RemoteMessageSimulatedEventLoopTest, BootUUIDTest) {
Austin Schuh20ac95d2020-12-05 17:24:19 -08001473 const Node *pi1 = configuration::GetNode(&config.message(), "pi1");
1474 const Node *pi2 = configuration::GetNode(&config.message(), "pi2");
1475
1476 SimulatedEventLoopFactory simulated_event_loop_factory(&config.message());
1477
1478 std::unique_ptr<EventLoop> ping_event_loop =
1479 simulated_event_loop_factory.MakeEventLoop("ping", pi1);
1480 Ping ping(ping_event_loop.get());
1481
1482 std::unique_ptr<EventLoop> pong_event_loop =
1483 simulated_event_loop_factory.MakeEventLoop("pong", pi2);
1484 Pong pong(pong_event_loop.get());
1485
1486 std::unique_ptr<EventLoop> pi1_remote_timestamp =
1487 simulated_event_loop_factory.MakeEventLoop("pi1_remote_timestamp", pi1);
1488 std::string expected_boot_uuid(
1489 simulated_event_loop_factory.GetNodeEventLoopFactory(pi2)
1490 ->boot_uuid()
1491 .string_view());
1492
1493 int timestamp_count = 0;
1494 pi1_remote_timestamp->MakeWatcher(
Austin Schuh89c9b812021-02-20 14:42:10 -08001495 shared() ? "/pi1/aos/remote_timestamps/pi2"
1496 : "/pi1/aos/remote_timestamps/pi2/test/aos-examples-Ping",
Austin Schuh20ac95d2020-12-05 17:24:19 -08001497 [&timestamp_count, &expected_boot_uuid](const RemoteMessage &header) {
1498 EXPECT_TRUE(header.has_boot_uuid());
1499 EXPECT_EQ(header.boot_uuid()->string_view(), expected_boot_uuid);
1500 VLOG(1) << aos::FlatbufferToJson(&header);
1501 ++timestamp_count;
1502 });
1503
1504 int pi1_server_statistics_count = 0;
1505 pi1_remote_timestamp->MakeWatcher(
1506 "/pi1/aos", [&pi1_server_statistics_count, &expected_boot_uuid](
1507 const message_bridge::ServerStatistics &stats) {
1508 VLOG(1) << "pi1 ServerStatistics " << FlatbufferToJson(&stats);
1509 for (const message_bridge::ServerConnection *connection :
1510 *stats.connections()) {
1511 EXPECT_TRUE(connection->has_boot_uuid());
1512 if (connection->node()->name()->string_view() == "pi2") {
1513 EXPECT_EQ(expected_boot_uuid,
1514 connection->boot_uuid()->string_view())
1515 << " : Got " << aos::FlatbufferToJson(&stats);
1516 ++pi1_server_statistics_count;
1517 }
1518 }
1519 });
1520
1521 // Let a couple of ServerStatistics messages show up before rebooting.
1522 simulated_event_loop_factory.RunFor(chrono::milliseconds(2001));
1523
1524 EXPECT_GT(timestamp_count, 100);
1525 EXPECT_GE(pi1_server_statistics_count, 1u);
1526
1527 // Confirm that reboot changes the UUID.
1528 simulated_event_loop_factory.GetNodeEventLoopFactory(pi2)->Reboot();
1529
1530 EXPECT_NE(expected_boot_uuid,
1531 simulated_event_loop_factory.GetNodeEventLoopFactory(pi2)
1532 ->boot_uuid()
1533 .string_view());
1534
1535 expected_boot_uuid =
1536 simulated_event_loop_factory.GetNodeEventLoopFactory(pi2)
1537 ->boot_uuid()
1538 .string_view();
1539 timestamp_count = 0;
1540 pi1_server_statistics_count = 0;
1541
1542 simulated_event_loop_factory.RunFor(chrono::milliseconds(2000));
1543 EXPECT_GT(timestamp_count, 100);
1544 EXPECT_GE(pi1_server_statistics_count, 1u);
1545}
1546
Austin Schuh89c9b812021-02-20 14:42:10 -08001547INSTANTIATE_TEST_CASE_P(
1548 All, RemoteMessageSimulatedEventLoopTest,
1549 ::testing::Values(
1550 Param{"multinode_pingpong_test_combined_config.json", true},
1551 Param{"multinode_pingpong_test_split_config.json", false}));
1552
Neil Balchc8f41ed2018-01-20 22:06:53 -08001553} // namespace testing
1554} // namespace aos