Merge "Update log_stats to handle reboots better"
diff --git a/aos/configuration.cc b/aos/configuration.cc
index 4de5332..3b50ab4 100644
--- a/aos/configuration.cc
+++ b/aos/configuration.cc
@@ -1636,6 +1636,13 @@
   return nullptr;
 }
 
+const Node *SourceNode(const Configuration *config, const Channel *channel) {
+  if (!MultiNode(config)) {
+    return nullptr;
+  }
+  return GetNode(config, channel->source_node()->string_view());
+}
+
 std::vector<size_t> SourceNodeIndex(const Configuration *config) {
   CHECK(config->has_channels());
   std::vector<size_t> result;
diff --git a/aos/configuration.h b/aos/configuration.h
index 7ef2e6e..80c0238 100644
--- a/aos/configuration.h
+++ b/aos/configuration.h
@@ -198,6 +198,9 @@
 // Returns the source node index for each channel in a config.
 std::vector<size_t> SourceNodeIndex(const Configuration *config);
 
+// Returns the source node for a channel.
+const Node *SourceNode(const Configuration *config, const Channel *channel);
+
 // Returns the all nodes that are logging timestamps on our node.
 std::vector<const Node *> TimestampNodes(const Configuration *config,
                                          const Node *my_node);
diff --git a/aos/configuration_test.cc b/aos/configuration_test.cc
index eb08d78..0ce6772 100644
--- a/aos/configuration_test.cc
+++ b/aos/configuration_test.cc
@@ -1022,6 +1022,38 @@
   EXPECT_THAT(result, ::testing::ElementsAreArray({0, 1, 0, 0}));
 }
 
+// Tests that SourceNode reasonably handles both single and multi-node configs.
+TEST_F(ConfigurationTest, SourceNode) {
+  {
+    FlatbufferDetachedBuffer<Configuration> config_single_node =
+        ReadConfig(ArtifactPath("aos/testdata/config1.json"));
+    const Node *result =
+        SourceNode(&config_single_node.message(),
+                   config_single_node.message().channels()->Get(0));
+    EXPECT_EQ(result, nullptr);
+  }
+
+  {
+    FlatbufferDetachedBuffer<Configuration> config_multi_node =
+        ReadConfig(ArtifactPath("aos/testdata/good_multinode.json"));
+    size_t pi1_channels = 0;
+    size_t pi2_channels = 0;
+    for (const aos::Channel *channel :
+         *config_multi_node.message().channels()) {
+      const Node *result = SourceNode(&config_multi_node.message(), channel);
+      if (channel->source_node()->string_view() == "pi1") {
+        ++pi1_channels;
+        EXPECT_EQ(result, config_multi_node.message().nodes()->Get(0));
+      } else {
+        ++pi2_channels;
+        EXPECT_EQ(result, config_multi_node.message().nodes()->Get(1));
+      }
+    }
+    EXPECT_GT(pi1_channels, 0u);
+    EXPECT_GT(pi2_channels, 0u);
+  }
+}
+
 // Tests that we reject invalid logging configurations.
 TEST_F(ConfigurationDeathTest, InvalidLoggerConfig) {
   EXPECT_DEATH(
diff --git a/aos/containers/resizeable_buffer.h b/aos/containers/resizeable_buffer.h
index 2f6ff32..9f9967f 100644
--- a/aos/containers/resizeable_buffer.h
+++ b/aos/containers/resizeable_buffer.h
@@ -118,6 +118,20 @@
   }
 };
 
+// Allocates aligned memory.
+template <size_t alignment>
+class AlignedReallocator {
+ public:
+  static void *Realloc(void *old, size_t old_size, size_t new_capacity) {
+    void *new_memory = std::aligned_alloc(alignment, new_capacity);
+    if (old) {
+      memcpy(new_memory, old, old_size);
+      free(old);
+    }
+    return new_memory;
+  }
+};
+
 // A resizable buffer which uses realloc when it needs to grow to attempt to
 // avoid full coppies.
 class ResizeableBuffer : public AllocatorResizeableBuffer<Reallocator> {};
diff --git a/aos/events/logging/BUILD b/aos/events/logging/BUILD
index eadefa8..3b5f57a 100644
--- a/aos/events/logging/BUILD
+++ b/aos/events/logging/BUILD
@@ -1014,7 +1014,9 @@
     srcs = [
         "ssd_profiler.cc",
     ],
