Move reboot and header responsibility into DataWriter
Now that we have a class wrapped around the raw flatbuffer writing to
disk, we can start to make it responsible for tracking and writing the
header. It can also track reboots. Each message that is to be written
now needs to be passed in with the corresponding boot UUID.
This removes a ton of code in LogWriter since it is simpler to do it in
DataWriter. We don't need to detect state and then trigger the right
behavior nearly as much because the bottom layers track it very easily
now and just work.
Side effects here of delaying writing a bunch of stuff is that we
shouldn't write headers with no body because we delay writing the header
until we know the body. This would potentially be a performance problem
with inline configuration, but with the separate configuration files,
this isn't a problem.
Change-Id: Iec91284fcceeb73e3b6c181aebd2fb7edc7bc1ac
Signed-off-by: Austin Schuh <austin.schuh@bluerivertech.com>
diff --git a/aos/events/logging/log_namer.cc b/aos/events/logging/log_namer.cc
index 7ff3bc4..ca7fc61 100644
--- a/aos/events/logging/log_namer.cc
+++ b/aos/events/logging/log_namer.cc
@@ -17,6 +17,76 @@
namespace aos {
namespace logger {
+NewDataWriter::NewDataWriter(LogNamer *log_namer, const Node *node,
+ std::function<void(NewDataWriter *)> reopen,
+ std::function<void(NewDataWriter *)> close)
+ : node_(node),
+ node_index_(configuration::GetNodeIndex(log_namer->configuration_, node)),
+ log_namer_(log_namer),
+ reopen_(std::move(reopen)),
+ close_(std::move(close)) {
+ reopen_(this);
+}
+
+NewDataWriter::~NewDataWriter() {
+ if (writer) {
+ Close();
+ }
+}
+
+void NewDataWriter::Rotate() {
+ ++parts_index_;
+ reopen_(this);
+ header_written_ = false;
+ QueueHeader(log_namer_->MakeHeader(node_index_, source_node_boot_uuid_,
+ parts_uuid(), parts_index_));
+}
+
+void NewDataWriter::Reboot() {
+ parts_uuid_ = UUID::Random();
+ ++parts_index_;
+ reopen_(this);
+ header_written_ = false;
+}
+
+void NewDataWriter::QueueSizedFlatbuffer(flatbuffers::FlatBufferBuilder *fbb,
+ const UUID &source_node_boot_uuid,
+ aos::monotonic_clock::time_point now) {
+ // TODO(austin): Handle remote nodes changing too, not just the source node.
+ if (source_node_boot_uuid_ != source_node_boot_uuid) {
+ if (header_written_) {
+ Reboot();
+ }
+
+ QueueHeader(log_namer_->MakeHeader(node_index_, source_node_boot_uuid,
+ parts_uuid(), parts_index_));
+ }
+ CHECK_EQ(source_node_boot_uuid_, source_node_boot_uuid);
+ CHECK(header_written_) << ": Attempting to write message before header to "
+ << writer->filename();
+ writer->QueueSizedFlatbuffer(fbb, now);
+}
+
+void NewDataWriter::QueueHeader(
+ aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> &&header) {
+ CHECK(!header_written_) << ": Attempting to write duplicate header to "
+ << writer->filename();
+ CHECK(header.message().has_source_node_boot_uuid());
+ source_node_boot_uuid_ =
+ UUID::FromString(header.message().source_node_boot_uuid());
+ // TODO(austin): This triggers a dummy allocation that we don't need as part
+ // of releasing. Can we skip it?
+ writer->QueueSizedFlatbuffer(header.Release());
+ header_written_ = true;
+}
+
+void NewDataWriter::Close() {
+ CHECK(writer);
+ close_(this);
+ writer.reset();
+ header_written_ = false;
+}
+
aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> LogNamer::MakeHeader(
size_t node_index, const UUID &source_node_boot_uuid,
const UUID &parts_uuid, int parts_index) const {
@@ -146,14 +216,6 @@
return result;
}
-void LocalLogNamer::WriteHeader(const Node *node) {
- CHECK_EQ(node, this->node());
- const size_t node_index = configuration::GetNodeIndex(configuration_, node);
- data_writer_.QueueHeader(
- MakeHeader(node_index, node_states_[node_index].source_node_boot_uuid,
- data_writer_.uuid(), data_writer_.part_number));
-}
-
NewDataWriter *LocalLogNamer::MakeWriter(const Channel *channel) {
CHECK(configuration::ChannelIsSendableOnNode(channel, node()))
<< ": " << configuration::CleanedChannelToString(channel);
@@ -162,12 +224,7 @@
void LocalLogNamer::Rotate(const Node *node) {
CHECK(node == this->node());
- const size_t node_index = configuration::GetNodeIndex(configuration_, node);
data_writer_.Rotate();
-
- data_writer_.QueueHeader(
- MakeHeader(node_index, node_states_[node_index].source_node_boot_uuid,
- data_writer_.uuid(), data_writer_.part_number));
}
void LocalLogNamer::WriteConfiguration(
@@ -181,11 +238,6 @@
writer->QueueSizedFlatbuffer(header->Release());
}
-void LocalLogNamer::Reboot(const Node * /*node*/
-) {
- LOG(FATAL) << "Can't reboot a single node.";
-}
-
NewDataWriter *LocalLogNamer::MakeTimestampWriter(const Channel *channel) {
CHECK(configuration::ChannelIsReadableOnNode(channel, node_))
<< ": Message is not delivered to this node.";
@@ -214,63 +266,16 @@
}
}
-void MultiNodeLogNamer::WriteHeader(const Node *node) {
- if (node == this->node()) {
- if (!data_writer_) {
- OpenDataWriter();
- }
-
- const size_t node_index = configuration::GetNodeIndex(configuration_, node);
- data_writer_->QueueHeader(
- MakeHeader(node_index, node_states_[node_index].source_node_boot_uuid,
- data_writer_->uuid(), data_writer_->part_number));
- } else {
- const size_t node_index = configuration::GetNodeIndex(configuration_, node);
- for (std::pair<const Channel *const, NewDataWriter> &data_writer :
- data_writers_) {
- if (node == data_writer.second.node) {
- data_writer.second.QueueHeader(MakeHeader(
- node_index, node_states_[node_index].source_node_boot_uuid,
- data_writer.second.uuid(), data_writer.second.part_number));
- }
- }
- }
-}
-
-void MultiNodeLogNamer::Rotate(const Node *node) { DoRotate(node, false); }
-
-void MultiNodeLogNamer::Reboot(const Node *node) { DoRotate(node, true); }
-
-void MultiNodeLogNamer::DoRotate(const Node *node, bool reboot) {
+void MultiNodeLogNamer::Rotate(const Node *node) {
if (node == this->node()) {
if (data_writer_) {
- if (reboot) {
- data_writer_->Reboot();
- } else {
- data_writer_->Rotate();
- }
- // TODO(austin): Move this logic down once we have a better ownership
- // model for the header.
-
- const size_t node_index =
- configuration::GetNodeIndex(configuration_, node);
- data_writer_->QueueHeader(
- MakeHeader(node_index, node_states_[node_index].source_node_boot_uuid,
- data_writer_->uuid(), data_writer_->part_number));
+ data_writer_->Rotate();
}
} else {
- const size_t node_index = configuration::GetNodeIndex(configuration_, node);
for (std::pair<const Channel *const, NewDataWriter> &data_writer :
data_writers_) {
- if (node == data_writer.second.node) {
- if (reboot) {
- data_writer.second.Reboot();
- } else {
- data_writer.second.Rotate();
- }
- data_writer.second.QueueHeader(MakeHeader(
- node_index, node_states_[node_index].source_node_boot_uuid,
- data_writer.second.uuid(), data_writer.second.part_number));
+ if (node == data_writer.second.node()) {
+ data_writer.second.Rotate();
}
}
}
@@ -336,15 +341,13 @@
nodes_.emplace_back(source_node);
}
- NewDataWriter data_writer(
- [this, channel](NewDataWriter *data_writer) {
- OpenWriter(channel, data_writer);
- },
- [this](NewDataWriter *data_writer) {
- CloseWriter(&data_writer->writer);
- });
- data_writer.node = source_node;
-
+ NewDataWriter data_writer(this, source_node,
+ [this, channel](NewDataWriter *data_writer) {
+ OpenWriter(channel, data_writer);
+ },
+ [this](NewDataWriter *data_writer) {
+ CloseWriter(&data_writer->writer);
+ });
return &(
data_writers_.emplace(channel, std::move(data_writer)).first->second);
}
@@ -362,15 +365,14 @@
nodes_.emplace_back(node);
}
- NewDataWriter data_writer(
- [this, channel](NewDataWriter *data_writer) {
- OpenForwardedTimestampWriter(channel, data_writer);
- },
- [this](NewDataWriter *data_writer) {
- CloseWriter(&data_writer->writer);
- });
- data_writer.node = node;
-
+ NewDataWriter data_writer(this, node,
+ [this, channel](NewDataWriter *data_writer) {
+ OpenForwardedTimestampWriter(channel,
+ data_writer);
+ },
+ [this](NewDataWriter *data_writer) {
+ CloseWriter(&data_writer->writer);
+ });
return &(
data_writers_.emplace(channel, std::move(data_writer)).first->second);
}
@@ -419,7 +421,7 @@
std::string filename =
absl::StrCat("timestamps", channel->name()->string_view(), "/",
channel->type()->string_view(), ".part",
- data_writer->part_number, ".bfbs", extension_);
+ data_writer->parts_index(), ".bfbs", extension_);
CreateBufferWriter(filename, &data_writer->writer);
}
@@ -428,19 +430,20 @@
const std::string filename = absl::StrCat(
CHECK_NOTNULL(channel->source_node())->string_view(), "_data",
channel->name()->string_view(), "/", channel->type()->string_view(),
- ".part", data_writer->part_number, ".bfbs", extension_);
+ ".part", data_writer->parts_index(), ".bfbs", extension_);
CreateBufferWriter(filename, &data_writer->writer);
}
void MultiNodeLogNamer::OpenDataWriter() {
if (!data_writer_) {
data_writer_ = std::make_unique<NewDataWriter>(
+ this, node_,
[this](NewDataWriter *writer) {
std::string name;
if (node() != nullptr) {
name = absl::StrCat(name, node()->name()->string_view(), "_");
}
- absl::StrAppend(&name, "data.part", writer->part_number, ".bfbs",
+ absl::StrAppend(&name, "data.part", writer->parts_index(), ".bfbs",
extension_);
CreateBufferWriter(name, &writer->writer);
},
diff --git a/aos/events/logging/log_namer.h b/aos/events/logging/log_namer.h
index 8349a09..2727e63 100644
--- a/aos/events/logging/log_namer.h
+++ b/aos/events/logging/log_namer.h
@@ -15,80 +15,75 @@
namespace aos {
namespace logger {
-// TODO(austin): Track if headers are written here much more carefully.
-//
+class LogNamer;
+
// TODO(austin): Rename this back to DataWriter once all other callers are of
// the old DataWriter.
+//
+// Class to manage writing data to log files. This lets us track which boot the
+// written header has in it, and if the header has been written or not.
class NewDataWriter {
public:
// Constructs a NewDataWriter.
+ // log_namer is the log namer which holds the config and any other data we
+ // need for our header.
+ // node is the node whom's prespective we are logging from.
// reopen is called whenever a file needs to be reopened.
// close is called to close that file and extract any statistics.
- NewDataWriter(std::function<void(NewDataWriter *)> reopen,
- std::function<void(NewDataWriter *)> close)
- : reopen_(std::move(reopen)), close_(std::move(close)) {
- reopen_(this);
- }
+ NewDataWriter(LogNamer *log_namer, const Node *node,
+ std::function<void(NewDataWriter *)> reopen,
+ std::function<void(NewDataWriter *)> close);
NewDataWriter(NewDataWriter &&other) = default;
aos::logger::NewDataWriter &operator=(NewDataWriter &&other) = default;
NewDataWriter(const NewDataWriter &) = delete;
void operator=(const NewDataWriter &) = delete;
- ~NewDataWriter() {
- if (writer) {
- Close();
- }
- }
+ ~NewDataWriter();
- void Rotate() {
- ++part_number;
- reopen_(this);
- }
+ // Rotates the log file, delaying writing the new header until data arrives.
+ void Rotate();
// TODO(austin): Copy header and add all UUIDs and such when available
// whenever data is written.
//
- // TODO(austin): Automatically write the header and update on boot UUID
- // change.
- //
// TODO(austin): Add known timestamps for each node every time we cycle a log
// for sorting.
+ // Queues up a message with the provided boot UUID.
void QueueSizedFlatbuffer(flatbuffers::FlatBufferBuilder *fbb,
- aos::monotonic_clock::time_point now) {
- writer->QueueSizedFlatbuffer(fbb, now);
- }
+ const UUID &source_node_boot_uuid,
+ aos::monotonic_clock::time_point now);
- void QueueHeader(
- aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> &&header) {
- // TODO(austin): This triggers a dummy allocation that we don't need as part
- // of releasing. Can we skip it?
- writer->QueueSizedFlatbuffer(header.Release());
- }
-
+ // Returns the filename of the writer.
std::string_view filename() const { return writer->filename(); }
- void Reboot() {
- uuid_ = UUID::Random();
- Rotate();
- }
+ // Signals that a node has rebooted.
+ void Reboot();
- void Close() {
- CHECK(writer);
- close_(this);
- writer.reset();
- }
+ void Close();
std::unique_ptr<DetachedBufferWriter> writer = nullptr;
- const Node *node = nullptr;
- size_t part_number = 0;
- const UUID &uuid() const { return uuid_; }
+
+ size_t node_index() const { return node_index_; }
+ const UUID &parts_uuid() const { return parts_uuid_; }
+ size_t parts_index() const { return parts_index_; }
+ const Node *node() const { return node_; }
private:
- UUID uuid_ = UUID::Random();
+ void QueueHeader(
+ aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> &&header);
+
+ const Node *const node_ = nullptr;
+ size_t node_index_ = 0;
+ LogNamer *log_namer_;
+ UUID parts_uuid_ = UUID::Random();
+ size_t parts_index_ = 0;
+
std::function<void(NewDataWriter *)> reopen_;
std::function<void(NewDataWriter *)> close_;
+ bool header_written_ = false;
+ UUID source_node_boot_uuid_ = UUID::Zero();
};
// Interface describing how to name, track, and add headers to log file parts.
@@ -112,14 +107,6 @@
// Only renaming the folder is supported, not the file base name.
virtual void set_base_name(std::string_view base_name) = 0;
- // Writes the header to all log files for a specific node. This function
- // needs to be called after all the writers are created.
- //
- // Modifies header to contain the uuid and part number for each writer as it
- // writes it. Since this is done unconditionally, it does not restore the
- // previous value at the end.
- virtual void WriteHeader(const Node *node) = 0;
-
// Returns a writer for writing data from messages on this channel (on the
// primary node).
//
@@ -146,9 +133,6 @@
// Rotates all log files for the provided node.
virtual void Rotate(const Node *node) = 0;
- // Reboots all log files for the provided node. Resets any parts UUIDs.
- virtual void Reboot(const Node *node) = 0;
-
// Returns all the nodes that data is being written for.
const std::vector<const Node *> &nodes() const { return nodes_; }
@@ -176,24 +160,15 @@
logger_monotonic_start_time;
node_states_[node_index].logger_realtime_start_time =
logger_realtime_start_time;
+
// TODO(austin): Track that the header has changed and needs to be
- // rewritten.
+ // rewritten down here rather than up in log_writer.
}
monotonic_clock::time_point monotonic_start_time(size_t node_index) const {
return node_states_[node_index].monotonic_start_time;
}
- // TODO(austin): I need to move header writing fully inside NewDataWriter and
- // delete this method.
- bool SetBootUUID(size_t node_index, const UUID &uuid) {
- if (node_states_[node_index].source_node_boot_uuid != uuid) {
- node_states_[node_index].source_node_boot_uuid = uuid;
- return true;
- }
- return false;
- }
-
protected:
// Creates a new header by copying fields out of the template and combining
// them with the arguments provided.
@@ -205,6 +180,8 @@
const Node *const node_;
std::vector<const Node *> nodes_;
+ friend NewDataWriter;
+
// Structure with state per node about times and such.
// TODO(austin): some of this lives better in NewDataWriter once we move
// ownership of deciding when to write headers into LogNamer.
@@ -219,8 +196,6 @@
monotonic_clock::min_time;
realtime_clock::time_point logger_realtime_start_time =
realtime_clock::min_time;
-
- UUID source_node_boot_uuid = UUID::Zero();
};
std::vector<NodeState> node_states_;
@@ -237,14 +212,14 @@
const Node *node)
: LogNamer(configuration, node),
base_name_(base_name),
- data_writer_(
- [this](NewDataWriter *writer) {
- writer->writer = std::make_unique<DetachedBufferWriter>(
- absl::StrCat(base_name_, ".part", writer->part_number,
- ".bfbs"),
- std::make_unique<aos::logger::DummyEncoder>());
- },
- [](NewDataWriter * /*writer*/) {}) {}
+ data_writer_(this, node,
+ [this](NewDataWriter *writer) {
+ writer->writer = std::make_unique<DetachedBufferWriter>(
+ absl::StrCat(base_name_, ".part",
+ writer->parts_index(), ".bfbs"),
+ std::make_unique<aos::logger::DummyEncoder>());
+ },
+ [](NewDataWriter * /*writer*/) {}) {}
LocalLogNamer(const LocalLogNamer &) = delete;
LocalLogNamer(LocalLogNamer &&) = delete;
@@ -259,14 +234,10 @@
base_name_ = base_name;
}
- void WriteHeader(const Node *node) override;
-
NewDataWriter *MakeWriter(const Channel *channel) override;
void Rotate(const Node *node) override;
- void Reboot(const Node *node) override;
-
NewDataWriter *MakeTimestampWriter(const Channel *channel) override;
NewDataWriter *MakeForwardedTimestampWriter(const Channel * /*channel*/,
@@ -327,12 +298,8 @@
return all_filenames_;
}
- void WriteHeader(const Node *node) override;
-
void Rotate(const Node *node) override;
- void Reboot(const Node *node) override;
-
void WriteConfiguration(
aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> *header,
std::string_view config_sha256) override;
@@ -439,10 +406,6 @@
void ResetStatistics();
private:
- // Implements Rotate and Reboot, controlled by the 'reboot' flag. The only
- // difference between the two is if DataWriter::uuid is reset or not.
- void DoRotate(const Node *node, bool reboot);
-
// Opens up a writer for timestamps forwarded back.
void OpenForwardedTimestampWriter(const Channel *channel,
NewDataWriter *data_writer);
diff --git a/aos/events/logging/log_writer.cc b/aos/events/logging/log_writer.cc
index 71c5750..580a126 100644
--- a/aos/events/logging/log_writer.cc
+++ b/aos/events/logging/log_writer.cc
@@ -254,11 +254,6 @@
}
}
- CHECK(node_state_.empty());
- node_state_.resize(configuration::MultiNode(configuration_)
- ? configuration_->nodes()->size()
- : 1u);
-
log_namer_->SetHeaderTemplate(MakeHeader(config_sha256));
const aos::monotonic_clock::time_point beginning_time =
@@ -278,7 +273,7 @@
}
// Clear out any old timestamps in case we are re-starting logging.
- for (size_t i = 0; i < node_state_.size(); ++i) {
+ for (size_t i = 0; i < configuration::NodesCount(configuration_); ++i) {
log_namer_->SetStartTimes(
i, monotonic_clock::min_time, realtime_clock::min_time,
monotonic_clock::min_time, realtime_clock::min_time);
@@ -330,7 +325,6 @@
f.timestamp_writer = nullptr;
f.contents_writer = nullptr;
}
- node_state_.clear();
log_event_uuid_ = UUID::Zero();
log_start_uuid_ = std::nullopt;
@@ -358,54 +352,9 @@
const int node_index = configuration::GetNodeIndex(configuration_, node);
MaybeUpdateTimestamp(node, node_index, monotonic_start_time,
realtime_start_time);
- MaybeWriteHeader(node_index, node);
}
}
-void Logger::MaybeWriteHeader(int node_index) {
- if (configuration::MultiNode(configuration_)) {
- return MaybeWriteHeader(node_index,
- configuration_->nodes()->Get(node_index));
- } else {
- return MaybeWriteHeader(node_index, nullptr);
- }
-}
-
-void Logger::MaybeWriteHeader(int node_index, const Node *node) {
- // This function is responsible for writing the header when the header both
- // has valid data, and when it needs to be written.
- if (node_state_[node_index].header_written &&
- node_state_[node_index].header_valid) {
- // The header has been written and is valid, nothing to do.
- return;
- }
- if (!node_state_[node_index].has_source_node_boot_uuid) {
- // Can't write a header if we don't have the boot UUID.
- return;
- }
-
- // WriteHeader writes the first header in a log file. We want to do this only
- // once.
- //
- // Rotate rewrites the same header with a new part ID, but keeps the same part
- // UUID. We don't want that when things reboot, because that implies that
- // parts go together across a reboot.
- //
- // Reboot resets the parts UUID. So, once we've written a header the first
- // time, we want to use Reboot to rotate the log and reset the parts UUID.
- //
- // header_valid is cleared whenever the remote reboots.
- if (node_state_[node_index].header_written) {
- VLOG(1) << "Rebooting";
- log_namer_->Reboot(node);
- } else {
- log_namer_->WriteHeader(node);
-
- node_state_[node_index].header_written = true;
- }
- node_state_[node_index].header_valid = true;
-}
-
void Logger::WriteMissingTimestamps() {
if (configuration::MultiNode(configuration_)) {
server_statistics_fetcher_.Fetch();
@@ -423,12 +372,8 @@
node, node_index,
server_statistics_fetcher_.context().monotonic_event_time,
server_statistics_fetcher_.context().realtime_event_time)) {
- CHECK(node_state_[node_index].header_written);
- CHECK(node_state_[node_index].header_valid);
VLOG(1) << "Rotating because timestamps changed";
log_namer_->Rotate(node);
- } else {
- MaybeWriteHeader(node_index, node);
}
}
}
@@ -448,9 +393,6 @@
log_namer_->SetStartTimes(node_index, monotonic_start_time,
realtime_start_time, monotonic_start_time,
realtime_start_time);
- log_namer_->SetBootUUID(node_index, event_loop_->boot_uuid());
- node_state_[node_index].header_valid = false;
- node_state_[node_index].has_source_node_boot_uuid = true;
return true;
} else if (server_statistics_fetcher_.get() != nullptr) {
// We must be a remote node now. Look for the connection and see if it is
@@ -623,18 +565,10 @@
break;
}
if (f.writer != nullptr) {
- // Only check if the boot UUID has changed if this is data from another
- // node. Our UUID can't change without restarting the application.
- if (our_node_index != f.data_node_index) {
- // And update our boot UUID if the UUID has changed.
- if (log_namer_->SetBootUUID(f.data_node_index,
- f.fetcher->context().source_boot_uuid)) {
- node_state_[f.data_node_index].header_valid = false;
- node_state_[f.data_node_index].has_source_node_boot_uuid = true;
- MaybeWriteHeader(f.data_node_index);
- }
- }
-
+ const UUID source_node_boot_uuid =
+ our_node_index != f.data_node_index
+ ? f.fetcher->context().source_boot_uuid
+ : event_loop_->boot_uuid();
// Write!
const auto start = event_loop_->monotonic_now();
flatbuffers::FlatBufferBuilder fbb(f.fetcher->context().size +
@@ -656,10 +590,7 @@
max_header_size_ = std::max(max_header_size_,
fbb.GetSize() - f.fetcher->context().size);
- CHECK(node_state_[f.data_node_index].header_valid)
- << ": Can't write data before the header on channel "
- << configuration::CleanedChannelToString(f.fetcher->channel());
- f.writer->QueueSizedFlatbuffer(&fbb, end);
+ f.writer->QueueSizedFlatbuffer(&fbb, source_node_boot_uuid, end);
}
if (f.timestamp_writer != nullptr) {
@@ -682,10 +613,11 @@
flatbuffers::GetSizePrefixedRoot<MessageHeader>(
fbb.GetBufferPointer()));
- CHECK(node_state_[f.timestamp_node_index].header_valid)
- << ": Can't write data before the header on channel "
- << configuration::CleanedChannelToString(f.fetcher->channel());
- f.timestamp_writer->QueueSizedFlatbuffer(&fbb, end);
+ // TODO(austin): How do I track remote timestamp boot UUIDs? I need to
+ // update the uuid list in the header when one changes and track
+ // timestamps.
+ f.timestamp_writer->QueueSizedFlatbuffer(&fbb, event_loop_->boot_uuid(),
+ end);
}
if (f.contents_writer != nullptr) {
@@ -701,12 +633,6 @@
flatbuffers::GetRoot<RemoteMessage>(f.fetcher->context().data);
CHECK(msg->has_boot_uuid()) << ": " << aos::FlatbufferToJson(msg);
- if (log_namer_->SetBootUUID(f.contents_node_index,
- UUID::FromVector(msg->boot_uuid()))) {
- node_state_[f.contents_node_index].header_valid = false;
- node_state_[f.contents_node_index].has_source_node_boot_uuid = true;
- MaybeWriteHeader(f.contents_node_index);
- }
logger::MessageHeader::Builder message_header_builder(fbb);
@@ -745,10 +671,8 @@
const auto end = event_loop_->monotonic_now();
RecordCreateMessageTime(start, end, &f);
- CHECK(node_state_[f.contents_node_index].header_valid)
- << ": Can't write data before the header on channel "
- << configuration::CleanedChannelToString(f.fetcher->channel());
- f.contents_writer->QueueSizedFlatbuffer(&fbb, end);
+ f.contents_writer->QueueSizedFlatbuffer(
+ &fbb, UUID::FromVector(msg->boot_uuid()), end);
}
f.written = true;
diff --git a/aos/events/logging/log_writer.h b/aos/events/logging/log_writer.h
index fa76b15..a93da7d 100644
--- a/aos/events/logging/log_writer.h
+++ b/aos/events/logging/log_writer.h
@@ -202,28 +202,12 @@
// channel index.
std::vector<int> event_loop_to_logged_channel_index_;
- struct NodeState {
- // Tracks if LogNamer has a source boot UUID set or not.
- bool has_source_node_boot_uuid = false;
-
- // True if a header has been written to the start of a log file.
- bool header_written = false;
- // True if the current written header represents the contents which will
- // follow. This is cleared when boot_uuid is known to not match anymore.
- bool header_valid = false;
- };
-
void WriteHeader();
// Makes a template header for all the follower nodes.
aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> MakeHeader(
std::string_view config_sha256);
- // Writes the header for the provided node if enough information is valid.
- void MaybeWriteHeader(int node_index);
- // Overload for when we already know node as well.
- void MaybeWriteHeader(int node_index, const Node *node);
-
bool MaybeUpdateTimestamp(
const Node *node, int node_index,
aos::monotonic_clock::time_point monotonic_start_time,
@@ -300,8 +284,6 @@
// Fetcher for all the statistics from all the nodes.
aos::Fetcher<message_bridge::ServerStatistics> server_statistics_fetcher_;
-
- std::vector<NodeState> node_state_;
};
} // namespace logger