blob: 5c7e49d43f7fb162c0258b60a9b46e24fc7e1693 [file] [log] [blame]
#include "aos/events/event_loop.h"
#include "aos/configuration.h"
#include "aos/configuration_generated.h"
#include "aos/logging/implementations.h"
#include "glog/logging.h"
DEFINE_bool(timing_reports, true, "Publish timing reports.");
DEFINE_int32(timing_report_ms, 1000,
"Period in milliseconds to publish timing reports at.");
namespace aos {
RawSender::RawSender(EventLoop *event_loop, const Channel *channel)
: event_loop_(event_loop),
channel_(channel),
ftrace_prefix_(configuration::StrippedChannelToString(channel)),
timing_(event_loop_->ChannelIndex(channel)) {
event_loop_->NewSender(this);
}
RawSender::~RawSender() { event_loop_->DeleteSender(this); }
RawFetcher::RawFetcher(EventLoop *event_loop, const Channel *channel)
: event_loop_(event_loop),
channel_(channel),
ftrace_prefix_(configuration::StrippedChannelToString(channel)),
timing_(event_loop_->ChannelIndex(channel)) {
context_.monotonic_event_time = monotonic_clock::min_time;
context_.monotonic_remote_time = monotonic_clock::min_time;
context_.realtime_event_time = realtime_clock::min_time;
context_.realtime_remote_time = realtime_clock::min_time;
context_.queue_index = 0xffffffff;
context_.size = 0;
context_.data = nullptr;
context_.buffer_index = -1;
event_loop_->NewFetcher(this);
}
RawFetcher::~RawFetcher() { event_loop_->DeleteFetcher(this); }
TimerHandler::TimerHandler(EventLoop *event_loop, std::function<void()> fn)
: event_loop_(event_loop), fn_(std::move(fn)) {}
TimerHandler::~TimerHandler() {}
PhasedLoopHandler::PhasedLoopHandler(EventLoop *event_loop,
std::function<void(int)> fn,
const monotonic_clock::duration interval,
const monotonic_clock::duration offset)
: event_loop_(event_loop),
fn_(std::move(fn)),
phased_loop_(interval, event_loop_->monotonic_now(), offset) {
event_loop_->OnRun([this]() {
const monotonic_clock::time_point monotonic_now =
event_loop_->monotonic_now();
phased_loop_.Reset(monotonic_now);
Reschedule(
[this](monotonic_clock::time_point sleep_time) {
Schedule(sleep_time);
},
monotonic_now);
// The first time, we'll double count. Reschedule here will count cycles
// elapsed before now, and then the reschedule before runing the handler
// will count the time that elapsed then. So clear the count here.
cycles_elapsed_ = 0;
});
}
PhasedLoopHandler::~PhasedLoopHandler() {}
EventLoop::EventLoop(const Configuration *configuration)
: timing_report_(flatbuffers::DetachedBuffer()),
configuration_(configuration) {
logging::Init();
}
EventLoop::~EventLoop() {
CHECK_EQ(senders_.size(), 0u) << ": Not all senders destroyed";
CHECK_EQ(events_.size(), 0u) << ": Not all events unregistered";
}
int EventLoop::ChannelIndex(const Channel *channel) {
return configuration::ChannelIndex(configuration_, channel);
}
WatcherState *EventLoop::GetWatcherState(const Channel *channel) {
const int channel_index = ChannelIndex(channel);
for (const std::unique_ptr<WatcherState> &watcher : watchers_) {
if (watcher->channel_index() == channel_index) {
return watcher.get();
}
}
LOG(FATAL) << "No watcher found for channel";
}
void EventLoop::NewSender(RawSender *sender) {
senders_.emplace_back(sender);
UpdateTimingReport();
}
void EventLoop::DeleteSender(RawSender *sender) {
CHECK(!is_running());
auto s = std::find(senders_.begin(), senders_.end(), sender);
CHECK(s != senders_.end()) << ": Sender not in senders list";
senders_.erase(s);
UpdateTimingReport();
}
TimerHandler *EventLoop::NewTimer(std::unique_ptr<TimerHandler> timer) {
timers_.emplace_back(std::move(timer));
UpdateTimingReport();
return timers_.back().get();
}
PhasedLoopHandler *EventLoop::NewPhasedLoop(
std::unique_ptr<PhasedLoopHandler> phased_loop) {
phased_loops_.emplace_back(std::move(phased_loop));
UpdateTimingReport();
return phased_loops_.back().get();
}
void EventLoop::NewFetcher(RawFetcher *fetcher) {
fetchers_.emplace_back(fetcher);
UpdateTimingReport();
}
void EventLoop::DeleteFetcher(RawFetcher *fetcher) {
CHECK(!is_running());
auto f = std::find(fetchers_.begin(), fetchers_.end(), fetcher);
CHECK(f != fetchers_.end()) << ": Fetcher not in fetchers list";
fetchers_.erase(f);
UpdateTimingReport();
}
WatcherState *EventLoop::NewWatcher(std::unique_ptr<WatcherState> watcher) {
watchers_.emplace_back(std::move(watcher));
UpdateTimingReport();
return watchers_.back().get();
}
void EventLoop::TakeWatcher(const Channel *channel) {
CHECK(!is_running()) << ": Cannot add new objects while running.";
ChannelIndex(channel);
CHECK(taken_senders_.find(channel) == taken_senders_.end())
<< ": " << configuration::CleanedChannelToString(channel)
<< " is already being used.";
auto result = taken_watchers_.insert(channel);
CHECK(result.second) << ": " << configuration::CleanedChannelToString(channel)
<< " is already being used.";
if (!configuration::ChannelIsReadableOnNode(channel, node())) {
LOG(FATAL) << ": " << configuration::CleanedChannelToString(channel)
<< " is not able to be watched on this node. Check your "
"configuration.";
}
}
void EventLoop::TakeSender(const Channel *channel) {
CHECK(!is_running()) << ": Cannot add new objects while running.";
ChannelIndex(channel);
CHECK(taken_watchers_.find(channel) == taken_watchers_.end())
<< ": Channel " << configuration::CleanedChannelToString(channel)
<< " is already being used.";
// We don't care if this is a duplicate.
taken_senders_.insert(channel);
}
void EventLoop::SendTimingReport() {
// We need to do a fancy dance here to get all the accounting to work right.
// We want to copy the memory here, but then send after resetting. Otherwise
// the send for the timing report won't be counted in the timing report.
//
// Also, flatbuffers build from the back end. So place this at the back end
// of the buffer. We only have to care because we are using this in a very
// raw fashion.
CHECK_LE(timing_report_.size(), timing_report_sender_->size())
<< ": Timing report bigger than the sender size.";
std::copy(timing_report_.data(),
timing_report_.data() + timing_report_.size(),
reinterpret_cast<uint8_t *>(timing_report_sender_->data()) +
timing_report_sender_->size() - timing_report_.size());
for (const std::unique_ptr<TimerHandler> &timer : timers_) {
timer->timing_.ResetTimingReport();
}
for (const std::unique_ptr<WatcherState> &watcher : watchers_) {
watcher->ResetReport();
}
for (const std::unique_ptr<PhasedLoopHandler> &phased_loop : phased_loops_) {
phased_loop->timing_.ResetTimingReport();
}
for (RawSender *sender : senders_) {
sender->timing_.ResetTimingReport();
}
for (RawFetcher *fetcher : fetchers_) {
fetcher->timing_.ResetTimingReport();
}
timing_report_sender_->Send(timing_report_.size());
}
void EventLoop::UpdateTimingReport() {
// We need to support senders and fetchers changing while we are setting up
// the event loop. Otherwise we can't fetch or send until the loop runs. This
// means that on each change, we need to redo all this work. This makes setup
// more expensive, but not by all that much on a modern processor.
// Now, build up a report with everything pre-filled out.
flatbuffers::FlatBufferBuilder fbb;
fbb.ForceDefaults(true);
// Pre-fill in the defaults for timers.
std::vector<flatbuffers::Offset<timing::Timer>> timer_offsets;
for (const std::unique_ptr<TimerHandler> &timer : timers_) {
flatbuffers::Offset<timing::Statistic> wakeup_latency_offset =
timing::CreateStatistic(fbb);
flatbuffers::Offset<timing::Statistic> handler_time_offset =
timing::CreateStatistic(fbb);
flatbuffers::Offset<flatbuffers::String> name_offset;
if (timer->name().size() != 0) {
name_offset = fbb.CreateString(timer->name());
}
timing::Timer::Builder timer_builder(fbb);
if (timer->name().size() != 0) {
timer_builder.add_name(name_offset);
}
timer_builder.add_wakeup_latency(wakeup_latency_offset);
timer_builder.add_handler_time(handler_time_offset);
timer_builder.add_count(0);
timer_offsets.emplace_back(timer_builder.Finish());
}
// Pre-fill in the defaults for phased_loops.
std::vector<flatbuffers::Offset<timing::Timer>> phased_loop_offsets;
for (const std::unique_ptr<PhasedLoopHandler> &phased_loop : phased_loops_) {
flatbuffers::Offset<timing::Statistic> wakeup_latency_offset =
timing::CreateStatistic(fbb);
flatbuffers::Offset<timing::Statistic> handler_time_offset =
timing::CreateStatistic(fbb);
flatbuffers::Offset<flatbuffers::String> name_offset;
if (phased_loop->name().size() != 0) {
name_offset = fbb.CreateString(phased_loop->name());
}
timing::Timer::Builder timer_builder(fbb);
if (phased_loop->name().size() != 0) {
timer_builder.add_name(name_offset);
}
timer_builder.add_wakeup_latency(wakeup_latency_offset);
timer_builder.add_handler_time(handler_time_offset);
timer_builder.add_count(0);
phased_loop_offsets.emplace_back(timer_builder.Finish());
}
// Pre-fill in the defaults for watchers.
std::vector<flatbuffers::Offset<timing::Watcher>> watcher_offsets;
for (const std::unique_ptr<WatcherState> &watcher : watchers_) {
flatbuffers::Offset<timing::Statistic> wakeup_latency_offset =
timing::CreateStatistic(fbb);
flatbuffers::Offset<timing::Statistic> handler_time_offset =
timing::CreateStatistic(fbb);
timing::Watcher::Builder watcher_builder(fbb);
watcher_builder.add_channel_index(watcher->channel_index());
watcher_builder.add_wakeup_latency(wakeup_latency_offset);
watcher_builder.add_handler_time(handler_time_offset);
watcher_builder.add_count(0);
watcher_offsets.emplace_back(watcher_builder.Finish());
}
// Pre-fill in the defaults for senders.
std::vector<flatbuffers::Offset<timing::Sender>> sender_offsets;
for (const RawSender *sender : senders_) {
flatbuffers::Offset<timing::Statistic> size_offset =
timing::CreateStatistic(fbb);
timing::Sender::Builder sender_builder(fbb);
sender_builder.add_channel_index(sender->timing_.channel_index);
sender_builder.add_size(size_offset);
sender_builder.add_count(0);
sender_offsets.emplace_back(sender_builder.Finish());
}
// Pre-fill in the defaults for fetchers.
std::vector<flatbuffers::Offset<timing::Fetcher>> fetcher_offsets;
for (RawFetcher *fetcher : fetchers_) {
flatbuffers::Offset<timing::Statistic> latency_offset =
timing::CreateStatistic(fbb);
timing::Fetcher::Builder fetcher_builder(fbb);
fetcher_builder.add_channel_index(fetcher->timing_.channel_index);
fetcher_builder.add_count(0);
fetcher_builder.add_latency(latency_offset);
fetcher_offsets.emplace_back(fetcher_builder.Finish());
}
// Then build the final report.
flatbuffers::Offset<flatbuffers::Vector<flatbuffers::Offset<timing::Timer>>>
timers_offset;
if (timer_offsets.size() > 0) {
timers_offset = fbb.CreateVector(timer_offsets);
}
flatbuffers::Offset<flatbuffers::Vector<flatbuffers::Offset<timing::Timer>>>
phased_loops_offset;
if (phased_loop_offsets.size() > 0) {
phased_loops_offset = fbb.CreateVector(phased_loop_offsets);
}
flatbuffers::Offset<flatbuffers::Vector<flatbuffers::Offset<timing::Watcher>>>
watchers_offset;
if (watcher_offsets.size() > 0) {
watchers_offset = fbb.CreateVector(watcher_offsets);
}
flatbuffers::Offset<flatbuffers::Vector<flatbuffers::Offset<timing::Sender>>>
senders_offset;
if (sender_offsets.size() > 0) {
senders_offset = fbb.CreateVector(sender_offsets);
}
flatbuffers::Offset<flatbuffers::Vector<flatbuffers::Offset<timing::Fetcher>>>
fetchers_offset;
if (fetcher_offsets.size() > 0) {
fetchers_offset = fbb.CreateVector(fetcher_offsets);
}
flatbuffers::Offset<flatbuffers::String> name_offset =
fbb.CreateString(name());
timing::Report::Builder report_builder(fbb);
report_builder.add_name(name_offset);
report_builder.add_pid(GetTid());
if (timer_offsets.size() > 0) {
report_builder.add_timers(timers_offset);
}
if (phased_loop_offsets.size() > 0) {
report_builder.add_phased_loops(phased_loops_offset);
}
if (watcher_offsets.size() > 0) {
report_builder.add_watchers(watchers_offset);
}
if (sender_offsets.size() > 0) {
report_builder.add_senders(senders_offset);
}
if (fetcher_offsets.size() > 0) {
report_builder.add_fetchers(fetchers_offset);
}
fbb.Finish(report_builder.Finish());
timing_report_ = FlatbufferDetachedBuffer<timing::Report>(fbb.Release());
// Now that the pointers are stable, pass them to the timers and watchers to
// be updated.
for (size_t i = 0; i < timers_.size(); ++i) {
timers_[i]->timing_.set_timing_report(
timing_report_.mutable_message()->mutable_timers()->GetMutableObject(
i));
}
for (size_t i = 0; i < phased_loops_.size(); ++i) {
phased_loops_[i]->timing_.set_timing_report(
timing_report_.mutable_message()
->mutable_phased_loops()
->GetMutableObject(i));
}
for (size_t i = 0; i < watchers_.size(); ++i) {
watchers_[i]->set_timing_report(
timing_report_.mutable_message()->mutable_watchers()->GetMutableObject(
i));
}
for (size_t i = 0; i < senders_.size(); ++i) {
senders_[i]->timing_.set_timing_report(
timing_report_.mutable_message()->mutable_senders()->GetMutableObject(
i));
}
for (size_t i = 0; i < fetchers_.size(); ++i) {
fetchers_[i]->timing_.set_timing_report(
timing_report_.mutable_message()->mutable_fetchers()->GetMutableObject(
i));
}
}
void EventLoop::MaybeScheduleTimingReports() {
if (FLAGS_timing_reports && !skip_timing_report_) {
CHECK(!timing_report_sender_) << ": Timing reports already scheduled.";
// Make a raw sender for the report.
const Channel *channel = configuration::GetChannel(
configuration(), "/aos", timing::Report::GetFullyQualifiedName(),
name(), node());
CHECK(channel != nullptr) << ": Failed to look up {\"name\": \"/aos\", "
"\"type\": \"aos.timing.Report\"} on node "
<< FlatbufferToJson(node());
// Since we are using a RawSender, validity isn't checked. So check it
// ourselves.
if (!configuration::ChannelIsSendableOnNode(channel, node())) {
LOG(FATAL) << "Channel { \"name\": \"/aos"
<< channel->name()->string_view() << "\", \"type\": \""
<< channel->type()->string_view()
<< "\" } is not able to be sent on this node. Check your "
"configuration.";
}
CHECK(channel != nullptr) << ": Channel { \"name\": \"/aos\", \"type\": \""
<< timing::Report::GetFullyQualifiedName()
<< "\" } not found in config.";
timing_report_sender_ = MakeRawSender(channel);
// Register a handler which sends the report out by copying the raw data
// from the prebuilt and subsequently modified report.
TimerHandler *timing_reports_timer =
AddTimer([this]() { SendTimingReport(); });
// Set it up to send once per second.
timing_reports_timer->set_name("timing_reports");
OnRun([this, timing_reports_timer]() {
timing_reports_timer->Setup(
monotonic_now() + std::chrono::milliseconds(FLAGS_timing_report_ms),
std::chrono::milliseconds(FLAGS_timing_report_ms));
});
UpdateTimingReport();
}
}
void EventLoop::ReserveEvents() {
events_.reserve(timers_.size() + phased_loops_.size() + watchers_.size());
}
namespace {
bool CompareEvents(const EventLoopEvent *first, const EventLoopEvent *second) {
if (first->event_time() > second->event_time()) {
return true;
}
if (first->event_time() < second->event_time()) {
return false;
}
return first->generation() > second->generation();
}
} // namespace
void EventLoop::AddEvent(EventLoopEvent *event) {
DCHECK(std::find(events_.begin(), events_.end(), event) == events_.end());
DCHECK(event->generation() == 0);
event->set_generation(++event_generation_);
events_.push_back(event);
std::push_heap(events_.begin(), events_.end(), CompareEvents);
}
void EventLoop::RemoveEvent(EventLoopEvent *event) {
auto e = std::find(events_.begin(), events_.end(), event);
if (e != events_.end()) {
DCHECK(event->generation() != 0);
events_.erase(e);
std::make_heap(events_.begin(), events_.end(), CompareEvents);
event->Invalidate();
}
}
EventLoopEvent *EventLoop::PopEvent() {
EventLoopEvent *result = events_.front();
std::pop_heap(events_.begin(), events_.end(), CompareEvents);
events_.pop_back();
result->Invalidate();
return result;
}
void WatcherState::set_timing_report(timing::Watcher *watcher) {
CHECK_NOTNULL(watcher);
watcher_ = watcher;
wakeup_latency_.set_statistic(watcher->mutable_wakeup_latency());
handler_time_.set_statistic(watcher->mutable_handler_time());
}
void WatcherState::ResetReport() {
wakeup_latency_.Reset();
handler_time_.Reset();
watcher_->mutate_count(0);
}
} // namespace aos