Split configuration into a separate file.
This saves space and CPU. The configuration can't change (by design) in
a log file, so it was previously being duplicated. In some cases with
lots of forwarded messages and nodes, rotating was triggering
recompression of enough configuration information that we were falling
behind.
Note: on purpose, we aren't storing a link from the log file to the
header. We don't store links between parts either. It up to the user
to provide the config in the list of files and folders provided.
Change-Id: I2cd600ed76c5f6f4b2bd6ba77d49bc739227756f
diff --git a/aos/events/logging/log_namer.cc b/aos/events/logging/log_namer.cc
index 2d1e111..44e8f5a 100644
--- a/aos/events/logging/log_namer.cc
+++ b/aos/events/logging/log_namer.cc
@@ -50,6 +50,18 @@
UpdateHeader(header, uuid_, part_number_);
data_writer_->QueueSpan(header->span());
}
+
+void LocalLogNamer::WriteConfiguration(
+ aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> *header,
+ std::string_view config_sha256) {
+ const std::string filename = absl::StrCat(base_name_, config_sha256, ".bfbs");
+
+ std::unique_ptr<DetachedBufferWriter> writer =
+ std::make_unique<DetachedBufferWriter>(
+ filename, std::make_unique<aos::logger::DummyEncoder>());
+ writer->QueueSizedFlatbuffer(header->Release());
+}
+
void LocalLogNamer::Reboot(
const Node * /*node*/,
aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> * /*header*/) {
@@ -120,7 +132,8 @@
void MultiNodeLogNamer::DoRotate(
const Node *node,
- aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> *header, bool reboot) {
+ aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> *header,
+ bool reboot) {
if (node == this->node()) {
if (data_writer_.writer) {
if (reboot) {
@@ -148,6 +161,28 @@
}
}
+void MultiNodeLogNamer::WriteConfiguration(
+ aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> *header,
+ std::string_view config_sha256) {
+ if (ran_out_of_space_) {
+ return;
+ }
+
+ const std::string_view separator = base_name_.back() == '/' ? "" : "_";
+ const std::string filename = absl::StrCat(
+ base_name_, separator, config_sha256, ".bfbs", extension_, temp_suffix_);
+
+ std::unique_ptr<DetachedBufferWriter> writer =
+ std::make_unique<DetachedBufferWriter>(filename, encoder_factory_());
+
+ writer->QueueSizedFlatbuffer(header->Release());
+
+ if (!writer->ran_out_of_space()) {
+ all_filenames_.emplace_back(filename);
+ }
+ CloseWriter(&writer);
+}
+
DetachedBufferWriter *MultiNodeLogNamer::MakeWriter(const Channel *channel) {
// See if we can read the data on this node at all.
const bool is_readable =
diff --git a/aos/events/logging/log_namer.h b/aos/events/logging/log_namer.h
index 7e58152..2d4c23e 100644
--- a/aos/events/logging/log_namer.h
+++ b/aos/events/logging/log_namer.h
@@ -74,6 +74,11 @@
// Returns the node the logger is running on.
const Node *node() const { return node_; }
+ // Writes out the nested Configuration object to the config file location.
+ virtual void WriteConfiguration(
+ aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> *header,
+ std::string_view config_sha256) = 0;
+
protected:
// Modifies the header to have the provided UUID and part id.
void UpdateHeader(
@@ -115,6 +120,10 @@
DetachedBufferWriter *MakeForwardedTimestampWriter(
const Channel * /*channel*/, const Node * /*node*/) override;
+ void WriteConfiguration(
+ aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> *header,
+ std::string_view config_sha256) override;
+
private:
// Creates a new data writer with the new part number.
std::unique_ptr<DetachedBufferWriter> OpenDataWriter() {
@@ -181,6 +190,10 @@
aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> *header)
override;
+ void WriteConfiguration(
+ aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> *header,
+ std::string_view config_sha256) override;
+
DetachedBufferWriter *MakeWriter(const Channel *channel) override;
DetachedBufferWriter *MakeForwardedTimestampWriter(const Channel *channel,
diff --git a/aos/events/logging/logfile_sorting.cc b/aos/events/logging/logfile_sorting.cc
index f03bb3d..b1a993e 100644
--- a/aos/events/logging/logfile_sorting.cc
+++ b/aos/events/logging/logfile_sorting.cc
@@ -12,8 +12,7 @@
#include "aos/flatbuffer_merge.h"
#include "aos/flatbuffers.h"
#include "aos/time/time.h"
-
-#include <openssl/sha.h>
+#include "openssl/sha.h"
namespace aos {
namespace logger {
@@ -47,6 +46,28 @@
return error == 0;
}
+bool ConfigOnly(const LogFileHeader *header) {
+ CHECK_EQ(LogFileHeader::MiniReflectTypeTable()->num_elems, 17u);
+ if (header->has_monotonic_start_time()) return false;
+ if (header->has_realtime_start_time()) return false;
+ if (header->has_max_out_of_order_duration()) return false;
+ if (header->has_configuration_sha256()) return false;
+ if (header->has_name()) return false;
+ if (header->has_node()) return false;
+ if (header->has_log_event_uuid()) return false;
+ if (header->has_logger_instance_uuid()) return false;
+ if (header->has_logger_node_boot_uuid()) return false;
+ if (header->has_source_node_boot_uuid()) return false;
+ if (header->has_logger_monotonic_start_time()) return false;
+ if (header->has_logger_realtime_start_time()) return false;
+ if (header->has_log_start_uuid()) return false;
+ if (header->has_parts_uuid()) return false;
+ if (header->has_parts_index()) return false;
+ if (header->has_logger_node()) return false;
+
+ return header->has_configuration();
+}
+
} // namespace
void FindLogs(std::vector<std::string> *files, std::string filename) {
@@ -90,6 +111,9 @@
std::vector<LogFile> SortParts(const std::vector<std::string> &parts) {
std::vector<std::string> corrupted;
+ std::map<std::string, std::shared_ptr<const Configuration>>
+ config_sha256_lookup;
+
// Start by grouping all parts by UUID, and extracting the part index.
// Datastructure to hold all the info extracted from a set of parts which go
// together so we can sort them afterwords.
@@ -111,6 +135,8 @@
// Pairs of the filename and the part index for sorting.
std::vector<std::pair<std::string, int>> parts;
+
+ std::string config_sha256;
};
// Struct to hold both the node, and the parts associated with it.
@@ -199,6 +225,31 @@
? log_header->message().source_node_boot_uuid()->string_view()
: "";
+ const std::string_view configuration_sha256 =
+ log_header->message().has_configuration_sha256()
+ ? log_header->message().configuration_sha256()->string_view()
+ : "";
+
+ if (ConfigOnly(&log_header->message())) {
+ const std::string hash = Sha256(log_header->span());
+
+ if (config_sha256_lookup.find(hash) == config_sha256_lookup.end()) {
+ auto header =
+ std::make_shared<SizePrefixedFlatbufferVector<LogFileHeader>>(
+ std::move(*log_header));
+ config_sha256_lookup.emplace(
+ hash, std::shared_ptr<const Configuration>(
+ header, header->message().configuration()));
+ }
+ continue;
+ }
+
+ if (configuration_sha256.empty()) {
+ CHECK(log_header->message().has_configuration());
+ } else {
+ CHECK(!log_header->message().has_configuration());
+ }
+
// Looks like an old log. No UUID, index, and also single node. We have
// little to no multi-node log files in the wild without part UUIDs and
// indexes which we care much about.
@@ -242,8 +293,6 @@
CHECK(log_header->message().has_parts_uuid());
CHECK(log_header->message().has_parts_index());
- CHECK(log_header->message().has_configuration());
-
CHECK_EQ(log_header->message().has_logger_node(),
log_header->message().has_node());
@@ -289,8 +338,10 @@
it->second.logger_realtime_start_time = logger_realtime_start_time;
it->second.node = std::string(node);
it->second.source_boot_uuid = source_boot_uuid;
+ it->second.config_sha256 = configuration_sha256;
} else {
CHECK_EQ(it->second.source_boot_uuid, source_boot_uuid);
+ CHECK_EQ(it->second.config_sha256, configuration_sha256);
}
// First part might be min_time. If it is, try to put a better time on it.
@@ -366,7 +417,7 @@
std::shared_ptr<const Configuration> config(
header, header->message().configuration());
- copied_config_sha256.emplace(config_copy_sha256, config);
+ copied_config_sha256.emplace(std::move(config_copy_sha256), config);
log_file.config = config;
}
@@ -399,6 +450,7 @@
new_file.name = logs.second.name;
new_file.corrupted = corrupted;
bool seen_part = false;
+ std::string config_sha256;
for (std::pair<const std::string, UnsortedLogParts> &parts :
logs.second.unsorted_parts) {
LogParts new_parts;
@@ -423,28 +475,60 @@
new_parts.parts.emplace_back(std::move(p.first));
}
- if (!seen_part) {
- auto header =
- std::make_shared<SizePrefixedFlatbufferVector<LogFileHeader>>(
- std::move(*ReadHeader(new_parts.parts[0])));
-
- std::shared_ptr<const Configuration> config(
- header, header->message().configuration());
-
- FlatbufferDetachedBuffer<Configuration> config_copy =
- RecursiveCopyFlatBuffer(header->message().configuration());
-
- std::string config_copy_sha256 = Sha256(config_copy.span());
-
- auto it = copied_config_sha256.find(config_copy_sha256);
- if (it != copied_config_sha256.end()) {
- new_file.config = it->second;
+ if (!parts.second.config_sha256.empty()) {
+ // The easy case. We've got a sha256 to point to, so go look it up.
+ // Abort if it doesn't exist.
+ auto it = config_sha256_lookup.find(parts.second.config_sha256);
+ CHECK(it != config_sha256_lookup.end())
+ << ": Failed to find a matching config with a SHA256 of "
+ << parts.second.config_sha256;
+ new_parts.config_sha256 = std::move(parts.second.config_sha256);
+ new_parts.config = it->second;
+ if (!seen_part) {
+ new_file.config_sha256 = new_parts.config_sha256;
+ new_file.config = new_parts.config;
+ config_sha256 = new_file.config_sha256;
} else {
- copied_config_sha256.emplace(config_copy_sha256, config);
- new_file.config = config;
+ CHECK_EQ(config_sha256, new_file.config_sha256)
+ << ": Mismatched configs in " << new_file;
}
+ } else {
+ CHECK(config_sha256.empty())
+ << ": Part " << new_parts
+ << " is missing a sha256 but other parts have one.";
+ if (!seen_part) {
+ // We want to use a single Configuration flatbuffer for all the parts
+ // to make downstream easier. Since this is an old log, it doesn't
+ // have a SHA256 in the header to rely on, so we need a way to detect
+ // duplicates.
+ //
+ // SHA256 is decently fast, so use that as a representative hash of
+ // the header.
+ auto header =
+ std::make_shared<SizePrefixedFlatbufferVector<LogFileHeader>>(
+ std::move(*ReadHeader(new_parts.parts[0])));
+
+ // Do a recursive copy to normalize the flatbuffer. Different
+ // configurations can be built different ways, and can even have their
+ // vtable out of order. Don't think and just trigger a copy.
+ FlatbufferDetachedBuffer<Configuration> config_copy =
+ RecursiveCopyFlatBuffer(header->message().configuration());
+
+ std::string config_copy_sha256 = Sha256(config_copy.span());
+
+ auto it = copied_config_sha256.find(config_copy_sha256);
+ if (it != copied_config_sha256.end()) {
+ new_file.config = it->second;
+ } else {
+ std::shared_ptr<const Configuration> config(
+ header, header->message().configuration());
+
+ copied_config_sha256.emplace(std::move(config_copy_sha256), config);
+ new_file.config = config;
+ }
+ }
+ new_parts.config = new_file.config;
}
- new_parts.config = new_file.config;
seen_part = true;
new_file.parts.emplace_back(std::move(new_parts));
@@ -493,6 +577,9 @@
stream << " \"logger_boot_uuid\": \"" << file.logger_boot_uuid << "\",\n";
}
stream << " \"config\": " << file.config.get();
+ if (!file.config_sha256.empty()) {
+ stream << ",\n \"config_sha256\": \"" << file.config_sha256 << "\"";
+ }
stream << ",\n \"monotonic_start_time\": " << file.monotonic_start_time
<< ",\n \"realtime_start_time\": " << file.realtime_start_time
<< ",\n";
@@ -521,6 +608,9 @@
stream << " \"source_boot_uuid\": \"" << parts.source_boot_uuid << "\",\n";
}
stream << " \"config\": " << parts.config.get();
+ if (!parts.config_sha256.empty()) {
+ stream << ",\n \"config_sha256\": \"" << parts.config_sha256 << "\"";
+ }
stream << ",\n \"monotonic_start_time\": " << parts.monotonic_start_time
<< ",\n \"realtime_start_time\": " << parts.realtime_start_time
<< ",\n \"parts\": [";
diff --git a/aos/events/logging/logfile_sorting.h b/aos/events/logging/logfile_sorting.h
index c3d1dea..65bb1de 100644
--- a/aos/events/logging/logfile_sorting.h
+++ b/aos/events/logging/logfile_sorting.h
@@ -43,6 +43,7 @@
// Configuration for all the log parts. This will be a single object for all
// log files with the same config.
+ std::string config_sha256;
std::shared_ptr<const aos::Configuration> config;
};
@@ -70,8 +71,9 @@
// A list of parts which were corrupted and are unknown where they should go.
std::vector<std::string> corrupted;
- // Configuration for all the log parts and files. This is a single
+ // Configuration for all the log parts and files. This will be a single
// object for log files with the same config.
+ std::string config_sha256;
std::shared_ptr<const aos::Configuration> config;
};
diff --git a/aos/events/logging/logger.cc b/aos/events/logging/logger.cc
index d750caf..962a107 100644
--- a/aos/events/logging/logger.cc
+++ b/aos/events/logging/logger.cc
@@ -22,6 +22,7 @@
#include "aos/time/time.h"
#include "aos/util/file.h"
#include "flatbuffers/flatbuffers.h"
+#include "openssl/sha.h"
#include "third_party/gmp/gmpxx.h"
DEFINE_bool(skip_missing_forwarding_entries, false,
@@ -40,6 +41,19 @@
namespace aos {
namespace logger {
namespace {
+std::string Sha256(const absl::Span<const uint8_t> str) {
+ unsigned char hash[SHA256_DIGEST_LENGTH];
+ SHA256_CTX sha256;
+ SHA256_Init(&sha256);
+ SHA256_Update(&sha256, str.data(), str.size());
+ SHA256_Final(hash, &sha256);
+ std::stringstream ss;
+ for (int i = 0; i < SHA256_DIGEST_LENGTH; i++) {
+ ss << std::hex << std::setw(2) << std::setfill('0') << (int)hash[i];
+ }
+ return ss.str();
+}
+
std::string LogFileVectorToString(std::vector<LogFile> log_files) {
std::stringstream ss;
for (const auto f : log_files) {
@@ -274,6 +288,22 @@
std::string_view log_start_uuid) {
CHECK(!log_namer_) << ": Already logging";
log_namer_ = std::move(log_namer);
+
+ std::string config_sha256;
+ if (separate_config_) {
+ flatbuffers::FlatBufferBuilder fbb;
+ flatbuffers::Offset<aos::Configuration> configuration_offset =
+ CopyFlatBuffer(configuration_, &fbb);
+ LogFileHeader::Builder log_file_header_builder(fbb);
+ log_file_header_builder.add_configuration(configuration_offset);
+ fbb.FinishSizePrefixed(log_file_header_builder.Finish());
+ aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> config_header(
+ fbb.Release());
+ config_sha256 = Sha256(config_header.span());
+ LOG(INFO) << "Config sha256 of " << config_sha256;
+ log_namer_->WriteConfiguration(&config_header, config_sha256);
+ }
+
log_event_uuid_ = UUID::Random();
log_start_uuid_ = log_start_uuid;
VLOG(1) << "Starting logger for " << FlatbufferToJson(event_loop_->node());
@@ -303,7 +333,8 @@
for (const Node *node : log_namer_->nodes()) {
const int node_index = configuration::GetNodeIndex(configuration_, node);
- node_state_[node_index].log_file_header = MakeHeader(node);
+ node_state_[node_index].log_file_header =
+ MakeHeader(node, config_sha256);
}
// Grab data from each channel right before we declare the log file started
@@ -570,15 +601,17 @@
}
aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> Logger::MakeHeader(
- const Node *node) {
+ const Node *node, std::string_view config_sha256) {
// Now write the header with this timestamp in it.
flatbuffers::FlatBufferBuilder fbb;
fbb.ForceDefaults(true);
- // TODO(austin): Compress this much more efficiently. There are a bunch of
- // duplicated schemas.
- const flatbuffers::Offset<aos::Configuration> configuration_offset =
- CopyFlatBuffer(configuration_, &fbb);
+ flatbuffers::Offset<aos::Configuration> configuration_offset;
+ if (!separate_config_) {
+ configuration_offset = CopyFlatBuffer(configuration_, &fbb);
+ } else {
+ CHECK(!config_sha256.empty());
+ }
const flatbuffers::Offset<flatbuffers::String> name_offset =
fbb.CreateString(name_);
@@ -595,6 +628,11 @@
log_start_uuid_offset = fbb.CreateString(log_start_uuid_);
}
+ flatbuffers::Offset<flatbuffers::String> config_sha256_offset;
+ if (!config_sha256.empty()) {
+ config_sha256_offset = fbb.CreateString(config_sha256);
+ }
+
const flatbuffers::Offset<flatbuffers::String> logger_node_boot_uuid_offset =
fbb.CreateString(event_loop_->boot_uuid().string_view());
@@ -608,7 +646,6 @@
flatbuffers::Offset<Node> logger_node_offset;
if (configuration::MultiNode(configuration_)) {
- // TODO(austin): Reuse the node we just copied in above.
node_offset = RecursiveCopyFlatBuffer(node, &fbb);
logger_node_offset = RecursiveCopyFlatBuffer(event_loop_->node(), &fbb);
}
@@ -623,7 +660,9 @@
log_file_header_builder.add_logger_node(logger_node_offset);
}
- log_file_header_builder.add_configuration(configuration_offset);
+ if (!configuration_offset.IsNull()) {
+ log_file_header_builder.add_configuration(configuration_offset);
+ }
// The worst case theoretical out of order is the polling period times 2.
// One message could get logged right after the boundary, but be for right
// before the next boundary. And the reverse could happen for another
@@ -665,6 +704,12 @@
log_file_header_builder.add_parts_uuid(parts_uuid_offset);
log_file_header_builder.add_parts_index(0);
+ log_file_header_builder.add_configuration_sha256(0);
+
+ if (!config_sha256_offset.IsNull()) {
+ log_file_header_builder.add_configuration_sha256(config_sha256_offset);
+ }
+
fbb.FinishSizePrefixed(log_file_header_builder.Finish());
aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> result(
fbb.Release());
diff --git a/aos/events/logging/logger.fbs b/aos/events/logging/logger.fbs
index fef68f4..33702e0 100644
--- a/aos/events/logging/logger.fbs
+++ b/aos/events/logging/logger.fbs
@@ -27,8 +27,13 @@
// find a message out of order.
max_out_of_order_duration:long (id: 2);
- // The configuration of the channels.
+ // The configuration of the channels. It is valid to have a log file with
+ // just this filled out. That is a config only file which will be pointed to
+ // by files using configuration_sha256 and optionally configuration_path.
configuration:aos.Configuration (id: 3);
+ // sha256 of the configuration used. If this is set, configuration will not
+ // be set.
+ configuration_sha256:string (id: 16);
// Name of the device which this log file is for.
name:string (id: 4);
diff --git a/aos/events/logging/logger.h b/aos/events/logging/logger.h
index 56b99e8..a117b5a 100644
--- a/aos/events/logging/logger.h
+++ b/aos/events/logging/logger.h
@@ -61,6 +61,10 @@
on_logged_period_ = std::move(on_logged_period);
}
+ void set_separate_config(bool separate_config) {
+ separate_config_ = separate_config;
+ }
+
// Sets the period between polling the data. Defaults to 100ms.
//
// Changing this while a set of files is being written may result in
@@ -232,7 +236,7 @@
void WriteHeader();
aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> MakeHeader(
- const Node *node);
+ const Node *node, std::string_view config_sha256);
// Writes the header for the provided node if enough information is valid.
void MaybeWriteHeader(int node_index);
@@ -314,6 +318,9 @@
// reserved in the builder to avoid reallocating.
size_t max_header_size_ = 0;
+ // If true, write the message header into a separate file.
+ bool separate_config_ = true;
+
// Fetcher for all the statistics from all the nodes.
aos::Fetcher<message_bridge::ServerStatistics> server_statistics_fetcher_;
diff --git a/aos/events/logging/logger_test.cc b/aos/events/logging/logger_test.cc
index 7313b8f..2b5d6ca 100644
--- a/aos/events/logging/logger_test.cc
+++ b/aos/events/logging/logger_test.cc
@@ -26,6 +26,9 @@
using aos::message_bridge::RemoteMessage;
using aos::testing::MessageCounter;
+constexpr std::string_view kConfigSha1(
+ "0000c81e444ac470b8d29fb864621ae93a0e294a7e90c0dc4840d0f0d40fd72e");
+
class LoggerTest : public ::testing::Test {
public:
LoggerTest()
@@ -70,6 +73,7 @@
event_loop_factory_.RunFor(chrono::milliseconds(95));
Logger logger(logger_event_loop.get());
+ logger.set_separate_config(false);
logger.set_polling_period(std::chrono::milliseconds(100));
logger.StartLoggingLocalNamerOnRun(base_name);
event_loop_factory_.RunFor(chrono::milliseconds(20000));
@@ -162,6 +166,7 @@
event_loop_factory_.RunFor(chrono::milliseconds(95));
Logger logger(logger_event_loop.get());
+ logger.set_separate_config(false);
logger.set_polling_period(std::chrono::milliseconds(100));
logger_event_loop->OnRun([base_name, &logger_event_loop, &logger]() {
logger.StartLogging(std::make_unique<LocalLogNamer>(
@@ -193,6 +198,7 @@
event_loop_factory_.RunFor(chrono::milliseconds(95));
Logger logger(logger_event_loop.get());
+ logger.set_separate_config(false);
logger.set_polling_period(std::chrono::milliseconds(100));
logger.StartLogging(
std::make_unique<LocalLogNamer>(base_name1, logger_event_loop->node()));
@@ -256,6 +262,7 @@
event_loop_factory_.RunFor(chrono::milliseconds(95));
Logger logger(logger_event_loop.get());
+ logger.set_separate_config(false);
logger.set_polling_period(std::chrono::milliseconds(100));
logger.StartLoggingLocalNamerOnRun(base_name);
event_loop_factory_.RunFor(chrono::milliseconds(10000));
@@ -354,6 +361,7 @@
});
Logger logger(logger_event_loop.get());
+ logger.set_separate_config(false);
logger.set_polling_period(std::chrono::milliseconds(100));
logger.StartLoggingLocalNamerOnRun(base_name);
@@ -361,28 +369,30 @@
}
}
-std::vector<std::string> MakeLogFiles(std::string logfile_base) {
+std::vector<std::string> MakeLogFiles(std::string logfile_base1, std::string logfile_base2) {
return std::vector<std::string>(
- {logfile_base + "_pi1_data.part0.bfbs",
- logfile_base + "_pi2_data/test/aos.examples.Pong.part0.bfbs",
- logfile_base + "_pi2_data/test/aos.examples.Pong.part1.bfbs",
- logfile_base + "_pi2_data.part0.bfbs",
- logfile_base + "_timestamps/pi1/aos/remote_timestamps/pi2/"
- "aos.message_bridge.RemoteMessage.part0.bfbs",
- logfile_base + "_timestamps/pi1/aos/remote_timestamps/pi2/"
- "aos.message_bridge.RemoteMessage.part1.bfbs",
- logfile_base + "_timestamps/pi2/aos/remote_timestamps/pi1/"
- "aos.message_bridge.RemoteMessage.part0.bfbs",
- logfile_base + "_timestamps/pi2/aos/remote_timestamps/pi1/"
- "aos.message_bridge.RemoteMessage.part1.bfbs",
- logfile_base +
+ {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_base +
+ logfile_base2 +
"_pi1_data/pi1/aos/aos.message_bridge.Timestamp.part1.bfbs",
- logfile_base +
+ logfile_base1 +
"_pi2_data/pi2/aos/aos.message_bridge.Timestamp.part0.bfbs",
- logfile_base +
- "_pi2_data/pi2/aos/aos.message_bridge.Timestamp.part1.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")});
}
class MultinodeLoggerTest : public ::testing::Test {
@@ -396,32 +406,35 @@
pi2_(
configuration::GetNode(event_loop_factory_.configuration(), "pi2")),
tmp_dir_(aos::testing::TestTmpDir()),
- logfile_base_(tmp_dir_ + "/multi_logfile"),
+ logfile_base1_(tmp_dir_ + "/multi_logfile1"),
+ logfile_base2_(tmp_dir_ + "/multi_logfile2"),
pi1_reboot_logfiles_(
- {logfile_base_ + "_pi1_data.part0.bfbs",
- logfile_base_ + "_pi2_data/test/aos.examples.Pong.part0.bfbs",
- logfile_base_ + "_pi2_data/test/aos.examples.Pong.part1.bfbs",
- logfile_base_ + "_pi2_data/test/aos.examples.Pong.part2.bfbs",
- logfile_base_ + "_timestamps/pi1/aos/remote_timestamps/pi2/"
- "aos.message_bridge.RemoteMessage.part0.bfbs",
- logfile_base_ + "_timestamps/pi1/aos/remote_timestamps/pi2/"
- "aos.message_bridge.RemoteMessage.part1.bfbs",
- logfile_base_ + "_timestamps/pi1/aos/remote_timestamps/pi2/"
- "aos.message_bridge.RemoteMessage.part2.bfbs",
- logfile_base_ +
+ {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_base_ +
+ logfile_base1_ +
"_pi2_data/pi2/aos/aos.message_bridge.Timestamp.part1.bfbs",
- logfile_base_ +
- "_pi2_data/pi2/aos/aos.message_bridge.Timestamp.part2.bfbs"}),
- logfiles_(MakeLogFiles(logfile_base_)),
+ logfile_base1_ +
+ "_pi2_data/pi2/aos/aos.message_bridge.Timestamp.part2.bfbs",
+ absl::StrCat(logfile_base1_, "_", kConfigSha1, ".bfbs")}),
+ logfiles_(MakeLogFiles(logfile_base1_, logfile_base2_)),
pi1_single_direction_logfiles_(
- {logfile_base_ + "_pi1_data.part0.bfbs",
- logfile_base_ + "_pi2_data/test/aos.examples.Pong.part0.bfbs",
- logfile_base_ + "_timestamps/pi1/aos/remote_timestamps/pi2/"
- "aos.message_bridge.RemoteMessage.part0.bfbs",
- logfile_base_ +
- "_pi2_data/pi2/aos/aos.message_bridge.Timestamp.part0.bfbs"}),
+ {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]},
@@ -440,7 +453,7 @@
unlink((file + ".xz").c_str());
}
- for (const auto file : MakeLogFiles("relogged")) {
+ for (const auto file : MakeLogFiles("relogged1", "relogged2")) {
unlink(file.c_str());
}
@@ -468,7 +481,11 @@
void StartLogger(LoggerState *logger, std::string logfile_base = "",
bool compress = false) {
if (logfile_base.empty()) {
- logfile_base = logfile_base_;
+ if (logger->event_loop->node()->name()->string_view() == "pi1") {
+ logfile_base = logfile_base1_;
+ } else {
+ logfile_base = logfile_base2_;
+ }
}
logger->logger = std::make_unique<Logger>(logger->event_loop.get());
@@ -598,7 +615,8 @@
const Node *pi2_;
std::string tmp_dir_;
- std::string logfile_base_;
+ std::string logfile_base1_;
+ std::string logfile_base2_;
std::vector<std::string> pi1_reboot_logfiles_;
std::vector<std::string> logfiles_;
std::vector<std::string> pi1_single_direction_logfiles_;
@@ -614,11 +632,11 @@
// Counts the number of messages on a channel. Returns (channel name, channel
// type, count) for every message matching matcher()
std::vector<std::tuple<std::string, std::string, int>> CountChannelsMatching(
+ std::shared_ptr<const aos::Configuration> config,
std::string_view filename,
std::function<bool(const MessageHeader *)> matcher) {
MessageReader message_reader(filename);
- std::vector<int> counts(
- message_reader.log_file_header()->configuration()->channels()->size(), 0);
+ std::vector<int> counts(config->channels()->size(), 0);
while (true) {
std::optional<SizePrefixedFlatbufferVector<MessageHeader>> msg =
@@ -636,8 +654,7 @@
int channel = 0;
for (size_t i = 0; i < counts.size(); ++i) {
if (counts[i] != 0) {
- const Channel *channel =
- message_reader.log_file_header()->configuration()->channels()->Get(i);
+ const Channel *channel = config->channels()->Get(i);
result.push_back(std::make_tuple(channel->name()->str(),
channel->type()->str(), counts[i]));
}
@@ -649,8 +666,9 @@
// Counts the number of messages (channel, count) for all data messages.
std::vector<std::tuple<std::string, std::string, int>> CountChannelsData(
+ std::shared_ptr<const aos::Configuration> config,
std::string_view filename) {
- return CountChannelsMatching(filename, [](const MessageHeader *msg) {
+ return CountChannelsMatching(config, filename, [](const MessageHeader *msg) {
if (msg->has_data()) {
CHECK(!msg->has_monotonic_remote_time());
CHECK(!msg->has_realtime_remote_time());
@@ -663,8 +681,9 @@
// Counts the number of messages (channel, count) for all timestamp messages.
std::vector<std::tuple<std::string, std::string, int>> CountChannelsTimestamp(
+ std::shared_ptr<const aos::Configuration> config,
std::string_view filename) {
- return CountChannelsMatching(filename, [](const MessageHeader *msg) {
+ return CountChannelsMatching(config, filename, [](const MessageHeader *msg) {
if (!msg->has_data()) {
CHECK(msg->has_monotonic_remote_time());
CHECK(msg->has_realtime_remote_time());
@@ -697,8 +716,11 @@
std::vector<SizePrefixedFlatbufferVector<LogFileHeader>> log_header;
for (std::string_view f : logfiles_) {
log_header.emplace_back(ReadHeader(f).value());
- logfile_uuids.insert(log_header.back().message().log_event_uuid()->str());
- parts_uuids.insert(log_header.back().message().parts_uuid()->str());
+ if (!log_header.back().message().has_configuration()) {
+ logfile_uuids.insert(
+ log_header.back().message().log_event_uuid()->str());
+ parts_uuids.insert(log_header.back().message().parts_uuid()->str());
+ }
}
EXPECT_EQ(logfile_uuids.size(), 2u);
@@ -733,12 +755,15 @@
EXPECT_EQ(log_header[11].message().parts_index(), 1);
}
+ const std::vector<LogFile> sorted_log_files = SortParts(logfiles_);
{
using ::testing::UnorderedElementsAre;
+ std::shared_ptr<const aos::Configuration> config =
+ sorted_log_files[0].config;
// Timing reports, pings
EXPECT_THAT(
- CountChannelsData(logfiles_[0]),
+ CountChannelsData(config, logfiles_[0]),
UnorderedElementsAre(
std::make_tuple("/pi1/aos", "aos.message_bridge.Timestamp", 200),
std::make_tuple("/pi1/aos", "aos.timing.Report", 40),
@@ -746,7 +771,7 @@
<< " : " << logfiles_[0];
// Timestamps for pong
EXPECT_THAT(
- CountChannelsTimestamp(logfiles_[0]),
+ CountChannelsTimestamp(config, logfiles_[0]),
UnorderedElementsAre(
std::make_tuple("/test", "aos.examples.Pong", 2001),
std::make_tuple("/pi2/aos", "aos.message_bridge.Timestamp", 200)))
@@ -754,23 +779,23 @@
// Pong data.
EXPECT_THAT(
- CountChannelsData(logfiles_[1]),
+ CountChannelsData(config, logfiles_[1]),
UnorderedElementsAre(std::make_tuple("/test", "aos.examples.Pong", 91)))
<< " : " << logfiles_[1];
- EXPECT_THAT(CountChannelsData(logfiles_[2]),
+ EXPECT_THAT(CountChannelsData(config, logfiles_[2]),
UnorderedElementsAre(
std::make_tuple("/test", "aos.examples.Pong", 1910)))
<< " : " << logfiles_[1];
// No timestamps
- EXPECT_THAT(CountChannelsTimestamp(logfiles_[1]), UnorderedElementsAre())
+ EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[1]), UnorderedElementsAre())
<< " : " << logfiles_[1];
- EXPECT_THAT(CountChannelsTimestamp(logfiles_[2]), UnorderedElementsAre())
+ EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[2]), UnorderedElementsAre())
<< " : " << logfiles_[2];
// Timing reports and pongs.
EXPECT_THAT(
- CountChannelsData(logfiles_[3]),
+ CountChannelsData(config, logfiles_[3]),
UnorderedElementsAre(
std::make_tuple("/pi2/aos", "aos.message_bridge.Timestamp", 200),
std::make_tuple("/pi2/aos", "aos.timing.Report", 40),
@@ -778,73 +803,73 @@
<< " : " << logfiles_[3];
// And ping timestamps.
EXPECT_THAT(
- CountChannelsTimestamp(logfiles_[3]),
+ 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(logfiles_[4]), UnorderedElementsAre())
+ EXPECT_THAT(CountChannelsData(config, logfiles_[4]), UnorderedElementsAre())
<< " : " << logfiles_[4];
- EXPECT_THAT(CountChannelsData(logfiles_[5]), UnorderedElementsAre())
+ EXPECT_THAT(CountChannelsData(config, logfiles_[5]), UnorderedElementsAre())
<< " : " << logfiles_[5];
- EXPECT_THAT(CountChannelsData(logfiles_[6]), UnorderedElementsAre())
+ EXPECT_THAT(CountChannelsData(config, logfiles_[6]), UnorderedElementsAre())
<< " : " << logfiles_[6];
- EXPECT_THAT(CountChannelsData(logfiles_[7]), UnorderedElementsAre())
+ EXPECT_THAT(CountChannelsData(config, logfiles_[7]), UnorderedElementsAre())
<< " : " << logfiles_[7];
EXPECT_THAT(
- CountChannelsTimestamp(logfiles_[4]),
+ 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];
EXPECT_THAT(
- CountChannelsTimestamp(logfiles_[5]),
+ CountChannelsTimestamp(config, logfiles_[5]),
UnorderedElementsAre(
std::make_tuple("/pi1/aos", "aos.message_bridge.Timestamp", 191),
std::make_tuple("/test", "aos.examples.Ping", 1910)))
<< " : " << logfiles_[5];
- EXPECT_THAT(CountChannelsTimestamp(logfiles_[6]),
+ EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[6]),
UnorderedElementsAre(std::make_tuple(
"/pi2/aos", "aos.message_bridge.Timestamp", 9)))
<< " : " << logfiles_[6];
- EXPECT_THAT(CountChannelsTimestamp(logfiles_[7]),
+ 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(logfiles_[8]), UnorderedElementsAre())
+ EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[8]), UnorderedElementsAre())
<< " : " << logfiles_[8];
- EXPECT_THAT(CountChannelsTimestamp(logfiles_[9]), UnorderedElementsAre())
+ EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[9]), UnorderedElementsAre())
<< " : " << logfiles_[9];
- EXPECT_THAT(CountChannelsTimestamp(logfiles_[10]), UnorderedElementsAre())
+ EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[10]), UnorderedElementsAre())
<< " : " << logfiles_[10];
- EXPECT_THAT(CountChannelsTimestamp(logfiles_[11]), UnorderedElementsAre())
+ EXPECT_THAT(CountChannelsTimestamp(config, logfiles_[11]), UnorderedElementsAre())
<< " : " << logfiles_[11];
- EXPECT_THAT(CountChannelsData(logfiles_[8]),
+ EXPECT_THAT(CountChannelsData(config, logfiles_[8]),
UnorderedElementsAre(std::make_tuple(
"/pi1/aos", "aos.message_bridge.Timestamp", 9)))
<< " : " << logfiles_[8];
- EXPECT_THAT(CountChannelsData(logfiles_[9]),
+ EXPECT_THAT(CountChannelsData(config, logfiles_[9]),
UnorderedElementsAre(std::make_tuple(
"/pi1/aos", "aos.message_bridge.Timestamp", 191)))
<< " : " << logfiles_[9];
- EXPECT_THAT(CountChannelsData(logfiles_[10]),
+ EXPECT_THAT(CountChannelsData(config, logfiles_[10]),
UnorderedElementsAre(std::make_tuple(
"/pi2/aos", "aos.message_bridge.Timestamp", 9)))
<< " : " << logfiles_[10];
- EXPECT_THAT(CountChannelsData(logfiles_[11]),
+ EXPECT_THAT(CountChannelsData(config, logfiles_[11]),
UnorderedElementsAre(std::make_tuple(
"/pi2/aos", "aos.message_bridge.Timestamp", 191)))
<< " : " << logfiles_[11];
}
- LogReader reader(SortParts(logfiles_));
+ LogReader reader(sorted_log_files);
SimulatedEventLoopFactory log_reader_factory(reader.configuration());
log_reader_factory.set_send_delay(chrono::microseconds(0));
@@ -1654,8 +1679,8 @@
configuration::GetNode(log_reader_factory.configuration(), pi2_),
&log_reader_factory);
- StartLogger(&pi1_logger, "relogged");
- StartLogger(&pi2_logger, "relogged");
+ StartLogger(&pi1_logger, "relogged1");
+ StartLogger(&pi2_logger, "relogged2");
log_reader_factory.Run();
}
diff --git a/aos/flatbuffers.h b/aos/flatbuffers.h
index b751a4e..aa1e279 100644
--- a/aos/flatbuffers.h
+++ b/aos/flatbuffers.h
@@ -416,6 +416,16 @@
return SizePrefixedFlatbufferDetachedBuffer<T>(fbb.Release());
}
+ flatbuffers::DetachedBuffer Release() {
+ flatbuffers::FlatBufferBuilder fbb;
+ fbb.ForceDefaults(true);
+ const auto end = fbb.EndTable(fbb.StartTable());
+ fbb.Finish(flatbuffers::Offset<flatbuffers::Table>(end));
+ flatbuffers::DetachedBuffer result = fbb.Release();
+ std::swap(result, buffer_);
+ return result;
+ }
+
// Returns references to the buffer, and the data.
absl::Span<uint8_t> span() override {
return absl::Span<uint8_t>(buffer_.data(), buffer_.size());