Add a BootTimestamp class and move some things over
This makes it so we can compare timestamps again across boots. Move
most things over to using it instead so we can more naturally track what
happens in what order.
Note: this is not complete and has no tests. Anywhere where I'm not
comfortable that the code will handle it today should now CHECK that the
count is 0 so we explode. Tests and tolerance of reboots will come in
future patches, starting with TimestampMapper and then with the
time recover code and then LogReader.
Change-Id: I0fdbbe1860de4fe5e48d1a6cd516e7118e5942dc
Signed-off-by: Austin Schuh <austin.schuh@bluerivertech.com>
diff --git a/aos/events/logging/BUILD b/aos/events/logging/BUILD
index 0123270..fead91d 100644
--- a/aos/events/logging/BUILD
+++ b/aos/events/logging/BUILD
@@ -15,10 +15,12 @@
cc_library(
name = "logfile_utils",
srcs = [
+ "boot_timestamp.cc",
"logfile_sorting.cc",
"logfile_utils.cc",
],
hdrs = [
+ "boot_timestamp.h",
"logfile_sorting.h",
"logfile_utils.h",
],
diff --git a/aos/events/logging/boot_timestamp.cc b/aos/events/logging/boot_timestamp.cc
new file mode 100644
index 0000000..c037633
--- /dev/null
+++ b/aos/events/logging/boot_timestamp.cc
@@ -0,0 +1,14 @@
+#include "aos/events/logging/boot_timestamp.h"
+
+#include <iostream>
+
+#include "aos/time/time.h"
+
+namespace aos::logger {
+std::ostream &operator<<(std::ostream &os,
+ const struct BootTimestamp ×tamp) {
+ return os << "{.boot=" << timestamp.boot << ", .time=" << timestamp.time
+ << "}";
+}
+
+} // namespace aos::logger
diff --git a/aos/events/logging/boot_timestamp.h b/aos/events/logging/boot_timestamp.h
new file mode 100644
index 0000000..ea66c8e
--- /dev/null
+++ b/aos/events/logging/boot_timestamp.h
@@ -0,0 +1,79 @@
+#ifndef AOS_EVENTS_LOGGING_BOOT_TIMESTAMP_H_
+#define AOS_EVENTS_LOGGING_BOOT_TIMESTAMP_H_
+
+#include <iostream>
+
+#include "aos/time/time.h"
+
+namespace aos::logger {
+
+// Simple class representing which boot and what monotonic time in that boot.
+// Boots are assumed to be sequential, and the monotonic clock resets on reboot
+// for all the compare operations.
+struct BootTimestamp {
+ // Boot number for this timestamp.
+ size_t boot = 0u;
+ // Monotonic time in that boot.
+ monotonic_clock::time_point time = monotonic_clock::min_time;
+
+ static constexpr BootTimestamp min_time() {
+ return BootTimestamp{.boot = 0u, .time = monotonic_clock::min_time};
+ }
+ static constexpr BootTimestamp max_time() {
+ return BootTimestamp{.boot = std::numeric_limits<size_t>::max(),
+ .time = monotonic_clock::max_time};
+ }
+
+ // Compare operators. These are implemented such that earlier boots always
+ // compare less than later boots, and the times are only compared in a single
+ // boot.
+ bool operator<(const BootTimestamp &m2) const;
+ bool operator<=(const BootTimestamp &m2) const;
+ bool operator>=(const BootTimestamp &m2) const;
+ bool operator>(const BootTimestamp &m2) const;
+ bool operator==(const BootTimestamp &m2) const;
+ bool operator!=(const BootTimestamp &m2) const;
+};
+
+std::ostream &operator<<(std::ostream &os,
+ const struct BootTimestamp ×tamp);
+
+inline bool BootTimestamp::operator<(const BootTimestamp &m2) const {
+ if (boot != m2.boot) {
+ return boot < m2.boot;
+ }
+
+ return time < m2.time;
+}
+inline bool BootTimestamp::operator<=(const BootTimestamp &m2) const {
+ if (boot != m2.boot) {
+ return boot <= m2.boot;
+ }
+
+ return time <= m2.time;
+}
+inline bool BootTimestamp::operator>=(const BootTimestamp &m2) const {
+ if (boot != m2.boot) {
+ return boot >= m2.boot;
+ }
+
+ return time >= m2.time;
+}
+inline bool BootTimestamp::operator>(const BootTimestamp &m2) const {
+ if (boot != m2.boot) {
+ return boot > m2.boot;
+ }
+
+ return time > m2.time;
+}
+
+inline bool BootTimestamp::operator==(const BootTimestamp &m2) const {
+ return boot == m2.boot && time == m2.time;
+}
+inline bool BootTimestamp::operator!=(const BootTimestamp &m2) const {
+ return boot != m2.boot || time != m2.time;
+}
+
+} // namespace aos::logger
+
+#endif // AOS_EVENTS_LOGGING_BOOT_TIMESTAMP_H_
diff --git a/aos/events/logging/log_reader.cc b/aos/events/logging/log_reader.cc
index e295fc1..3cf7fd5 100644
--- a/aos/events/logging/log_reader.cc
+++ b/aos/events/logging/log_reader.cc
@@ -11,6 +11,7 @@
#include "absl/strings/escaping.h"
#include "absl/types/span.h"
#include "aos/events/event_loop.h"
+#include "aos/events/logging/boot_timestamp.h"
#include "aos/events/logging/logfile_sorting.h"
#include "aos/events/logging/logger_generated.h"
#include "aos/flatbuffer_merge.h"
@@ -280,7 +281,8 @@
states_[configuration::GetNodeIndex(configuration(), node)].get();
CHECK(state != nullptr) << ": Unknown node " << FlatbufferToJson(node);
- return state->monotonic_start_time();
+ // TODO(austin): Un-hard-code the 0 boot count.
+ return state->monotonic_start_time(0);
}
realtime_clock::time_point LogReader::realtime_start_time(
@@ -289,7 +291,8 @@
states_[configuration::GetNodeIndex(configuration(), node)].get();
CHECK(state != nullptr) << ": Unknown node " << FlatbufferToJson(node);
- return state->realtime_start_time();
+ // TODO(austin): Un-hard-code the 0 boot count.
+ return state->realtime_start_time(0);
}
void LogReader::Register() {
@@ -393,16 +396,16 @@
// running until the last node.
for (std::unique_ptr<State> &state : states_) {
- VLOG(1) << "Start time is " << state->monotonic_start_time() << " for node "
+ VLOG(1) << "Start time is " << state->monotonic_start_time(0) << " for node "
<< MaybeNodeName(state->event_loop()->node()) << "now "
<< state->monotonic_now();
- if (state->monotonic_start_time() == monotonic_clock::min_time) {
+ if (state->monotonic_start_time(0) == monotonic_clock::min_time) {
continue;
}
// And start computing the start time on the distributed clock now that
// that works.
start_time = std::max(
- start_time, state->ToDistributedClock(state->monotonic_start_time()));
+ start_time, state->ToDistributedClock(state->monotonic_start_time(0)));
}
// TODO(austin): If a node doesn't have a start time, we might not queue
@@ -452,11 +455,11 @@
for (std::unique_ptr<State> &state : states_) {
// Make the RT clock be correct before handing it to the user.
- if (state->realtime_start_time() != realtime_clock::min_time) {
- state->SetRealtimeOffset(state->monotonic_start_time(),
- state->realtime_start_time());
+ if (state->realtime_start_time(0) != realtime_clock::min_time) {
+ state->SetRealtimeOffset(state->monotonic_start_time(0),
+ state->realtime_start_time(0));
}
- VLOG(1) << "Start time is " << state->monotonic_start_time() << " for node "
+ VLOG(1) << "Start time is " << state->monotonic_start_time(0) << " for node "
<< MaybeNodeName(state->event_loop()->node()) << "now "
<< state->monotonic_now();
}
@@ -555,16 +558,20 @@
}
TimestampedMessage timestamped_message = state->PopOldest();
+ CHECK_EQ(timestamped_message.monotonic_event_time.boot, 0u);
+ CHECK_EQ(timestamped_message.monotonic_remote_time.boot, 0u);
+ CHECK_EQ(timestamped_message.monotonic_timestamp_time.boot, 0u);
const monotonic_clock::time_point monotonic_now =
state->event_loop()->context().monotonic_event_time;
if (!FLAGS_skip_order_validation) {
- CHECK(monotonic_now == timestamped_message.monotonic_event_time)
+ CHECK(monotonic_now == timestamped_message.monotonic_event_time.time)
<< ": " << FlatbufferToJson(state->event_loop()->node()) << " Now "
<< monotonic_now << " trying to send "
<< timestamped_message.monotonic_event_time << " failure "
<< state->DebugString();
- } else if (monotonic_now != timestamped_message.monotonic_event_time) {
+ } else if (BootTimestamp{.boot = 0u, .time = monotonic_now} !=
+ timestamped_message.monotonic_event_time) {
LOG(WARNING) << "Check failed: monotonic_now == "
"timestamped_message.monotonic_event_time) ("
<< monotonic_now << " vs. "
@@ -575,12 +582,13 @@
<< state->DebugString();
}
- if (timestamped_message.monotonic_event_time >
- state->monotonic_start_time() ||
+ if (timestamped_message.monotonic_event_time.time >
+ state->monotonic_start_time(
+ timestamped_message.monotonic_event_time.boot) ||
event_loop_factory_ != nullptr) {
if (timestamped_message.data.span().size() != 0u) {
if (timestamped_message.monotonic_remote_time !=
- monotonic_clock::min_time) {
+ BootTimestamp::min_time()) {
// Confirm that the message was sent on the sending node before the
// destination node (this node). As a proxy, do this by making sure
// that time on the source node is past when the message was sent.
@@ -591,7 +599,7 @@
// fix that.
if (!FLAGS_skip_order_validation) {
CHECK_LE(
- timestamped_message.monotonic_remote_time,
+ timestamped_message.monotonic_remote_time.time,
state->monotonic_remote_now(timestamped_message.channel_index))
<< state->event_loop()->node()->name()->string_view() << " to "
<< state->remote_node(timestamped_message.channel_index)
@@ -602,7 +610,7 @@
logged_configuration()->channels()->Get(
timestamped_message.channel_index))
<< " " << state->DebugString();
- } else if (timestamped_message.monotonic_remote_time >
+ } else if (timestamped_message.monotonic_remote_time.time >
state->monotonic_remote_now(
timestamped_message.channel_index)) {
LOG(WARNING)
@@ -620,18 +628,18 @@
<< " currently " << timestamped_message.monotonic_event_time
<< " ("
<< state->ToDistributedClock(
- timestamped_message.monotonic_event_time)
+ timestamped_message.monotonic_event_time.time)
<< ") remote event time "
<< timestamped_message.monotonic_remote_time << " ("
<< state->RemoteToDistributedClock(
timestamped_message.channel_index,
- timestamped_message.monotonic_remote_time)
+ timestamped_message.monotonic_remote_time.time)
<< ") " << state->DebugString();
}
}
// If we have access to the factory, use it to fix the realtime time.
- state->SetRealtimeOffset(timestamped_message.monotonic_event_time,
+ state->SetRealtimeOffset(timestamped_message.monotonic_event_time.time,
timestamped_message.realtime_event_time);
VLOG(1) << MaybeNodeName(state->event_loop()->node()) << "Sending "
@@ -643,8 +651,9 @@
// When starting up, we can have data which was sent before the
// log starts, but the timestamp was after the log starts. This
// is unreasonable to avoid, so ignore the missing data.
- timestamped_message.monotonic_remote_time >=
+ timestamped_message.monotonic_remote_time.time >=
state->monotonic_remote_start_time(
+ timestamped_message.monotonic_remote_time.boot,
timestamped_message.channel_index) &&
!FLAGS_skip_missing_forwarding_entries) {
// We've found a timestamp without data that we expect to have data for.
@@ -725,13 +734,15 @@
}
}
} else {
- LOG(WARNING)
- << "Not sending data from before the start of the log file. "
- << timestamped_message.monotonic_event_time.time_since_epoch().count()
- << " start " << monotonic_start_time().time_since_epoch().count()
- << " "
- << FlatbufferToJson(timestamped_message.data,
- {.multi_line = false, .max_vector_size = 100});
+ LOG(WARNING) << "Not sending data from before the start of the log file. "
+ << timestamped_message.monotonic_event_time.time
+ .time_since_epoch()
+ .count()
+ << " start "
+ << monotonic_start_time().time_since_epoch().count() << " "
+ << FlatbufferToJson(
+ timestamped_message.data,
+ {.multi_line = false, .max_vector_size = 100});
}
const monotonic_clock::time_point next_time = state->OldestMessageTime();
@@ -1176,7 +1187,8 @@
uint32_t queue_index;
} search;
- search.monotonic_event_time = timestamped_message.monotonic_remote_time;
+ CHECK_EQ(timestamped_message.monotonic_remote_time.boot, 0u);
+ search.monotonic_event_time = timestamped_message.monotonic_remote_time.time;
search.queue_index = timestamped_message.remote_queue_index;
// Find the sent time if available.
@@ -1207,9 +1219,11 @@
// other node isn't done yet. So there is no send time, but there is a
// receive time.
if (element != queue_index_map->end()) {
- CHECK_GE(timestamped_message.monotonic_remote_time,
+ CHECK_EQ(timestamped_message.monotonic_remote_time.boot, 0u);
+
+ CHECK_GE(timestamped_message.monotonic_remote_time.time,
element->starting_monotonic_event_time);
- CHECK_LE(timestamped_message.monotonic_remote_time,
+ CHECK_LE(timestamped_message.monotonic_remote_time.time,
element->ending_monotonic_event_time);
CHECK_GE(timestamped_message.remote_queue_index,
element->starting_queue_index);
@@ -1226,10 +1240,11 @@
// Send! Use the replayed queue index here instead of the logged queue index
// for the remote queue index. This makes re-logging work.
+ CHECK_EQ(timestamped_message.monotonic_remote_time.boot, 0u);
const bool sent = sender->Send(
timestamped_message.data.message().data()->Data(),
timestamped_message.data.message().data()->size(),
- timestamped_message.monotonic_remote_time,
+ timestamped_message.monotonic_remote_time.time,
timestamped_message.realtime_remote_time, remote_queue_index,
(channel_source_state_[timestamped_message.channel_index] != nullptr
? CHECK_NOTNULL(
@@ -1242,9 +1257,10 @@
if (queue_index_map_[timestamped_message.channel_index]->empty()) {
// Nothing here, start a range with 0 length.
ContiguousSentTimestamp timestamp;
+ CHECK_EQ(timestamped_message.monotonic_event_time.boot, 0u);
timestamp.starting_monotonic_event_time =
timestamp.ending_monotonic_event_time =
- timestamped_message.monotonic_event_time;
+ timestamped_message.monotonic_event_time.time;
timestamp.starting_queue_index = timestamp.ending_queue_index =
timestamped_message.queue_index;
timestamp.actual_queue_index = sender->sent_queue_index();
@@ -1258,14 +1274,16 @@
if ((back->starting_queue_index - back->actual_queue_index) ==
(timestamped_message.queue_index - sender->sent_queue_index())) {
back->ending_queue_index = timestamped_message.queue_index;
+ CHECK_EQ(timestamped_message.monotonic_event_time.boot, 0u);
back->ending_monotonic_event_time =
- timestamped_message.monotonic_event_time;
+ timestamped_message.monotonic_event_time.time;
} else {
// Otherwise, make a new one.
+ CHECK_EQ(timestamped_message.monotonic_event_time.boot, 0u);
ContiguousSentTimestamp timestamp;
timestamp.starting_monotonic_event_time =
timestamp.ending_monotonic_event_time =
- timestamped_message.monotonic_event_time;
+ timestamped_message.monotonic_event_time.time;
timestamp.starting_queue_index = timestamp.ending_queue_index =
timestamped_message.queue_index;
timestamp.actual_queue_index = sender->sent_queue_index();
@@ -1297,8 +1315,10 @@
sender->realtime_sent_time().time_since_epoch().count());
message_header_builder.add_queue_index(sender->sent_queue_index());
+ CHECK_EQ(timestamped_message.monotonic_remote_time.boot, 0u);
message_header_builder.add_monotonic_remote_time(
- timestamped_message.monotonic_remote_time.time_since_epoch().count());
+ timestamped_message.monotonic_remote_time.time.time_since_epoch()
+ .count());
message_header_builder.add_realtime_remote_time(
timestamped_message.realtime_remote_time.time_since_epoch().count());
@@ -1307,9 +1327,10 @@
fbb.Finish(message_header_builder.Finish());
+ CHECK_EQ(timestamped_message.monotonic_timestamp_time.boot, 0u);
remote_timestamp_senders_[timestamped_message.channel_index]->Send(
FlatbufferDetachedBuffer<RemoteMessage>(fbb.Release()),
- timestamped_message.monotonic_timestamp_time);
+ timestamped_message.monotonic_timestamp_time.time);
}
return true;
@@ -1435,7 +1456,9 @@
timestamp_mapper_->PopFront();
SeedSortedMessages();
- if (result.monotonic_remote_time != monotonic_clock::min_time) {
+ CHECK_EQ(result.monotonic_remote_time.boot, 0u);
+
+ if (result.monotonic_remote_time.time != monotonic_clock::min_time) {
message_bridge::NoncausalOffsetEstimator *filter =
filters_[result.channel_index];
CHECK(filter != nullptr);
@@ -1459,9 +1482,10 @@
if (result_ptr == nullptr) {
return monotonic_clock::max_time;
}
+ CHECK_EQ(result_ptr->monotonic_event_time.boot, 0u);
VLOG(2) << MaybeNodeName(event_loop_->node()) << "oldest message at "
- << result_ptr->monotonic_event_time;
- return result_ptr->monotonic_event_time;
+ << result_ptr->monotonic_event_time.time;
+ return result_ptr->monotonic_event_time.time;
}
void LogReader::State::SeedSortedMessages() {
diff --git a/aos/events/logging/log_reader.h b/aos/events/logging/log_reader.h
index 37163c1..5cfb5df 100644
--- a/aos/events/logging/log_reader.h
+++ b/aos/events/logging/log_reader.h
@@ -258,13 +258,15 @@
void SeedSortedMessages();
// Returns the starting time for this node.
- monotonic_clock::time_point monotonic_start_time() const {
- return timestamp_mapper_ ? timestamp_mapper_->monotonic_start_time()
- : monotonic_clock::min_time;
+ monotonic_clock::time_point monotonic_start_time(size_t boot_count) const {
+ return timestamp_mapper_
+ ? timestamp_mapper_->monotonic_start_time(boot_count)
+ : monotonic_clock::min_time;
}
- realtime_clock::time_point realtime_start_time() const {
- return timestamp_mapper_ ? timestamp_mapper_->realtime_start_time()
- : realtime_clock::min_time;
+ realtime_clock::time_point realtime_start_time(size_t boot_count) const {
+ return timestamp_mapper_
+ ? timestamp_mapper_->realtime_start_time(boot_count)
+ : realtime_clock::min_time;
}
// Sets the node event loop factory for replaying into a
@@ -307,8 +309,10 @@
// Returns the start time of the remote for the provided channel.
monotonic_clock::time_point monotonic_remote_start_time(
+ size_t boot_count,
size_t channel_index) {
- return channel_source_state_[channel_index]->monotonic_start_time();
+ return channel_source_state_[channel_index]->monotonic_start_time(
+ boot_count);
}
distributed_clock::time_point RemoteToDistributedClock(
diff --git a/aos/events/logging/logfile_utils.cc b/aos/events/logging/logfile_utils.cc
index f4d4501..f008ac6 100644
--- a/aos/events/logging/logfile_utils.cc
+++ b/aos/events/logging/logfile_utils.cc
@@ -527,11 +527,11 @@
}
bool Message::operator<(const Message &m2) const {
- CHECK_EQ(this->boot_count, m2.boot_count);
+ CHECK_EQ(this->timestamp.boot, m2.timestamp.boot);
- if (this->timestamp < m2.timestamp) {
+ if (this->timestamp.time < m2.timestamp.time) {
return true;
- } else if (this->timestamp > m2.timestamp) {
+ } else if (this->timestamp.time > m2.timestamp.time) {
return false;
}
@@ -546,16 +546,15 @@
bool Message::operator>=(const Message &m2) const { return !(*this < m2); }
bool Message::operator==(const Message &m2) const {
- CHECK_EQ(this->boot_count, m2.boot_count);
+ CHECK_EQ(this->timestamp.boot, m2.timestamp.boot);
- return timestamp == m2.timestamp && channel_index == m2.channel_index &&
- queue_index == m2.queue_index;
+ return timestamp.time == m2.timestamp.time &&
+ channel_index == m2.channel_index && queue_index == m2.queue_index;
}
std::ostream &operator<<(std::ostream &os, const Message &m) {
os << "{.channel_index=" << m.channel_index
- << ", .queue_index=" << m.queue_index << ", .timestamp=" << m.timestamp
- << ", .boot_count=" << m.boot_count;
+ << ", .queue_index=" << m.queue_index << ", .timestamp=" << m.timestamp;
if (m.data.Verify()) {
os << ", .data="
<< aos::FlatbufferToJson(m.data,
@@ -573,13 +572,13 @@
if (m.remote_queue_index != 0xffffffff) {
os << ", .remote_queue_index=" << m.remote_queue_index;
}
- if (m.monotonic_remote_time != monotonic_clock::min_time) {
+ if (m.monotonic_remote_time != BootTimestamp::min_time()) {
os << ", .monotonic_remote_time=" << m.monotonic_remote_time;
}
if (m.realtime_remote_time != realtime_clock::min_time) {
os << ", .realtime_remote_time=" << m.realtime_remote_time;
}
- if (m.monotonic_timestamp_time != monotonic_clock::min_time) {
+ if (m.monotonic_timestamp_time != BootTimestamp::min_time()) {
os << ", .monotonic_timestamp_time=" << m.monotonic_timestamp_time;
}
if (m.data.Verify()) {
@@ -600,7 +599,7 @@
// sure the nothing path is checked quickly.
if (sorted_until() != monotonic_clock::max_time) {
while (true) {
- if (!messages_.empty() && messages_.begin()->timestamp < sorted_until() &&
+ if (!messages_.empty() && messages_.begin()->timestamp.time < sorted_until() &&
sorted_until() >= monotonic_start_time()) {
break;
}
@@ -616,9 +615,11 @@
messages_.insert(Message{
.channel_index = m.value().message().channel_index(),
.queue_index = m.value().message().queue_index(),
- .timestamp = monotonic_clock::time_point(std::chrono::nanoseconds(
- m.value().message().monotonic_sent_time())),
- .boot_count = parts().boot_count,
+ .timestamp =
+ BootTimestamp{
+ .boot = parts().boot_count,
+ .time = monotonic_clock::time_point(std::chrono::nanoseconds(
+ m.value().message().monotonic_sent_time()))},
.data = std::move(m.value())});
// Now, update sorted_until_ to match the new message.
@@ -640,9 +641,9 @@
return nullptr;
}
- CHECK_GE(messages_.begin()->timestamp, last_message_time_)
+ CHECK_GE(messages_.begin()->timestamp.time, last_message_time_)
<< DebugString() << " reading " << parts_message_reader_.filename();
- last_message_time_ = messages_.begin()->timestamp;
+ last_message_time_ = messages_.begin()->timestamp.time;
return &(*messages_.begin());
}
@@ -707,7 +708,7 @@
// Return the current Front if we have one, otherwise go compute one.
if (current_ != nullptr) {
Message *result = current_->Front();
- CHECK_GE(result->timestamp, last_message_time_);
+ CHECK_GE(result->timestamp.time, last_message_time_);
return result;
}
@@ -745,8 +746,8 @@
}
if (oldest) {
- CHECK_GE(oldest->timestamp, last_message_time_);
- last_message_time_ = oldest->timestamp;
+ CHECK_GE(oldest->timestamp.time, last_message_time_);
+ last_message_time_ = oldest->timestamp.time;
} else {
last_message_time_ = monotonic_clock::max_time;
}
@@ -805,10 +806,22 @@
void BootMerger::PopFront() { node_mergers_[index_]->PopFront(); }
+std::vector<const LogParts *> BootMerger::Parts() const {
+ std::vector<const LogParts *> results;
+ for (const std::unique_ptr<NodeMerger> &node_merger : node_mergers_) {
+ std::vector<const LogParts *> node_parts = node_merger->Parts();
+
+ results.insert(results.end(), std::make_move_iterator(node_parts.begin()),
+ std::make_move_iterator(node_parts.end()));
+ }
+
+ return results;
+}
+
TimestampMapper::TimestampMapper(std::vector<LogParts> parts)
- : node_merger_(std::move(parts)),
+ : boot_merger_(std::move(parts)),
timestamp_callback_([](TimestampedMessage *) {}) {
- for (const LogParts *part : node_merger_.Parts()) {
+ for (const LogParts *part : boot_merger_.Parts()) {
if (!configuration_) {
configuration_ = part->config;
} else {
@@ -873,9 +886,9 @@
.realtime_event_time = aos::realtime_clock::time_point(
std::chrono::nanoseconds(m->data.message().realtime_sent_time())),
.remote_queue_index = 0xffffffff,
- .monotonic_remote_time = monotonic_clock::min_time,
+ .monotonic_remote_time = BootTimestamp::min_time(),
.realtime_remote_time = realtime_clock::min_time,
- .monotonic_timestamp_time = monotonic_clock::min_time,
+ .monotonic_timestamp_time = BootTimestamp::min_time(),
.data = std::move(m->data)});
}
@@ -904,7 +917,7 @@
if (nodes_data_.empty()) {
// Simple path. We are single node, so there are no timestamps to match!
CHECK_EQ(messages_.size(), 0u);
- Message *m = node_merger_.Front();
+ Message *m = boot_merger_.Front();
if (!m) {
return false;
}
@@ -916,7 +929,7 @@
last_message_time_ = matched_messages_.back().monotonic_event_time;
// We are thin wrapper around node_merger. Call it directly.
- node_merger_.PopFront();
+ boot_merger_.PopFront();
timestamp_callback_(&matched_messages_.back());
return true;
}
@@ -931,7 +944,7 @@
}
// Now that it has been added (and cannibalized), forget about it upstream.
- node_merger_.PopFront();
+ boot_merger_.PopFront();
}
Message *m = &(messages_.front());
@@ -958,13 +971,14 @@
std::chrono::nanoseconds(m->data.message().realtime_sent_time())),
.remote_queue_index = m->data.message().remote_queue_index(),
.monotonic_remote_time =
- monotonic_clock::time_point(std::chrono::nanoseconds(
- m->data.message().monotonic_remote_time())),
+ // TODO(austin): 0 is wrong...
+ {0, monotonic_clock::time_point(std::chrono::nanoseconds(
+ m->data.message().monotonic_remote_time()))},
.realtime_remote_time = realtime_clock::time_point(
std::chrono::nanoseconds(m->data.message().realtime_remote_time())),
.monotonic_timestamp_time =
- monotonic_clock::time_point(std::chrono::nanoseconds(
- m->data.message().monotonic_timestamp_time())),
+ {0, monotonic_clock::time_point(std::chrono::nanoseconds(
+ m->data.message().monotonic_timestamp_time()))},
.data = std::move(data.data)});
CHECK_GE(matched_messages_.back().monotonic_event_time, last_message_time_);
last_message_time_ = matched_messages_.back().monotonic_event_time;
@@ -975,7 +989,7 @@
}
}
-void TimestampMapper::QueueUntil(monotonic_clock::time_point queue_time) {
+void TimestampMapper::QueueUntil(BootTimestamp queue_time) {
while (last_message_time_ <= queue_time) {
if (!QueueMatched()) {
return;
@@ -984,6 +998,11 @@
}
void TimestampMapper::QueueFor(chrono::nanoseconds time_estimation_buffer) {
+ // Note: queueing for time doesn't really work well across boots. So we just
+ // assume that if you are using this, you only care about the current boot.
+ //
+ // TODO(austin): Is that the right concept?
+ //
// Make sure we have something queued first. This makes the end time
// calculation simpler, and is typically what folks want regardless.
if (matched_messages_.empty()) {
@@ -993,14 +1012,15 @@
}
const aos::monotonic_clock::time_point end_queue_time =
- std::max(monotonic_start_time(),
- matched_messages_.front().monotonic_event_time) +
+ std::max(monotonic_start_time(
+ matched_messages_.front().monotonic_event_time.boot),
+ matched_messages_.front().monotonic_event_time.time) +
time_estimation_buffer;
// Place sorted messages on the list until we have
// --time_estimation_buffer_seconds seconds queued up (but queue at least
// until the log starts).
- while (end_queue_time >= last_message_time_) {
+ while (end_queue_time >= last_message_time_.time) {
if (!QueueMatched()) {
return;
}
@@ -1023,8 +1043,10 @@
CHECK(message.data.message().has_monotonic_remote_time());
CHECK(message.data.message().has_realtime_remote_time());
- const monotonic_clock::time_point monotonic_remote_time(
- std::chrono::nanoseconds(message.data.message().monotonic_remote_time()));
+ const BootTimestamp monotonic_remote_time{
+ .boot = 0,
+ .time = monotonic_clock::time_point(std::chrono::nanoseconds(
+ message.data.message().monotonic_remote_time()))};
const realtime_clock::time_point realtime_remote_time(
std::chrono::nanoseconds(message.data.message().realtime_remote_time()));
@@ -1034,11 +1056,12 @@
// asked to pull a timestamp from a peer which doesn't exist, return an empty
// message.
if (peer == nullptr) {
+ // TODO(austin): Make sure the tests hit all these paths with a boot count
+ // of 1...
return Message{
.channel_index = message.channel_index,
.queue_index = remote_queue_index,
.timestamp = monotonic_remote_time,
- .boot_count = 0,
.data = SizePrefixedFlatbufferVector<MessageHeader>::Empty()};
}
@@ -1053,7 +1076,6 @@
.channel_index = message.channel_index,
.queue_index = remote_queue_index,
.timestamp = monotonic_remote_time,
- .boot_count = 0,
.data = SizePrefixedFlatbufferVector<MessageHeader>::Empty()};
}
@@ -1063,7 +1085,6 @@
.channel_index = message.channel_index,
.queue_index = remote_queue_index,
.timestamp = monotonic_remote_time,
- .boot_count = 0,
.data = SizePrefixedFlatbufferVector<MessageHeader>::Empty()};
}
@@ -1098,7 +1119,6 @@
.channel_index = message.channel_index,
.queue_index = remote_queue_index,
.timestamp = monotonic_remote_time,
- .boot_count = 0,
.data = SizePrefixedFlatbufferVector<MessageHeader>::Empty()};
}
@@ -1117,7 +1137,7 @@
}
}
-void TimestampMapper::QueueUnmatchedUntil(monotonic_clock::time_point t) {
+void TimestampMapper::QueueUnmatchedUntil(BootTimestamp t) {
if (queued_until_ > t) {
return;
}
@@ -1129,17 +1149,17 @@
if (!Queue()) {
// Found nothing to add, we are out of data!
- queued_until_ = monotonic_clock::max_time;
+ queued_until_ = BootTimestamp::max_time();
return;
}
// Now that it has been added (and cannibalized), forget about it upstream.
- node_merger_.PopFront();
+ boot_merger_.PopFront();
}
}
bool TimestampMapper::Queue() {
- Message *m = node_merger_.Front();
+ Message *m = boot_merger_.Front();
if (m == nullptr) {
return false;
}
diff --git a/aos/events/logging/logfile_utils.h b/aos/events/logging/logfile_utils.h
index 2632302..4bb481d 100644
--- a/aos/events/logging/logfile_utils.h
+++ b/aos/events/logging/logfile_utils.h
@@ -18,6 +18,7 @@
#include "absl/types/span.h"
#include "aos/containers/resizeable_buffer.h"
#include "aos/events/event_loop.h"
+#include "aos/events/logging/boot_timestamp.h"
#include "aos/events/logging/buffer_encoder.h"
#include "aos/events/logging/logfile_sorting.h"
#include "aos/events/logging/logger_generated.h"
@@ -348,10 +349,8 @@
uint32_t channel_index = 0xffffffff;
// The local queue index.
uint32_t queue_index = 0xffffffff;
- // The local timestamp on the monotonic clock.
- monotonic_clock::time_point timestamp = monotonic_clock::min_time;
- // The current boot count added on by SortParts.
- size_t boot_count = 0;
+ // The local timestamp.
+ BootTimestamp timestamp;
// The data (either a timestamp header, or a data header).
SizePrefixedFlatbufferVector<MessageHeader> data;
@@ -370,15 +369,14 @@
uint32_t channel_index = 0xffffffff;
uint32_t queue_index = 0xffffffff;
- monotonic_clock::time_point monotonic_event_time = monotonic_clock::min_time;
+ BootTimestamp monotonic_event_time;
realtime_clock::time_point realtime_event_time = realtime_clock::min_time;
uint32_t remote_queue_index = 0xffffffff;
- monotonic_clock::time_point monotonic_remote_time = monotonic_clock::min_time;
+ BootTimestamp monotonic_remote_time;
realtime_clock::time_point realtime_remote_time = realtime_clock::min_time;
- monotonic_clock::time_point monotonic_timestamp_time =
- monotonic_clock::min_time;
+ BootTimestamp monotonic_timestamp_time;
SizePrefixedFlatbufferVector<MessageHeader> data;
};
@@ -511,11 +509,13 @@
return node_mergers_[0]->configuration();
}
- monotonic_clock::time_point monotonic_start_time() const {
- return node_mergers_[index_]->monotonic_start_time();
+ monotonic_clock::time_point monotonic_start_time(size_t boot) const {
+ CHECK_LT(boot, node_mergers_.size());
+ return node_mergers_[boot]->monotonic_start_time();
}
- realtime_clock::time_point realtime_start_time() const {
- return node_mergers_[index_]->realtime_start_time();
+ realtime_clock::time_point realtime_start_time(size_t boot) const {
+ CHECK_LT(boot, node_mergers_.size());
+ return node_mergers_[boot]->realtime_start_time();
}
bool started() const {
@@ -561,14 +561,16 @@
const Configuration *configuration() const { return configuration_.get(); }
// Returns which node this is sorting for.
- size_t node() const { return node_merger_.node(); }
+ size_t node() const { return boot_merger_.node(); }
// The start time of this log.
- monotonic_clock::time_point monotonic_start_time() const {
- return node_merger_.monotonic_start_time();
+ // TODO(austin): This concept is probably wrong... We have start times per
+ // boot, and an order of them.
+ monotonic_clock::time_point monotonic_start_time(size_t boot) const {
+ return boot_merger_.monotonic_start_time(boot);
}
- realtime_clock::time_point realtime_start_time() const {
- return node_merger_.realtime_start_time();
+ realtime_clock::time_point realtime_start_time(size_t boot) const {
+ return boot_merger_.realtime_start_time(boot);
}
// Uses timestamp_mapper as the peer for its node. Only one mapper may be set
@@ -577,9 +579,7 @@
void AddPeer(TimestampMapper *timestamp_mapper);
// Returns true if anything has been queued up.
- bool started() const {
- return node_merger_.sorted_until() != monotonic_clock::min_time;
- }
+ bool started() const { return boot_merger_.started(); }
// Returns the next message for this node.
TimestampedMessage *Front();
@@ -590,7 +590,7 @@
std::string DebugString() const;
// Queues data the provided time.
- void QueueUntil(monotonic_clock::time_point queue_time);
+ void QueueUntil(BootTimestamp queue_time);
// Queues until we have time_estimation_buffer of data in the queue.
void QueueFor(std::chrono::nanoseconds time_estimation_buffer);
@@ -653,13 +653,13 @@
// Queues up data until we have at least one message >= to time t.
// Useful for triggering a remote node to read enough data to have the
// timestamp you care about available.
- void QueueUnmatchedUntil(monotonic_clock::time_point t);
+ void QueueUnmatchedUntil(BootTimestamp t);
// Queues m into matched_messages_.
void QueueMessage(Message *m);
// The node merger to source messages from.
- NodeMerger node_merger_;
+ BootMerger boot_merger_;
std::shared_ptr<const Configuration> configuration_;
@@ -686,9 +686,9 @@
// Timestamp of the last message returned. Used to make sure nothing goes
// backwards.
- monotonic_clock::time_point last_message_time_ = monotonic_clock::min_time;
+ BootTimestamp last_message_time_ = BootTimestamp::min_time();
// Time this node is queued up until. Used for caching.
- monotonic_clock::time_point queued_until_ = monotonic_clock::min_time;
+ BootTimestamp queued_until_ = BootTimestamp::min_time();
std::function<void(TimestampedMessage *)> timestamp_callback_;
};
diff --git a/aos/events/logging/logfile_utils_test.cc b/aos/events/logging/logfile_utils_test.cc
index d5cb41b..90a716e 100644
--- a/aos/events/logging/logfile_utils_test.cc
+++ b/aos/events/logging/logfile_utils_test.cc
@@ -223,20 +223,20 @@
Message m1{.channel_index = 0,
.queue_index = 0,
- .timestamp = e + chrono::milliseconds(1),
- .boot_count = 0,
+ .timestamp =
+ BootTimestamp{.boot = 0, .time = e + chrono::milliseconds(1)},
.data = SizePrefixedFlatbufferVector<MessageHeader>::Empty()};
Message m2{.channel_index = 0,
.queue_index = 0,
- .timestamp = e + chrono::milliseconds(2),
- .boot_count = 0,
+ .timestamp =
+ BootTimestamp{.boot = 0, .time = e + chrono::milliseconds(2)},
.data = SizePrefixedFlatbufferVector<MessageHeader>::Empty()};
EXPECT_LT(m1, m2);
EXPECT_GE(m2, m1);
- m1.timestamp = e;
- m2.timestamp = e;
+ m1.timestamp.time = e;
+ m2.timestamp.time = e;
m1.channel_index = 1;
m2.channel_index = 2;
@@ -548,10 +548,14 @@
ASSERT_TRUE(parts_sorter.Front() == nullptr);
- EXPECT_EQ(output[0].timestamp, e + chrono::milliseconds(1000));
- EXPECT_EQ(output[1].timestamp, e + chrono::milliseconds(1000));
- EXPECT_EQ(output[2].timestamp, e + chrono::milliseconds(1901));
- EXPECT_EQ(output[3].timestamp, e + chrono::milliseconds(2000));
+ EXPECT_EQ(output[0].timestamp.boot, 0);
+ EXPECT_EQ(output[0].timestamp.time, e + chrono::milliseconds(1000));
+ EXPECT_EQ(output[1].timestamp.boot, 0);
+ EXPECT_EQ(output[1].timestamp.time, e + chrono::milliseconds(1000));
+ EXPECT_EQ(output[2].timestamp.boot, 0);
+ EXPECT_EQ(output[2].timestamp.time, e + chrono::milliseconds(1901));
+ EXPECT_EQ(output[3].timestamp.boot, 0);
+ EXPECT_EQ(output[3].timestamp.time, e + chrono::milliseconds(2000));
}
// Tests that we can pull messages out of a log sorted in order.
@@ -594,11 +598,16 @@
ASSERT_TRUE(parts_sorter.Front() == nullptr);
- EXPECT_EQ(output[0].timestamp, e - chrono::milliseconds(1000));
- EXPECT_EQ(output[1].timestamp, e - chrono::milliseconds(500));
- EXPECT_EQ(output[2].timestamp, e - chrono::milliseconds(10));
- EXPECT_EQ(output[3].timestamp, e + chrono::milliseconds(1901));
- EXPECT_EQ(output[4].timestamp, e + chrono::milliseconds(2000));
+ EXPECT_EQ(output[0].timestamp.boot, 0u);
+ EXPECT_EQ(output[0].timestamp.time, e - chrono::milliseconds(1000));
+ EXPECT_EQ(output[1].timestamp.boot, 0u);
+ EXPECT_EQ(output[1].timestamp.time, e - chrono::milliseconds(500));
+ EXPECT_EQ(output[2].timestamp.boot, 0u);
+ EXPECT_EQ(output[2].timestamp.time, e - chrono::milliseconds(10));
+ EXPECT_EQ(output[3].timestamp.boot, 0u);
+ EXPECT_EQ(output[3].timestamp.time, e + chrono::milliseconds(1901));
+ EXPECT_EQ(output[4].timestamp.boot, 0u);
+ EXPECT_EQ(output[4].timestamp.time, e + chrono::milliseconds(2000));
}
// Tests that messages too far out of order trigger death.
@@ -709,12 +718,18 @@
ASSERT_TRUE(merger.Front() == nullptr);
- EXPECT_EQ(output[0].timestamp, e + chrono::milliseconds(1000));
- EXPECT_EQ(output[1].timestamp, e + chrono::milliseconds(1001));
- EXPECT_EQ(output[2].timestamp, e + chrono::milliseconds(1002));
- EXPECT_EQ(output[3].timestamp, e + chrono::milliseconds(2000));
- EXPECT_EQ(output[4].timestamp, e + chrono::milliseconds(3000));
- EXPECT_EQ(output[5].timestamp, e + chrono::milliseconds(3002));
+ EXPECT_EQ(output[0].timestamp.boot, 0u);
+ EXPECT_EQ(output[0].timestamp.time, e + chrono::milliseconds(1000));
+ EXPECT_EQ(output[1].timestamp.boot, 0u);
+ EXPECT_EQ(output[1].timestamp.time, e + chrono::milliseconds(1001));
+ EXPECT_EQ(output[2].timestamp.boot, 0u);
+ EXPECT_EQ(output[2].timestamp.time, e + chrono::milliseconds(1002));
+ EXPECT_EQ(output[3].timestamp.boot, 0u);
+ EXPECT_EQ(output[3].timestamp.time, e + chrono::milliseconds(2000));
+ EXPECT_EQ(output[4].timestamp.boot, 0u);
+ EXPECT_EQ(output[4].timestamp.time, e + chrono::milliseconds(3000));
+ EXPECT_EQ(output[5].timestamp.boot, 0u);
+ EXPECT_EQ(output[5].timestamp.time, e + chrono::milliseconds(3002));
}
// Tests that we can merge timestamps with various combinations of
@@ -776,15 +791,22 @@
}
ASSERT_TRUE(merger.Front() == nullptr);
- EXPECT_EQ(output[0].timestamp, e + chrono::milliseconds(101000));
+ EXPECT_EQ(output[0].timestamp.boot, 0u);
+ EXPECT_EQ(output[0].timestamp.time, e + chrono::milliseconds(101000));
EXPECT_FALSE(output[0].data.message().has_monotonic_timestamp_time());
- EXPECT_EQ(output[1].timestamp, e + chrono::milliseconds(101001));
+
+ EXPECT_EQ(output[1].timestamp.boot, 0u);
+ EXPECT_EQ(output[1].timestamp.time, e + chrono::milliseconds(101001));
EXPECT_TRUE(output[1].data.message().has_monotonic_timestamp_time());
EXPECT_EQ(output[1].data.message().monotonic_timestamp_time(), 971);
- EXPECT_EQ(output[2].timestamp, e + chrono::milliseconds(101002));
+
+ EXPECT_EQ(output[2].timestamp.boot, 0u);
+ EXPECT_EQ(output[2].timestamp.time, e + chrono::milliseconds(101002));
EXPECT_TRUE(output[2].data.message().has_monotonic_timestamp_time());
EXPECT_EQ(output[2].data.message().monotonic_timestamp_time(), 972);
- EXPECT_EQ(output[3].timestamp, e + chrono::milliseconds(101003));
+
+ EXPECT_EQ(output[3].timestamp.boot, 0u);
+ EXPECT_EQ(output[3].timestamp.time, e + chrono::milliseconds(101003));
EXPECT_TRUE(output[3].data.message().has_monotonic_timestamp_time());
EXPECT_EQ(output[3].data.message().monotonic_timestamp_time(), 973);
}
@@ -862,11 +884,19 @@
ASSERT_TRUE(mapper0.Front() == nullptr);
- EXPECT_EQ(output0[0].monotonic_event_time, e + chrono::milliseconds(1000));
+ EXPECT_EQ(output0[0].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output0[0].monotonic_event_time.time,
+ e + chrono::milliseconds(1000));
EXPECT_TRUE(output0[0].data.Verify());
- EXPECT_EQ(output0[1].monotonic_event_time, e + chrono::milliseconds(2000));
+
+ EXPECT_EQ(output0[1].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output0[1].monotonic_event_time.time,
+ e + chrono::milliseconds(2000));
EXPECT_TRUE(output0[1].data.Verify());
- EXPECT_EQ(output0[2].monotonic_event_time, e + chrono::milliseconds(3000));
+
+ EXPECT_EQ(output0[2].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output0[2].monotonic_event_time.time,
+ e + chrono::milliseconds(3000));
EXPECT_TRUE(output0[2].data.Verify());
}
@@ -906,13 +936,18 @@
EXPECT_EQ(mapper0_count, 3u);
EXPECT_EQ(mapper1_count, 3u);
- EXPECT_EQ(output1[0].monotonic_event_time,
+ EXPECT_EQ(output1[0].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[0].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(1000));
EXPECT_TRUE(output1[0].data.Verify());
- EXPECT_EQ(output1[1].monotonic_event_time,
+
+ EXPECT_EQ(output1[1].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[1].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(2000));
EXPECT_TRUE(output1[1].data.Verify());
- EXPECT_EQ(output1[2].monotonic_event_time,
+
+ EXPECT_EQ(output1[2].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[2].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(3000));
EXPECT_TRUE(output1[2].data.Verify());
}
@@ -977,14 +1012,28 @@
ASSERT_TRUE(mapper0.Front() == nullptr);
- EXPECT_EQ(output0[0].monotonic_event_time, e + chrono::milliseconds(1000));
- EXPECT_EQ(output0[0].monotonic_timestamp_time, monotonic_clock::min_time);
+ EXPECT_EQ(output0[0].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output0[0].monotonic_event_time.time,
+ e + chrono::milliseconds(1000));
+ EXPECT_EQ(output0[0].monotonic_timestamp_time.boot, 0u);
+ EXPECT_EQ(output0[0].monotonic_timestamp_time.time,
+ monotonic_clock::min_time);
EXPECT_TRUE(output0[0].data.Verify());
- EXPECT_EQ(output0[1].monotonic_event_time, e + chrono::milliseconds(2000));
- EXPECT_EQ(output0[1].monotonic_timestamp_time, monotonic_clock::min_time);
+
+ EXPECT_EQ(output0[1].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output0[1].monotonic_event_time.time,
+ e + chrono::milliseconds(2000));
+ EXPECT_EQ(output0[1].monotonic_timestamp_time.boot, 0u);
+ EXPECT_EQ(output0[1].monotonic_timestamp_time.time,
+ monotonic_clock::min_time);
EXPECT_TRUE(output0[1].data.Verify());
- EXPECT_EQ(output0[2].monotonic_event_time, e + chrono::milliseconds(3000));
- EXPECT_EQ(output0[2].monotonic_timestamp_time, monotonic_clock::min_time);
+
+ EXPECT_EQ(output0[2].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output0[2].monotonic_event_time.time,
+ e + chrono::milliseconds(3000));
+ EXPECT_EQ(output0[2].monotonic_timestamp_time.boot, 0u);
+ EXPECT_EQ(output0[2].monotonic_timestamp_time.time,
+ monotonic_clock::min_time);
EXPECT_TRUE(output0[2].data.Verify());
}
@@ -1000,19 +1049,28 @@
ASSERT_TRUE(mapper1.Front() == nullptr);
- EXPECT_EQ(output1[0].monotonic_event_time,
+ EXPECT_EQ(output1[0].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[0].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(1000));
- EXPECT_EQ(output1[0].monotonic_timestamp_time,
+ EXPECT_EQ(output1[0].monotonic_timestamp_time.boot, 0u);
+ EXPECT_EQ(output1[0].monotonic_timestamp_time.time,
e + chrono::nanoseconds(971));
EXPECT_TRUE(output1[0].data.Verify());
- EXPECT_EQ(output1[1].monotonic_event_time,
+
+ EXPECT_EQ(output1[1].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[1].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(2000));
- EXPECT_EQ(output1[1].monotonic_timestamp_time,
+ EXPECT_EQ(output1[1].monotonic_timestamp_time.boot, 0u);
+ EXPECT_EQ(output1[1].monotonic_timestamp_time.time,
e + chrono::nanoseconds(5458));
EXPECT_TRUE(output1[1].data.Verify());
- EXPECT_EQ(output1[2].monotonic_event_time,
+
+ EXPECT_EQ(output1[2].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[2].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(3000));
- EXPECT_EQ(output1[2].monotonic_timestamp_time, monotonic_clock::min_time);
+ EXPECT_EQ(output1[2].monotonic_timestamp_time.boot, 0u);
+ EXPECT_EQ(output1[2].monotonic_timestamp_time.time,
+ monotonic_clock::min_time);
EXPECT_TRUE(output1[2].data.Verify());
}
@@ -1085,13 +1143,18 @@
ASSERT_TRUE(mapper1.Front() == nullptr);
- EXPECT_EQ(output1[0].monotonic_event_time,
+ EXPECT_EQ(output1[0].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[0].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(1000));
EXPECT_TRUE(output1[0].data.Verify());
- EXPECT_EQ(output1[1].monotonic_event_time,
+
+ EXPECT_EQ(output1[1].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[1].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(2000));
EXPECT_TRUE(output1[1].data.Verify());
- EXPECT_EQ(output1[2].monotonic_event_time,
+
+ EXPECT_EQ(output1[2].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[2].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(3000));
EXPECT_TRUE(output1[2].data.Verify());
}
@@ -1116,11 +1179,19 @@
ASSERT_TRUE(mapper0.Front() == nullptr);
- EXPECT_EQ(output0[0].monotonic_event_time, e + chrono::milliseconds(1000));
+ EXPECT_EQ(output0[0].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output0[0].monotonic_event_time.time,
+ e + chrono::milliseconds(1000));
EXPECT_TRUE(output0[0].data.Verify());
- EXPECT_EQ(output0[1].monotonic_event_time, e + chrono::milliseconds(2000));
+
+ EXPECT_EQ(output0[1].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output0[1].monotonic_event_time.time,
+ e + chrono::milliseconds(2000));
EXPECT_TRUE(output0[1].data.Verify());
- EXPECT_EQ(output0[2].monotonic_event_time, e + chrono::milliseconds(3000));
+
+ EXPECT_EQ(output0[2].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output0[2].monotonic_event_time.time,
+ e + chrono::milliseconds(3000));
EXPECT_TRUE(output0[2].data.Verify());
}
@@ -1191,13 +1262,18 @@
ASSERT_TRUE(mapper1.Front() == nullptr);
- EXPECT_EQ(output1[0].monotonic_event_time,
+ EXPECT_EQ(output1[0].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[0].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(1000));
EXPECT_FALSE(output1[0].data.Verify());
- EXPECT_EQ(output1[1].monotonic_event_time,
+
+ EXPECT_EQ(output1[1].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[1].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(2000));
EXPECT_TRUE(output1[1].data.Verify());
- EXPECT_EQ(output1[2].monotonic_event_time,
+
+ EXPECT_EQ(output1[2].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[2].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(3000));
EXPECT_TRUE(output1[2].data.Verify());
}
@@ -1269,13 +1345,18 @@
ASSERT_TRUE(mapper1.Front() == nullptr);
- EXPECT_EQ(output1[0].monotonic_event_time,
+ EXPECT_EQ(output1[0].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[0].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(1000));
EXPECT_TRUE(output1[0].data.Verify());
- EXPECT_EQ(output1[1].monotonic_event_time,
+
+ EXPECT_EQ(output1[1].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[1].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(2000));
EXPECT_TRUE(output1[1].data.Verify());
- EXPECT_EQ(output1[2].monotonic_event_time,
+
+ EXPECT_EQ(output1[2].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[2].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(3000));
EXPECT_FALSE(output1[2].data.Verify());
}
@@ -1339,10 +1420,13 @@
ASSERT_FALSE(mapper1.Front() == nullptr);
- EXPECT_EQ(output1[0].monotonic_event_time,
+ EXPECT_EQ(output1[0].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[0].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(1000));
EXPECT_TRUE(output1[0].data.Verify());
- EXPECT_EQ(output1[1].monotonic_event_time,
+
+ EXPECT_EQ(output1[1].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[1].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(3000));
EXPECT_TRUE(output1[1].data.Verify());
}
@@ -1409,16 +1493,23 @@
}
ASSERT_TRUE(mapper1.Front() == nullptr);
- EXPECT_EQ(output1[0].monotonic_event_time,
+ EXPECT_EQ(output1[0].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[0].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(1000));
EXPECT_TRUE(output1[0].data.Verify());
- EXPECT_EQ(output1[1].monotonic_event_time,
+
+ EXPECT_EQ(output1[1].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[1].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(2000));
EXPECT_TRUE(output1[1].data.Verify());
- EXPECT_EQ(output1[2].monotonic_event_time,
+
+ EXPECT_EQ(output1[2].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[2].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(2000));
EXPECT_TRUE(output1[2].data.Verify());
- EXPECT_EQ(output1[3].monotonic_event_time,
+
+ EXPECT_EQ(output1[3].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[3].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(3000));
EXPECT_TRUE(output1[3].data.Verify());
}
@@ -1427,7 +1518,7 @@
EXPECT_EQ(mapper1_count, 4u);
}
-// Tests that we properly sort log files with duplicate timestamps.
+// Tests that we properly produce a valid start time.
TEST_F(TimestampMapperTest, StartTime) {
const aos::monotonic_clock::time_point e = monotonic_clock::epoch();
{
@@ -1447,8 +1538,8 @@
mapper0.set_timestamp_callback(
[&](TimestampedMessage *) { ++mapper0_count; });
- EXPECT_EQ(mapper0.monotonic_start_time(), e + chrono::milliseconds(1));
- EXPECT_EQ(mapper0.realtime_start_time(),
+ EXPECT_EQ(mapper0.monotonic_start_time(0), e + chrono::milliseconds(1));
+ EXPECT_EQ(mapper0.realtime_start_time(0),
realtime_clock::time_point(chrono::seconds(1000)));
EXPECT_EQ(mapper0_count, 0u);
}
@@ -1502,13 +1593,18 @@
mapper1.PopFront();
ASSERT_TRUE(mapper1.Front() == nullptr);
- EXPECT_EQ(output1[0].monotonic_event_time,
+ EXPECT_EQ(output1[0].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[0].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(1000));
EXPECT_FALSE(output1[0].data.Verify());
- EXPECT_EQ(output1[1].monotonic_event_time,
+
+ EXPECT_EQ(output1[1].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[1].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(2000));
EXPECT_FALSE(output1[1].data.Verify());
- EXPECT_EQ(output1[2].monotonic_event_time,
+
+ EXPECT_EQ(output1[2].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[2].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(3000));
EXPECT_FALSE(output1[2].data.Verify());
}
@@ -1568,7 +1664,8 @@
EXPECT_EQ(mapper0_count, 0u);
EXPECT_EQ(mapper1_count, 0u);
- mapper0.QueueUntil(e + chrono::milliseconds(1000));
+ mapper0.QueueUntil(
+ BootTimestamp{.boot = 0, .time = e + chrono::milliseconds(1000)});
EXPECT_EQ(mapper0_count, 3u);
EXPECT_EQ(mapper1_count, 0u);
@@ -1576,11 +1673,13 @@
EXPECT_EQ(mapper0_count, 3u);
EXPECT_EQ(mapper1_count, 0u);
- mapper0.QueueUntil(e + chrono::milliseconds(1500));
+ mapper0.QueueUntil(
+ BootTimestamp{.boot = 0, .time = e + chrono::milliseconds(1500)});
EXPECT_EQ(mapper0_count, 3u);
EXPECT_EQ(mapper1_count, 0u);
- mapper0.QueueUntil(e + chrono::milliseconds(2500));
+ mapper0.QueueUntil(
+ BootTimestamp{.boot = 0, .time = e + chrono::milliseconds(2500)});
EXPECT_EQ(mapper0_count, 4u);
EXPECT_EQ(mapper1_count, 0u);
@@ -1598,13 +1697,24 @@
ASSERT_TRUE(mapper0.Front() == nullptr);
- EXPECT_EQ(output0[0].monotonic_event_time, e + chrono::milliseconds(1000));
+ EXPECT_EQ(output0[0].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output0[0].monotonic_event_time.time,
+ e + chrono::milliseconds(1000));
EXPECT_TRUE(output0[0].data.Verify());
- EXPECT_EQ(output0[1].monotonic_event_time, e + chrono::milliseconds(1000));
+
+ EXPECT_EQ(output0[1].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output0[1].monotonic_event_time.time,
+ e + chrono::milliseconds(1000));
EXPECT_TRUE(output0[1].data.Verify());
- EXPECT_EQ(output0[2].monotonic_event_time, e + chrono::milliseconds(2000));
+
+ EXPECT_EQ(output0[2].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output0[2].monotonic_event_time.time,
+ e + chrono::milliseconds(2000));
EXPECT_TRUE(output0[2].data.Verify());
- EXPECT_EQ(output0[3].monotonic_event_time, e + chrono::milliseconds(3000));
+
+ EXPECT_EQ(output0[3].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output0[3].monotonic_event_time.time,
+ e + chrono::milliseconds(3000));
EXPECT_TRUE(output0[3].data.Verify());
}
@@ -1614,7 +1724,9 @@
EXPECT_EQ(mapper0_count, 4u);
EXPECT_EQ(mapper1_count, 0u);
- mapper1.QueueUntil(e + chrono::seconds(100) + chrono::milliseconds(1000));
+ mapper1.QueueUntil(BootTimestamp{
+ .boot = 0,
+ .time = e + chrono::seconds(100) + chrono::milliseconds(1000)});
EXPECT_EQ(mapper0_count, 4u);
EXPECT_EQ(mapper1_count, 3u);
@@ -1622,11 +1734,15 @@
EXPECT_EQ(mapper0_count, 4u);
EXPECT_EQ(mapper1_count, 3u);
- mapper1.QueueUntil(e + chrono::seconds(100) + chrono::milliseconds(1500));
+ mapper1.QueueUntil(BootTimestamp{
+ .boot = 0,
+ .time = e + chrono::seconds(100) + chrono::milliseconds(1500)});
EXPECT_EQ(mapper0_count, 4u);
EXPECT_EQ(mapper1_count, 3u);
- mapper1.QueueUntil(e + chrono::seconds(100) + chrono::milliseconds(2500));
+ mapper1.QueueUntil(BootTimestamp{
+ .boot = 0,
+ .time = e + chrono::seconds(100) + chrono::milliseconds(2500)});
EXPECT_EQ(mapper0_count, 4u);
EXPECT_EQ(mapper1_count, 4u);
@@ -1654,16 +1770,23 @@
EXPECT_EQ(mapper0_count, 4u);
EXPECT_EQ(mapper1_count, 4u);
- EXPECT_EQ(output1[0].monotonic_event_time,
+ EXPECT_EQ(output1[0].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[0].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(1000));
EXPECT_TRUE(output1[0].data.Verify());
- EXPECT_EQ(output1[1].monotonic_event_time,
+
+ EXPECT_EQ(output1[1].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[1].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(1000));
EXPECT_TRUE(output1[1].data.Verify());
- EXPECT_EQ(output1[2].monotonic_event_time,
+
+ EXPECT_EQ(output1[2].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[2].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(2000));
EXPECT_TRUE(output1[2].data.Verify());
- EXPECT_EQ(output1[3].monotonic_event_time,
+
+ EXPECT_EQ(output1[3].monotonic_event_time.boot, 0u);
+ EXPECT_EQ(output1[3].monotonic_event_time.time,
e + chrono::seconds(100) + chrono::milliseconds(3000));
EXPECT_TRUE(output1[3].data.Verify());
}
@@ -1784,14 +1907,15 @@
ASSERT_TRUE(merger.Front() == nullptr);
- EXPECT_EQ(output[0].timestamp, e + chrono::milliseconds(1000));
- EXPECT_EQ(output[0].boot_count, 0u);
- EXPECT_EQ(output[1].timestamp, e + chrono::milliseconds(2000));
- EXPECT_EQ(output[1].boot_count, 0u);
- EXPECT_EQ(output[2].timestamp, e + chrono::milliseconds(100));
- EXPECT_EQ(output[2].boot_count, 1u);
- EXPECT_EQ(output[3].timestamp, e + chrono::milliseconds(200));
- EXPECT_EQ(output[3].boot_count, 1u);
+ EXPECT_EQ(output[0].timestamp.boot, 0u);
+ EXPECT_EQ(output[0].timestamp.time, e + chrono::milliseconds(1000));
+ EXPECT_EQ(output[1].timestamp.boot, 0u);
+ EXPECT_EQ(output[1].timestamp.time, e + chrono::milliseconds(2000));
+
+ EXPECT_EQ(output[2].timestamp.boot, 1u);
+ EXPECT_EQ(output[2].timestamp.time, e + chrono::milliseconds(100));
+ EXPECT_EQ(output[3].timestamp.boot, 1u);
+ EXPECT_EQ(output[3].timestamp.time, e + chrono::milliseconds(200));
}
} // namespace testing
diff --git a/aos/network/multinode_timestamp_filter.cc b/aos/network/multinode_timestamp_filter.cc
index cafb940..a43fdfd 100644
--- a/aos/network/multinode_timestamp_filter.cc
+++ b/aos/network/multinode_timestamp_filter.cc
@@ -6,6 +6,7 @@
#include "absl/strings/str_join.h"
#include "aos/configuration.h"
+#include "aos/events/logging/boot_timestamp.h"
#include "aos/events/simulated_event_loop.h"
#include "aos/network/timestamp_filter.h"
#include "aos/time/time.h"
@@ -23,6 +24,7 @@
namespace message_bridge {
namespace {
namespace chrono = std::chrono;
+using aos::logger::BootTimestamp;
const Eigen::IOFormat kHeavyFormat(Eigen::StreamPrecision, Eigen::DontAlignCols,
", ", ";\n", "[", "]", "[", "]");
@@ -678,7 +680,9 @@
<< ": Timestamps queued before we registered the timestamp hooks.";
timestamp_mapper->set_timestamp_callback(
[this, node_index](logger::TimestampedMessage *msg) {
- if (msg->monotonic_remote_time != monotonic_clock::min_time) {
+ // TODO(austin): Funnel the boot index through the offset estimator.
+ CHECK_EQ(msg->monotonic_remote_time.boot, 0u);
+ if (msg->monotonic_remote_time.time != monotonic_clock::min_time) {
// Got a forwarding timestamp!
NoncausalOffsetEstimator *filter =
filters_per_channel_[node_index][msg->channel_index];
@@ -687,15 +691,18 @@
// Call the correct method depending on if we are the forward or
// reverse direction here.
- filter->Sample(node, msg->monotonic_event_time,
- msg->monotonic_remote_time);
+ CHECK_EQ(msg->monotonic_event_time.boot, 0u);
+ filter->Sample(node, msg->monotonic_event_time.time,
+ msg->monotonic_remote_time.time);
- if (msg->monotonic_timestamp_time != monotonic_clock::min_time) {
+ CHECK_EQ(msg->monotonic_timestamp_time.boot, 0u);
+ if (msg->monotonic_timestamp_time.time !=
+ monotonic_clock::min_time) {
// TODO(austin): This assumes that this timestamp is only logged
// on the node which sent the data. That is correct for now,
// but should be explicitly checked somewhere.
- filter->ReverseSample(node, msg->monotonic_event_time,
- msg->monotonic_timestamp_time);
+ filter->ReverseSample(node, msg->monotonic_event_time.time,
+ msg->monotonic_timestamp_time.time);
}
}
});
@@ -998,14 +1005,17 @@
// invalidate the point. Do this for both nodes to pick up all the
// timestamps.
if (filter.filter->has_unobserved_line()) {
+ // TODO(austin): Handle boots properly...
timestamp_mappers_[node_a_index]->QueueUntil(
- filter.filter->unobserved_line_end() +
- time_estimation_buffer_seconds_);
+ BootTimestamp{.boot = 0u,
+ .time = filter.filter->unobserved_line_end() +
+ time_estimation_buffer_seconds_});
if (timestamp_mappers_[node_b_index] != nullptr) {
- timestamp_mappers_[node_b_index]->QueueUntil(
- filter.filter->unobserved_line_remote_end() +
- time_estimation_buffer_seconds_);
+ timestamp_mappers_[node_b_index]->QueueUntil(BootTimestamp{
+ .boot = 0u,
+ .time = filter.filter->unobserved_line_remote_end() +
+ time_estimation_buffer_seconds_});
}
}
}