Move log reader and writer over to split timestamp channels
Reuse TimestampChannel (and split it up further) to implement naming and
finding channels for log reader and writer. Test both the old and new
timestamp configuration.
Once the config changes go in for y2020, this should fix reading the log
file James broke.
Change-Id: I6b09eec69c064ded3b3c149e0fdf23162bd352cf
diff --git a/aos/events/logging/BUILD b/aos/events/logging/BUILD
index fcf7683..586e932 100644
--- a/aos/events/logging/BUILD
+++ b/aos/events/logging/BUILD
@@ -320,8 +320,23 @@
)
aos_config(
- name = "multinode_pingpong_config",
- src = "multinode_pingpong.json",
+ name = "multinode_pingpong_split_config",
+ src = "multinode_pingpong_split.json",
+ flatbuffers = [
+ "//aos/events:ping_fbs",
+ "//aos/events:pong_fbs",
+ "//aos/network:message_bridge_client_fbs",
+ "//aos/network:message_bridge_server_fbs",
+ "//aos/network:remote_message_fbs",
+ "//aos/network:timestamp_fbs",
+ ],
+ target_compatible_with = ["@platforms//os:linux"],
+ deps = ["//aos/events:config"],
+)
+
+aos_config(
+ name = "multinode_pingpong_combined_config",
+ src = "multinode_pingpong_combined.json",
flatbuffers = [
"//aos/events:ping_fbs",
"//aos/events:pong_fbs",
@@ -343,7 +358,8 @@
"//conditions:default": [],
}),
data = [
- ":multinode_pingpong_config",
+ ":multinode_pingpong_combined_config",
+ ":multinode_pingpong_split_config",
"//aos/events:pingpong_config",
],
shard_count = 5,
diff --git a/aos/events/logging/log_reader.cc b/aos/events/logging/log_reader.cc
index c111a73..6e92568 100644
--- a/aos/events/logging/log_reader.cc
+++ b/aos/events/logging/log_reader.cc
@@ -19,6 +19,7 @@
#include "aos/network/remote_message_generated.h"
#include "aos/network/remote_message_schema.h"
#include "aos/network/team_number.h"
+#include "aos/network/timestamp_channel.h"
#include "aos/time/time.h"
#include "aos/util/file.h"
#include "flatbuffers/flatbuffers.h"
@@ -143,11 +144,48 @@
// Remap all existing remote timestamp channels. They will be recreated, and
// the data logged isn't relevant anymore.
for (const Node *node : configuration::GetNodes(logged_configuration())) {
+ message_bridge::ChannelTimestampFinder finder(logged_configuration(),
+ "log_reader", node);
+
+ absl::btree_set<std::string_view> remote_nodes;
+
+ for (const Channel *channel : *logged_configuration()->channels()) {
+ if (!configuration::ChannelIsSendableOnNode(channel, node)) {
+ continue;
+ }
+ if (!channel->has_destination_nodes()) {
+ continue;
+ }
+ for (const Connection *connection : *channel->destination_nodes()) {
+ if (configuration::ConnectionDeliveryTimeIsLoggedOnNode(connection,
+ node)) {
+ // Start by seeing if the split timestamp channels are being used for
+ // this message. If so, remap them.
+ const Channel *timestamp_channel = configuration::GetChannel(
+ logged_configuration(),
+ finder.SplitChannelName(channel, connection),
+ RemoteMessage::GetFullyQualifiedName(), "", node, true);
+
+ if (timestamp_channel != nullptr) {
+ if (timestamp_channel->logger() != LoggerConfig::NOT_LOGGED) {
+ RemapLoggedChannel<RemoteMessage>(
+ timestamp_channel->name()->string_view(), node);
+ }
+ continue;
+ }
+
+ // Otherwise collect this one up as a node to look for a combined
+ // channel from. It is more efficient to compare nodes than channels.
+ remote_nodes.insert(connection->name()->string_view());
+ }
+ }
+ }
+
std::vector<const Node *> timestamp_logger_nodes =
configuration::TimestampNodes(logged_configuration(), node);
- for (const Node *remote_node : timestamp_logger_nodes) {
- const std::string channel = absl::StrCat(
- "/aos/remote_timestamps/", remote_node->name()->string_view());
+ for (const std::string_view remote_node : remote_nodes) {
+ const std::string channel = finder.CombinedChannelName(remote_node);
+
// See if the log file is an old log with MessageHeader channels in it, or
// a newer log with RemoteMessage. If we find an older log, rename the
// type too along with the name.
@@ -467,9 +505,11 @@
// Delivery timestamps are supposed to be logged back on the source node.
// Configure remote timestamps to be sent.
+ const Connection *connection =
+ configuration::ConnectionToNode(channel, event_loop->node());
const bool delivery_time_is_logged =
- configuration::ConnectionDeliveryTimeIsLoggedOnNode(
- channel, event_loop->node(), source_node);
+ configuration::ConnectionDeliveryTimeIsLoggedOnNode(connection,
+ source_node);
source_state =
states_[configuration::GetNodeIndex(configuration(), source_node)]
@@ -477,7 +517,7 @@
if (delivery_time_is_logged) {
remote_timestamp_sender =
- source_state->RemoteTimestampSender(event_loop->node());
+ source_state->RemoteTimestampSender(channel, connection);
}
}
@@ -1233,22 +1273,44 @@
}
LogReader::RemoteMessageSender *LogReader::State::RemoteTimestampSender(
- const Node *delivered_node) {
- auto sender = remote_timestamp_senders_map_.find(delivered_node);
-
- if (sender == remote_timestamp_senders_map_.end()) {
- sender =
- remote_timestamp_senders_map_
- .emplace(delivered_node,
- std::make_unique<RemoteMessageSender>(
- event_loop()->MakeSender<RemoteMessage>(absl::StrCat(
- "/aos/remote_timestamps/",
- delivered_node->name()->string_view())),
- event_loop()))
- .first;
+ const Channel *channel, const Connection *connection) {
+ message_bridge::ChannelTimestampFinder finder(event_loop_);
+ // Look at any pre-created channel/connection pairs.
+ {
+ auto it =
+ channel_timestamp_loggers_.find(std::make_pair(channel, connection));
+ if (it != channel_timestamp_loggers_.end()) {
+ return it->second.get();
+ }
}
- return sender->second.get();
+ // That failed, so resolve the RemoteMessage channel timestamps will be logged
+ // to.
+ const Channel *timestamp_channel = finder.ForChannel(channel, connection);
+
+ {
+ // See if that has been created before. If so, cache it in
+ // channel_timestamp_loggers_ and return.
+ auto it = timestamp_loggers_.find(timestamp_channel);
+ if (it != timestamp_loggers_.end()) {
+ CHECK(channel_timestamp_loggers_
+ .try_emplace(std::make_pair(channel, connection), it->second)
+ .second);
+ return it->second.get();
+ }
+ }
+
+ // Otherwise, make a sender, save it, and cache it.
+ auto result = channel_timestamp_loggers_.try_emplace(
+ std::make_pair(channel, connection),
+ std::make_shared<RemoteMessageSender>(
+ event_loop()->MakeSender<RemoteMessage>(
+ timestamp_channel->name()->string_view()),
+ event_loop()));
+
+ CHECK(timestamp_loggers_.try_emplace(timestamp_channel, result.first->second)
+ .second);
+ return result.first->second.get();
}
TimestampedMessage LogReader::State::PopOldest() {
@@ -1303,7 +1365,8 @@
for (size_t i = 0; i < channels_.size(); ++i) {
channels_[i].reset();
}
- remote_timestamp_senders_map_.clear();
+ channel_timestamp_loggers_.clear();
+ timestamp_loggers_.clear();
event_loop_unique_ptr_.reset();
event_loop_ = nullptr;
timer_handler_ = nullptr;
diff --git a/aos/events/logging/log_reader.h b/aos/events/logging/log_reader.h
index 875c90d..24ac5c9 100644
--- a/aos/events/logging/log_reader.h
+++ b/aos/events/logging/log_reader.h
@@ -288,7 +288,8 @@
// Returns the MessageHeader sender to log delivery timestamps to for the
// provided remote node.
- RemoteMessageSender *RemoteTimestampSender(const Node *delivered_node);
+ RemoteMessageSender *RemoteTimestampSender(const Channel *channel,
+ const Connection *connection);
// Converts a timestamp from the monotonic clock on this node to the
// distributed clock.
@@ -417,8 +418,16 @@
// channel) which correspond to the originating node.
std::vector<State *> channel_source_state_;
- std::map<const Node *, std::unique_ptr<RemoteMessageSender>>
- remote_timestamp_senders_map_;
+ // This is a cache for channel, connection mapping to the corresponding
+ // sender.
+ absl::btree_map<std::pair<const Channel *, const Connection *>,
+ std::shared_ptr<RemoteMessageSender>>
+ channel_timestamp_loggers_;
+
+ // Mapping from resolved RemoteMessage channel to RemoteMessage sender. This
+ // is the channel that timestamps are published to.
+ absl::btree_map<const Channel *, std::shared_ptr<RemoteMessageSender>>
+ timestamp_loggers_;
};
// Node index -> State.
diff --git a/aos/events/logging/log_writer.cc b/aos/events/logging/log_writer.cc
index cb0f7c6..96995e9 100644
--- a/aos/events/logging/log_writer.cc
+++ b/aos/events/logging/log_writer.cc
@@ -8,6 +8,7 @@
#include "aos/events/event_loop.h"
#include "aos/network/message_bridge_server_generated.h"
#include "aos/network/team_number.h"
+#include "aos/network/timestamp_channel.h"
namespace aos {
namespace logger {
@@ -29,35 +30,28 @@
: aos::Fetcher<message_bridge::ServerStatistics>()) {
VLOG(1) << "Creating logger for " << FlatbufferToJson(event_loop_->node());
- // Find all the nodes which are logging timestamps on our node. This may
- // over-estimate if should_log is specified.
- std::vector<const Node *> timestamp_logger_nodes =
- configuration::TimestampNodes(configuration_, event_loop_->node());
-
std::map<const Channel *, const Node *> timestamp_logger_channels;
- // Now that we have all the nodes accumulated, make remote timestamp loggers
- // for them.
- for (const Node *node : timestamp_logger_nodes) {
- // Note: since we are doing a find using the event loop channel, we need to
- // make sure this channel pointer is part of the event loop configuration,
- // not configuration_. This only matters when configuration_ !=
- // event_loop->configuration();
- const Channel *channel = configuration::GetChannel(
- event_loop->configuration(),
- absl::StrCat("/aos/remote_timestamps/", node->name()->string_view()),
- RemoteMessage::GetFullyQualifiedName(), event_loop_->name(),
- event_loop_->node());
-
- CHECK(channel != nullptr)
- << ": Remote timestamps are logged on "
- << event_loop_->node()->name()->string_view()
- << " but can't find channel /aos/remote_timestamps/"
- << node->name()->string_view();
- if (!should_log(channel)) {
+ message_bridge::ChannelTimestampFinder finder(event_loop_);
+ for (const Channel *channel : *event_loop_->configuration()->channels()) {
+ if (!configuration::ChannelIsSendableOnNode(channel, event_loop_->node())) {
continue;
}
- timestamp_logger_channels.insert(std::make_pair(channel, node));
+ if (!channel->has_destination_nodes()) {
+ continue;
+ }
+ for (const Connection *connection : *channel->destination_nodes()) {
+ if (configuration::ConnectionDeliveryTimeIsLoggedOnNode(
+ connection, event_loop_->node())) {
+ const Node *other_node = configuration::GetNode(
+ event_loop_->configuration(), connection->name()->string_view());
+
+ VLOG(1) << "Timestamps are logged from "
+ << FlatbufferToJson(other_node);
+ timestamp_logger_channels.insert(
+ std::make_pair(finder.ForChannel(channel, connection), other_node));
+ }
+ }
}
const size_t our_node_index =
diff --git a/aos/events/logging/logger_test.cc b/aos/events/logging/logger_test.cc
index bd41663..cf8a373 100644
--- a/aos/events/logging/logger_test.cc
+++ b/aos/events/logging/logger_test.cc
@@ -30,8 +30,6 @@
constexpr std::string_view kSingleConfigSha1(
"bc8c9c2e31589eae6f0e36d766f6a437643e861d9568b7483106841cf7504dea");
-constexpr std::string_view kConfigSha1(
- "47511a1906dbb59cf9f8ad98ad08e568c718a4deb204c8bbce81ff76cef9095c");
std::vector<std::vector<std::string>> ToLogReaderVector(
const std::vector<LogFile> &log_files) {
@@ -412,38 +410,23 @@
}
}
-std::vector<std::string> MakeLogFiles(std::string logfile_base1,
- std::string logfile_base2) {
- return std::vector<std::string>(
- {logfile_base1 + "_pi1_data.part0.bfbs",
- logfile_base1 + "_pi2_data/test/aos.examples.Pong.part0.bfbs",
- logfile_base1 + "_pi2_data/test/aos.examples.Pong.part1.bfbs",
- logfile_base2 + "_pi2_data.part0.bfbs",
- logfile_base1 + "_timestamps/pi1/aos/remote_timestamps/pi2/"
- "aos.message_bridge.RemoteMessage.part0.bfbs",
- logfile_base1 + "_timestamps/pi1/aos/remote_timestamps/pi2/"
- "aos.message_bridge.RemoteMessage.part1.bfbs",
- logfile_base2 + "_timestamps/pi2/aos/remote_timestamps/pi1/"
- "aos.message_bridge.RemoteMessage.part0.bfbs",
- logfile_base2 + "_timestamps/pi2/aos/remote_timestamps/pi1/"
- "aos.message_bridge.RemoteMessage.part1.bfbs",
- logfile_base2 +
- "_pi1_data/pi1/aos/aos.message_bridge.Timestamp.part0.bfbs",
- logfile_base2 +
- "_pi1_data/pi1/aos/aos.message_bridge.Timestamp.part1.bfbs",
- logfile_base1 +
- "_pi2_data/pi2/aos/aos.message_bridge.Timestamp.part0.bfbs",
- logfile_base1 +
- "_pi2_data/pi2/aos/aos.message_bridge.Timestamp.part1.bfbs",
- absl::StrCat(logfile_base1, "_", kConfigSha1, ".bfbs"),
- absl::StrCat(logfile_base2, "_", kConfigSha1, ".bfbs")});
-}
+// Parameters to run all the tests with.
+struct Param {
+ // The config file to use.
+ std::string config;
+ // If true, the RemoteMessage channel should be shared between all the remote
+ // channels. If false, there will be 1 RemoteMessage channel per remote
+ // channel.
+ bool shared;
+ // sha256 of the config.
+ std::string sha256;
+};
-class MultinodeLoggerTest : public ::testing::Test {
+class MultinodeLoggerTest : public ::testing::TestWithParam<struct Param> {
public:
MultinodeLoggerTest()
: config_(aos::configuration::ReadConfig(
- "aos/events/logging/multinode_pingpong_config.json")),
+ absl::StrCat("aos/events/logging/", GetParam().config))),
time_converter_(configuration::NodesCount(&config_.message())),
event_loop_factory_(&config_.message()),
pi1_(
@@ -457,45 +440,15 @@
tmp_dir_(aos::testing::TestTmpDir()),
logfile_base1_(tmp_dir_ + "/multi_logfile1"),
logfile_base2_(tmp_dir_ + "/multi_logfile2"),
- pi1_reboot_logfiles_(
- {logfile_base1_ + "_pi1_data.part0.bfbs",
- logfile_base1_ + "_pi2_data/test/aos.examples.Pong.part0.bfbs",
- logfile_base1_ + "_pi2_data/test/aos.examples.Pong.part1.bfbs",
- logfile_base1_ + "_pi2_data/test/aos.examples.Pong.part2.bfbs",
- logfile_base1_ + "_timestamps/pi1/aos/remote_timestamps/pi2/"
- "aos.message_bridge.RemoteMessage.part0.bfbs",
- logfile_base1_ + "_timestamps/pi1/aos/remote_timestamps/pi2/"
- "aos.message_bridge.RemoteMessage.part1.bfbs",
- logfile_base1_ + "_timestamps/pi1/aos/remote_timestamps/pi2/"
- "aos.message_bridge.RemoteMessage.part2.bfbs",
- logfile_base1_ +
- "_pi2_data/pi2/aos/aos.message_bridge.Timestamp.part0.bfbs",
- logfile_base1_ +
- "_pi2_data/pi2/aos/aos.message_bridge.Timestamp.part1.bfbs",
- logfile_base1_ +
- "_pi2_data/pi2/aos/aos.message_bridge.Timestamp.part2.bfbs",
- absl::StrCat(logfile_base1_, "_", kConfigSha1, ".bfbs")}),
+ pi1_reboot_logfiles_(MakePi1RebootLogfiles()),
logfiles_(MakeLogFiles(logfile_base1_, logfile_base2_)),
- pi1_single_direction_logfiles_(
- {logfile_base1_ + "_pi1_data.part0.bfbs",
- logfile_base1_ + "_pi2_data/test/aos.examples.Pong.part0.bfbs",
- logfile_base1_ + "_timestamps/pi1/aos/remote_timestamps/pi2/"
- "aos.message_bridge.RemoteMessage.part0.bfbs",
- logfile_base1_ +
- "_pi2_data/pi2/aos/aos.message_bridge.Timestamp.part0.bfbs",
- absl::StrCat(logfile_base1_, "_", kConfigSha1, ".bfbs")}),
- structured_logfiles_{
- std::vector<std::string>{logfiles_[0]},
- std::vector<std::string>{logfiles_[1], logfiles_[2]},
- std::vector<std::string>{logfiles_[3]},
- std::vector<std::string>{logfiles_[4], logfiles_[5]},
- std::vector<std::string>{logfiles_[6], logfiles_[7]},
- std::vector<std::string>{logfiles_[8], logfiles_[9]},
- std::vector<std::string>{logfiles_[10], logfiles_[11]}},
+ pi1_single_direction_logfiles_(MakePi1SingleDirectionLogfiles()),
+ structured_logfiles_(StructureLogFiles()),
ping_event_loop_(event_loop_factory_.MakeEventLoop("ping", pi1_)),
ping_(ping_event_loop_.get()),
pong_event_loop_(event_loop_factory_.MakeEventLoop("pong", pi2_)),
pong_(pong_event_loop_.get()) {
+ LOG(INFO) << "Config " << GetParam().config;
event_loop_factory_.SetTimeConverter(&time_converter_);
// Go through and remove the logfiles if they already exist.
@@ -517,6 +470,182 @@
<< " and " << logfiles_[2];
}
+ bool shared() const { return GetParam().shared; }
+
+ std::vector<std::string> MakeLogFiles(std::string logfile_base1,
+ std::string logfile_base2) {
+ std::vector<std::string> result;
+ result.emplace_back(
+ absl::StrCat(logfile_base1, "_", GetParam().sha256, ".bfbs"));
+ result.emplace_back(
+ absl::StrCat(logfile_base2, "_", GetParam().sha256, ".bfbs"));
+ result.emplace_back(logfile_base1 + "_pi1_data.part0.bfbs");
+ result.emplace_back(logfile_base1 +
+ "_pi2_data/test/aos.examples.Pong.part0.bfbs");
+ result.emplace_back(logfile_base1 +
+ "_pi2_data/test/aos.examples.Pong.part1.bfbs");
+ result.emplace_back(logfile_base2 + "_pi2_data.part0.bfbs");
+ result.emplace_back(
+ logfile_base2 +
+ "_pi1_data/pi1/aos/aos.message_bridge.Timestamp.part0.bfbs");
+ result.emplace_back(
+ logfile_base2 +
+ "_pi1_data/pi1/aos/aos.message_bridge.Timestamp.part1.bfbs");
+ result.emplace_back(
+ logfile_base1 +
+ "_pi2_data/pi2/aos/aos.message_bridge.Timestamp.part0.bfbs");
+ result.emplace_back(
+ logfile_base1 +
+ "_pi2_data/pi2/aos/aos.message_bridge.Timestamp.part1.bfbs");
+ if (shared()) {
+ result.emplace_back(logfile_base1 +
+ "_timestamps/pi1/aos/remote_timestamps/pi2/"
+ "aos.message_bridge.RemoteMessage.part0.bfbs");
+ result.emplace_back(logfile_base1 +
+ "_timestamps/pi1/aos/remote_timestamps/pi2/"
+ "aos.message_bridge.RemoteMessage.part1.bfbs");
+ result.emplace_back(logfile_base2 +
+ "_timestamps/pi2/aos/remote_timestamps/pi1/"
+ "aos.message_bridge.RemoteMessage.part0.bfbs");
+ result.emplace_back(logfile_base2 +
+ "_timestamps/pi2/aos/remote_timestamps/pi1/"
+ "aos.message_bridge.RemoteMessage.part1.bfbs");
+ } else {
+ result.emplace_back(logfile_base1 +
+ "_timestamps/pi1/aos/remote_timestamps/pi2/pi1/aos/"
+ "aos-message_bridge-Timestamp/"
+ "aos.message_bridge.RemoteMessage.part0.bfbs");
+ result.emplace_back(logfile_base1 +
+ "_timestamps/pi1/aos/remote_timestamps/pi2/pi1/aos/"
+ "aos-message_bridge-Timestamp/"
+ "aos.message_bridge.RemoteMessage.part1.bfbs");
+ result.emplace_back(logfile_base2 +
+ "_timestamps/pi2/aos/remote_timestamps/pi1/pi2/aos/"
+ "aos-message_bridge-Timestamp/"
+ "aos.message_bridge.RemoteMessage.part0.bfbs");
+ result.emplace_back(logfile_base2 +
+ "_timestamps/pi2/aos/remote_timestamps/pi1/pi2/aos/"
+ "aos-message_bridge-Timestamp/"
+ "aos.message_bridge.RemoteMessage.part1.bfbs");
+ result.emplace_back(logfile_base1 +
+ "_timestamps/pi1/aos/remote_timestamps/pi2/test/"
+ "aos-examples-Ping/"
+ "aos.message_bridge.RemoteMessage.part0.bfbs");
+ result.emplace_back(logfile_base1 +
+ "_timestamps/pi1/aos/remote_timestamps/pi2/test/"
+ "aos-examples-Ping/"
+ "aos.message_bridge.RemoteMessage.part1.bfbs");
+ }
+
+ return result;
+ }
+
+ std::vector<std::string> MakePi1RebootLogfiles() {
+ std::vector<std::string> result;
+ result.emplace_back(logfile_base1_ + "_pi1_data.part0.bfbs");
+ result.emplace_back(logfile_base1_ +
+ "_pi2_data/test/aos.examples.Pong.part0.bfbs");
+ result.emplace_back(logfile_base1_ +
+ "_pi2_data/test/aos.examples.Pong.part1.bfbs");
+ result.emplace_back(logfile_base1_ +
+ "_pi2_data/test/aos.examples.Pong.part2.bfbs");
+ result.emplace_back(
+ logfile_base1_ +
+ "_pi2_data/pi2/aos/aos.message_bridge.Timestamp.part0.bfbs");
+ result.emplace_back(
+ logfile_base1_ +
+ "_pi2_data/pi2/aos/aos.message_bridge.Timestamp.part1.bfbs");
+ result.emplace_back(
+ logfile_base1_ +
+ "_pi2_data/pi2/aos/aos.message_bridge.Timestamp.part2.bfbs");
+ result.emplace_back(
+ absl::StrCat(logfile_base1_, "_", GetParam().sha256, ".bfbs"));
+ if (shared()) {
+ result.emplace_back(logfile_base1_ +
+ "_timestamps/pi1/aos/remote_timestamps/pi2/"
+ "aos.message_bridge.RemoteMessage.part0.bfbs");
+ result.emplace_back(logfile_base1_ +
+ "_timestamps/pi1/aos/remote_timestamps/pi2/"
+ "aos.message_bridge.RemoteMessage.part1.bfbs");
+ result.emplace_back(logfile_base1_ +
+ "_timestamps/pi1/aos/remote_timestamps/pi2/"
+ "aos.message_bridge.RemoteMessage.part2.bfbs");
+ } else {
+ result.emplace_back(logfile_base1_ +
+ "_timestamps/pi1/aos/remote_timestamps/pi2/pi1/aos/"
+ "aos-message_bridge-Timestamp/"
+ "aos.message_bridge.RemoteMessage.part0.bfbs");
+ result.emplace_back(logfile_base1_ +
+ "_timestamps/pi1/aos/remote_timestamps/pi2/pi1/aos/"
+ "aos-message_bridge-Timestamp/"
+ "aos.message_bridge.RemoteMessage.part1.bfbs");
+ result.emplace_back(logfile_base1_ +
+ "_timestamps/pi1/aos/remote_timestamps/pi2/pi1/aos/"
+ "aos-message_bridge-Timestamp/"
+ "aos.message_bridge.RemoteMessage.part2.bfbs");
+
+ result.emplace_back(logfile_base1_ +
+ "_timestamps/pi1/aos/remote_timestamps/pi2/test/"
+ "aos-examples-Ping/"
+ "aos.message_bridge.RemoteMessage.part0.bfbs");
+ result.emplace_back(logfile_base1_ +
+ "_timestamps/pi1/aos/remote_timestamps/pi2/test/"
+ "aos-examples-Ping/"
+ "aos.message_bridge.RemoteMessage.part1.bfbs");
+ result.emplace_back(logfile_base1_ +
+ "_timestamps/pi1/aos/remote_timestamps/pi2/test/"
+ "aos-examples-Ping/"
+ "aos.message_bridge.RemoteMessage.part2.bfbs");
+ }
+ return result;
+ }
+
+ std::vector<std::string> MakePi1SingleDirectionLogfiles() {
+ std::vector<std::string> result;
+ result.emplace_back(logfile_base1_ + "_pi1_data.part0.bfbs");
+ result.emplace_back(logfile_base1_ +
+ "_pi2_data/test/aos.examples.Pong.part0.bfbs");
+ result.emplace_back(
+ logfile_base1_ +
+ "_pi2_data/pi2/aos/aos.message_bridge.Timestamp.part0.bfbs");
+ result.emplace_back(
+ absl::StrCat(logfile_base1_, "_", GetParam().sha256, ".bfbs"));
+
+ if (shared()) {
+ result.emplace_back(logfile_base1_ +
+ "_timestamps/pi1/aos/remote_timestamps/pi2/"
+ "aos.message_bridge.RemoteMessage.part0.bfbs");
+ } else {
+ result.emplace_back(logfile_base1_ +
+ "_timestamps/pi1/aos/remote_timestamps/pi2/pi1/aos/"
+ "aos-message_bridge-Timestamp/"
+ "aos.message_bridge.RemoteMessage.part0.bfbs");
+ result.emplace_back(logfile_base1_ +
+ "_timestamps/pi1/aos/remote_timestamps/pi2/test/"
+ "aos-examples-Ping/"
+ "aos.message_bridge.RemoteMessage.part0.bfbs");
+ }
+ return result;
+ }
+
+ std::vector<std::vector<std::string>> StructureLogFiles() {
+ std::vector<std::vector<std::string>> result{
+ std::vector<std::string>{logfiles_[2]},
+ std::vector<std::string>{logfiles_[3], logfiles_[4]},
+ std::vector<std::string>{logfiles_[5]},
+ std::vector<std::string>{logfiles_[6], logfiles_[7]},
+ std::vector<std::string>{logfiles_[8], logfiles_[9]},
+ std::vector<std::string>{logfiles_[10], logfiles_[11]},
+ std::vector<std::string>{logfiles_[12], logfiles_[13]}};
+
+ if (!shared()) {
+ result.emplace_back(
+ std::vector<std::string>{logfiles_[14], logfiles_[15]});
+ }
+
+ return result;
+ }
+
struct LoggerState {
std::unique_ptr<EventLoop> event_loop;
std::unique_ptr<Logger> logger;
@@ -600,10 +729,13 @@
}
}
- // We won't have RT timestamps for 5 log files. We don't log the RT start
- // time on remote nodes because we don't know it and would be guessing. And
- // the log reader can actually do a better job.
- EXPECT_EQ(missing_rt_count, 5u);
+ // We won't have RT timestamps for 5 or 6 log files. We don't log the RT
+ // start time on remote nodes because we don't know it and would be
+ // guessing. And the log reader can actually do a better job. The number
+ // depends on if we have the remote timestamps split across 2 files, or just
+ // across 1, depending on if we are using a split or combined timestamp
+ // channel config.
+ EXPECT_EQ(missing_rt_count, shared() ? 5u : 6u);
EXPECT_EQ(log_event_uuids.size(), 2u);
EXPECT_EQ(parts_uuids.size(), ToLogReaderVector(sorted_parts).size());
@@ -749,7 +881,7 @@
}
// Tests that we can write and read simple multi-node log files.
-TEST_F(MultinodeLoggerTest, SimpleMultiNode) {
+TEST_P(MultinodeLoggerTest, SimpleMultiNode) {
time_converter_.StartEqual();
{
LoggerState pi1_logger = MakeLogger(pi1_);
@@ -779,35 +911,47 @@
}
EXPECT_EQ(logfile_uuids.size(), 2u);
- EXPECT_EQ(parts_uuids.size(), 7u);
+ if (shared()) {
+ EXPECT_EQ(parts_uuids.size(), 7u);
+ } else {
+ EXPECT_EQ(parts_uuids.size(), 8u);
+ }
// And confirm everything is on the correct node.
- EXPECT_EQ(log_header[0].message().node()->name()->string_view(), "pi1");
- EXPECT_EQ(log_header[1].message().node()->name()->string_view(), "pi2");
- EXPECT_EQ(log_header[2].message().node()->name()->string_view(), "pi2");
+ EXPECT_EQ(log_header[2].message().node()->name()->string_view(), "pi1");
EXPECT_EQ(log_header[3].message().node()->name()->string_view(), "pi2");
EXPECT_EQ(log_header[4].message().node()->name()->string_view(), "pi2");
EXPECT_EQ(log_header[5].message().node()->name()->string_view(), "pi2");
EXPECT_EQ(log_header[6].message().node()->name()->string_view(), "pi1");
EXPECT_EQ(log_header[7].message().node()->name()->string_view(), "pi1");
- EXPECT_EQ(log_header[8].message().node()->name()->string_view(), "pi1");
- EXPECT_EQ(log_header[9].message().node()->name()->string_view(), "pi1");
+ EXPECT_EQ(log_header[8].message().node()->name()->string_view(), "pi2");
+ EXPECT_EQ(log_header[9].message().node()->name()->string_view(), "pi2");
EXPECT_EQ(log_header[10].message().node()->name()->string_view(), "pi2");
EXPECT_EQ(log_header[11].message().node()->name()->string_view(), "pi2");
+ EXPECT_EQ(log_header[12].message().node()->name()->string_view(), "pi1");
+ EXPECT_EQ(log_header[13].message().node()->name()->string_view(), "pi1");
+ if (!shared()) {
+ EXPECT_EQ(log_header[14].message().node()->name()->string_view(), "pi2");
+ EXPECT_EQ(log_header[15].message().node()->name()->string_view(), "pi2");
+ }
// And the parts index matches.
- EXPECT_EQ(log_header[0].message().parts_index(), 0);
- EXPECT_EQ(log_header[1].message().parts_index(), 0);
- EXPECT_EQ(log_header[2].message().parts_index(), 1);
+ EXPECT_EQ(log_header[2].message().parts_index(), 0);
EXPECT_EQ(log_header[3].message().parts_index(), 0);
- EXPECT_EQ(log_header[4].message().parts_index(), 0);
- EXPECT_EQ(log_header[5].message().parts_index(), 1);
+ EXPECT_EQ(log_header[4].message().parts_index(), 1);
+ EXPECT_EQ(log_header[5].message().parts_index(), 0);
EXPECT_EQ(log_header[6].message().parts_index(), 0);
EXPECT_EQ(log_header[7].message().parts_index(), 1);
EXPECT_EQ(log_header[8].message().parts_index(), 0);
EXPECT_EQ(log_header[9].message().parts_index(), 1);
EXPECT_EQ(log_header[10].message().parts_index(), 0);
EXPECT_EQ(log_header[11].message().parts_index(), 1);
+ EXPECT_EQ(log_header[12].message().parts_index(), 0);
+ EXPECT_EQ(log_header[13].message().parts_index(), 1);
+ if (!shared()) {
+ EXPECT_EQ(log_header[14].message().parts_index(), 0);
+ EXPECT_EQ(log_header[15].message().parts_index(), 1);
+ }
}
const std::vector<LogFile> sorted_log_files = SortParts(logfiles_);
@@ -818,7 +962,7 @@
// Timing reports, pings
EXPECT_THAT(
- CountChannelsData(config, logfiles_[0]),
+ CountChannelsData(config, logfiles_[2]),
UnorderedElementsAre(
std::make_tuple("/pi1/aos", "aos.message_bridge.Timestamp", 200),
std::make_tuple("/pi1/aos", "aos.message_bridge.ServerStatistics",
@@ -827,36 +971,36 @@
200),
std::make_tuple("/pi1/aos", "aos.timing.Report", 40),
std::make_tuple("/test", "aos.examples.Ping", 2001)))
- << " : " << logfiles_[0];
+ << " : " << logfiles_[2];
// Timestamps for pong
EXPECT_THAT(
- CountChannelsTimestamp(config, logfiles_[0]),
+ CountChannelsTimestamp(config, logfiles_[2]),
UnorderedElementsAre(
std::make_tuple("/test", "aos.examples.Pong", 2001),
std::make_tuple("/pi2/aos", "aos.message_bridge.Timestamp", 200)))
- << " : " << logfiles_[0];
+ << " : " << logfiles_[2];
// Pong data.
EXPECT_THAT(
- CountChannelsData(config, logfiles_[1]),
+ CountChannelsData(config, logfiles_[3]),
UnorderedElementsAre(std::make_tuple("/test", "aos.examples.Pong", 91)))
- << " : " << logfiles_[1];
- EXPECT_THAT(CountChannelsData(config, logfiles_[2]),
+ << " : " << logfiles_[3];
+ EXPECT_THAT(CountChannelsData(config, logfiles_[4]),
UnorderedElementsAre(
std::make_tuple("/test", "aos.examples.Pong", 1910)))
- << " : " << logfiles_[1];
+ << " : " << logfiles_[3];
// No timestamps
- EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[1]),
+ EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[3]),
UnorderedElementsAre())
- << " : " << logfiles_[1];
- EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[2]),
+ << " : " << logfiles_[3];
+ EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[4]),
UnorderedElementsAre())
- << " : " << logfiles_[2];
+ << " : " << logfiles_[4];
// Timing reports and pongs.
EXPECT_THAT(
- CountChannelsData(config, logfiles_[3]),
+ CountChannelsData(config, logfiles_[5]),
UnorderedElementsAre(
std::make_tuple("/pi2/aos", "aos.message_bridge.Timestamp", 200),
std::make_tuple("/pi2/aos", "aos.message_bridge.ServerStatistics",
@@ -865,77 +1009,117 @@
200),
std::make_tuple("/pi2/aos", "aos.timing.Report", 40),
std::make_tuple("/test", "aos.examples.Pong", 2001)))
- << " : " << logfiles_[3];
- // And ping timestamps.
- EXPECT_THAT(
- CountChannelsTimestamp(config, logfiles_[3]),
- UnorderedElementsAre(
- std::make_tuple("/test", "aos.examples.Ping", 2001),
- std::make_tuple("/pi1/aos", "aos.message_bridge.Timestamp", 200)))
- << " : " << logfiles_[3];
-
- // Timestamps from pi2 on pi1, and the other way.
- EXPECT_THAT(CountChannelsData(config, logfiles_[4]), UnorderedElementsAre())
- << " : " << logfiles_[4];
- EXPECT_THAT(CountChannelsData(config, logfiles_[5]), UnorderedElementsAre())
<< " : " << logfiles_[5];
- EXPECT_THAT(CountChannelsData(config, logfiles_[6]), UnorderedElementsAre())
- << " : " << logfiles_[6];
- EXPECT_THAT(CountChannelsData(config, logfiles_[7]), UnorderedElementsAre())
- << " : " << logfiles_[7];
- EXPECT_THAT(
- CountChannelsTimestamp(config, logfiles_[4]),
- UnorderedElementsAre(
- std::make_tuple("/pi1/aos", "aos.message_bridge.Timestamp", 9),
- std::make_tuple("/test", "aos.examples.Ping", 91)))
- << " : " << logfiles_[4];
+ // And ping timestamps.
EXPECT_THAT(
CountChannelsTimestamp(config, logfiles_[5]),
UnorderedElementsAre(
- std::make_tuple("/pi1/aos", "aos.message_bridge.Timestamp", 191),
- std::make_tuple("/test", "aos.examples.Ping", 1910)))
+ std::make_tuple("/test", "aos.examples.Ping", 2001),
+ std::make_tuple("/pi1/aos", "aos.message_bridge.Timestamp", 200)))
<< " : " << logfiles_[5];
- EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[6]),
- UnorderedElementsAre(std::make_tuple(
- "/pi2/aos", "aos.message_bridge.Timestamp", 9)))
- << " : " << logfiles_[6];
- EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[7]),
- UnorderedElementsAre(std::make_tuple(
- "/pi2/aos", "aos.message_bridge.Timestamp", 191)))
- << " : " << logfiles_[7];
// And then test that the remotely logged timestamp data files only have
// timestamps in them.
+ EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[6]),
+ UnorderedElementsAre())
+ << " : " << logfiles_[6];
+ EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[7]),
+ UnorderedElementsAre())
+ << " : " << logfiles_[7];
EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[8]),
UnorderedElementsAre())
<< " : " << logfiles_[8];
EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[9]),
UnorderedElementsAre())
<< " : " << logfiles_[9];
- EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[10]),
- UnorderedElementsAre())
- << " : " << logfiles_[10];
- EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[11]),
- UnorderedElementsAre())
- << " : " << logfiles_[11];
+
+ EXPECT_THAT(CountChannelsData(config, logfiles_[6]),
+ UnorderedElementsAre(std::make_tuple(
+ "/pi1/aos", "aos.message_bridge.Timestamp", 9)))
+ << " : " << logfiles_[6];
+ EXPECT_THAT(CountChannelsData(config, logfiles_[7]),
+ UnorderedElementsAre(std::make_tuple(
+ "/pi1/aos", "aos.message_bridge.Timestamp", 191)))
+ << " : " << logfiles_[7];
EXPECT_THAT(CountChannelsData(config, logfiles_[8]),
UnorderedElementsAre(std::make_tuple(
- "/pi1/aos", "aos.message_bridge.Timestamp", 9)))
+ "/pi2/aos", "aos.message_bridge.Timestamp", 9)))
<< " : " << logfiles_[8];
EXPECT_THAT(CountChannelsData(config, logfiles_[9]),
UnorderedElementsAre(std::make_tuple(
- "/pi1/aos", "aos.message_bridge.Timestamp", 191)))
+ "/pi2/aos", "aos.message_bridge.Timestamp", 191)))
<< " : " << logfiles_[9];
+ // Timestamps from pi2 on pi1, and the other way.
EXPECT_THAT(CountChannelsData(config, logfiles_[10]),
- UnorderedElementsAre(std::make_tuple(
- "/pi2/aos", "aos.message_bridge.Timestamp", 9)))
+ UnorderedElementsAre())
<< " : " << logfiles_[10];
EXPECT_THAT(CountChannelsData(config, logfiles_[11]),
- UnorderedElementsAre(std::make_tuple(
- "/pi2/aos", "aos.message_bridge.Timestamp", 191)))
+ UnorderedElementsAre())
<< " : " << logfiles_[11];
+ EXPECT_THAT(CountChannelsData(config, logfiles_[12]),
+ UnorderedElementsAre())
+ << " : " << logfiles_[12];
+ EXPECT_THAT(CountChannelsData(config, logfiles_[13]),
+ UnorderedElementsAre())
+ << " : " << logfiles_[13];
+ if (!shared()) {
+ EXPECT_THAT(CountChannelsData(config, logfiles_[14]),
+ UnorderedElementsAre())
+ << " : " << logfiles_[14];
+ EXPECT_THAT(CountChannelsData(config, logfiles_[15]),
+ UnorderedElementsAre())
+ << " : " << logfiles_[15];
+ }
+
+ if (shared()) {
+ EXPECT_THAT(
+ CountChannelsTimestamp(config, logfiles_[10]),
+ UnorderedElementsAre(
+ std::make_tuple("/pi1/aos", "aos.message_bridge.Timestamp", 9),
+ std::make_tuple("/test", "aos.examples.Ping", 91)))
+ << " : " << logfiles_[10];
+ EXPECT_THAT(
+ CountChannelsTimestamp(config, logfiles_[11]),
+ UnorderedElementsAre(
+ std::make_tuple("/pi1/aos", "aos.message_bridge.Timestamp", 191),
+ std::make_tuple("/test", "aos.examples.Ping", 1910)))
+ << " : " << logfiles_[11];
+ EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[12]),
+ UnorderedElementsAre(std::make_tuple(
+ "/pi2/aos", "aos.message_bridge.Timestamp", 9)))
+ << " : " << logfiles_[12];
+ EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[13]),
+ UnorderedElementsAre(std::make_tuple(
+ "/pi2/aos", "aos.message_bridge.Timestamp", 191)))
+ << " : " << logfiles_[13];
+ } else {
+ EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[10]),
+ UnorderedElementsAre(std::make_tuple(
+ "/pi1/aos", "aos.message_bridge.Timestamp", 9)))
+ << " : " << logfiles_[10];
+ EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[11]),
+ UnorderedElementsAre(std::make_tuple(
+ "/pi1/aos", "aos.message_bridge.Timestamp", 191)))
+ << " : " << logfiles_[11];
+ EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[12]),
+ UnorderedElementsAre(std::make_tuple(
+ "/pi2/aos", "aos.message_bridge.Timestamp", 9)))
+ << " : " << logfiles_[12];
+ EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[13]),
+ UnorderedElementsAre(std::make_tuple(
+ "/pi2/aos", "aos.message_bridge.Timestamp", 191)))
+ << " : " << logfiles_[13];
+ EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[14]),
+ UnorderedElementsAre(
+ std::make_tuple("/test", "aos.examples.Ping", 91)))
+ << " : " << logfiles_[14];
+ EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[15]),
+ UnorderedElementsAre(
+ std::make_tuple("/test", "aos.examples.Ping", 1910)))
+ << " : " << logfiles_[15];
+ }
}
LogReader reader(sorted_log_files);
@@ -1092,7 +1276,7 @@
// Test that if we feed the replay with a mismatched node list that we die on
// the LogReader constructor.
-TEST_F(MultinodeLoggerDeathTest, MultiNodeBadReplayConfig) {
+TEST_P(MultinodeLoggerDeathTest, MultiNodeBadReplayConfig) {
time_converter_.StartEqual();
{
LoggerState pi1_logger = MakeLogger(pi1_);
@@ -1125,7 +1309,7 @@
// Tests that we can read log files where they don't start at the same monotonic
// time.
-TEST_F(MultinodeLoggerTest, StaggeredStart) {
+TEST_P(MultinodeLoggerTest, StaggeredStart) {
time_converter_.StartEqual();
{
LoggerState pi1_logger = MakeLogger(pi1_);
@@ -1230,7 +1414,7 @@
// Tests that we can read log files where the monotonic clocks drift and don't
// match correctly. While we are here, also test that different ending times
// also is readable.
-TEST_F(MultinodeLoggerTest, MismatchedClocks) {
+TEST_P(MultinodeLoggerTest, MismatchedClocks) {
// TODO(austin): Negate...
const chrono::nanoseconds initial_pi2_offset = chrono::seconds(1000);
@@ -1407,7 +1591,7 @@
}
// Tests that we can sort a bunch of parts into the pre-determined sorted parts.
-TEST_F(MultinodeLoggerTest, SortParts) {
+TEST_P(MultinodeLoggerTest, SortParts) {
time_converter_.StartEqual();
// Make a bunch of parts.
{
@@ -1428,7 +1612,7 @@
// Tests that we can sort a bunch of parts with an empty part. We should ignore
// it and remove it from the sorted list.
-TEST_F(MultinodeLoggerTest, SortEmptyParts) {
+TEST_P(MultinodeLoggerTest, SortEmptyParts) {
time_converter_.StartEqual();
// Make a bunch of parts.
{
@@ -1456,7 +1640,7 @@
#ifdef LZMA
// Tests that we can sort a bunch of parts with an empty .xz file in there. The
// empty file should be ignored.
-TEST_F(MultinodeLoggerTest, SortEmptyCompressedParts) {
+TEST_P(MultinodeLoggerTest, SortEmptyCompressedParts) {
time_converter_.StartEqual();
// Make a bunch of parts.
{
@@ -1485,7 +1669,7 @@
// Tests that we can sort a bunch of parts with the end missing off a compressed
// file. We should use the part we can read.
-TEST_F(MultinodeLoggerTest, SortTruncatedCompressedParts) {
+TEST_P(MultinodeLoggerTest, SortTruncatedCompressedParts) {
time_converter_.StartEqual();
// Make a bunch of parts.
{
@@ -1505,10 +1689,10 @@
// Strip off the end of one of the files. Pick one with a lot of data.
::std::string compressed_contents =
- aos::util::ReadFileToStringOrDie(logfiles_[0]);
+ aos::util::ReadFileToStringOrDie(logfiles_[2]);
aos::util::WriteStringToFileOrDie(
- logfiles_[0],
+ logfiles_[2],
compressed_contents.substr(0, compressed_contents.size() - 100));
const std::vector<LogFile> sorted_parts = SortParts(logfiles_);
@@ -1517,7 +1701,7 @@
#endif
// Tests that if we remap a remapped channel, it shows up correctly.
-TEST_F(MultinodeLoggerTest, RemapLoggedChannel) {
+TEST_P(MultinodeLoggerTest, RemapLoggedChannel) {
time_converter_.StartEqual();
{
LoggerState pi1_logger = MakeLogger(pi1_);
@@ -1584,7 +1768,7 @@
// Tests that we properly recreate forwarded timestamps when replaying a log.
// This should be enough that we can then re-run the logger and get a valid log
// back.
-TEST_F(MultinodeLoggerTest, MessageHeader) {
+TEST_P(MultinodeLoggerTest, MessageHeader) {
time_converter_.StartEqual();
{
LoggerState pi1_logger = MakeLogger(pi1_);
@@ -1664,134 +1848,168 @@
const chrono::nanoseconds network_delay = event_loop_factory_.network_delay();
const chrono::nanoseconds send_delay = event_loop_factory_.send_delay();
- pi1_event_loop->MakeWatcher(
- "/aos/remote_timestamps/pi2",
- [&pi1_event_loop, &pi2_event_loop, pi1_timestamp_channel,
- ping_timestamp_channel, &pi1_timestamp_on_pi1_fetcher,
- &pi1_timestamp_on_pi2_fetcher, &ping_on_pi1_fetcher,
- &ping_on_pi2_fetcher, network_delay,
- send_delay](const RemoteMessage &header) {
- const aos::monotonic_clock::time_point header_monotonic_sent_time(
- chrono::nanoseconds(header.monotonic_sent_time()));
- const aos::realtime_clock::time_point header_realtime_sent_time(
- chrono::nanoseconds(header.realtime_sent_time()));
- const aos::monotonic_clock::time_point header_monotonic_remote_time(
- chrono::nanoseconds(header.monotonic_remote_time()));
- const aos::realtime_clock::time_point header_realtime_remote_time(
- chrono::nanoseconds(header.realtime_remote_time()));
+ for (std::pair<int, std::string> channel :
+ shared()
+ ? std::vector<
+ std::pair<int, std::string>>{{-1,
+ "/aos/remote_timestamps/pi2"}}
+ : std::vector<std::pair<int, std::string>>{
+ {pi1_timestamp_channel,
+ "/aos/remote_timestamps/pi2/pi1/aos/"
+ "aos-message_bridge-Timestamp"},
+ {ping_timestamp_channel,
+ "/aos/remote_timestamps/pi2/test/aos-examples-Ping"}}) {
+ pi1_event_loop->MakeWatcher(
+ channel.second,
+ [&pi1_event_loop, &pi2_event_loop, pi1_timestamp_channel,
+ ping_timestamp_channel, &pi1_timestamp_on_pi1_fetcher,
+ &pi1_timestamp_on_pi2_fetcher, &ping_on_pi1_fetcher,
+ &ping_on_pi2_fetcher, network_delay, send_delay,
+ channel_index = channel.first](const RemoteMessage &header) {
+ const aos::monotonic_clock::time_point header_monotonic_sent_time(
+ chrono::nanoseconds(header.monotonic_sent_time()));
+ const aos::realtime_clock::time_point header_realtime_sent_time(
+ chrono::nanoseconds(header.realtime_sent_time()));
+ const aos::monotonic_clock::time_point header_monotonic_remote_time(
+ chrono::nanoseconds(header.monotonic_remote_time()));
+ const aos::realtime_clock::time_point header_realtime_remote_time(
+ chrono::nanoseconds(header.realtime_remote_time()));
- const Context *pi1_context = nullptr;
- const Context *pi2_context = nullptr;
+ if (channel_index != -1) {
+ ASSERT_EQ(channel_index, header.channel_index());
+ }
- if (header.channel_index() == pi1_timestamp_channel) {
- ASSERT_TRUE(pi1_timestamp_on_pi1_fetcher.FetchNext());
- ASSERT_TRUE(pi1_timestamp_on_pi2_fetcher.FetchNext());
- pi1_context = &pi1_timestamp_on_pi1_fetcher.context();
- pi2_context = &pi1_timestamp_on_pi2_fetcher.context();
- } else if (header.channel_index() == ping_timestamp_channel) {
- ASSERT_TRUE(ping_on_pi1_fetcher.FetchNext());
- ASSERT_TRUE(ping_on_pi2_fetcher.FetchNext());
- pi1_context = &ping_on_pi1_fetcher.context();
- pi2_context = &ping_on_pi2_fetcher.context();
- } else {
- LOG(FATAL) << "Unknown channel " << FlatbufferToJson(&header) << " "
- << configuration::CleanedChannelToString(
- pi1_event_loop->configuration()->channels()->Get(
- header.channel_index()));
- }
+ const Context *pi1_context = nullptr;
+ const Context *pi2_context = nullptr;
- ASSERT_TRUE(header.has_boot_uuid());
- EXPECT_EQ(header.boot_uuid()->string_view(),
- pi2_event_loop->boot_uuid().string_view());
+ if (header.channel_index() == pi1_timestamp_channel) {
+ ASSERT_TRUE(pi1_timestamp_on_pi1_fetcher.FetchNext());
+ ASSERT_TRUE(pi1_timestamp_on_pi2_fetcher.FetchNext());
+ pi1_context = &pi1_timestamp_on_pi1_fetcher.context();
+ pi2_context = &pi1_timestamp_on_pi2_fetcher.context();
+ } else if (header.channel_index() == ping_timestamp_channel) {
+ ASSERT_TRUE(ping_on_pi1_fetcher.FetchNext());
+ ASSERT_TRUE(ping_on_pi2_fetcher.FetchNext());
+ pi1_context = &ping_on_pi1_fetcher.context();
+ pi2_context = &ping_on_pi2_fetcher.context();
+ } else {
+ LOG(FATAL) << "Unknown channel " << FlatbufferToJson(&header) << " "
+ << configuration::CleanedChannelToString(
+ pi1_event_loop->configuration()->channels()->Get(
+ header.channel_index()));
+ }
- EXPECT_EQ(pi1_context->queue_index, header.remote_queue_index());
- EXPECT_EQ(pi2_context->remote_queue_index, header.remote_queue_index());
- EXPECT_EQ(pi2_context->queue_index, header.queue_index());
+ ASSERT_TRUE(header.has_boot_uuid());
+ EXPECT_EQ(header.boot_uuid()->string_view(),
+ pi2_event_loop->boot_uuid().string_view());
- EXPECT_EQ(pi2_context->monotonic_event_time,
- header_monotonic_sent_time);
- EXPECT_EQ(pi2_context->realtime_event_time, header_realtime_sent_time);
- EXPECT_EQ(pi2_context->realtime_remote_time,
- header_realtime_remote_time);
- EXPECT_EQ(pi2_context->monotonic_remote_time,
- header_monotonic_remote_time);
+ EXPECT_EQ(pi1_context->queue_index, header.remote_queue_index());
+ EXPECT_EQ(pi2_context->remote_queue_index,
+ header.remote_queue_index());
+ EXPECT_EQ(pi2_context->queue_index, header.queue_index());
- EXPECT_EQ(pi1_context->realtime_event_time,
- header_realtime_remote_time);
- EXPECT_EQ(pi1_context->monotonic_event_time,
- header_monotonic_remote_time);
+ EXPECT_EQ(pi2_context->monotonic_event_time,
+ header_monotonic_sent_time);
+ EXPECT_EQ(pi2_context->realtime_event_time,
+ header_realtime_sent_time);
+ EXPECT_EQ(pi2_context->realtime_remote_time,
+ header_realtime_remote_time);
+ EXPECT_EQ(pi2_context->monotonic_remote_time,
+ header_monotonic_remote_time);
- // Time estimation isn't perfect, but we know the clocks were identical
- // when logged, so we know when this should have come back. Confirm we
- // got it when we expected.
- EXPECT_EQ(
- pi1_event_loop->context().monotonic_event_time,
- pi1_context->monotonic_event_time + 2 * network_delay + send_delay);
- });
- pi2_event_loop->MakeWatcher(
- "/aos/remote_timestamps/pi1",
- [&pi2_event_loop, &pi1_event_loop, pi2_timestamp_channel,
- pong_timestamp_channel, &pi2_timestamp_on_pi2_fetcher,
- &pi2_timestamp_on_pi1_fetcher, &pong_on_pi2_fetcher,
- &pong_on_pi1_fetcher, network_delay,
- send_delay](const RemoteMessage &header) {
- const aos::monotonic_clock::time_point header_monotonic_sent_time(
- chrono::nanoseconds(header.monotonic_sent_time()));
- const aos::realtime_clock::time_point header_realtime_sent_time(
- chrono::nanoseconds(header.realtime_sent_time()));
- const aos::monotonic_clock::time_point header_monotonic_remote_time(
- chrono::nanoseconds(header.monotonic_remote_time()));
- const aos::realtime_clock::time_point header_realtime_remote_time(
- chrono::nanoseconds(header.realtime_remote_time()));
+ EXPECT_EQ(pi1_context->realtime_event_time,
+ header_realtime_remote_time);
+ EXPECT_EQ(pi1_context->monotonic_event_time,
+ header_monotonic_remote_time);
- const Context *pi2_context = nullptr;
- const Context *pi1_context = nullptr;
+ // Time estimation isn't perfect, but we know the clocks were
+ // identical when logged, so we know when this should have come back.
+ // Confirm we got it when we expected.
+ EXPECT_EQ(pi1_event_loop->context().monotonic_event_time,
+ pi1_context->monotonic_event_time + 2 * network_delay +
+ send_delay);
+ });
+ }
+ for (std::pair<int, std::string> channel :
+ shared()
+ ? std::vector<
+ std::pair<int, std::string>>{{-1,
+ "/aos/remote_timestamps/pi1"}}
+ : std::vector<std::pair<int, std::string>>{
+ {pi2_timestamp_channel,
+ "/aos/remote_timestamps/pi1/pi2/aos/"
+ "aos-message_bridge-Timestamp"}}) {
+ pi2_event_loop->MakeWatcher(
+ channel.second,
+ [&pi2_event_loop, &pi1_event_loop, pi2_timestamp_channel,
+ pong_timestamp_channel, &pi2_timestamp_on_pi2_fetcher,
+ &pi2_timestamp_on_pi1_fetcher, &pong_on_pi2_fetcher,
+ &pong_on_pi1_fetcher, network_delay, send_delay,
+ channel_index = channel.first](const RemoteMessage &header) {
+ const aos::monotonic_clock::time_point header_monotonic_sent_time(
+ chrono::nanoseconds(header.monotonic_sent_time()));
+ const aos::realtime_clock::time_point header_realtime_sent_time(
+ chrono::nanoseconds(header.realtime_sent_time()));
+ const aos::monotonic_clock::time_point header_monotonic_remote_time(
+ chrono::nanoseconds(header.monotonic_remote_time()));
+ const aos::realtime_clock::time_point header_realtime_remote_time(
+ chrono::nanoseconds(header.realtime_remote_time()));
- if (header.channel_index() == pi2_timestamp_channel) {
- ASSERT_TRUE(pi2_timestamp_on_pi2_fetcher.FetchNext());
- ASSERT_TRUE(pi2_timestamp_on_pi1_fetcher.FetchNext());
- pi2_context = &pi2_timestamp_on_pi2_fetcher.context();
- pi1_context = &pi2_timestamp_on_pi1_fetcher.context();
- } else if (header.channel_index() == pong_timestamp_channel) {
- ASSERT_TRUE(pong_on_pi2_fetcher.FetchNext());
- ASSERT_TRUE(pong_on_pi1_fetcher.FetchNext());
- pi2_context = &pong_on_pi2_fetcher.context();
- pi1_context = &pong_on_pi1_fetcher.context();
- } else {
- LOG(FATAL) << "Unknown channel " << FlatbufferToJson(&header) << " "
- << configuration::CleanedChannelToString(
- pi2_event_loop->configuration()->channels()->Get(
- header.channel_index()));
- }
+ if (channel_index != -1) {
+ ASSERT_EQ(channel_index, header.channel_index());
+ }
- ASSERT_TRUE(header.has_boot_uuid());
- EXPECT_EQ(header.boot_uuid()->string_view(),
- pi1_event_loop->boot_uuid().string_view());
+ const Context *pi2_context = nullptr;
+ const Context *pi1_context = nullptr;
- EXPECT_EQ(pi2_context->queue_index, header.remote_queue_index());
- EXPECT_EQ(pi1_context->remote_queue_index, header.remote_queue_index());
- EXPECT_EQ(pi1_context->queue_index, header.queue_index());
+ if (header.channel_index() == pi2_timestamp_channel) {
+ ASSERT_TRUE(pi2_timestamp_on_pi2_fetcher.FetchNext());
+ ASSERT_TRUE(pi2_timestamp_on_pi1_fetcher.FetchNext());
+ pi2_context = &pi2_timestamp_on_pi2_fetcher.context();
+ pi1_context = &pi2_timestamp_on_pi1_fetcher.context();
+ } else if (header.channel_index() == pong_timestamp_channel) {
+ ASSERT_TRUE(pong_on_pi2_fetcher.FetchNext());
+ ASSERT_TRUE(pong_on_pi1_fetcher.FetchNext());
+ pi2_context = &pong_on_pi2_fetcher.context();
+ pi1_context = &pong_on_pi1_fetcher.context();
+ } else {
+ LOG(FATAL) << "Unknown channel " << FlatbufferToJson(&header) << " "
+ << configuration::CleanedChannelToString(
+ pi2_event_loop->configuration()->channels()->Get(
+ header.channel_index()));
+ }
- EXPECT_EQ(pi1_context->monotonic_event_time,
- header_monotonic_sent_time);
- EXPECT_EQ(pi1_context->realtime_event_time, header_realtime_sent_time);
- EXPECT_EQ(pi1_context->realtime_remote_time,
- header_realtime_remote_time);
- EXPECT_EQ(pi1_context->monotonic_remote_time,
- header_monotonic_remote_time);
+ ASSERT_TRUE(header.has_boot_uuid());
+ EXPECT_EQ(header.boot_uuid()->string_view(),
+ pi1_event_loop->boot_uuid().string_view());
- EXPECT_EQ(pi2_context->realtime_event_time,
- header_realtime_remote_time);
- EXPECT_EQ(pi2_context->monotonic_event_time,
- header_monotonic_remote_time);
+ EXPECT_EQ(pi2_context->queue_index, header.remote_queue_index());
+ EXPECT_EQ(pi1_context->remote_queue_index,
+ header.remote_queue_index());
+ EXPECT_EQ(pi1_context->queue_index, header.queue_index());
- // Time estimation isn't perfect, but we know the clocks were identical
- // when logged, so we know when this should have come back. Confirm we
- // got it when we expected.
- EXPECT_EQ(
- pi2_event_loop->context().monotonic_event_time,
- pi2_context->monotonic_event_time + 2 * network_delay + send_delay);
- });
+ EXPECT_EQ(pi1_context->monotonic_event_time,
+ header_monotonic_sent_time);
+ EXPECT_EQ(pi1_context->realtime_event_time,
+ header_realtime_sent_time);
+ EXPECT_EQ(pi1_context->realtime_remote_time,
+ header_realtime_remote_time);
+ EXPECT_EQ(pi1_context->monotonic_remote_time,
+ header_monotonic_remote_time);
+
+ EXPECT_EQ(pi2_context->realtime_event_time,
+ header_realtime_remote_time);
+ EXPECT_EQ(pi2_context->monotonic_event_time,
+ header_monotonic_remote_time);
+
+ // Time estimation isn't perfect, but we know the clocks were
+ // identical when logged, so we know when this should have come back.
+ // Confirm we got it when we expected.
+ EXPECT_EQ(pi2_event_loop->context().monotonic_event_time,
+ pi2_context->monotonic_event_time + 2 * network_delay +
+ send_delay);
+ });
+ }
// And confirm we can re-create a log again, while checking the contents.
{
@@ -1823,7 +2041,7 @@
// Tests that we properly populate and extract the logger_start time by setting
// up a clock difference between 2 nodes and looking at the resulting parts.
-TEST_F(MultinodeLoggerTest, LoggerStartTime) {
+TEST_P(MultinodeLoggerTest, LoggerStartTime) {
time_converter_.AddMonotonic(
{monotonic_clock::epoch(),
monotonic_clock::epoch() + chrono::seconds(1000)});
@@ -1865,7 +2083,7 @@
// Tests that we properly recreate forwarded timestamps when replaying a log.
// This should be enough that we can then re-run the logger and get a valid log
// back.
-TEST_F(MultinodeLoggerDeathTest, RemoteReboot) {
+TEST_P(MultinodeLoggerDeathTest, RemoteReboot) {
time_converter_.StartEqual();
std::string pi2_boot1;
std::string pi2_boot2;
@@ -1908,7 +2126,7 @@
// Tests that we properly handle one direction of message_bridge being
// unavailable.
-TEST_F(MultinodeLoggerTest, OneDirectionWithNegativeSlope) {
+TEST_P(MultinodeLoggerTest, OneDirectionWithNegativeSlope) {
event_loop_factory_.GetNodeEventLoopFactory(pi1_)->Disconnect(pi2_);
time_converter_.AddMonotonic(
{monotonic_clock::epoch(),
@@ -1934,7 +2152,7 @@
// Tests that we properly handle one direction of message_bridge being
// unavailable.
-TEST_F(MultinodeLoggerTest, OneDirectionWithPositiveSlope) {
+TEST_P(MultinodeLoggerTest, OneDirectionWithPositiveSlope) {
event_loop_factory_.GetNodeEventLoopFactory(pi1_)->Disconnect(pi2_);
time_converter_.AddMonotonic(
{monotonic_clock::epoch(),
@@ -1960,7 +2178,7 @@
// Tests that we properly handle a dead node. Do this by just disconnecting it
// and only using one nodes of logs.
-TEST_F(MultinodeLoggerTest, DeadNode) {
+TEST_P(MultinodeLoggerTest, DeadNode) {
event_loop_factory_.GetNodeEventLoopFactory(pi1_)->Disconnect(pi2_);
event_loop_factory_.GetNodeEventLoopFactory(pi2_)->Disconnect(pi1_);
time_converter_.AddMonotonic(
@@ -1981,6 +2199,26 @@
ConfirmReadable(pi1_single_direction_logfiles_);
}
+INSTANTIATE_TEST_CASE_P(
+ All, MultinodeLoggerTest,
+ ::testing::Values(
+ Param{
+ "multinode_pingpong_combined_config.json", true,
+ "47511a1906dbb59cf9f8ad98ad08e568c718a4deb204c8bbce81ff76cef9095c"},
+ Param{"multinode_pingpong_split_config.json", false,
+ "ce3ec411a089e5b80d6868bdb2ff8ce86467053b41469e50a09edf3c0110d80"
+ "f"}));
+
+INSTANTIATE_TEST_CASE_P(
+ All, MultinodeLoggerDeathTest,
+ ::testing::Values(
+ Param{
+ "multinode_pingpong_combined_config.json", true,
+ "47511a1906dbb59cf9f8ad98ad08e568c718a4deb204c8bbce81ff76cef9095c"},
+ Param{"multinode_pingpong_split_config.json", false,
+ "ce3ec411a089e5b80d6868bdb2ff8ce86467053b41469e50a09edf3c0110d80"
+ "f"}));
+
// TODO(austin): Make a log file where the remote node has no start time.
} // namespace testing
diff --git a/aos/events/logging/multinode_pingpong.json b/aos/events/logging/multinode_pingpong_combined.json
similarity index 100%
rename from aos/events/logging/multinode_pingpong.json
rename to aos/events/logging/multinode_pingpong_combined.json
diff --git a/aos/events/logging/multinode_pingpong_split.json b/aos/events/logging/multinode_pingpong_split.json
new file mode 100644
index 0000000..4ead71e
--- /dev/null
+++ b/aos/events/logging/multinode_pingpong_split.json
@@ -0,0 +1,178 @@
+{
+ "channels": [
+ {
+ "name": "/pi1/aos",
+ "type": "aos.logging.LogMessageFbs",
+ "source_node": "pi1",
+ "frequency": 200,
+ "num_senders": 20,
+ "max_size": 2048
+ },
+ {
+ "name": "/pi2/aos",
+ "type": "aos.logging.LogMessageFbs",
+ "source_node": "pi2",
+ "frequency": 200,
+ "num_senders": 20,
+ "max_size": 2048
+ },
+ /* Logged on pi1 locally */
+ {
+ "name": "/pi1/aos",
+ "type": "aos.timing.Report",
+ "source_node": "pi1",
+ "frequency": 50,
+ "num_senders": 20,
+ "max_size": 2048
+ },
+ {
+ "name": "/pi2/aos",
+ "type": "aos.timing.Report",
+ "source_node": "pi2",
+ "frequency": 50,
+ "num_senders": 20,
+ "max_size": 2048
+ },
+ {
+ "name": "/pi1/aos",
+ "type": "aos.message_bridge.ServerStatistics",
+ "logger": "LOCAL_LOGGER",
+ "source_node": "pi1"
+ },
+ {
+ "name": "/pi2/aos",
+ "type": "aos.message_bridge.ServerStatistics",
+ "logger": "LOCAL_LOGGER",
+ "source_node": "pi2"
+ },
+ {
+ "name": "/pi1/aos",
+ "type": "aos.message_bridge.ClientStatistics",
+ "logger": "LOCAL_LOGGER",
+ "source_node": "pi1"
+ },
+ {
+ "name": "/pi2/aos",
+ "type": "aos.message_bridge.ClientStatistics",
+ "logger": "LOCAL_LOGGER",
+ "source_node": "pi2"
+ },
+ {
+ "name": "/pi1/aos",
+ "type": "aos.message_bridge.Timestamp",
+ "logger": "LOCAL_AND_REMOTE_LOGGER",
+ "logger_nodes": ["pi2"],
+ "source_node": "pi1",
+ "destination_nodes": [
+ {
+ "name": "pi2",
+ "timestamp_logger": "LOCAL_AND_REMOTE_LOGGER",
+ "timestamp_logger_nodes": ["pi1"]
+ }
+ ]
+ },
+ {
+ "name": "/pi2/aos",
+ "type": "aos.message_bridge.Timestamp",
+ "logger": "LOCAL_AND_REMOTE_LOGGER",
+ "logger_nodes": ["pi1"],
+ "source_node": "pi2",
+ "destination_nodes": [
+ {
+ "name": "pi1",
+ "timestamp_logger": "LOCAL_AND_REMOTE_LOGGER",
+ "timestamp_logger_nodes": ["pi2"]
+ }
+ ]
+ },
+ {
+ "name": "/pi1/aos/remote_timestamps/pi2/pi1/aos/aos-message_bridge-Timestamp",
+ "type": "aos.message_bridge.RemoteMessage",
+ "logger": "NOT_LOGGED",
+ "num_senders": 2,
+ "source_node": "pi1"
+ },
+ {
+ "name": "/pi2/aos/remote_timestamps/pi1/pi2/aos/aos-message_bridge-Timestamp",
+ "type": "aos.message_bridge.RemoteMessage",
+ "logger": "NOT_LOGGED",
+ "num_senders": 2,
+ "source_node": "pi2"
+ },
+ {
+ "name": "/pi1/aos/remote_timestamps/pi2/test/aos-examples-Ping",
+ "type": "aos.message_bridge.RemoteMessage",
+ "logger": "NOT_LOGGED",
+ "num_senders": 2,
+ "source_node": "pi1"
+ },
+ /* Forwarded to pi2 */
+ {
+ "name": "/test",
+ "type": "aos.examples.Ping",
+ "source_node": "pi1",
+ "destination_nodes": [
+ {
+ "name": "pi2",
+ "priority": 1,
+ "timestamp_logger": "LOCAL_AND_REMOTE_LOGGER",
+ "timestamp_logger_nodes": ["pi1"],
+ "time_to_live": 5000000
+ }
+ ]
+ },
+ /* Forwarded back to pi1.
+ * The message is logged both on the sending node and the receiving node
+ * (to make it easier to look at the results for now).
+ *
+ * The timestamps are logged on the receiving node.
+ */
+ {
+ "name": "/test",
+ "type": "aos.examples.Pong",
+ "source_node": "pi2",
+ "logger": "LOCAL_AND_REMOTE_LOGGER",
+ "logger_nodes": ["pi1"],
+ "destination_nodes": [
+ {
+ "name": "pi1",
+ "priority": 1,
+ "timestamp_logger": "LOCAL_LOGGER",
+ "time_to_live": 5000000
+ }
+ ]
+ }
+ ],
+ "maps": [
+ {
+ "match": {
+ "name": "/aos*",
+ "source_node": "pi1"
+ },
+ "rename": {
+ "name": "/pi1/aos"
+ }
+ },
+ {
+ "match": {
+ "name": "/aos*",
+ "source_node": "pi2"
+ },
+ "rename": {
+ "name": "/pi2/aos"
+ }
+ }
+ ],
+ "nodes": [
+ {
+ "name": "pi1",
+ "hostname": "raspberrypi",
+ "port": 9971
+ },
+ {
+ "name": "pi2",
+ "hostname": "raspberrypi2",
+ "port": 9971
+ }
+ ]
+}
diff --git a/aos/network/message_bridge_test.cc b/aos/network/message_bridge_test.cc
index 1f926c8..e2dddc4 100644
--- a/aos/network/message_bridge_test.cc
+++ b/aos/network/message_bridge_test.cc
@@ -391,46 +391,45 @@
// Wait until we are connected, then send.
int ping_count = 0;
int pi1_server_statistics_count = 0;
- ping_event_loop.MakeWatcher(
- "/pi1/aos",
- [this, &ping_count, &ping_sender,
- &pi1_server_statistics_count](const ServerStatistics &stats) {
- VLOG(1) << "/pi1/aos ServerStatistics " << FlatbufferToJson(&stats);
+ ping_event_loop.MakeWatcher("/pi1/aos", [this, &ping_count, &ping_sender,
+ &pi1_server_statistics_count](
+ const ServerStatistics &stats) {
+ VLOG(1) << "/pi1/aos ServerStatistics " << FlatbufferToJson(&stats);
- ASSERT_TRUE(stats.has_connections());
- EXPECT_EQ(stats.connections()->size(), 1);
+ ASSERT_TRUE(stats.has_connections());
+ EXPECT_EQ(stats.connections()->size(), 1);
- bool connected = false;
- for (const ServerConnection *connection : *stats.connections()) {
- // Confirm that we are estimating the server time offset correctly. It
- // should be about 0 since we are on the same machine here.
- if (connection->has_monotonic_offset()) {
- EXPECT_LT(chrono::nanoseconds(connection->monotonic_offset()),
- chrono::milliseconds(1));
- EXPECT_GT(chrono::nanoseconds(connection->monotonic_offset()),
- chrono::milliseconds(-1));
- ++pi1_server_statistics_count;
- }
+ bool connected = false;
+ for (const ServerConnection *connection : *stats.connections()) {
+ // Confirm that we are estimating the server time offset correctly. It
+ // should be about 0 since we are on the same machine here.
+ if (connection->has_monotonic_offset()) {
+ EXPECT_LT(chrono::nanoseconds(connection->monotonic_offset()),
+ chrono::milliseconds(1));
+ EXPECT_GT(chrono::nanoseconds(connection->monotonic_offset()),
+ chrono::milliseconds(-1));
+ ++pi1_server_statistics_count;
+ }
- if (connection->node()->name()->string_view() ==
- pi2_client_event_loop->node()->name()->string_view()) {
- if (connection->state() == State::CONNECTED) {
- EXPECT_TRUE(connection->has_boot_uuid());
- connected = true;
- }
- }
+ if (connection->node()->name()->string_view() ==
+ pi2_client_event_loop->node()->name()->string_view()) {
+ if (connection->state() == State::CONNECTED) {
+ EXPECT_TRUE(connection->has_boot_uuid());
+ connected = true;
}
+ }
+ }
- if (connected) {
- VLOG(1) << "Connected! Sent ping.";
- auto builder = ping_sender.MakeBuilder();
- examples::Ping::Builder ping_builder =
- builder.MakeBuilder<examples::Ping>();
- ping_builder.add_value(ping_count + 971);
- builder.Send(ping_builder.Finish());
- ++ping_count;
- }
- });
+ if (connected) {
+ VLOG(1) << "Connected! Sent ping.";
+ auto builder = ping_sender.MakeBuilder();
+ examples::Ping::Builder ping_builder =
+ builder.MakeBuilder<examples::Ping>();
+ ping_builder.add_value(ping_count + 971);
+ builder.Send(ping_builder.Finish());
+ ++ping_count;
+ }
+ });
// Confirm both client and server statistics messages have decent offsets in
// them.
@@ -1005,7 +1004,7 @@
pi1_remote_timestamp_event_loop.MakeWatcher(
channel_name, [this, channel_name, ping_channel_index,
&ping_timestamp_count](const RemoteMessage &header) {
- VLOG(1) <<channel_name << " RemoteMessage "
+ VLOG(1) << channel_name << " RemoteMessage "
<< aos::FlatbufferToJson(&header);
EXPECT_TRUE(header.has_boot_uuid());
if (shared() && header.channel_index() != ping_channel_index) {
diff --git a/aos/network/timestamp_channel.cc b/aos/network/timestamp_channel.cc
index 12c6713..d022b60 100644
--- a/aos/network/timestamp_channel.cc
+++ b/aos/network/timestamp_channel.cc
@@ -5,62 +5,99 @@
namespace aos {
namespace message_bridge {
+ChannelTimestampFinder::ChannelTimestampFinder(
+ const Configuration *configuration, const std::string_view name,
+ const Node *node)
+ : configuration_(configuration), name_(name), node_(node) {}
+
+std::string ChannelTimestampFinder::SplitChannelName(
+ const Channel *channel, const Connection *connection) {
+ std::string type(channel->type()->string_view());
+ std::replace(type.begin(), type.end(), '.', '-');
+ return absl::StrCat("/aos/remote_timestamps/",
+ connection->name()->string_view(),
+ channel->name()->string_view(), "/", type);
+}
+
+std::string ChannelTimestampFinder::CombinedChannelName(
+ std::string_view remote_node) {
+ return absl::StrCat("/aos/remote_timestamps/", remote_node);
+}
+
+const Channel *ChannelTimestampFinder::ForChannel(
+ const Channel *channel, const Connection *connection) {
+ const std::string split_timestamp_channel_name =
+ SplitChannelName(channel, connection);
+ const Channel *split_timestamp_channel = configuration::GetChannel(
+ configuration_, split_timestamp_channel_name,
+ RemoteMessage::GetFullyQualifiedName(), name_, node_, true);
+ if (split_timestamp_channel != nullptr) {
+ return split_timestamp_channel;
+ }
+
+ const std::string shared_timestamp_channel_name =
+ CombinedChannelName(connection->name()->string_view());
+ const Channel *shared_timestamp_channel = configuration::GetChannel(
+ configuration_, shared_timestamp_channel_name,
+ RemoteMessage::GetFullyQualifiedName(), name_, node_, true);
+ if (shared_timestamp_channel != nullptr) {
+ LOG(WARNING) << "Failed to find timestamp channel {\"name\": \""
+ << split_timestamp_channel << "\", \"type\": \""
+ << RemoteMessage::GetFullyQualifiedName()
+ << "\"}, falling back to old version.";
+ return shared_timestamp_channel;
+ }
+
+ CHECK(shared_timestamp_channel != nullptr)
+ << ": Remote timestamp channel { \"name\": \""
+ << split_timestamp_channel_name << "\", \"type\": \""
+ << RemoteMessage::GetFullyQualifiedName()
+ << "\" } not found in config for " << name_
+ << (configuration::MultiNode(configuration_)
+ ? absl::StrCat(" on node ", node_->name()->string_view())
+ : ".");
+
+ return nullptr;
+}
+
ChannelTimestampSender::ChannelTimestampSender(aos::EventLoop *event_loop)
: event_loop_(event_loop) {
CHECK(configuration::MultiNode(event_loop_->configuration()));
- timestamp_loggers_.resize(event_loop_->configuration()->nodes()->size());
}
aos::Sender<RemoteMessage> *ChannelTimestampSender::SenderForChannel(
const Channel *channel, const Connection *connection) {
+ ChannelTimestampFinder finder(event_loop_);
// Look at any pre-created channel/connection pairs.
- auto it =
- channel_timestamp_loggers_.find(std::make_pair(channel, connection));
- if (it != channel_timestamp_loggers_.end()) {
- return it->second.get();
+ {
+ auto it =
+ channel_timestamp_loggers_.find(std::make_pair(channel, connection));
+ if (it != channel_timestamp_loggers_.end()) {
+ return it->second.get();
+ }
}
- const Node *other_node = configuration::GetNode(
- event_loop_->configuration(), connection->name()->string_view());
- const size_t other_node_index =
- configuration::GetNodeIndex(event_loop_->configuration(), other_node);
+ const Channel *timestamp_channel = finder.ForChannel(channel, connection);
- std::string type(channel->type()->string_view());
- std::replace(type.begin(), type.end(), '.', '-');
- const std::string single_timestamp_channel =
- absl::StrCat("/aos/remote_timestamps/", connection->name()->string_view(),
- channel->name()->string_view(), "/", type);
- if (event_loop_->HasChannel<RemoteMessage>(single_timestamp_channel)) {
- LOG(INFO) << "Making RemoteMessage channel " << single_timestamp_channel;
- auto result = channel_timestamp_loggers_.try_emplace(
- std::make_pair(channel, connection),
- std::make_unique<aos::Sender<RemoteMessage>>(
- event_loop_->MakeSender<RemoteMessage>(single_timestamp_channel)));
- return result.first->second.get();
- } else {
- // Then look for any per-remote-node channels.
- if (timestamp_loggers_[other_node_index]) {
- return ×tamp_loggers_[other_node_index];
+ {
+ auto it = timestamp_loggers_.find(timestamp_channel);
+ if (it != timestamp_loggers_.end()) {
+ CHECK(channel_timestamp_loggers_
+ .try_emplace(std::make_pair(channel, connection), it->second)
+ .second);
+ return it->second.get();
}
- const std::string shared_timestamp_channel = absl::StrCat(
- "/aos/remote_timestamps/", connection->name()->string_view());
- LOG(INFO) << "Looking for " << shared_timestamp_channel;
- if (event_loop_->HasChannel<RemoteMessage>(shared_timestamp_channel)) {
- LOG(WARNING) << "Failed to find timestamp channel {\"name\": \""
- << single_timestamp_channel << "\", \"type\": \""
- << RemoteMessage::GetFullyQualifiedName()
- << "\"}, falling back to old version.";
- timestamp_loggers_[other_node_index] =
- event_loop_->MakeSender<RemoteMessage>(shared_timestamp_channel);
- return ×tamp_loggers_[other_node_index];
- } else {
- LOG(ERROR) << "Failed";
- }
-
- // Explode with an error about the new channel.
- event_loop_->MakeSender<RemoteMessage>(single_timestamp_channel);
- return nullptr;
}
+
+ auto result = channel_timestamp_loggers_.try_emplace(
+ std::make_pair(channel, connection),
+ std::make_shared<aos::Sender<RemoteMessage>>(
+ event_loop_->MakeSender<RemoteMessage>(
+ timestamp_channel->name()->string_view())));
+
+ CHECK(timestamp_loggers_.try_emplace(timestamp_channel, result.first->second)
+ .second);
+ return result.first->second.get();
}
} // namespace message_bridge
diff --git a/aos/network/timestamp_channel.h b/aos/network/timestamp_channel.h
index b62c5e9..1702632 100644
--- a/aos/network/timestamp_channel.h
+++ b/aos/network/timestamp_channel.h
@@ -1,6 +1,7 @@
#ifndef AOS_NETWORK_TIMESTAMP_CHANNEL_
#define AOS_NETWORK_TIMESTAMP_CHANNEL_
+#include <string_view>
#include <vector>
#include "absl/container/btree_map.h"
@@ -12,6 +13,30 @@
namespace aos {
namespace message_bridge {
+// Class to find the corresponding channel where timestamps for a specified data
+// channel and connection will be logged.
+class ChannelTimestampFinder {
+ public:
+ ChannelTimestampFinder(aos::EventLoop *event_loop)
+ : ChannelTimestampFinder(event_loop->configuration(), event_loop->name(),
+ event_loop->node()) {}
+ ChannelTimestampFinder(const Configuration *configuration,
+ const std::string_view name, const Node *node);
+
+ // Finds the timestamp logging channel for the provided data channel and
+ // connection.
+ const Channel *ForChannel(const Channel *channel,
+ const Connection *connection);
+ std::string SplitChannelName(const Channel *channel,
+ const Connection *connection);
+ std::string CombinedChannelName(std::string_view remote_node);
+
+ private:
+ const Configuration *configuration_;
+ const std::string_view name_;
+ const Node *node_;
+};
+
// Class to manage lifetime, and creating senders for channels and connections.
class ChannelTimestampSender {
public:
@@ -35,13 +60,15 @@
// I'd prefer 3) to be an error, but don't have strong opinions. We will
// still be correct if it gets used, as long as everything is consistent.
- // List of Senders per node.
- std::vector<aos::Sender<RemoteMessage>> timestamp_loggers_;
-
// Mapping from channel and connection to logger.
absl::btree_map<std::pair<const Channel *, const Connection *>,
- std::unique_ptr<aos::Sender<RemoteMessage>>>
+ std::shared_ptr<aos::Sender<RemoteMessage>>>
channel_timestamp_loggers_;
+
+ // Mapping from channel to RemoteMessage sender. This is the channel that
+ // timestamps are published to.
+ absl::btree_map<const Channel *, std::shared_ptr<aos::Sender<RemoteMessage>>>
+ timestamp_loggers_;
};
} // namespace message_bridge