+    visibility = ["//visibility:public"],
     deps = [
+        ":log_backend",
         "//aos:init",
         "//aos/containers:resizeable_buffer",
         "//aos/time",
diff --git a/aos/events/logging/buffer_encoder.h b/aos/events/logging/buffer_encoder.h
index 394992c..a155bb5 100644
--- a/aos/events/logging/buffer_encoder.h
+++ b/aos/events/logging/buffer_encoder.h
@@ -15,6 +15,10 @@
  public:
   virtual ~DataEncoder() = default;
 
+  // Size of an aligned sector used to detect when the data is aligned enough to
+  // use O_DIRECT instead.
+  static constexpr size_t kSector = 512u;
+
   // Interface to copy data into a buffer.
   class Copier {
    public:
@@ -114,21 +118,7 @@
  private:
   size_t total_bytes_ = 0;
 
-  // A class which uses aligned_alloc to allocate sector aligned blocks of
-  // memory.
-  class AlignedReallocator {
-   public:
-    static void *Realloc(void *old, size_t old_size, size_t new_capacity) {
-      void *new_memory = std::aligned_alloc(512, new_capacity);
-      if (old) {
-        memcpy(new_memory, old, old_size);
-        free(old);
-      }
-      return new_memory;
-    }
-  };
-
-  AllocatorResizeableBuffer<AlignedReallocator> input_buffer_;
+  AllocatorResizeableBuffer<aos::AlignedReallocator<kSector>> input_buffer_;
   std::vector<absl::Span<const uint8_t>> return_queue_;
 };
 
diff --git a/aos/events/logging/log_backend_test.cc b/aos/events/logging/log_backend_test.cc
index 1e95c10..0603b33 100644
--- a/aos/events/logging/log_backend_test.cc
+++ b/aos/events/logging/log_backend_test.cc
@@ -246,19 +246,10 @@
 struct FileWriteTestBase : public ::testing::Test {
   uint8_t NextRandom() { return distribution(engine); }
 
-  class AlignedReallocator {
-   public:
-    static void *Realloc(void *old, size_t old_size, size_t new_capacity) {
-      void *new_memory = std::aligned_alloc(512, new_capacity);
-      if (old) {
-        memcpy(new_memory, old, old_size);
-        free(old);
-      }
-      return new_memory;
-    }
-  };
+  AllocatorResizeableBuffer<AlignedReallocator<aos::logger::FileHandler::kSector
 
-  AllocatorResizeableBuffer<AlignedReallocator> buffer;
+                                               >>
+      buffer;
 
   void TestRecipe(const WriteRecipe &recipe) {
     VLOG(1) << "Starting";
@@ -374,8 +365,8 @@
 
 // Test an aligned to unaligned transition to make sure everything works.
 TEST_F(FileWriteTestBase, AlignedToUnaligned) {
-  AllocatorResizeableBuffer<AlignedReallocator> aligned_buffer;
-  AllocatorResizeableBuffer<AlignedReallocator> unaligned_buffer;
+  AllocatorResizeableBuffer<AlignedReallocator<512>> aligned_buffer;
+  AllocatorResizeableBuffer<AlignedReallocator<512>> unaligned_buffer;
 
   aligned_buffer.resize(FileHandler::kSector * 4);
   std::generate(std::begin(aligned_buffer), std::end(aligned_buffer),
diff --git a/aos/events/logging/ssd_profiler.cc b/aos/events/logging/ssd_profiler.cc
index e7493ff..093e17b 100644
--- a/aos/events/logging/ssd_profiler.cc
+++ b/aos/events/logging/ssd_profiler.cc
@@ -12,6 +12,7 @@
 #include "glog/logging.h"
 
 #include "aos/containers/resizeable_buffer.h"
+#include "aos/events/logging/log_backend.h"
 #include "aos/init.h"
 #include "aos/realtime.h"
 #include "aos/time/time.h"
@@ -39,19 +40,6 @@
               "Write speed in MB/s to simulate. This is only used when "
               "--rate_limit is specified.");
 
-// Stolen from aos/events/logging/DummyEncoder
-class AlignedReallocator {
- public:
-  static void *Realloc(void *old, size_t old_size, size_t new_capacity) {
-    void *new_memory = std::aligned_alloc(512, new_capacity);
-    if (old) {
-      memcpy(new_memory, old, old_size);
-      free(old);
-    }
-    return new_memory;
-  }
-};
-
 void trap_sig(int signum) { exit(signum); }
 
 aos::monotonic_clock::time_point start_time = aos::monotonic_clock::min_time;
@@ -82,7 +70,9 @@
     std::signal(SIGHUP, trap_sig);
     std::atexit(cleanup);
   }
-  aos::AllocatorResizeableBuffer<AlignedReallocator> data;
+  aos::AllocatorResizeableBuffer<
+      aos::AlignedReallocator<aos::logger::FileHandler::kSector>>
+      data;
 
   {
     // We want uncompressible data.  The easiest way to do this is to grab a