Merge changes I9638e90b,Ib2636dfc

* changes:
  logd: replace std::vector<uint8_t> in SerializedLogChunk
  logd: fix use after resize of contents_ vector

Author: Tom Cherry

logd/CompressionEngine.cpp

@@ -27,7 +27,7 @@ CompressionEngine& CompressionEngine::GetInstance() {
     return *engine;
 }
 
-bool ZlibCompressionEngine::Compress(std::span<uint8_t> in, std::vector<uint8_t>& out) {
+bool ZlibCompressionEngine::Compress(SerializedData& in, size_t data_length, SerializedData& out) {
     z_stream strm;
     strm.zalloc = Z_NULL;
     strm.zfree = Z_NULL;
@@ -37,34 +37,34 @@ bool ZlibCompressionEngine::Compress(std::span<uint8_t> in, std::vector<uint8_t>
         LOG(FATAL) << "deflateInit() failed";
     }
 
-    CHECK_LE(in.size(), static_cast<int64_t>(std::numeric_limits<uint32_t>::max()));
-    uint32_t out_size = deflateBound(&strm, in.size());
+    CHECK_LE(data_length, in.size());
+    CHECK_LE(in.size(), std::numeric_limits<uint32_t>::max());
+    uint32_t deflate_bound = deflateBound(&strm, in.size());
 
-    out.resize(out_size);
-    strm.avail_in = in.size();
-    strm.next_in = const_cast<uint8_t*>(in.data());
-    strm.avail_out = out_size;
+    out.Resize(deflate_bound);
+
+    strm.avail_in = data_length;
+    strm.next_in = in.data();
+    strm.avail_out = out.size();
     strm.next_out = out.data();
     ret = deflate(&strm, Z_FINISH);
     CHECK_EQ(ret, Z_STREAM_END);
 
-    uint32_t compressed_data_size = strm.total_out;
+    uint32_t compressed_size = strm.total_out;
     deflateEnd(&strm);
-    out.resize(compressed_data_size);
+
+    out.Resize(compressed_size);
 
     return true;
 }
 
-bool ZlibCompressionEngine::Decompress(const std::vector<uint8_t>& in, std::vector<uint8_t>& out,
-                                       size_t out_size) {
-    out.resize(out_size);
-
+bool ZlibCompressionEngine::Decompress(SerializedData& in, SerializedData& out) {
     z_stream strm;
     strm.zalloc = Z_NULL;
     strm.zfree = Z_NULL;
     strm.opaque = Z_NULL;
     strm.avail_in = in.size();
-    strm.next_in = const_cast<uint8_t*>(in.data());
+    strm.next_in = in.data();
     strm.avail_out = out.size();
     strm.next_out = out.data();
 
@@ -79,22 +79,22 @@ bool ZlibCompressionEngine::Decompress(const std::vector<uint8_t>& in, std::vect
     return true;
 }
 
-bool ZstdCompressionEngine::Compress(std::span<uint8_t> in, std::vector<uint8_t>& out) {
-    size_t out_size = ZSTD_compressBound(in.size());
-    out.resize(out_size);
+bool ZstdCompressionEngine::Compress(SerializedData& in, size_t data_length, SerializedData& out) {
+    CHECK_LE(data_length, in.size());
+
+    size_t compress_bound = ZSTD_compressBound(data_length);
+    out.Resize(compress_bound);
 
-    out_size = ZSTD_compress(out.data(), out_size, in.data(), in.size(), 1);
+    size_t out_size = ZSTD_compress(out.data(), out.size(), in.data(), data_length, 1);
     if (ZSTD_isError(out_size)) {
         LOG(FATAL) << "ZSTD_compress failed: " << ZSTD_getErrorName(out_size);
     }
-    out.resize(out_size);
+    out.Resize(out_size);
 
     return true;
 }
 
-bool ZstdCompressionEngine::Decompress(const std::vector<uint8_t>& in, std::vector<uint8_t>& out,
-                                       size_t out_size) {
-    out.resize(out_size);
+bool ZstdCompressionEngine::Decompress(SerializedData& in, SerializedData& out) {
     size_t result = ZSTD_decompress(out.data(), out.size(), in.data(), in.size());
     if (ZSTD_isError(result)) {
         LOG(FATAL) << "ZSTD_decompress failed: " << ZSTD_getErrorName(result);

logd/CompressionEngine.h

@@ -16,32 +16,30 @@
 
 #pragma once
 
-#include <span>
-#include <vector>
+#include <memory>
+
+#include "SerializedData.h"
 
 class CompressionEngine {
   public:
     static CompressionEngine& GetInstance();
 
     virtual ~CompressionEngine(){};
 
-    virtual bool Compress(std::span<uint8_t> in, std::vector<uint8_t>& out) = 0;
-    // Decompress the contents of `in` into `out`.  `out_size` must be set to the decompressed size
-    // of the contents.
-    virtual bool Decompress(const std::vector<uint8_t>& in, std::vector<uint8_t>& out,
-                            size_t out_size) = 0;
+    virtual bool Compress(SerializedData& in, size_t data_length, SerializedData& out) = 0;
+    // Decompress the contents of `in` into `out`.  `out.size()` must be set to the decompressed
+    // size of the contents.
+    virtual bool Decompress(SerializedData& in, SerializedData& out) = 0;
 };
 
 class ZlibCompressionEngine : public CompressionEngine {
   public:
-    bool Compress(std::span<uint8_t> in, std::vector<uint8_t>& out) override;
-    bool Decompress(const std::vector<uint8_t>& in, std::vector<uint8_t>& out,
-                    size_t out_size) override;
+    bool Compress(SerializedData& in, size_t data_length, SerializedData& out) override;
+    bool Decompress(SerializedData& in, SerializedData& out) override;
 };
 
 class ZstdCompressionEngine : public CompressionEngine {
   public:
-    bool Compress(std::span<uint8_t> in, std::vector<uint8_t>& out) override;
-    bool Decompress(const std::vector<uint8_t>& in, std::vector<uint8_t>& out,
-                    size_t out_size) override;
+    bool Compress(SerializedData& in, size_t data_length, SerializedData& out) override;
+    bool Decompress(SerializedData& in, SerializedData& out) override;
 };

logd/SerializedData.h

@@ -0,0 +1,55 @@
+/*
+ * Copyright (C) 2020 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <sys/types.h>
+
+#include <algorithm>
+#include <memory>
+
+// This class is used instead of std::string or std::vector because their clear(), erase(), etc
+// functions don't actually deallocate.  shrink_to_fit() does deallocate but is not guaranteed to
+// work and swapping with an empty string/vector is clunky.
+class SerializedData {
+  public:
+    SerializedData() {}
+    SerializedData(size_t size) : data_(new uint8_t[size]), size_(size) {}
+
+    void Resize(size_t new_size) {
+        if (size_ == 0) {
+            data_.reset(new uint8_t[new_size]);
+            size_ = new_size;
+        } else if (new_size == 0) {
+            data_.reset();
+            size_ = 0;
+        } else if (new_size != size_) {
+            std::unique_ptr<uint8_t[]> new_data(new uint8_t[new_size]);
+            size_t copy_size = std::min(size_, new_size);
+            memcpy(new_data.get(), data_.get(), copy_size);
+            data_.swap(new_data);
+            size_ = new_size;
+        }
+    }
+
+    uint8_t* data() { return data_.get(); }
+    const uint8_t* data() const { return data_.get(); }
+    size_t size() const { return size_; }
+
+  private:
+    std::unique_ptr<uint8_t[]> data_;
+    size_t size_ = 0;
+};

logd/SerializedFlushToState.cpp

@@ -121,7 +121,7 @@ MinHeapElement SerializedFlushToState::PopNextUnreadLog() {
 
     log_positions_[log_id]->read_offset += entry->total_len();
 
-    AddMinHeapEntry(log_id);
+    logs_needed_from_next_position_[log_id] = true;
 
     return top;
 }

logd/SerializedFlushToState.h

@@ -61,15 +61,13 @@ class SerializedFlushToState : public FlushToState {
         if (logs_ == nullptr) logs_ = logs;
     }
 
-    // Checks to see if any log buffers set in logs_needed_from_next_position_ have new logs and
-    // calls AddMinHeapEntry() if so.
-    void CheckForNewLogs();
-
-    bool HasUnreadLogs() { return !min_heap_.empty(); }
+    bool HasUnreadLogs() {
+        CheckForNewLogs();
+        return !min_heap_.empty();
+    }
 
-    // Pops the next unread log from the min heap.  Add the next log for that log_id to the min heap
-    // if one is available otherwise set logs_needed_from_next_position_ to indicate that we're
-    // waiting for more logs.
+    // Pops the next unread log from the min heap and sets logs_needed_from_next_position_ to
+    // indicate that we're waiting for more logs from the associated log buffer.
     MinHeapElement PopNextUnreadLog();
 
     // If the parent log buffer prunes logs, the reference that this class contains may become
@@ -86,6 +84,10 @@ class SerializedFlushToState : public FlushToState {
     // first chunk and then first log entry within that chunk that is greater or equal to start().
     void CreateLogPosition(log_id_t log_id);
 
+    // Checks to see if any log buffers set in logs_needed_from_next_position_ have new logs and
+    // calls AddMinHeapEntry() if so.
+    void CheckForNewLogs();
+
     std::list<SerializedLogChunk>* logs_ = nullptr;
     // An optional structure that contains an iterator to the serialized log buffer and offset into
     // it that this logger should handle next.

logd/SerializedFlushToStateTest.cpp

@@ -89,7 +89,6 @@ class SerializedFlushToStateTest : public testing::Test {
             for (uint32_t mask = 0; mask < max_mask; ++mask) {
                 auto state = SerializedFlushToState{sequence, mask};
                 state.InitializeLogs(log_chunks_);
-                state.CheckForNewLogs();
                 TestReading(sequence, mask, state);
             }
         }
@@ -109,7 +108,6 @@ class SerializedFlushToStateTest : public testing::Test {
                 state.InitializeLogs(log_chunks_);
                 int loop_count = 0;
                 while (write_logs(loop_count++)) {
-                    state.CheckForNewLogs();
                     TestReading(sequence, mask, state);
                     sequence_numbers_per_buffer_.clear();
                 }
@@ -149,7 +147,7 @@ class SerializedFlushToStateTest : public testing::Test {
 
     // Add a chunk with the given messages to the a given log buffer.  Keep track of the sequence
     // numbers for future validation.  Optionally mark the block as having finished writing.
-    void AddChunkWithMessages(int buffer, bool finish_writing,
+    void AddChunkWithMessages(bool finish_writing, int buffer,
                               const std::vector<std::string>& messages) {
         auto chunk = SerializedLogChunk{kChunkSize};
         for (const auto& message : messages) {
@@ -252,3 +250,41 @@ TEST_F(SerializedFlushToStateTest, future_writes) {
 
     TestAllReadingWithFutureMessages(write_logs);
 }
+
+TEST_F(SerializedFlushToStateTest, no_dangling_references) {
+    AddChunkWithMessages(true, 0, {"1st", "2nd"});
+    AddChunkWithMessages(true, 0, {"3rd", "4th"});
+
+    auto state = SerializedFlushToState{1, kLogMaskAll};
+    state.InitializeLogs(log_chunks_);
+
+    ASSERT_EQ(log_chunks_[0].size(), 2U);
+    auto first_chunk = log_chunks_[0].begin();
+    auto second_chunk = std::next(first_chunk);
+
+    ASSERT_TRUE(state.HasUnreadLogs());
+    auto first_log = state.PopNextUnreadLog();
+    EXPECT_STREQ(first_log.entry->msg(), "1st");
+    EXPECT_EQ(first_chunk->reader_ref_count(), 1U);
+    EXPECT_EQ(second_chunk->reader_ref_count(), 0U);
+
+    ASSERT_TRUE(state.HasUnreadLogs());
+    auto second_log = state.PopNextUnreadLog();
+    EXPECT_STREQ(second_log.entry->msg(), "2nd");
+    EXPECT_EQ(first_chunk->reader_ref_count(), 1U);
+    EXPECT_EQ(second_chunk->reader_ref_count(), 0U);
+
+    ASSERT_TRUE(state.HasUnreadLogs());
+    auto third_log = state.PopNextUnreadLog();
+    EXPECT_STREQ(third_log.entry->msg(), "3rd");
+    EXPECT_EQ(first_chunk->reader_ref_count(), 0U);
+    EXPECT_EQ(second_chunk->reader_ref_count(), 1U);
+
+    ASSERT_TRUE(state.HasUnreadLogs());
+    auto fourth_log = state.PopNextUnreadLog();
+    EXPECT_STREQ(fourth_log.entry->msg(), "4th");
+    EXPECT_EQ(first_chunk->reader_ref_count(), 0U);
+    EXPECT_EQ(second_chunk->reader_ref_count(), 1U);
+
+    EXPECT_FALSE(state.HasUnreadLogs());
+}

logd/SerializedLogBuffer.cpp

@@ -276,7 +276,6 @@ bool SerializedLogBuffer::FlushTo(
 
     auto& state = reinterpret_cast<SerializedFlushToState&>(abstract_state);
     state.InitializeLogs(logs_);
-    state.CheckForNewLogs();
 
     while (state.HasUnreadLogs()) {
         MinHeapElement top = state.PopNextUnreadLog();
@@ -309,10 +308,6 @@ bool SerializedLogBuffer::FlushTo(
             return false;
         }
         lock.lock();
-
-        // Since we released the log above, buffers that aren't in our min heap may now have had
-        // logs added, so re-check them.
-        state.CheckForNewLogs();
     }
 
     state.set_start(state.start() + 1);

logd/SerializedLogChunk.cpp

@@ -25,14 +25,13 @@ SerializedLogChunk::~SerializedLogChunk() {
 }
 
 void SerializedLogChunk::Compress() {
-    if (compressed_log_.empty()) {
-        CompressionEngine::GetInstance().Compress({contents_.data(), write_offset_},
-                                                  compressed_log_);
+    if (compressed_log_.size() == 0) {
+        CompressionEngine::GetInstance().Compress(contents_, write_offset_, compressed_log_);
         LOG(INFO) << "Compressed Log, buffer max size: " << contents_.size()
                   << " size used: " << write_offset_
                   << " compressed size: " << compressed_log_.size();
     }
-    contents_.resize(0);
+    contents_.Resize(0);
 }
 
 // TODO: Develop a better reference counting strategy to guard against the case where the writer is
@@ -41,7 +40,8 @@ void SerializedLogChunk::IncReaderRefCount() {
     if (++reader_ref_count_ != 1 || writer_active_) {
         return;
     }
-    CompressionEngine::GetInstance().Decompress(compressed_log_, contents_, write_offset_);
+    contents_.Resize(write_offset_);
+    CompressionEngine::GetInstance().Decompress(compressed_log_, contents_);
 }
 
 void SerializedLogChunk::DecReaderRefCount(bool compress) {
@@ -90,7 +90,7 @@ bool SerializedLogChunk::ClearUidLogs(uid_t uid, log_id_t log_id, LogStatistics*
     // Clear the old compressed logs and set write_offset_ appropriately for DecReaderRefCount()
     // to compress the new partially cleared log.
     if (new_write_offset != write_offset_) {
-        compressed_log_.clear();
+        compressed_log_.Resize(0);
         write_offset_ = new_write_offset;
     }
 

logd/SerializedLogChunk.h

@@ -18,14 +18,14 @@
 
 #include <sys/types.h>
 
-#include <vector>
-
 #include "LogWriter.h"
+#include "SerializedData.h"
 #include "SerializedLogEntry.h"
 
 class SerializedLogChunk {
   public:
     explicit SerializedLogChunk(size_t size) : contents_(size) {}
+    SerializedLogChunk(SerializedLogChunk&& other) noexcept = default;
     ~SerializedLogChunk();
 
     void Compress();
@@ -60,13 +60,16 @@ class SerializedLogChunk {
     int write_offset() const { return write_offset_; }
     uint64_t highest_sequence_number() const { return highest_sequence_number_; }
 
+    // Exposed for testing
+    uint32_t reader_ref_count() const { return reader_ref_count_; }
+
   private:
     // The decompressed contents of this log buffer.  Deallocated when the ref_count reaches 0 and
     // writer_active_ is false.
-    std::vector<uint8_t> contents_;
+    SerializedData contents_;
     int write_offset_ = 0;
     uint32_t reader_ref_count_ = 0;
     bool writer_active_ = true;
     uint64_t highest_sequence_number_ = 1;
-    std::vector<uint8_t> compressed_log_;
+    SerializedData compressed_log_;
 };

logd/fuzz/Android.bp

@@ -44,4 +44,7 @@ cc_fuzz {
     srcs: [
         "serialized_log_buffer_fuzzer.cpp",
     ],
+    corpus: [
+        "corpus/logentry_use_after_compress",
+    ]
 }