pw_multisink: Add UnsafeForEachEntryFromEnd()

UnsafeForEachEntryFromEnd() captures the latest (last) entries
from the multisink upto a specified max bytes. Earlier entries
will be dropped.

Bug: b/375653884
Change-Id: I1d840a8881d53d4268489d888dca415667e01052
Reviewed-on: https://pigweed-review.googlesource.com/c/pigweed/pigweed/+/244556
Commit-Queue: Dave Roth <davidroth@google.com>
Lint: Lint 🤖 <android-build-ayeaye@system.gserviceaccount.com>
Reviewed-by: Wyatt Hepler <hepler@google.com>

pw_multisink/docs.rst

@@ -150,7 +150,10 @@ iterator to be used even if no drains have been previously attached.
 As an alternative to using the ``UnsafeIterationWrapper``,
 ``MultiSink::UnsafeForEachEntry()`` may be used to run a callback for each
 entry in the buffer. This helper also provides a way to limit the iteration to
-the ``N`` most recent entries.
+the ``N`` most recent entries. In certain cases such as when there isn't
+enough space to copy the entire buffer, it is desirable to capture
+the latest entries rather than the first entries. In this case
+``MultiSink::UnsafeForEachEntryFromEnd`` can be used.
 
 Peek & Pop
 ==========

pw_multisink/multisink.cc

@@ -223,6 +223,54 @@ Status MultiSink::UnsafeForEachEntry(
   return OkStatus();
 }
 
+Status MultiSink::UnsafeForEachEntryFromEnd(
+    const Function<void(ConstByteSpan)>& callback, size_t max_size_bytes) {
+  MultiSink::UnsafeIterationWrapper multisink_iteration = UnsafeIteration();
+
+  // First count the number of entries and total size of the entries.
+  size_t num_entries = 0;
+  size_t total_bytes = 0;
+  iterator it = multisink_iteration.begin();
+  iterator last_elem_it;
+  for (; it != multisink_iteration.end(); ++it) {
+    num_entries++;
+    total_bytes += (*it).size();
+    last_elem_it = it;
+  }
+
+  size_t max_num_entries = std::numeric_limits<size_t>::max();
+  // All entries won't fit in the available space, so reverse iterate
+  // from the end to calculate the number of elements from the end
+  // which will fit in the available space.
+  if (total_bytes > max_size_bytes) {
+    total_bytes = 0;
+    max_num_entries = 0;
+    while (total_bytes <= max_size_bytes) {
+      total_bytes += (*last_elem_it).size();
+      last_elem_it--;
+      max_num_entries++;
+    }
+  }
+
+  // Log up to the max number of logs to avoid overflowing the crash log
+  // writer.
+  const size_t first_logged_offset =
+      max_num_entries > num_entries ? 0 : num_entries - max_num_entries;
+  it = multisink_iteration.begin();
+  for (size_t offset = 0; it != multisink_iteration.end(); ++it, ++offset) {
+    if (offset < first_logged_offset) {
+      continue;  // Skip this log.
+    }
+    callback(*it);
+  }
+  if (!it.status().ok()) {
+    PW_LOG_WARN("Multisink corruption detected, some entries may be missing");
+    return Status::DataLoss();
+  }
+
+  return OkStatus();
+}
+
 Status MultiSink::Drain::PopEntry(const PeekedEntry& entry) {
   PW_DCHECK_NOTNULL(multisink_);
   return multisink_->PopEntry(*this, entry);

pw_multisink/multisink_test.cc

@@ -664,5 +664,66 @@ TEST(UnsafeIteration, Subset) {
   EXPECT_EQ(kExpectedEntriesMaxEntries, entry_count);
 }
 
+TEST(UnsafeIterationFromEnd, NoTruncation) {
+  constexpr std::array<std::string_view, 5> kExpectedEntries{
+      "one", "two", "three", "four", "five"};
+  constexpr size_t buffer_size = 32;
+  std::array<std::byte, buffer_size> buffer;
+  MultiSink multisink(buffer);
+
+  for (std::string_view entry : kExpectedEntries) {
+    multisink.HandleEntry(as_bytes(span<const char>(entry)));
+  }
+
+  size_t entry_count = 0;
+  struct {
+    size_t& entry_count;
+    span<const std::string_view> expected_results;
+  } ctx{entry_count, kExpectedEntries};
+  auto cb = [&ctx](ConstByteSpan data) {
+    std::string_view expected_entry = ctx.expected_results[ctx.entry_count];
+    EXPECT_EQ(data.size(), expected_entry.size());
+    const int result =
+        memcmp(data.data(), expected_entry.data(), expected_entry.size());
+    EXPECT_EQ(0, result);
+    ctx.entry_count++;
+  };
+
+  EXPECT_EQ(OkStatus(), multisink.UnsafeForEachEntryFromEnd(cb, buffer_size));
+  EXPECT_EQ(kExpectedEntries.size(), entry_count);
+}
+
+TEST(UnsafeIterationFromEnd, Truncation) {
+  constexpr std::array<std::string_view, 5> kEntries{
+      "one", "two", "three", "four", "five"};
+  constexpr std::array<std::string_view, 3> kExpectedEntries{
+      "three", "four", "five"};
+  constexpr size_t buffer_size = 32;
+  std::array<std::byte, buffer_size> buffer;
+  MultiSink multisink(buffer);
+
+  for (std::string_view entry : kEntries) {
+    multisink.HandleEntry(as_bytes(span<const char>(entry)));
+  }
+
+  size_t entry_count = 0;
+  struct {
+    size_t& entry_count;
+    span<const std::string_view> expected_results;
+  } ctx{entry_count, kExpectedEntries};
+  auto cb = [&ctx](ConstByteSpan data) {
+    std::string_view expected_entry = ctx.expected_results[ctx.entry_count];
+    EXPECT_EQ(data.size(), expected_entry.size());
+    const int result =
+        memcmp(data.data(), expected_entry.data(), expected_entry.size());
+    EXPECT_EQ(0, result);
+    ctx.entry_count++;
+  };
+
+  EXPECT_EQ(OkStatus(),
+            multisink.UnsafeForEachEntryFromEnd(cb, buffer_size / 2));
+  EXPECT_EQ(kExpectedEntries.size(), entry_count);
+}
+
 }  // namespace
 }  // namespace pw::multisink

