| #include "aos/events/logging/log_namer.h" |
| |
| #include <functional> |
| #include <map> |
| #include <memory> |
| #include <string_view> |
| #include <vector> |
| |
| #include "absl/strings/str_cat.h" |
| #include "flatbuffers/flatbuffers.h" |
| #include "glog/logging.h" |
| |
| #include "aos/containers/error_list.h" |
| #include "aos/containers/sized_array.h" |
| #include "aos/events/logging/logfile_utils.h" |
| #include "aos/events/logging/logger_generated.h" |
| #include "aos/flatbuffer_merge.h" |
| #include "aos/uuid.h" |
| |
| DECLARE_int32(flush_size); |
| |
| namespace aos::logger { |
| |
| NewDataWriter::NewDataWriter(LogNamer *log_namer, const Node *node, |
| const Node *logger_node, |
| std::function<void(NewDataWriter *)> reopen, |
| std::function<void(NewDataWriter *)> close, |
| size_t max_message_size, |
| std::initializer_list<StoredDataType> types) |
| : node_(node), |
| node_index_(configuration::GetNodeIndex(log_namer->configuration_, node)), |
| logger_node_index_( |
| configuration::GetNodeIndex(log_namer->configuration_, logger_node)), |
| log_namer_(log_namer), |
| reopen_(std::move(reopen)), |
| close_(std::move(close)), |
| max_message_size_(max_message_size), |
| max_out_of_order_duration_(log_namer_->base_max_out_of_order_duration()) { |
| allowed_data_types_.fill(false); |
| |
| state_.resize(configuration::NodesCount(log_namer->configuration_)); |
| CHECK_LT(node_index_, state_.size()); |
| for (StoredDataType type : types) { |
| CHECK_LT(static_cast<size_t>(type), allowed_data_types_.size()); |
| allowed_data_types_[static_cast<size_t>(type)] = true; |
| } |
| } |
| |
| NewDataWriter::~NewDataWriter() { |
| if (writer) { |
| Close(); |
| } |
| } |
| |
| void NewDataWriter::Rotate() { |
| // No need to rotate if nothing has been written. |
| if (header_written_) { |
| VLOG(1) << "Rotated " << name(); |
| ++parts_index_; |
| |
| aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> header = |
| MakeHeader(); |
| |
| if (header.span().size() > max_message_size_) { |
| max_message_size_ = header.span().size(); |
| } |
| |
| reopen_(this); |
| header_written_ = false; |
| QueueHeader(std::move(header)); |
| } |
| } |
| |
| void NewDataWriter::Reboot(const UUID &source_node_boot_uuid) { |
| parts_uuid_ = UUID::Random(); |
| ++parts_index_; |
| reopen_(this); |
| header_written_ = false; |
| for (State &state : state_) { |
| state.boot_uuid = UUID::Zero(); |
| state.oldest_remote_monotonic_timestamp = monotonic_clock::max_time; |
| state.oldest_local_monotonic_timestamp = monotonic_clock::max_time; |
| state.oldest_remote_unreliable_monotonic_timestamp = |
| monotonic_clock::max_time; |
| state.oldest_local_unreliable_monotonic_timestamp = |
| monotonic_clock::max_time; |
| state.oldest_remote_reliable_monotonic_timestamp = |
| monotonic_clock::max_time; |
| state.oldest_local_reliable_monotonic_timestamp = monotonic_clock::max_time; |
| state.oldest_logger_remote_unreliable_monotonic_timestamp = |
| monotonic_clock::max_time; |
| state.oldest_logger_local_unreliable_monotonic_timestamp = |
| monotonic_clock::max_time; |
| } |
| |
| state_[node_index_].boot_uuid = source_node_boot_uuid; |
| |
| VLOG(1) << "Rebooted " << name(); |
| newest_message_time_ = monotonic_clock::min_time; |
| // When a node reboots, parts_uuid changes but the same writer continues to |
| // write the data, so we can reset the max out of order duration. If we don't |
| // do this, the max out of order duration can grow to an unreasonable value. |
| max_out_of_order_duration_ = log_namer_->base_max_out_of_order_duration(); |
| } |
| |
| void NewDataWriter::UpdateBoot(const UUID &source_node_boot_uuid) { |
| if (state_[node_index_].boot_uuid != source_node_boot_uuid) { |
| state_[node_index_].boot_uuid = source_node_boot_uuid; |
| if (header_written_) { |
| Reboot(source_node_boot_uuid); |
| } |
| } |
| } |
| |
| void NewDataWriter::UpdateRemote( |
| const size_t remote_node_index, const UUID &remote_node_boot_uuid, |
| const monotonic_clock::time_point monotonic_remote_time, |
| const monotonic_clock::time_point monotonic_event_time, const bool reliable, |
| monotonic_clock::time_point monotonic_timestamp_time) { |
| // Trigger rotation if anything in the header changes. |
| bool rotate = false; |
| CHECK_LT(remote_node_index, state_.size()); |
| State &state = state_[remote_node_index]; |
| |
| // Did the remote boot UUID change? |
| if (state.boot_uuid != remote_node_boot_uuid) { |
| VLOG(1) << name() << " Remote " << remote_node_index << " updated to " |
| << remote_node_boot_uuid << " from " << state.boot_uuid; |
| state.boot_uuid = remote_node_boot_uuid; |
| state.oldest_remote_monotonic_timestamp = monotonic_clock::max_time; |
| state.oldest_local_monotonic_timestamp = monotonic_clock::max_time; |
| state.oldest_remote_unreliable_monotonic_timestamp = |
| monotonic_clock::max_time; |
| state.oldest_local_unreliable_monotonic_timestamp = |
| monotonic_clock::max_time; |
| state.oldest_remote_reliable_monotonic_timestamp = |
| monotonic_clock::max_time; |
| state.oldest_local_reliable_monotonic_timestamp = monotonic_clock::max_time; |
| state.oldest_logger_remote_unreliable_monotonic_timestamp = |
| monotonic_clock::max_time; |
| state.oldest_logger_local_unreliable_monotonic_timestamp = |
| monotonic_clock::max_time; |
| rotate = true; |
| } |
| |
| // Did the unreliable timestamps change? |
| if (!reliable) { |
| if (state.oldest_remote_unreliable_monotonic_timestamp > |
| monotonic_remote_time) { |
| VLOG(1) << name() << " Remote " << remote_node_index |
| << " oldest_remote_unreliable_monotonic_timestamp updated from " |
| << state.oldest_remote_unreliable_monotonic_timestamp << " to " |
| << monotonic_remote_time; |
| state.oldest_remote_unreliable_monotonic_timestamp = |
| monotonic_remote_time; |
| state.oldest_local_unreliable_monotonic_timestamp = monotonic_event_time; |
| rotate = true; |
| } |
| } else { |
| if (state.oldest_remote_reliable_monotonic_timestamp > |
| monotonic_remote_time) { |
| VLOG(1) << name() << " Remote " << remote_node_index |
| << " oldest_remote_reliable_monotonic_timestamp updated from " |
| << state.oldest_remote_reliable_monotonic_timestamp << " to " |
| << monotonic_remote_time; |
| state.oldest_remote_reliable_monotonic_timestamp = monotonic_remote_time; |
| state.oldest_local_reliable_monotonic_timestamp = monotonic_event_time; |
| rotate = true; |
| } |
| } |
| |
| // Track the logger timestamps too. |
| if (monotonic_timestamp_time != monotonic_clock::min_time) { |
| State &logger_state = state_[node_index_]; |
| CHECK_EQ(remote_node_index, logger_node_index_); |
| if (monotonic_event_time < |
| logger_state.oldest_logger_remote_unreliable_monotonic_timestamp) { |
| VLOG(1) |
| << name() << " Remote " << node_index_ |
| << " oldest_logger_remote_unreliable_monotonic_timestamp updated " |
| "from " |
| << logger_state.oldest_logger_remote_unreliable_monotonic_timestamp |
| << " to " << monotonic_event_time; |
| logger_state.oldest_logger_remote_unreliable_monotonic_timestamp = |
| monotonic_event_time; |
| logger_state.oldest_logger_local_unreliable_monotonic_timestamp = |
| monotonic_timestamp_time; |
| |
| rotate = true; |
| } |
| } |
| |
| // Did any of the timestamps change? |
| if (state.oldest_remote_monotonic_timestamp > monotonic_remote_time) { |
| VLOG(1) << name() << " Remote " << remote_node_index |
| << " oldest_remote_monotonic_timestamp updated from " |
| << state.oldest_remote_monotonic_timestamp << " to " |
| << monotonic_remote_time; |
| state.oldest_remote_monotonic_timestamp = monotonic_remote_time; |
| state.oldest_local_monotonic_timestamp = monotonic_event_time; |
| rotate = true; |
| } |
| |
| if (rotate) { |
| Rotate(); |
| } |
| } |
| |
| void NewDataWriter::CopyDataMessage( |
| DataEncoder::Copier *coppier, const UUID &source_node_boot_uuid, |
| aos::monotonic_clock::time_point now, |
| aos::monotonic_clock::time_point message_time) { |
| CHECK(allowed_data_types_[static_cast<size_t>(StoredDataType::DATA)]) |
| << ": Tried to write data on non-data writer."; |
| CopyMessage(coppier, source_node_boot_uuid, now, message_time); |
| } |
| |
| void NewDataWriter::CopyTimestampMessage( |
| DataEncoder::Copier *coppier, const UUID &source_node_boot_uuid, |
| aos::monotonic_clock::time_point now, |
| aos::monotonic_clock::time_point message_time) { |
| CHECK(allowed_data_types_[static_cast<size_t>(StoredDataType::TIMESTAMPS)]) |
| << ": Tried to write timestamps on non-timestamp writer."; |
| CopyMessage(coppier, source_node_boot_uuid, now, message_time); |
| } |
| |
| void NewDataWriter::CopyRemoteTimestampMessage( |
| DataEncoder::Copier *coppier, const UUID &source_node_boot_uuid, |
| aos::monotonic_clock::time_point now, |
| aos::monotonic_clock::time_point message_time) { |
| CHECK(allowed_data_types_[static_cast<size_t>( |
| StoredDataType::REMOTE_TIMESTAMPS)]) |
| << ": Tried to write remote timestamps on non-remote timestamp writer."; |
| CopyMessage(coppier, source_node_boot_uuid, now, message_time); |
| } |
| |
| void NewDataWriter::CopyMessage(DataEncoder::Copier *coppier, |
| const UUID &source_node_boot_uuid, |
| aos::monotonic_clock::time_point now, |
| aos::monotonic_clock::time_point message_time) { |
| // Trigger a reboot if we detect the boot UUID change. |
| UpdateBoot(source_node_boot_uuid); |
| |
| if (!header_written_) { |
| QueueHeader(MakeHeader()); |
| } |
| |
| bool max_out_of_order_duration_exceeded = false; |
| // Enforce max out of duration contract. |
| // |
| // Updates the newest message time. |
| // Rotate the part file if current message is more than |
| // max_out_of_order_duration behind the newest message we've logged so far. |
| if (message_time > newest_message_time_) { |
| newest_message_time_ = message_time; |
| } |
| |
| // Don't consider messages before start up when checking for max out of order |
| // duration. |
| monotonic_clock::time_point monotonic_start_time = |
| log_namer_->monotonic_start_time(node_index_, source_node_boot_uuid); |
| |
| if (std::chrono::nanoseconds((newest_message_time_ - |
| std::max(monotonic_start_time, message_time))) > |
| max_out_of_order_duration_) { |
| // If the new message is older than 2 * max_out_order_duration, doubling it |
| // won't be sufficient. |
| // |
| // Example: newest_message_time = 10, logged_message_time = 5, |
| // max_ooo_duration = 2 |
| // |
| // In this case actual max_ooo_duration = 10 - 5 = 5, but we double the |
| // existing max_ooo_duration we get 4 which is not sufficient. |
| // |
| // Take the max of the two values. |
| max_out_of_order_duration_ = |
| 2 * std::max(max_out_of_order_duration_, |
| std::chrono::nanoseconds( |
| (newest_message_time_ - message_time))); |
| max_out_of_order_duration_exceeded = true; |
| } |
| |
| // If the start time has changed for this node, trigger a rotation. |
| if ((monotonic_start_time != monotonic_start_time_) || |
| max_out_of_order_duration_exceeded) { |
| // If we just received a start time now, we will rotate parts shortly. Use a |
| // reasonable max out of order durationin the new header based on start time |
| // information available now. |
| if ((monotonic_start_time_ == monotonic_clock::min_time) && |
| (monotonic_start_time != monotonic_clock::min_time)) { |
| // If we're writing current messages but we receive an older start time, |
| // we can pick a reasonable max ooo duration number for the next part. |
| // |
| // For example - Our current max ooo duration is 0.3 seconds. We're |
| // writing messages at 20 seconds and recieve a start time of 1 second. We |
| // don't need max ooo duration to be (20 - 1) = 19 seconds although that |
| // would still work. |
| // |
| // Pick the minimum max out of duration value that satisifies the |
| // requirement but bound the minimum at the base value we started with. |
| max_out_of_order_duration_ = |
| std::max(log_namer_->base_max_out_of_order_duration(), |
| std::min(max_out_of_order_duration_, |
| std::chrono::nanoseconds(newest_message_time_ - |
| monotonic_start_time))); |
| } |
| CHECK(header_written_); |
| Rotate(); |
| } |
| |
| CHECK_EQ(log_namer_->monotonic_start_time(node_index_, source_node_boot_uuid), |
| monotonic_start_time_); |
| CHECK_EQ(state_[node_index_].boot_uuid, source_node_boot_uuid); |
| CHECK(writer); |
| CHECK(header_written_) << ": Attempting to write message before header to " |
| << writer->name(); |
| CHECK_LE(coppier->size(), max_message_size_); |
| writer->CopyMessage(coppier, now); |
| } |
| |
| aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> |
| NewDataWriter::MakeHeader() { |
| const size_t logger_node_index = log_namer_->logger_node_index(); |
| const UUID &logger_node_boot_uuid = log_namer_->logger_node_boot_uuid(); |
| if (state_[logger_node_index].boot_uuid == UUID::Zero()) { |
| VLOG(1) << name() << " Logger node is " << logger_node_index |
| << " and uuid is " << logger_node_boot_uuid; |
| state_[logger_node_index].boot_uuid = logger_node_boot_uuid; |
| } else { |
| CHECK_EQ(state_[logger_node_index].boot_uuid, logger_node_boot_uuid); |
| } |
| return log_namer_->MakeHeader(node_index_, state_, parts_uuid(), parts_index_, |
| max_out_of_order_duration_, |
| allowed_data_types_); |
| } |
| |
| void NewDataWriter::QueueHeader( |
| aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> &&header) { |
| CHECK(!header_written_) << ": Attempting to write duplicate header to " |
| << writer->name(); |
| CHECK(header.message().has_source_node_boot_uuid()); |
| CHECK_EQ(state_[node_index_].boot_uuid, |
| UUID::FromString(header.message().source_node_boot_uuid())); |
| if (!writer) { |
| // Since we haven't opened the first time, it's still not too late to update |
| // the max message size. Make sure the header fits. |
| // |
| // This won't work well on reboots, but the structure of the header is fixed |
| // by that point in time, so it's size is fixed too. |
| // |
| // Most of the time, the minimum buffer size inside the encoder of around |
| // 128k will make this a non-issue. |
| UpdateMaxMessageSize(header.span().size()); |
| |
| reopen_(this); |
| } |
| |
| VLOG(1) << "Writing to " << name() << " " |
| << aos::FlatbufferToJson( |
| header, {.multi_line = false, .max_vector_size = 100}); |
| |
| CHECK(writer); |
| DataEncoder::SpanCopier coppier(header.span()); |
| CHECK_LE(coppier.size(), max_message_size_); |
| writer->CopyMessage(&coppier, aos::monotonic_clock::now()); |
| header_written_ = true; |
| monotonic_start_time_ = log_namer_->monotonic_start_time( |
| node_index_, state_[node_index_].boot_uuid); |
| } |
| |
| void NewDataWriter::Close() { |
| CHECK(writer); |
| close_(this); |
| writer.reset(); |
| header_written_ = false; |
| } |
| |
| LogNamer::NodeState *LogNamer::GetNodeState(size_t node_index, |
| const UUID &boot_uuid) { |
| auto it = node_states_.find(std::make_pair(node_index, boot_uuid)); |
| if (it == node_states_.end()) { |
| it = |
| node_states_.emplace(std::make_pair(node_index, boot_uuid), NodeState()) |
| .first; |
| } |
| return &it->second; |
| } |
| |
| aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> LogNamer::MakeHeader( |
| size_t node_index, const std::vector<NewDataWriter::State> &state, |
| const UUID &parts_uuid, int parts_index, |
| std::chrono::nanoseconds max_out_of_order_duration, |
| const std::array<bool, static_cast<size_t>(StoredDataType::MAX) + 1> |
| &allowed_data_types) { |
| const UUID &source_node_boot_uuid = state[node_index].boot_uuid; |
| const Node *const source_node = |
| configuration::GetNode(configuration_, node_index); |
| CHECK_EQ(LogFileHeader::MiniReflectTypeTable()->num_elems, 35u); |
| flatbuffers::FlatBufferBuilder fbb; |
| fbb.ForceDefaults(true); |
| |
| flatbuffers::Offset<flatbuffers::String> config_sha256_offset; |
| flatbuffers::Offset<aos::Configuration> configuration_offset; |
| if (header_.message().has_configuration()) { |
| CHECK(!header_.message().has_configuration_sha256()); |
| configuration_offset = |
| CopyFlatBuffer(header_.message().configuration(), &fbb); |
| } else { |
| CHECK(!header_.message().has_configuration()); |
| CHECK(header_.message().has_configuration_sha256()); |
| config_sha256_offset = fbb.CreateString( |
| header_.message().configuration_sha256()->string_view()); |
| } |
| |
| CHECK(header_.message().has_name()); |
| const flatbuffers::Offset<flatbuffers::String> name_offset = |
| fbb.CreateString(header_.message().name()->string_view()); |
| const flatbuffers::Offset<flatbuffers::String> logger_sha1_offset = |
| header_.message().has_logger_sha1() |
| ? fbb.CreateString(header_.message().logger_sha1()->string_view()) |
| : 0; |
| const flatbuffers::Offset<flatbuffers::String> logger_version_offset = |
| header_.message().has_logger_version() |
| ? fbb.CreateString(header_.message().logger_version()->string_view()) |
| : 0; |
| |
| CHECK(header_.message().has_log_event_uuid()); |
| const flatbuffers::Offset<flatbuffers::String> log_event_uuid_offset = |
| fbb.CreateString(header_.message().log_event_uuid()->string_view()); |
| |
| CHECK(header_.message().has_logger_instance_uuid()); |
| const flatbuffers::Offset<flatbuffers::String> logger_instance_uuid_offset = |
| fbb.CreateString(header_.message().logger_instance_uuid()->string_view()); |
| |
| flatbuffers::Offset<flatbuffers::String> log_start_uuid_offset; |
| if (header_.message().has_log_start_uuid()) { |
| log_start_uuid_offset = |
| fbb.CreateString(header_.message().log_start_uuid()->string_view()); |
| } |
| |
| CHECK(header_.message().has_logger_node_boot_uuid()); |
| const flatbuffers::Offset<flatbuffers::String> logger_node_boot_uuid_offset = |
| fbb.CreateString( |
| header_.message().logger_node_boot_uuid()->string_view()); |
| |
| CHECK_NE(source_node_boot_uuid, UUID::Zero()); |
| const flatbuffers::Offset<flatbuffers::String> source_node_boot_uuid_offset = |
| source_node_boot_uuid.PackString(&fbb); |
| |
| const flatbuffers::Offset<flatbuffers::String> parts_uuid_offset = |
| parts_uuid.PackString(&fbb); |
| |
| flatbuffers::Offset<Node> node_offset; |
| flatbuffers::Offset<Node> logger_node_offset; |
| |
| if (configuration::MultiNode(configuration_)) { |
| node_offset = RecursiveCopyFlatBuffer(source_node, &fbb); |
| logger_node_offset = RecursiveCopyFlatBuffer(node_, &fbb); |
| } |
| |
| std::vector<flatbuffers::Offset<flatbuffers::String>> boot_uuid_offsets; |
| boot_uuid_offsets.reserve(state.size()); |
| |
| int64_t *unused; |
| flatbuffers::Offset<flatbuffers::Vector<int64_t>> |
| oldest_remote_monotonic_timestamps_offset = |
| fbb.CreateUninitializedVector(state.size(), &unused); |
| |
| flatbuffers::Offset<flatbuffers::Vector<int64_t>> |
| oldest_local_monotonic_timestamps_offset = |
| fbb.CreateUninitializedVector(state.size(), &unused); |
| |
| flatbuffers::Offset<flatbuffers::Vector<int64_t>> |
| oldest_remote_unreliable_monotonic_timestamps_offset = |
| fbb.CreateUninitializedVector(state.size(), &unused); |
| |
| flatbuffers::Offset<flatbuffers::Vector<int64_t>> |
| oldest_local_unreliable_monotonic_timestamps_offset = |
| fbb.CreateUninitializedVector(state.size(), &unused); |
| |
| flatbuffers::Offset<flatbuffers::Vector<int64_t>> |
| oldest_remote_reliable_monotonic_timestamps_offset = |
| fbb.CreateUninitializedVector(state.size(), &unused); |
| |
| flatbuffers::Offset<flatbuffers::Vector<int64_t>> |
| oldest_local_reliable_monotonic_timestamps_offset = |
| fbb.CreateUninitializedVector(state.size(), &unused); |
| |
| flatbuffers::Offset<flatbuffers::Vector<int64_t>> |
| oldest_logger_remote_unreliable_monotonic_timestamps_offset = |
| fbb.CreateUninitializedVector(state.size(), &unused); |
| |
| flatbuffers::Offset<flatbuffers::Vector<int64_t>> |
| oldest_logger_local_unreliable_monotonic_timestamps_offset = |
| fbb.CreateUninitializedVector(state.size(), &unused); |
| |
| for (size_t i = 0; i < state.size(); ++i) { |
| if (state[i].boot_uuid != UUID::Zero()) { |
| boot_uuid_offsets.emplace_back(state[i].boot_uuid.PackString(&fbb)); |
| } else { |
| boot_uuid_offsets.emplace_back(fbb.CreateString("")); |
| } |
| if (state[i].boot_uuid == UUID::Zero()) { |
| CHECK_EQ(state[i].oldest_remote_monotonic_timestamp, |
| monotonic_clock::max_time); |
| CHECK_EQ(state[i].oldest_local_monotonic_timestamp, |
| monotonic_clock::max_time); |
| CHECK_EQ(state[i].oldest_remote_unreliable_monotonic_timestamp, |
| monotonic_clock::max_time); |
| CHECK_EQ(state[i].oldest_local_unreliable_monotonic_timestamp, |
| monotonic_clock::max_time); |
| CHECK_EQ(state[i].oldest_remote_reliable_monotonic_timestamp, |
| monotonic_clock::max_time); |
| CHECK_EQ(state[i].oldest_local_reliable_monotonic_timestamp, |
| monotonic_clock::max_time); |
| CHECK_EQ(state[i].oldest_logger_remote_unreliable_monotonic_timestamp, |
| monotonic_clock::max_time); |
| CHECK_EQ(state[i].oldest_logger_local_unreliable_monotonic_timestamp, |
| monotonic_clock::max_time); |
| } |
| |
| flatbuffers::GetMutableTemporaryPointer( |
| fbb, oldest_remote_monotonic_timestamps_offset) |
| ->Mutate(i, state[i] |
| .oldest_remote_monotonic_timestamp.time_since_epoch() |
| .count()); |
| flatbuffers::GetMutableTemporaryPointer( |
| fbb, oldest_local_monotonic_timestamps_offset) |
| ->Mutate(i, state[i] |
| .oldest_local_monotonic_timestamp.time_since_epoch() |
| .count()); |
| flatbuffers::GetMutableTemporaryPointer( |
| fbb, oldest_remote_unreliable_monotonic_timestamps_offset) |
| ->Mutate(i, state[i] |
| .oldest_remote_unreliable_monotonic_timestamp |
| .time_since_epoch() |
| .count()); |
| flatbuffers::GetMutableTemporaryPointer( |
| fbb, oldest_local_unreliable_monotonic_timestamps_offset) |
| ->Mutate(i, state[i] |
| .oldest_local_unreliable_monotonic_timestamp |
| .time_since_epoch() |
| .count()); |
| |
| flatbuffers::GetMutableTemporaryPointer( |
| fbb, oldest_remote_reliable_monotonic_timestamps_offset) |
| ->Mutate(i, state[i] |
| .oldest_remote_reliable_monotonic_timestamp |
| .time_since_epoch() |
| .count()); |
| flatbuffers::GetMutableTemporaryPointer( |
| fbb, oldest_local_reliable_monotonic_timestamps_offset) |
| ->Mutate( |
| i, state[i] |
| .oldest_local_reliable_monotonic_timestamp.time_since_epoch() |
| .count()); |
| |
| flatbuffers::GetMutableTemporaryPointer( |
| fbb, oldest_logger_remote_unreliable_monotonic_timestamps_offset) |
| ->Mutate(i, state[i] |
| .oldest_logger_remote_unreliable_monotonic_timestamp |
| .time_since_epoch() |
| .count()); |
| flatbuffers::GetMutableTemporaryPointer( |
| fbb, oldest_logger_local_unreliable_monotonic_timestamps_offset) |
| ->Mutate(i, state[i] |
| .oldest_logger_local_unreliable_monotonic_timestamp |
| .time_since_epoch() |
| .count()); |
| } |
| |
| flatbuffers::Offset< |
| flatbuffers::Vector<flatbuffers::Offset<flatbuffers::String>>> |
| boot_uuids_offset = fbb.CreateVector(boot_uuid_offsets); |
| |
| aos::ErrorList<StoredDataType> allowed_data_types_vector; |
| for (size_t type = static_cast<size_t>(StoredDataType::MIN); |
| type <= static_cast<size_t>(StoredDataType::MAX); ++type) { |
| if (allowed_data_types[type]) { |
| allowed_data_types_vector.Set(static_cast<StoredDataType>(type)); |
| } |
| } |
| |
| flatbuffers::Offset<flatbuffers::Vector<StoredDataType>> data_stored_offset = |
| fbb.CreateVector(allowed_data_types_vector.data(), |
| allowed_data_types_vector.size()); |
| |
| aos::logger::LogFileHeader::Builder log_file_header_builder(fbb); |
| |
| log_file_header_builder.add_name(name_offset); |
| if (!logger_sha1_offset.IsNull()) { |
| log_file_header_builder.add_logger_sha1(logger_sha1_offset); |
| } |
| if (!logger_version_offset.IsNull()) { |
| log_file_header_builder.add_logger_version(logger_version_offset); |
| } |
| |
| // Only add the node if we are running in a multinode configuration. |
| if (!logger_node_offset.IsNull()) { |
| log_file_header_builder.add_node(node_offset); |
| log_file_header_builder.add_logger_node(logger_node_offset); |
| } |
| |
| if (!configuration_offset.IsNull()) { |
| log_file_header_builder.add_configuration(configuration_offset); |
| } |
| |
| log_file_header_builder.add_max_out_of_order_duration( |
| max_out_of_order_duration.count()); |
| |
| NodeState *node_state = GetNodeState(node_index, source_node_boot_uuid); |
| log_file_header_builder.add_monotonic_start_time( |
| std::chrono::duration_cast<std::chrono::nanoseconds>( |
| node_state->monotonic_start_time.time_since_epoch()) |
| .count()); |
| if (source_node == node_) { |
| log_file_header_builder.add_realtime_start_time( |
| std::chrono::duration_cast<std::chrono::nanoseconds>( |
| node_state->realtime_start_time.time_since_epoch()) |
| .count()); |
| } else { |
| // Fill out the legacy start times. Since these were implemented to never |
| // change on reboot, they aren't very helpful in tracking what happened. |
| log_file_header_builder.add_logger_monotonic_start_time( |
| std::chrono::duration_cast<std::chrono::nanoseconds>( |
| node_state->logger_monotonic_start_time.time_since_epoch()) |
| .count()); |
| log_file_header_builder.add_logger_realtime_start_time( |
| std::chrono::duration_cast<std::chrono::nanoseconds>( |
| node_state->logger_realtime_start_time.time_since_epoch()) |
| .count()); |
| } |
| |
| // TODO(austin): Add more useful times. When was this part started? What do |
| // we know about both the logger and remote then? |
| |
| log_file_header_builder.add_log_event_uuid(log_event_uuid_offset); |
| log_file_header_builder.add_logger_instance_uuid(logger_instance_uuid_offset); |
| if (!log_start_uuid_offset.IsNull()) { |
| log_file_header_builder.add_log_start_uuid(log_start_uuid_offset); |
| } |
| log_file_header_builder.add_logger_node_boot_uuid( |
| logger_node_boot_uuid_offset); |
| log_file_header_builder.add_source_node_boot_uuid( |
| source_node_boot_uuid_offset); |
| |
| log_file_header_builder.add_parts_uuid(parts_uuid_offset); |
| log_file_header_builder.add_parts_index(parts_index); |
| |
| if (!config_sha256_offset.IsNull()) { |
| log_file_header_builder.add_configuration_sha256(config_sha256_offset); |
| } |
| |
| log_file_header_builder.add_boot_uuids(boot_uuids_offset); |
| log_file_header_builder.add_logger_part_monotonic_start_time( |
| std::chrono::duration_cast<std::chrono::nanoseconds>( |
| event_loop_->monotonic_now().time_since_epoch()) |
| .count()); |
| log_file_header_builder.add_logger_part_realtime_start_time( |
| std::chrono::duration_cast<std::chrono::nanoseconds>( |
| event_loop_->realtime_now().time_since_epoch()) |
| .count()); |
| log_file_header_builder.add_oldest_remote_monotonic_timestamps( |
| oldest_remote_monotonic_timestamps_offset); |
| log_file_header_builder.add_oldest_local_monotonic_timestamps( |
| oldest_local_monotonic_timestamps_offset); |
| log_file_header_builder.add_oldest_remote_unreliable_monotonic_timestamps( |
| oldest_remote_unreliable_monotonic_timestamps_offset); |
| log_file_header_builder.add_oldest_local_unreliable_monotonic_timestamps( |
| oldest_local_unreliable_monotonic_timestamps_offset); |
| log_file_header_builder.add_oldest_remote_reliable_monotonic_timestamps( |
| oldest_remote_reliable_monotonic_timestamps_offset); |
| log_file_header_builder.add_oldest_local_reliable_monotonic_timestamps( |
| oldest_local_reliable_monotonic_timestamps_offset); |
| log_file_header_builder |
| .add_oldest_logger_remote_unreliable_monotonic_timestamps( |
| oldest_logger_remote_unreliable_monotonic_timestamps_offset); |
| log_file_header_builder |
| .add_oldest_logger_local_unreliable_monotonic_timestamps( |
| oldest_logger_local_unreliable_monotonic_timestamps_offset); |
| |
| log_file_header_builder.add_data_stored(data_stored_offset); |
| fbb.FinishSizePrefixed(log_file_header_builder.Finish()); |
| aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> result( |
| fbb.Release()); |
| |
| CHECK(result.Verify()) << ": Built a corrupted header."; |
| |
| return result; |
| } |
| |
| MultiNodeLogNamer::MultiNodeLogNamer(std::unique_ptr<LogBackend> log_backend, |
| EventLoop *event_loop) |
| : MultiNodeLogNamer(std::move(log_backend), event_loop->configuration(), |
| event_loop, event_loop->node()) {} |
| |
| MultiNodeLogNamer::MultiNodeLogNamer(std::unique_ptr<LogBackend> log_backend, |
| const Configuration *configuration, |
| EventLoop *event_loop, const Node *node) |
| : LogNamer(configuration, event_loop, node), |
| log_backend_(std::move(log_backend)), |
| encoder_factory_([](size_t max_message_size) { |
| // TODO(austin): For slow channels, can we allocate less memory? |
| return std::make_unique<DummyEncoder>(max_message_size, |
| FLAGS_flush_size); |
| }) {} |
| |
| MultiNodeLogNamer::~MultiNodeLogNamer() { |
| if (!ran_out_of_space_) { |
| // This handles renaming temporary files etc. |
| Close(); |
| } |
| } |
| |
| void MultiNodeLogNamer::Rotate(const Node *node) { |
| for (auto &data_map : {&node_data_writers_, &node_timestamp_writers_}) { |
| auto it = data_map->find(node); |
| if (it != data_map->end()) { |
| it->second.Rotate(); |
| } |
| } |
| } |
| |
| void MultiNodeLogNamer::WriteConfiguration( |
| aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> *header, |
| std::string_view config_sha256) { |
| if (ran_out_of_space_) { |
| return; |
| } |
| |
| const std::string filename = absl::StrCat(config_sha256, ".bfbs", extension_); |
| auto file_handle = log_backend_->RequestFile(filename); |
| std::unique_ptr<DetachedBufferWriter> writer = |
| std::make_unique<DetachedBufferWriter>( |
| std::move(file_handle), encoder_factory_(header->span().size())); |
| |
| DataEncoder::SpanCopier coppier(header->span()); |
| writer->CopyMessage(&coppier, aos::monotonic_clock::now()); |
| |
| if (!writer->ran_out_of_space()) { |
| all_filenames_.emplace_back(filename); |
| } |
| // Close the file and maybe rename it too. |
| CloseWriter(&writer); |
| } |
| |
| void MultiNodeLogNamer::NoticeNode(const Node *source_node) { |
| if (std::find(nodes_.begin(), nodes_.end(), source_node) == nodes_.end()) { |
| nodes_.emplace_back(source_node); |
| } |
| } |
| |
| NewDataWriter *MultiNodeLogNamer::FindNodeDataWriter(const Node *source_node, |
| size_t max_message_size) { |
| NoticeNode(source_node); |
| |
| auto it = node_data_writers_.find(source_node); |
| if (it != node_data_writers_.end()) { |
| it->second.UpdateMaxMessageSize(max_message_size); |
| return &(it->second); |
| } |
| return nullptr; |
| } |
| |
| NewDataWriter *MultiNodeLogNamer::FindNodeTimestampWriter( |
| const Node *source_node, size_t max_message_size) { |
| NoticeNode(source_node); |
| |
| auto it = node_timestamp_writers_.find(source_node); |
| if (it != node_timestamp_writers_.end()) { |
| it->second.UpdateMaxMessageSize(max_message_size); |
| return &(it->second); |
| } |
| return nullptr; |
| } |
| |
| NewDataWriter *MultiNodeLogNamer::AddNodeDataWriter(const Node *source_node, |
| NewDataWriter &&writer) { |
| auto result = node_data_writers_.emplace(source_node, std::move(writer)); |
| CHECK(result.second); |
| return &(result.first->second); |
| } |
| |
| NewDataWriter *MultiNodeLogNamer::AddNodeTimestampWriter( |
| const Node *source_node, NewDataWriter &&writer) { |
| auto result = node_timestamp_writers_.emplace(source_node, std::move(writer)); |
| CHECK(result.second); |
| return &(result.first->second); |
| } |
| |
| NewDataWriter *MultiNodeLogNamer::MakeWriter(const Channel *channel) { |
| // See if we can read the data on this node at all. |
| const bool is_readable = |
| configuration::ChannelIsReadableOnNode(channel, this->node()); |
| if (!is_readable) { |
| return nullptr; |
| } |
| |
| // Then, see if we are supposed to log the data here. |
| const bool log_message = |
| configuration::ChannelMessageIsLoggedOnNode(channel, this->node()); |
| |
| if (!log_message) { |
| return nullptr; |
| } |
| |
| // Ok, we have data that is being forwarded to us that we are supposed to |
| // log. It needs to be logged with send timestamps, but be sorted enough |
| // to be able to be processed. |
| |
| const Node *source_node = |
| configuration::MultiNode(configuration_) |
| ? configuration::GetNode(configuration_, |
| channel->source_node()->string_view()) |
| : nullptr; |
| |
| // If we already have a data writer for the node, then use the same writer for |
| // all channels of that node. |
| NewDataWriter *result = FindNodeDataWriter( |
| source_node, |
| PackMessageSize(LogType::kLogRemoteMessage, channel->max_size())); |
| if (result != nullptr) { |
| return result; |
| } |
| |
| // If we don't have a data writer for the node, create one. |
| return AddNodeDataWriter( |
| source_node, |
| NewDataWriter{ |
| this, |
| source_node, |
| node_, |
| [this, source_node](NewDataWriter *data_writer) { |
| OpenDataWriter(source_node, data_writer); |
| }, |
| [this](NewDataWriter *data_writer) { |
| CloseWriter(&data_writer->writer); |
| }, |
| PackMessageSize(LogType::kLogRemoteMessage, channel->max_size()), |
| {StoredDataType::DATA}}); |
| } |
| |
| NewDataWriter *MultiNodeLogNamer::MakeForwardedTimestampWriter( |
| const Channel *channel, const Node *node) { |
| // See if we can read the data on this node at all. |
| const bool is_readable = |
| configuration::ChannelIsReadableOnNode(channel, this->node()); |
| CHECK(is_readable) << ": " << configuration::CleanedChannelToString(channel); |
| |
| CHECK_NE(node, this->node()); |
| |
| // If we have a remote timestamp writer for a particular node, use the same |
| // writer for all remote timestamp channels of that node. |
| NewDataWriter *result = |
| FindNodeTimestampWriter(node, PackRemoteMessageSize()); |
| if (result != nullptr) { |
| return result; |
| } |
| |
| // If there are no remote timestamp writers for the node, create one. |
| return AddNodeTimestampWriter( |
| node, NewDataWriter{this, |
| configuration::GetNode(configuration_, node), |
| node_, |
| [this](NewDataWriter *data_writer) { |
| OpenForwardedTimestampWriter(node_, data_writer); |
| }, |
| [this](NewDataWriter *data_writer) { |
| CloseWriter(&data_writer->writer); |
| }, |
| PackRemoteMessageSize(), |
| {StoredDataType::REMOTE_TIMESTAMPS}}); |
| } |
| |
| NewDataWriter *MultiNodeLogNamer::MakeTimestampWriter(const Channel *channel) { |
| bool log_delivery_times = false; |
| if (this->node() != nullptr) { |
| log_delivery_times = configuration::ConnectionDeliveryTimeIsLoggedOnNode( |
| channel, this->node(), this->node()); |
| } |
| if (!log_delivery_times) { |
| return nullptr; |
| } |
| |
| // There is only one of these. |
| NewDataWriter *result = FindNodeTimestampWriter( |
| this->node(), PackMessageSize(LogType::kLogDeliveryTimeOnly, 0)); |
| if (result != nullptr) { |
| return result; |
| } |
| |
| return AddNodeTimestampWriter( |
| node_, NewDataWriter{this, |
| node_, |
| node_, |
| [this](NewDataWriter *data_writer) { |
| OpenTimestampWriter(data_writer); |
| }, |
| [this](NewDataWriter *data_writer) { |
| CloseWriter(&data_writer->writer); |
| }, |
| PackMessageSize(LogType::kLogDeliveryTimeOnly, 0), |
| {StoredDataType::TIMESTAMPS}}); |
| } |
| |
| WriteCode MultiNodeLogNamer::Close() { |
| node_data_writers_.clear(); |
| node_timestamp_writers_.clear(); |
| if (ran_out_of_space_) { |
| return WriteCode::kOutOfSpace; |
| } |
| return WriteCode::kOk; |
| } |
| |
| void MultiNodeLogNamer::ResetStatistics() { |
| for (std::pair<const Node *const, NewDataWriter> &data_writer : |
| node_data_writers_) { |
| if (!data_writer.second.writer) continue; |
| data_writer.second.writer->WriteStatistics()->ResetStats(); |
| } |
| for (std::pair<const Node *const, NewDataWriter> &data_writer : |
| node_timestamp_writers_) { |
| if (!data_writer.second.writer) continue; |
| data_writer.second.writer->WriteStatistics()->ResetStats(); |
| } |
| max_write_time_ = std::chrono::nanoseconds::zero(); |
| max_write_time_bytes_ = -1; |
| max_write_time_messages_ = -1; |
| total_write_time_ = std::chrono::nanoseconds::zero(); |
| total_write_count_ = 0; |
| total_write_messages_ = 0; |
| total_write_bytes_ = 0; |
| } |
| |
| void MultiNodeLogNamer::OpenForwardedTimestampWriter( |
| const Node * /*source_node*/, NewDataWriter *data_writer) { |
| const std::string filename = absl::StrCat( |
| "timestamps/remote_", data_writer->node()->name()->string_view(), ".part", |
| data_writer->parts_index(), ".bfbs", extension_); |
| CreateBufferWriter(filename, data_writer->max_message_size(), |
| &data_writer->writer); |
| } |
| |
| void MultiNodeLogNamer::OpenDataWriter(const Node *source_node, |
| NewDataWriter *data_writer) { |
| std::string filename; |
| |
| if (source_node != nullptr) { |
| if (source_node == node_) { |
| filename = absl::StrCat(source_node->name()->string_view(), "_"); |
| } else { |
| filename = absl::StrCat("data/", source_node->name()->string_view(), "_"); |
| } |
| } |
| |
| absl::StrAppend(&filename, "data.part", data_writer->parts_index(), ".bfbs", |
| extension_); |
| CreateBufferWriter(filename, data_writer->max_message_size(), |
| &data_writer->writer); |
| } |
| |
| void MultiNodeLogNamer::OpenTimestampWriter(NewDataWriter *data_writer) { |
| std::string filename = |
| absl::StrCat(node()->name()->string_view(), "_timestamps.part", |
| data_writer->parts_index(), ".bfbs", extension_); |
| CreateBufferWriter(filename, data_writer->max_message_size(), |
| &data_writer->writer); |
| } |
| |
| void MultiNodeLogNamer::CreateBufferWriter( |
| std::string_view path, size_t max_message_size, |
| std::unique_ptr<DetachedBufferWriter> *destination) { |
| if (ran_out_of_space_) { |
| // Refuse to open any new files, which might skip data. Any existing files |
| // are in the same folder, which means they're on the same filesystem, which |
| // means they're probably going to run out of space and get stuck too. |
| if (!(*destination)) { |
| // But avoid leaving a nullptr writer if we're out of space when |
| // attempting to open the first file. |
| *destination = std::make_unique<DetachedBufferWriter>( |
| DetachedBufferWriter::already_out_of_space_t()); |
| } |
| return; |
| } |
| |
| // Let's check that we need to close and replace current driver. |
| if (*destination) { |
| // Let's close the current writer. |
| CloseWriter(destination); |
| // Are we out of space now? |
| if (ran_out_of_space_) { |
| *destination = std::make_unique<DetachedBufferWriter>( |
| DetachedBufferWriter::already_out_of_space_t()); |
| return; |
| } |
| } |
| |
| const std::string filename(path); |
| *destination = std::make_unique<DetachedBufferWriter>( |
| log_backend_->RequestFile(filename), encoder_factory_(max_message_size)); |
| if (!(*destination)->ran_out_of_space()) { |
| all_filenames_.emplace_back(path); |
| } |
| } |
| |
| void MultiNodeLogNamer::CloseWriter( |
| std::unique_ptr<DetachedBufferWriter> *writer_pointer) { |
| CHECK_NOTNULL(writer_pointer); |
| if (!(*writer_pointer)) { |
| return; |
| } |
| DetachedBufferWriter *const writer = writer_pointer->get(); |
| writer->Close(); |
| |
| const auto *stats = writer->WriteStatistics(); |
| if (stats->max_write_time() > max_write_time_) { |
| max_write_time_ = stats->max_write_time(); |
| max_write_time_bytes_ = stats->max_write_time_bytes(); |
| max_write_time_messages_ = stats->max_write_time_messages(); |
| } |
| total_write_time_ += stats->total_write_time(); |
| total_write_count_ += stats->total_write_count(); |
| total_write_messages_ += stats->total_write_messages(); |
| total_write_bytes_ += stats->total_write_bytes(); |
| |
| if (writer->ran_out_of_space()) { |
| ran_out_of_space_ = true; |
| writer->acknowledge_out_of_space(); |
| } |
| } |
| |
| NewDataWriter *MinimalFileMultiNodeLogNamer::MakeWriter( |
| const Channel *channel) { |
| // See if we can read the data on this node at all. |
| const bool is_readable = |
| configuration::ChannelIsReadableOnNode(channel, this->node()); |
| if (!is_readable) { |
| return nullptr; |
| } |
| |
| // Then, see if we are supposed to log the data here. |
| const bool log_message = |
| configuration::ChannelMessageIsLoggedOnNode(channel, this->node()); |
| |
| if (!log_message) { |
| return nullptr; |
| } |
| |
| // Ok, we have data that is being forwarded to us that we are supposed to |
| // log. It needs to be logged with send timestamps, but be sorted enough |
| // to be able to be processed. |
| |
| const Node *source_node = |
| configuration::MultiNode(configuration_) |
| ? configuration::GetNode(configuration_, |
| channel->source_node()->string_view()) |
| : nullptr; |
| |
| // If we don't have a data writer for the node, create one. |
| if (this->node() == source_node) { |
| // If we already have a data writer for the node, then use the same writer |
| // for all channels of that node. |
| NewDataWriter *result = FindNodeDataWriter( |
| source_node, |
| PackMessageSize(LogType::kLogMessage, channel->max_size())); |
| if (result != nullptr) { |
| return result; |
| } |
| |
| return AddNodeDataWriter( |
| source_node, |
| NewDataWriter{ |
| this, |
| source_node, |
| node_, |
| [this, source_node](NewDataWriter *data_writer) { |
| OpenNodeWriter(source_node, data_writer); |
| }, |
| [this](NewDataWriter *data_writer) { |
| CloseWriter(&data_writer->writer); |
| }, |
| PackMessageSize(LogType::kLogMessage, channel->max_size()), |
| {StoredDataType::DATA, StoredDataType::TIMESTAMPS}}); |
| } else { |
| // If we already have a data writer for the node, then use the same writer |
| // for all channels of that node. |
| NewDataWriter *result = FindNodeDataWriter( |
| source_node, |
| PackMessageSize(LogType::kLogRemoteMessage, channel->max_size())); |
| if (result != nullptr) { |
| return result; |
| } |
| |
| return AddNodeDataWriter( |
| source_node, |
| NewDataWriter{ |
| this, |
| source_node, |
| node_, |
| [this, source_node](NewDataWriter *data_writer) { |
| OpenNodeWriter(source_node, data_writer); |
| }, |
| [this](NewDataWriter *data_writer) { |
| CloseWriter(&data_writer->writer); |
| }, |
| PackMessageSize(LogType::kLogRemoteMessage, channel->max_size()), |
| {StoredDataType::DATA, StoredDataType::REMOTE_TIMESTAMPS}}); |
| } |
| } |
| |
| NewDataWriter *MinimalFileMultiNodeLogNamer::MakeTimestampWriter( |
| const Channel *channel) { |
| bool log_delivery_times = false; |
| if (this->node() != nullptr) { |
| log_delivery_times = configuration::ConnectionDeliveryTimeIsLoggedOnNode( |
| channel, this->node(), this->node()); |
| } |
| if (!log_delivery_times) { |
| return nullptr; |
| } |
| |
| // There is only one of these. |
| NewDataWriter *result = FindNodeDataWriter( |
| this->node(), PackMessageSize(LogType::kLogDeliveryTimeOnly, 0)); |
| if (result != nullptr) { |
| return result; |
| } |
| |
| return AddNodeDataWriter( |
| node_, NewDataWriter{this, |
| node_, |
| node_, |
| [this](NewDataWriter *data_writer) { |
| OpenNodeWriter(node_, data_writer); |
| }, |
| [this](NewDataWriter *data_writer) { |
| CloseWriter(&data_writer->writer); |
| }, |
| PackMessageSize(LogType::kLogDeliveryTimeOnly, 0), |
| {StoredDataType::DATA, StoredDataType::TIMESTAMPS}}); |
| } |
| |
| NewDataWriter *MinimalFileMultiNodeLogNamer::MakeForwardedTimestampWriter( |
| const Channel *channel, const Node *node) { |
| // See if we can read the data on this node at all. |
| const bool is_readable = |
| configuration::ChannelIsReadableOnNode(channel, this->node()); |
| CHECK(is_readable) << ": " << configuration::CleanedChannelToString(channel); |
| |
| CHECK_NE(node, this->node()); |
| |
| // If we have a remote timestamp writer for a particular node, use the same |
| // writer for all remote timestamp channels of that node. |
| NewDataWriter *result = FindNodeDataWriter(node, PackRemoteMessageSize()); |
| if (result != nullptr) { |
| return result; |
| } |
| |
| // If there are no remote timestamp writers for the node, create one. |
| return AddNodeDataWriter( |
| node, |
| NewDataWriter{this, |
| configuration::GetNode(configuration_, node), |
| node_, |
| [this, node](NewDataWriter *data_writer) { |
| OpenNodeWriter(node, data_writer); |
| }, |
| [this](NewDataWriter *data_writer) { |
| CloseWriter(&data_writer->writer); |
| }, |
| PackRemoteMessageSize(), |
| {StoredDataType::DATA, StoredDataType::REMOTE_TIMESTAMPS}}); |
| } |
| |
| void MinimalFileMultiNodeLogNamer::OpenNodeWriter(const Node *source_node, |
| NewDataWriter *data_writer) { |
| std::string filename; |
| |
| if (node() != nullptr) { |
| filename = absl::StrCat(node()->name()->string_view(), "_"); |
| } |
| |
| if (source_node != nullptr) { |
| absl::StrAppend(&filename, source_node->name()->string_view(), "_"); |
| } |
| |
| absl::StrAppend(&filename, "all.part", data_writer->parts_index(), ".bfbs", |
| extension_); |
| VLOG(1) << "Going to open " << filename; |
| CreateBufferWriter(filename, data_writer->max_message_size(), |
| &data_writer->writer); |
| } |
| |
| } // namespace aos::logger |