blob: 21228145338c870cd6f82e02210e5bd82e4e58fe [file] [log] [blame]
Austin Schuh54cf95f2019-11-29 13:14:18 -08001#include "aos/events/event_loop.h"
2
Philipp Schrader790cb542023-07-05 21:06:52 -07003#include "glog/logging.h"
4
Austin Schuh54cf95f2019-11-29 13:14:18 -08005#include "aos/configuration.h"
6#include "aos/configuration_generated.h"
Tyler Chatow67ddb032020-01-12 14:30:04 -08007#include "aos/logging/implementations.h"
Austin Schuh070019a2022-12-20 22:23:09 -08008#include "aos/realtime.h"
Austin Schuh54cf95f2019-11-29 13:14:18 -08009
Austin Schuh39788ff2019-12-01 18:22:57 -080010DEFINE_bool(timing_reports, true, "Publish timing reports.");
11DEFINE_int32(timing_report_ms, 1000,
12 "Period in milliseconds to publish timing reports at.");
13
Austin Schuh54cf95f2019-11-29 13:14:18 -080014namespace aos {
Austin Schuhd54780b2020-10-03 16:26:02 -070015namespace {
16void CheckAlignment(const Channel *channel) {
17 if (channel->max_size() % alignof(flatbuffers::largest_scalar_t) != 0) {
18 LOG(FATAL) << "max_size() (" << channel->max_size()
19 << ") is not a multiple of alignment ("
20 << alignof(flatbuffers::largest_scalar_t) << ") for channel "
21 << configuration::CleanedChannelToString(channel) << ".";
22 }
23}
milind1f1dca32021-07-03 13:50:07 -070024
25std::string_view ErrorToString(const RawSender::Error err) {
26 switch (err) {
27 case RawSender::Error::kOk:
28 return "RawSender::Error::kOk";
29 case RawSender::Error::kMessagesSentTooFast:
30 return "RawSender::Error::kMessagesSentTooFast";
Eric Schmiedebergef44b8a2022-02-28 17:30:38 -070031 case RawSender::Error::kInvalidRedzone:
32 return "RawSender::Error::kInvalidRedzone";
milind1f1dca32021-07-03 13:50:07 -070033 }
34 LOG(FATAL) << "Unknown error given with code " << static_cast<int>(err);
35}
Austin Schuhd54780b2020-10-03 16:26:02 -070036} // namespace
Austin Schuh54cf95f2019-11-29 13:14:18 -080037
James Kuszmaul762e8692023-07-31 14:57:53 -070038std::optional<std::string> EventLoop::default_version_string_;
39
Austin Schuhe0ab4de2023-05-03 08:05:08 -070040std::pair<SharedSpan, absl::Span<uint8_t>> MakeSharedSpan(size_t size) {
41 AlignedOwningSpan *const span = reinterpret_cast<AlignedOwningSpan *>(
42 malloc(sizeof(AlignedOwningSpan) + size + kChannelDataAlignment - 1));
43
44 absl::Span<uint8_t> mutable_span(
45 reinterpret_cast<uint8_t *>(RoundChannelData(span->data(), size)), size);
46 // Use the placement new operator to construct an actual absl::Span in place.
47 new (span) AlignedOwningSpan(mutable_span);
48
49 return std::make_pair(
50 SharedSpan(std::shared_ptr<AlignedOwningSpan>(span,
51 [](AlignedOwningSpan *s) {
52 s->~AlignedOwningSpan();
53 free(s);
54 }),
55 &span->span),
56 mutable_span);
57}
58
milind1f1dca32021-07-03 13:50:07 -070059std::ostream &operator<<(std::ostream &os, const RawSender::Error err) {
60 os << ErrorToString(err);
61 return os;
62}
63
64void RawSender::CheckOk(const RawSender::Error err) {
65 CHECK_EQ(err, Error::kOk) << "Messages were sent too fast on channel: "
66 << configuration::CleanedChannelToString(channel_);
67}
68
Austin Schuh39788ff2019-12-01 18:22:57 -080069RawSender::RawSender(EventLoop *event_loop, const Channel *channel)
70 : event_loop_(event_loop),
71 channel_(channel),
Brian Silverman79ec7fc2020-06-08 20:11:22 -050072 ftrace_prefix_(configuration::StrippedChannelToString(channel)),
Austin Schuh39788ff2019-12-01 18:22:57 -080073 timing_(event_loop_->ChannelIndex(channel)) {
74 event_loop_->NewSender(this);
75}
76
77RawSender::~RawSender() { event_loop_->DeleteSender(this); }
78
milind1f1dca32021-07-03 13:50:07 -070079RawSender::Error RawSender::DoSend(
80 const SharedSpan data, monotonic_clock::time_point monotonic_remote_time,
81 realtime_clock::time_point realtime_remote_time,
82 uint32_t remote_queue_index, const UUID &source_boot_uuid) {
Tyler Chatowb7c6eba2021-07-28 14:43:23 -070083 return DoSend(data->data(), data->size(), monotonic_remote_time,
84 realtime_remote_time, remote_queue_index, source_boot_uuid);
85}
86
James Kuszmaul93abac12022-04-14 15:05:10 -070087void RawSender::RecordSendResult(const Error error, size_t message_size) {
88 switch (error) {
89 case Error::kOk: {
90 if (timing_.sender) {
91 timing_.size.Add(message_size);
92 timing_.sender->mutate_count(timing_.sender->count() + 1);
93 }
94 break;
95 }
96 case Error::kMessagesSentTooFast:
97 timing_.IncrementError(timing::SendError::MESSAGE_SENT_TOO_FAST);
98 break;
99 case Error::kInvalidRedzone:
100 timing_.IncrementError(timing::SendError::INVALID_REDZONE);
101 break;
102 }
103}
104
Austin Schuh39788ff2019-12-01 18:22:57 -0800105RawFetcher::RawFetcher(EventLoop *event_loop, const Channel *channel)
106 : event_loop_(event_loop),
107 channel_(channel),
Brian Silverman79ec7fc2020-06-08 20:11:22 -0500108 ftrace_prefix_(configuration::StrippedChannelToString(channel)),
Austin Schuh39788ff2019-12-01 18:22:57 -0800109 timing_(event_loop_->ChannelIndex(channel)) {
Austin Schuhad154822019-12-27 15:45:13 -0800110 context_.monotonic_event_time = monotonic_clock::min_time;
111 context_.monotonic_remote_time = monotonic_clock::min_time;
112 context_.realtime_event_time = realtime_clock::min_time;
113 context_.realtime_remote_time = realtime_clock::min_time;
Austin Schuh39788ff2019-12-01 18:22:57 -0800114 context_.queue_index = 0xffffffff;
Austin Schuh0debde12022-08-17 16:25:17 -0700115 context_.remote_queue_index = 0xffffffffu;
Austin Schuh39788ff2019-12-01 18:22:57 -0800116 context_.size = 0;
117 context_.data = nullptr;
Brian Silverman4f4e0612020-08-12 19:54:41 -0700118 context_.buffer_index = -1;
Austin Schuh39788ff2019-12-01 18:22:57 -0800119 event_loop_->NewFetcher(this);
120}
121
122RawFetcher::~RawFetcher() { event_loop_->DeleteFetcher(this); }
123
124TimerHandler::TimerHandler(EventLoop *event_loop, std::function<void()> fn)
125 : event_loop_(event_loop), fn_(std::move(fn)) {}
126
127TimerHandler::~TimerHandler() {}
128
129PhasedLoopHandler::PhasedLoopHandler(EventLoop *event_loop,
130 std::function<void(int)> fn,
131 const monotonic_clock::duration interval,
132 const monotonic_clock::duration offset)
133 : event_loop_(event_loop),
134 fn_(std::move(fn)),
135 phased_loop_(interval, event_loop_->monotonic_now(), offset) {
136 event_loop_->OnRun([this]() {
137 const monotonic_clock::time_point monotonic_now =
138 event_loop_->monotonic_now();
139 phased_loop_.Reset(monotonic_now);
James Kuszmaul20dcc7c2023-01-20 11:06:31 -0800140 Reschedule(monotonic_now);
Milind Upadhyay42589bb2021-05-19 20:05:16 -0700141 // Reschedule here will count cycles elapsed before now, and then the
142 // reschedule before running the handler will count the time that elapsed
143 // then. So clear the count here.
Austin Schuh39788ff2019-12-01 18:22:57 -0800144 cycles_elapsed_ = 0;
145 });
146}
147
148PhasedLoopHandler::~PhasedLoopHandler() {}
149
Austin Schuh83c7f702021-01-19 22:36:29 -0800150EventLoop::EventLoop(const Configuration *configuration)
James Kuszmaul762e8692023-07-31 14:57:53 -0700151 : version_string_(default_version_string_),
152 timing_report_(flatbuffers::DetachedBuffer()),
Austin Schuh56196432020-10-24 20:15:21 -0700153 configuration_(configuration) {}
Tyler Chatow67ddb032020-01-12 14:30:04 -0800154
Austin Schuh39788ff2019-12-01 18:22:57 -0800155EventLoop::~EventLoop() {
Brian Silvermanbf889922021-11-10 12:41:57 -0800156 if (!senders_.empty()) {
Austin Schuh58646e22021-08-23 23:51:46 -0700157 for (const RawSender *sender : senders_) {
158 LOG(ERROR) << " Sender "
159 << configuration::StrippedChannelToString(sender->channel())
160 << " still open";
161 }
162 }
Austin Schuh39788ff2019-12-01 18:22:57 -0800163 CHECK_EQ(senders_.size(), 0u) << ": Not all senders destroyed";
Austin Schuh7d87b672019-12-01 20:23:49 -0800164 CHECK_EQ(events_.size(), 0u) << ": Not all events unregistered";
Austin Schuh39788ff2019-12-01 18:22:57 -0800165}
166
Brian Silvermanbf889922021-11-10 12:41:57 -0800167void EventLoop::SkipTimingReport() {
168 skip_timing_report_ = true;
169 timing_report_ = flatbuffers::DetachedBuffer();
170
171 for (size_t i = 0; i < timers_.size(); ++i) {
172 timers_[i]->timing_.set_timing_report(nullptr);
173 }
174
175 for (size_t i = 0; i < phased_loops_.size(); ++i) {
176 phased_loops_[i]->timing_.set_timing_report(nullptr);
177 }
178
179 for (size_t i = 0; i < watchers_.size(); ++i) {
180 watchers_[i]->set_timing_report(nullptr);
181 }
182
183 for (size_t i = 0; i < senders_.size(); ++i) {
184 senders_[i]->timing_.set_timing_report(nullptr);
185 }
186
187 for (size_t i = 0; i < fetchers_.size(); ++i) {
188 fetchers_[i]->timing_.set_timing_report(nullptr);
189 }
190}
191
Austin Schuh39788ff2019-12-01 18:22:57 -0800192int EventLoop::ChannelIndex(const Channel *channel) {
Austin Schuhc9e10ec2020-01-26 16:08:28 -0800193 return configuration::ChannelIndex(configuration_, channel);
Austin Schuh39788ff2019-12-01 18:22:57 -0800194}
195
Brian Silverman5120afb2020-01-31 17:44:35 -0800196WatcherState *EventLoop::GetWatcherState(const Channel *channel) {
197 const int channel_index = ChannelIndex(channel);
198 for (const std::unique_ptr<WatcherState> &watcher : watchers_) {
199 if (watcher->channel_index() == channel_index) {
200 return watcher.get();
201 }
202 }
203 LOG(FATAL) << "No watcher found for channel";
204}
205
Austin Schuh39788ff2019-12-01 18:22:57 -0800206void EventLoop::NewSender(RawSender *sender) {
207 senders_.emplace_back(sender);
208 UpdateTimingReport();
209}
210void EventLoop::DeleteSender(RawSender *sender) {
211 CHECK(!is_running());
212 auto s = std::find(senders_.begin(), senders_.end(), sender);
213 CHECK(s != senders_.end()) << ": Sender not in senders list";
214 senders_.erase(s);
215 UpdateTimingReport();
216}
217
218TimerHandler *EventLoop::NewTimer(std::unique_ptr<TimerHandler> timer) {
219 timers_.emplace_back(std::move(timer));
220 UpdateTimingReport();
221 return timers_.back().get();
222}
223
224PhasedLoopHandler *EventLoop::NewPhasedLoop(
225 std::unique_ptr<PhasedLoopHandler> phased_loop) {
226 phased_loops_.emplace_back(std::move(phased_loop));
227 UpdateTimingReport();
228 return phased_loops_.back().get();
229}
230
231void EventLoop::NewFetcher(RawFetcher *fetcher) {
Austin Schuhd54780b2020-10-03 16:26:02 -0700232 CheckAlignment(fetcher->channel());
233
Austin Schuh39788ff2019-12-01 18:22:57 -0800234 fetchers_.emplace_back(fetcher);
235 UpdateTimingReport();
236}
237
238void EventLoop::DeleteFetcher(RawFetcher *fetcher) {
239 CHECK(!is_running());
240 auto f = std::find(fetchers_.begin(), fetchers_.end(), fetcher);
241 CHECK(f != fetchers_.end()) << ": Fetcher not in fetchers list";
242 fetchers_.erase(f);
243 UpdateTimingReport();
244}
245
246WatcherState *EventLoop::NewWatcher(std::unique_ptr<WatcherState> watcher) {
247 watchers_.emplace_back(std::move(watcher));
248
249 UpdateTimingReport();
250
251 return watchers_.back().get();
252}
253
Brian Silverman0fc69932020-01-24 21:54:02 -0800254void EventLoop::TakeWatcher(const Channel *channel) {
255 CHECK(!is_running()) << ": Cannot add new objects while running.";
256 ChannelIndex(channel);
257
Austin Schuhd54780b2020-10-03 16:26:02 -0700258 CheckAlignment(channel);
259
Brian Silverman0fc69932020-01-24 21:54:02 -0800260 CHECK(taken_senders_.find(channel) == taken_senders_.end())
Austin Schuh8072f922020-02-16 21:51:47 -0800261 << ": " << configuration::CleanedChannelToString(channel)
milind-u5dbdba42023-02-04 17:48:43 -0800262 << " is already being used for sending. Can't make a watcher on the "
263 "same event loop.";
Brian Silverman0fc69932020-01-24 21:54:02 -0800264
265 auto result = taken_watchers_.insert(channel);
Austin Schuh8072f922020-02-16 21:51:47 -0800266 CHECK(result.second) << ": " << configuration::CleanedChannelToString(channel)
Brian Silverman0fc69932020-01-24 21:54:02 -0800267 << " is already being used.";
268
269 if (!configuration::ChannelIsReadableOnNode(channel, node())) {
Austin Schuh8072f922020-02-16 21:51:47 -0800270 LOG(FATAL) << ": " << configuration::CleanedChannelToString(channel)
Brian Silverman0fc69932020-01-24 21:54:02 -0800271 << " is not able to be watched on this node. Check your "
272 "configuration.";
273 }
274}
275
276void EventLoop::TakeSender(const Channel *channel) {
277 CHECK(!is_running()) << ": Cannot add new objects while running.";
278 ChannelIndex(channel);
279
Austin Schuhd54780b2020-10-03 16:26:02 -0700280 CheckAlignment(channel);
281
Brian Silverman0fc69932020-01-24 21:54:02 -0800282 CHECK(taken_watchers_.find(channel) == taken_watchers_.end())
Austin Schuh8072f922020-02-16 21:51:47 -0800283 << ": Channel " << configuration::CleanedChannelToString(channel)
284 << " is already being used.";
Brian Silverman0fc69932020-01-24 21:54:02 -0800285
286 // We don't care if this is a duplicate.
287 taken_senders_.insert(channel);
288}
289
Austin Schuh39788ff2019-12-01 18:22:57 -0800290void EventLoop::SendTimingReport() {
Brian Silvermance418d02021-11-03 11:25:52 -0700291 if (!timing_report_sender_) {
292 // Timing reports are disabled, so nothing for us to do.
293 return;
294 }
295
Austin Schuh39788ff2019-12-01 18:22:57 -0800296 // We need to do a fancy dance here to get all the accounting to work right.
297 // We want to copy the memory here, but then send after resetting. Otherwise
298 // the send for the timing report won't be counted in the timing report.
299 //
300 // Also, flatbuffers build from the back end. So place this at the back end
301 // of the buffer. We only have to care because we are using this in a very
302 // raw fashion.
Austin Schuhadd6eb32020-11-09 21:24:26 -0800303 CHECK_LE(timing_report_.span().size(), timing_report_sender_->size())
Austin Schuhf9402e32023-07-10 12:55:50 -0700304 << ": Timing report bigger than the sender size for " << name() << ".";
Austin Schuhadd6eb32020-11-09 21:24:26 -0800305 std::copy(timing_report_.span().data(),
306 timing_report_.span().data() + timing_report_.span().size(),
Austin Schuh39788ff2019-12-01 18:22:57 -0800307 reinterpret_cast<uint8_t *>(timing_report_sender_->data()) +
Austin Schuhadd6eb32020-11-09 21:24:26 -0800308 timing_report_sender_->size() - timing_report_.span().size());
Austin Schuh39788ff2019-12-01 18:22:57 -0800309
310 for (const std::unique_ptr<TimerHandler> &timer : timers_) {
311 timer->timing_.ResetTimingReport();
312 }
313 for (const std::unique_ptr<WatcherState> &watcher : watchers_) {
314 watcher->ResetReport();
315 }
316 for (const std::unique_ptr<PhasedLoopHandler> &phased_loop : phased_loops_) {
317 phased_loop->timing_.ResetTimingReport();
318 }
319 for (RawSender *sender : senders_) {
320 sender->timing_.ResetTimingReport();
321 }
322 for (RawFetcher *fetcher : fetchers_) {
323 fetcher->timing_.ResetTimingReport();
324 }
milind1f1dca32021-07-03 13:50:07 -0700325 // TODO(milind): If we fail to send, we don't want to reset the timing report.
326 // We would need to move the reset after the send, and then find the correct
327 // timing report and set the reports with it instead of letting the sender do
328 // this. If we failed to send, we wouldn't reset or set the reports, so they
329 // can accumalate until the next send.
330 timing_report_failure_counter_.Count(
331 timing_report_sender_->Send(timing_report_.span().size()));
Austin Schuh39788ff2019-12-01 18:22:57 -0800332}
333
334void EventLoop::UpdateTimingReport() {
Brian Silvermanbf889922021-11-10 12:41:57 -0800335 if (skip_timing_report_) {
336 return;
337 }
338
Austin Schuh39788ff2019-12-01 18:22:57 -0800339 // We need to support senders and fetchers changing while we are setting up
340 // the event loop. Otherwise we can't fetch or send until the loop runs. This
341 // means that on each change, we need to redo all this work. This makes setup
342 // more expensive, but not by all that much on a modern processor.
343
344 // Now, build up a report with everything pre-filled out.
345 flatbuffers::FlatBufferBuilder fbb;
Austin Schuhd7b15da2020-02-17 15:06:11 -0800346 fbb.ForceDefaults(true);
Austin Schuh39788ff2019-12-01 18:22:57 -0800347
348 // Pre-fill in the defaults for timers.
349 std::vector<flatbuffers::Offset<timing::Timer>> timer_offsets;
350 for (const std::unique_ptr<TimerHandler> &timer : timers_) {
351 flatbuffers::Offset<timing::Statistic> wakeup_latency_offset =
352 timing::CreateStatistic(fbb);
353 flatbuffers::Offset<timing::Statistic> handler_time_offset =
354 timing::CreateStatistic(fbb);
355 flatbuffers::Offset<flatbuffers::String> name_offset;
356 if (timer->name().size() != 0) {
357 name_offset = fbb.CreateString(timer->name());
358 }
359
360 timing::Timer::Builder timer_builder(fbb);
361
362 if (timer->name().size() != 0) {
363 timer_builder.add_name(name_offset);
364 }
365 timer_builder.add_wakeup_latency(wakeup_latency_offset);
366 timer_builder.add_handler_time(handler_time_offset);
367 timer_builder.add_count(0);
368 timer_offsets.emplace_back(timer_builder.Finish());
369 }
370
371 // Pre-fill in the defaults for phased_loops.
372 std::vector<flatbuffers::Offset<timing::Timer>> phased_loop_offsets;
373 for (const std::unique_ptr<PhasedLoopHandler> &phased_loop : phased_loops_) {
374 flatbuffers::Offset<timing::Statistic> wakeup_latency_offset =
375 timing::CreateStatistic(fbb);
376 flatbuffers::Offset<timing::Statistic> handler_time_offset =
377 timing::CreateStatistic(fbb);
378 flatbuffers::Offset<flatbuffers::String> name_offset;
379 if (phased_loop->name().size() != 0) {
380 name_offset = fbb.CreateString(phased_loop->name());
381 }
382
383 timing::Timer::Builder timer_builder(fbb);
384
385 if (phased_loop->name().size() != 0) {
386 timer_builder.add_name(name_offset);
387 }
388 timer_builder.add_wakeup_latency(wakeup_latency_offset);
389 timer_builder.add_handler_time(handler_time_offset);
390 timer_builder.add_count(0);
391 phased_loop_offsets.emplace_back(timer_builder.Finish());
392 }
393
394 // Pre-fill in the defaults for watchers.
395 std::vector<flatbuffers::Offset<timing::Watcher>> watcher_offsets;
396 for (const std::unique_ptr<WatcherState> &watcher : watchers_) {
397 flatbuffers::Offset<timing::Statistic> wakeup_latency_offset =
398 timing::CreateStatistic(fbb);
399 flatbuffers::Offset<timing::Statistic> handler_time_offset =
400 timing::CreateStatistic(fbb);
401
402 timing::Watcher::Builder watcher_builder(fbb);
403
404 watcher_builder.add_channel_index(watcher->channel_index());
405 watcher_builder.add_wakeup_latency(wakeup_latency_offset);
406 watcher_builder.add_handler_time(handler_time_offset);
407 watcher_builder.add_count(0);
408 watcher_offsets.emplace_back(watcher_builder.Finish());
409 }
410
411 // Pre-fill in the defaults for senders.
412 std::vector<flatbuffers::Offset<timing::Sender>> sender_offsets;
James Kuszmaulcc94ed42022-08-24 11:36:17 -0700413 for (RawSender *sender : senders_) {
Austin Schuh39788ff2019-12-01 18:22:57 -0800414 flatbuffers::Offset<timing::Statistic> size_offset =
415 timing::CreateStatistic(fbb);
416
James Kuszmaul78514332022-04-06 15:08:34 -0700417 const flatbuffers::Offset<
418 flatbuffers::Vector<flatbuffers::Offset<timing::SendErrorCount>>>
James Kuszmaulcc94ed42022-08-24 11:36:17 -0700419 error_counts_offset = sender->timing_.error_counter.Initialize(&fbb);
James Kuszmaul78514332022-04-06 15:08:34 -0700420
Austin Schuh39788ff2019-12-01 18:22:57 -0800421 timing::Sender::Builder sender_builder(fbb);
422
423 sender_builder.add_channel_index(sender->timing_.channel_index);
424 sender_builder.add_size(size_offset);
James Kuszmaul78514332022-04-06 15:08:34 -0700425 sender_builder.add_error_counts(error_counts_offset);
Austin Schuh39788ff2019-12-01 18:22:57 -0800426 sender_builder.add_count(0);
427 sender_offsets.emplace_back(sender_builder.Finish());
428 }
429
430 // Pre-fill in the defaults for fetchers.
431 std::vector<flatbuffers::Offset<timing::Fetcher>> fetcher_offsets;
432 for (RawFetcher *fetcher : fetchers_) {
433 flatbuffers::Offset<timing::Statistic> latency_offset =
434 timing::CreateStatistic(fbb);
435
436 timing::Fetcher::Builder fetcher_builder(fbb);
437
438 fetcher_builder.add_channel_index(fetcher->timing_.channel_index);
439 fetcher_builder.add_count(0);
440 fetcher_builder.add_latency(latency_offset);
441 fetcher_offsets.emplace_back(fetcher_builder.Finish());
442 }
443
444 // Then build the final report.
445 flatbuffers::Offset<flatbuffers::Vector<flatbuffers::Offset<timing::Timer>>>
446 timers_offset;
447 if (timer_offsets.size() > 0) {
448 timers_offset = fbb.CreateVector(timer_offsets);
449 }
450
451 flatbuffers::Offset<flatbuffers::Vector<flatbuffers::Offset<timing::Timer>>>
452 phased_loops_offset;
453 if (phased_loop_offsets.size() > 0) {
454 phased_loops_offset = fbb.CreateVector(phased_loop_offsets);
455 }
456
457 flatbuffers::Offset<flatbuffers::Vector<flatbuffers::Offset<timing::Watcher>>>
458 watchers_offset;
459 if (watcher_offsets.size() > 0) {
460 watchers_offset = fbb.CreateVector(watcher_offsets);
461 }
462
463 flatbuffers::Offset<flatbuffers::Vector<flatbuffers::Offset<timing::Sender>>>
464 senders_offset;
465 if (sender_offsets.size() > 0) {
466 senders_offset = fbb.CreateVector(sender_offsets);
467 }
468
469 flatbuffers::Offset<flatbuffers::Vector<flatbuffers::Offset<timing::Fetcher>>>
470 fetchers_offset;
471 if (fetcher_offsets.size() > 0) {
472 fetchers_offset = fbb.CreateVector(fetcher_offsets);
473 }
474
475 flatbuffers::Offset<flatbuffers::String> name_offset =
476 fbb.CreateString(name());
477
James Kuszmaul762e8692023-07-31 14:57:53 -0700478 const flatbuffers::Offset<flatbuffers::String> version_offset =
479 version_string_.has_value() ? fbb.CreateString(version_string_.value())
480 : flatbuffers::Offset<flatbuffers::String>();
481
Austin Schuh39788ff2019-12-01 18:22:57 -0800482 timing::Report::Builder report_builder(fbb);
483 report_builder.add_name(name_offset);
James Kuszmaul762e8692023-07-31 14:57:53 -0700484 report_builder.add_version(version_offset);
Austin Schuh39788ff2019-12-01 18:22:57 -0800485 report_builder.add_pid(GetTid());
486 if (timer_offsets.size() > 0) {
487 report_builder.add_timers(timers_offset);
488 }
489 if (phased_loop_offsets.size() > 0) {
490 report_builder.add_phased_loops(phased_loops_offset);
491 }
492 if (watcher_offsets.size() > 0) {
493 report_builder.add_watchers(watchers_offset);
494 }
495 if (sender_offsets.size() > 0) {
496 report_builder.add_senders(senders_offset);
497 }
498 if (fetcher_offsets.size() > 0) {
499 report_builder.add_fetchers(fetchers_offset);
500 }
milind1f1dca32021-07-03 13:50:07 -0700501 report_builder.add_send_failures(timing_report_failure_counter_.failures());
Austin Schuh39788ff2019-12-01 18:22:57 -0800502 fbb.Finish(report_builder.Finish());
503
504 timing_report_ = FlatbufferDetachedBuffer<timing::Report>(fbb.Release());
505
506 // Now that the pointers are stable, pass them to the timers and watchers to
507 // be updated.
508 for (size_t i = 0; i < timers_.size(); ++i) {
509 timers_[i]->timing_.set_timing_report(
510 timing_report_.mutable_message()->mutable_timers()->GetMutableObject(
511 i));
512 }
513
514 for (size_t i = 0; i < phased_loops_.size(); ++i) {
515 phased_loops_[i]->timing_.set_timing_report(
516 timing_report_.mutable_message()
517 ->mutable_phased_loops()
518 ->GetMutableObject(i));
519 }
520
521 for (size_t i = 0; i < watchers_.size(); ++i) {
522 watchers_[i]->set_timing_report(
523 timing_report_.mutable_message()->mutable_watchers()->GetMutableObject(
524 i));
525 }
526
527 for (size_t i = 0; i < senders_.size(); ++i) {
528 senders_[i]->timing_.set_timing_report(
529 timing_report_.mutable_message()->mutable_senders()->GetMutableObject(
530 i));
531 }
532
533 for (size_t i = 0; i < fetchers_.size(); ++i) {
534 fetchers_[i]->timing_.set_timing_report(
535 timing_report_.mutable_message()->mutable_fetchers()->GetMutableObject(
536 i));
537 }
538}
539
540void EventLoop::MaybeScheduleTimingReports() {
541 if (FLAGS_timing_reports && !skip_timing_report_) {
542 CHECK(!timing_report_sender_) << ": Timing reports already scheduled.";
543 // Make a raw sender for the report.
544 const Channel *channel = configuration::GetChannel(
545 configuration(), "/aos", timing::Report::GetFullyQualifiedName(),
Austin Schuhbca6cf02019-12-22 17:28:34 -0800546 name(), node());
Austin Schuh196a4452020-03-15 23:12:03 -0700547 CHECK(channel != nullptr) << ": Failed to look up {\"name\": \"/aos\", "
548 "\"type\": \"aos.timing.Report\"} on node "
549 << FlatbufferToJson(node());
Austin Schuhbca6cf02019-12-22 17:28:34 -0800550
551 // Since we are using a RawSender, validity isn't checked. So check it
552 // ourselves.
Austin Schuhca4828c2019-12-28 14:21:35 -0800553 if (!configuration::ChannelIsSendableOnNode(channel, node())) {
554 LOG(FATAL) << "Channel { \"name\": \"/aos"
555 << channel->name()->string_view() << "\", \"type\": \""
556 << channel->type()->string_view()
557 << "\" } is not able to be sent on this node. Check your "
558 "configuration.";
Austin Schuhbca6cf02019-12-22 17:28:34 -0800559 }
Austin Schuh39788ff2019-12-01 18:22:57 -0800560 CHECK(channel != nullptr) << ": Channel { \"name\": \"/aos\", \"type\": \""
561 << timing::Report::GetFullyQualifiedName()
562 << "\" } not found in config.";
563 timing_report_sender_ = MakeRawSender(channel);
564
565 // Register a handler which sends the report out by copying the raw data
566 // from the prebuilt and subsequently modified report.
567 TimerHandler *timing_reports_timer =
568 AddTimer([this]() { SendTimingReport(); });
569
570 // Set it up to send once per second.
571 timing_reports_timer->set_name("timing_reports");
572 OnRun([this, timing_reports_timer]() {
Philipp Schradera6712522023-07-05 20:25:11 -0700573 timing_reports_timer->Schedule(
Austin Schuh39788ff2019-12-01 18:22:57 -0800574 monotonic_now() + std::chrono::milliseconds(FLAGS_timing_report_ms),
575 std::chrono::milliseconds(FLAGS_timing_report_ms));
576 });
577
578 UpdateTimingReport();
579 }
580}
581
Austin Schuh7d87b672019-12-01 20:23:49 -0800582void EventLoop::ReserveEvents() {
583 events_.reserve(timers_.size() + phased_loops_.size() + watchers_.size());
584}
585
586namespace {
587bool CompareEvents(const EventLoopEvent *first, const EventLoopEvent *second) {
Brian Silvermanbd405c02020-06-23 16:25:23 -0700588 if (first->event_time() > second->event_time()) {
589 return true;
590 }
591 if (first->event_time() < second->event_time()) {
592 return false;
593 }
594 return first->generation() > second->generation();
Austin Schuh7d87b672019-12-01 20:23:49 -0800595}
596} // namespace
597
598void EventLoop::AddEvent(EventLoopEvent *event) {
599 DCHECK(std::find(events_.begin(), events_.end(), event) == events_.end());
Brian Silvermanbd405c02020-06-23 16:25:23 -0700600 DCHECK(event->generation() == 0);
601 event->set_generation(++event_generation_);
Austin Schuh7d87b672019-12-01 20:23:49 -0800602 events_.push_back(event);
603 std::push_heap(events_.begin(), events_.end(), CompareEvents);
604}
605
606void EventLoop::RemoveEvent(EventLoopEvent *event) {
607 auto e = std::find(events_.begin(), events_.end(), event);
608 if (e != events_.end()) {
Brian Silvermanbd405c02020-06-23 16:25:23 -0700609 DCHECK(event->generation() != 0);
Austin Schuh7d87b672019-12-01 20:23:49 -0800610 events_.erase(e);
611 std::make_heap(events_.begin(), events_.end(), CompareEvents);
612 event->Invalidate();
613 }
614}
615
616EventLoopEvent *EventLoop::PopEvent() {
617 EventLoopEvent *result = events_.front();
618 std::pop_heap(events_.begin(), events_.end(), CompareEvents);
619 events_.pop_back();
620 result->Invalidate();
621 return result;
622}
623
Austin Schuh0debde12022-08-17 16:25:17 -0700624void EventLoop::ClearContext() {
625 context_.monotonic_event_time = monotonic_clock::min_time;
626 context_.monotonic_remote_time = monotonic_clock::min_time;
627 context_.realtime_event_time = realtime_clock::min_time;
628 context_.realtime_remote_time = realtime_clock::min_time;
629 context_.queue_index = 0xffffffffu;
630 context_.remote_queue_index = 0xffffffffu;
631 context_.size = 0u;
632 context_.data = nullptr;
633 context_.buffer_index = -1;
634 context_.source_boot_uuid = boot_uuid();
635}
636
Austin Schuha9012be2021-07-21 15:19:11 -0700637void EventLoop::SetTimerContext(
638 monotonic_clock::time_point monotonic_event_time) {
639 context_.monotonic_event_time = monotonic_event_time;
640 context_.monotonic_remote_time = monotonic_clock::min_time;
641 context_.realtime_event_time = realtime_clock::min_time;
642 context_.realtime_remote_time = realtime_clock::min_time;
643 context_.queue_index = 0xffffffffu;
Austin Schuh0debde12022-08-17 16:25:17 -0700644 context_.remote_queue_index = 0xffffffffu;
Austin Schuha9012be2021-07-21 15:19:11 -0700645 context_.size = 0u;
646 context_.data = nullptr;
647 context_.buffer_index = -1;
648 context_.source_boot_uuid = boot_uuid();
649}
650
Austin Schuh070019a2022-12-20 22:23:09 -0800651cpu_set_t EventLoop::DefaultAffinity() { return aos::DefaultAffinity(); }
652
James Kuszmaul762e8692023-07-31 14:57:53 -0700653void EventLoop::SetDefaultVersionString(std::string_view version) {
654 default_version_string_ = version;
655}
656
657void EventLoop::SetVersionString(std::string_view version) {
658 CHECK(!is_running())
659 << ": Can't do things that might alter the timing report while running.";
660 version_string_ = version;
661
662 UpdateTimingReport();
663}
664
Austin Schuh39788ff2019-12-01 18:22:57 -0800665void WatcherState::set_timing_report(timing::Watcher *watcher) {
Austin Schuh39788ff2019-12-01 18:22:57 -0800666 watcher_ = watcher;
Brian Silvermanbf889922021-11-10 12:41:57 -0800667 if (!watcher) {
668 wakeup_latency_.set_statistic(nullptr);
669 handler_time_.set_statistic(nullptr);
670 } else {
671 wakeup_latency_.set_statistic(watcher->mutable_wakeup_latency());
672 handler_time_.set_statistic(watcher->mutable_handler_time());
673 }
Austin Schuh39788ff2019-12-01 18:22:57 -0800674}
675
676void WatcherState::ResetReport() {
Brian Silvermanbf889922021-11-10 12:41:57 -0800677 if (!watcher_) {
678 return;
679 }
680
Austin Schuh39788ff2019-12-01 18:22:57 -0800681 wakeup_latency_.Reset();
682 handler_time_.Reset();
683 watcher_->mutate_count(0);
Austin Schuh54cf95f2019-11-29 13:14:18 -0800684}
685
686} // namespace aos