pw_multisink/public/pw_multisink/multisink.h

@@ -274,6 +274,16 @@ class MultiSink {
       return original;
     }
 
+    iterator& operator--() {
+      it_--;
+      return *this;
+    }
+    iterator operator--(int) {
+      iterator original = *this;
+      --*this;
+      return original;
+    }
+
     ConstByteSpan& operator*() {
       entry_ = (*it_).buffer;
       return entry_;
@@ -411,6 +421,16 @@ class MultiSink {
       const Function<void(ConstByteSpan)>& callback,
       size_t max_num_entries = std::numeric_limits<size_t>::max());
 
+  // Uses MultiSink's unsafe iteration to dump the contents to a user-provided
+  // callback. UnsafeForEachEntryFromEnd dumps the latest entries, up to the
+  // aggregate element size of max_size_bytes.
+  //
+  // Returns:
+  //   OK - Successfully dumped entire multisink.
+  //   DATA_LOSS - Corruption detected, some entries may have been lost.
+  Status UnsafeForEachEntryFromEnd(
+      const Function<void(ConstByteSpan)>& callback, size_t max_size_bytes);
+
  protected:
   friend Drain;
 

pw_multisink/public/pw_multisink/util.h

@@ -30,4 +30,12 @@ Status UnsafeDumpMultiSinkLogs(
     pw::log::pwpb::LogEntries::StreamEncoder& encoder,
     size_t max_num_entries = std::numeric_limits<size_t>::max());
 
+// Uses MultiSink's unsafe iteration to dump the contents as a series of log
+// entries. max_size_bytes is the total size of the captured log entries. This
+// can be used to dump proto-encoded logs to a pw.snapshot.Snapshot.
+Status UnsafeDumpMultiSinkLogsFromEnd(
+    MultiSink& sink,
+    pw::log::pwpb::LogEntries::StreamEncoder& encoder,
+    size_t max_size_bytes);
+
 }  // namespace pw::multisink

pw_multisink/util.cc

@@ -35,4 +35,18 @@ Status UnsafeDumpMultiSinkLogs(
   return sink.UnsafeForEachEntry(callback, max_num_entries);
 }
 
+Status UnsafeDumpMultiSinkLogsFromEnd(
+    MultiSink& sink,
+    pw::log::pwpb::LogEntries::StreamEncoder& encoder,
+    size_t max_size_bytes) {
+  auto callback = [&encoder](ConstByteSpan entry) {
+    encoder
+        .WriteBytes(
+            static_cast<uint32_t>(pw::log::pwpb::LogEntries::Fields::kEntries),
+            entry)
+        .IgnoreError();
+  };
+  return sink.UnsafeForEachEntryFromEnd(callback, max_size_bytes);
+}
+
 }  // namespace pw::multisink