libsnapshot: implement resume buffer

Add resume space to cow v3. Resume buffer goes after header and scratch
space, and is currently set to contain 4 resume points. When AddLabel is
called, the oldest label is replaced with newest one.

Parser will parse up until the last resumable op from a given label.

Test: cow_api_test
Change-Id: Ie072f245721776887d59c96dad296965ad31a5cc

fs_mgr/libsnapshot/include/libsnapshot/cow_format.h

@@ -94,6 +94,16 @@ struct CowHeader {
 
 } __attribute__((packed));
 
+// Resume point structure used for resume buffer
+struct ResumePoint {
+    // monotonically increasing value used by update_engine
+    uint64_t label;
+    // Index of last CowOperation guaranteed to be resumable
+    uint32_t op_index;
+} __attribute__((packed));
+
+static constexpr uint8_t kNumResumePoints = 4;
+
 struct CowHeaderV3 : public CowHeader {
     // Location of sequence buffer in COW.
     uint64_t sequence_buffer_offset;
@@ -245,6 +255,8 @@ std::ostream& operator<<(std::ostream& os, CowOperationV2 const& arg);
 
 std::ostream& operator<<(std::ostream& os, CowOperation const& arg);
 
+std::ostream& operator<<(std::ostream& os, ResumePoint const& arg);
+
 int64_t GetNextOpOffset(const CowOperationV2& op, uint32_t cluster_size);
 int64_t GetNextDataOffset(const CowOperationV2& op, uint32_t cluster_size);
 

fs_mgr/libsnapshot/include/libsnapshot/cow_reader.h

@@ -16,6 +16,7 @@
 
 #include <stdint.h>
 
+#include <deque>
 #include <functional>
 #include <memory>
 #include <optional>
@@ -192,6 +193,5 @@ class CowReader final : public ICowReader {
 // The extra fields will just be filled as 0. V3 header is strictly a superset of v1/v2 header and
 // contains all of the latter's field
 bool ReadCowHeader(android::base::borrowed_fd fd, CowHeaderV3* header);
-
 }  // namespace snapshot
 }  // namespace android

fs_mgr/libsnapshot/libsnapshot_cow/cow_format.cpp

@@ -100,6 +100,11 @@ std::ostream& operator<<(std::ostream& os, CowOperation const& op) {
     return os;
 }
 
+std::ostream& operator<<(std::ostream& os, ResumePoint const& resume_point) {
+    os << "ResumePoint(" << resume_point.label << " , " << resume_point.op_index << ")";
+    return os;
+}
+
 int64_t GetNextOpOffset(const CowOperationV2& op, uint32_t cluster_ops) {
     if (op.type == kCowClusterOp) {
         return op.source;

fs_mgr/libsnapshot/libsnapshot_cow/parser_v3.cpp

@@ -15,8 +15,10 @@
 
 #include <android-base/file.h>
 #include <android-base/logging.h>
+#include <android-base/strings.h>
 
 #include <libsnapshot/cow_format.h>
+#include <libsnapshot/cow_reader.h>
 
 namespace android {
 namespace snapshot {
@@ -36,6 +38,7 @@ bool CowParserV3::Parse(borrowed_fd fd, const CowHeaderV3& header, std::optional
         LOG(ERROR) << "Footer size isn't 0, read " << header_.footer_size;
         return false;
     }
+
     if (header_.op_size != sizeof(CowOperationV3)) {
         LOG(ERROR) << "Operation size unknown, read " << header_.op_size << ", expected "
                    << sizeof(CowOperationV3);
@@ -55,18 +58,54 @@ bool CowParserV3::Parse(borrowed_fd fd, const CowHeaderV3& header, std::optional
         return false;
     }
 
-    return ParseOps(fd, label);
+    std::optional<uint32_t> op_index = header_.op_count;
+    if (label) {
+        if (!ReadResumeBuffer(fd)) {
+            PLOG(ERROR) << "Failed to read resume buffer";
+            return false;
+        }
+        op_index = FindResumeOp(label.value());
+        if (op_index == std::nullopt) {
+            LOG(ERROR) << "failed to get op index from given label: " << label.value();
+            return false;
+        }
+    }
+
+    return ParseOps(fd, op_index.value());
+}
+
+bool CowParserV3::ReadResumeBuffer(borrowed_fd fd) {
+    resume_points_ = std::make_shared<std::vector<ResumePoint>>(header_.resume_buffer_size);
+
+    return android::base::ReadFullyAtOffset(fd, resume_points_->data(),
+                                            header_.resume_buffer_size * sizeof(ResumePoint),
+                                            header_.prefix.header_size + header_.buffer_size);
+}
+
+std::optional<uint32_t> CowParserV3::FindResumeOp(const uint32_t label) {
+    for (auto& resume_point : *resume_points_) {
+        if (resume_point.label == label) {
+            return resume_point.op_index;
+        }
+    }
+    LOG(ERROR) << "failed to find label: " << label << "from following labels";
+    LOG(ERROR) << android::base::Join(*resume_points_, " ");
+
+    return std::nullopt;
 }
 
 off_t CowParserV3::GetDataOffset() const {
-    return sizeof(CowHeaderV3) + header_.buffer_size + header_.op_count_max * sizeof(CowOperation);
+    return sizeof(CowHeaderV3) + header_.buffer_size +
+           header_.resume_buffer_size * sizeof(ResumePoint) +
+           header_.op_count_max * sizeof(CowOperation);
 }
 
-bool CowParserV3::ParseOps(borrowed_fd fd, std::optional<uint64_t> label) {
+bool CowParserV3::ParseOps(borrowed_fd fd, const uint32_t op_index) {
     ops_ = std::make_shared<std::vector<CowOperationV3>>();
-    ops_->resize(header_.op_count);
+    ops_->resize(op_index);
 
-    const off_t offset = header_.prefix.header_size + header_.buffer_size;
+    const off_t offset = header_.prefix.header_size + header_.buffer_size +
+                         header_.resume_buffer_size * sizeof(ResumePoint);
     if (!android::base::ReadFullyAtOffset(fd, ops_->data(), ops_->size() * sizeof(CowOperationV3),
                                           offset)) {
         PLOG(ERROR) << "read ops failed";
@@ -87,7 +126,6 @@ bool CowParserV3::ParseOps(borrowed_fd fd, std::optional<uint64_t> label) {
     // :TODO: sequence buffer & resume buffer follow
     // Once we implement labels, we'll have to discard unused ops and adjust
     // the header as needed.
-    CHECK(!label);
 
     ops_->shrink_to_fit();
 

fs_mgr/libsnapshot/libsnapshot_cow/parser_v3.h

@@ -45,12 +45,16 @@ class CowParserV3 final : public CowParserBase {
     bool Parse(android::base::borrowed_fd fd, const CowHeaderV3& header,
                std::optional<uint64_t> label = {}) override;
     bool Translate(TranslatedCowOps* out) override;
+    std::shared_ptr<std::vector<ResumePoint>> resume_points() const { return resume_points_; }
 
   private:
-    bool ParseOps(android::base::borrowed_fd fd, std::optional<uint64_t> label);
+    bool ParseOps(android::base::borrowed_fd fd, const uint32_t op_index);
+    std::optional<uint32_t> FindResumeOp(const uint32_t label);
     off_t GetDataOffset() const;
     CowHeaderV3 header_ = {};
     std::shared_ptr<std::vector<CowOperationV3>> ops_;
+    bool ReadResumeBuffer(android::base::borrowed_fd fd);
+    std::shared_ptr<std::vector<ResumePoint>> resume_points_;
 };
 
 }  // namespace snapshot

fs_mgr/libsnapshot/libsnapshot_cow/test_v3.cpp

@@ -457,5 +457,30 @@ TEST_F(CowTestV3, GzCompression) {
     iter->Next();
     ASSERT_TRUE(iter->AtEnd());
 }
+
+TEST_F(CowTestV3, ResumePointTest) {
+    CowOptions options;
+    options.op_count_max = 100;
+    auto writer = CreateCowWriter(3, options, GetCowFd());
+
+    ASSERT_TRUE(writer->AddZeroBlocks(0, 15));
+    ASSERT_TRUE(writer->AddLabel(0));
+    ASSERT_TRUE(writer->AddZeroBlocks(15, 15));
+    ASSERT_TRUE(writer->Finalize());
+
+    CowReader reader;
+    ASSERT_TRUE(reader.Parse(cow_->fd));
+
+    auto header = reader.header_v3();
+    ASSERT_EQ(header.op_count, 30);
+
+    CowWriterV3 second_writer(options, GetCowFd());
+    ASSERT_TRUE(second_writer.Initialize(0));
+    ASSERT_TRUE(second_writer.Finalize());
+
+    ASSERT_TRUE(reader.Parse(cow_->fd));
+    header = reader.header_v3();
+    ASSERT_EQ(header.op_count, 15);
+}
 }  // namespace snapshot
 }  // namespace android

fs_mgr/libsnapshot/libsnapshot_cow/writer_v3.cpp

@@ -34,6 +34,7 @@
 #include <zlib.h>
 
 #include <fcntl.h>
+#include <libsnapshot_cow/parser_v3.h>
 #include <linux/fs.h>
 #include <sys/ioctl.h>
 #include <unistd.h>
@@ -77,7 +78,7 @@ void CowWriterV3::SetupHeaders() {
     // WIP: not quite sure how some of these are calculated yet, assuming buffer_size is determined
     // during COW size estimation
     header_.sequence_buffer_offset = 0;
-    header_.resume_buffer_size = 0;
+    header_.resume_buffer_size = kNumResumePoints;
     header_.op_count = 0;
     header_.op_count_max = 0;
     header_.compression_algorithm = kCowCompressNone;
@@ -120,11 +121,14 @@ bool CowWriterV3::Initialize(std::optional<uint64_t> label) {
     if (!InitFd() || !ParseOptions()) {
         return false;
     }
-
-    CHECK(!label.has_value());
-
-    if (!OpenForWrite()) {
-        return false;
+    if (!label) {
+        if (!OpenForWrite()) {
+            return false;
+        }
+    } else {
+        if (!OpenForAppend(*label)) {
+            return false;
+        }
     }
 
     return true;
@@ -159,13 +163,44 @@ bool CowWriterV3::OpenForWrite() {
         }
     }
     header_.op_count_max = options_.op_count_max;
+    resume_points_ = std::make_shared<std::vector<ResumePoint>>();
 
     if (!Sync()) {
         LOG(ERROR) << "Header sync failed";
         return false;
     }
-    next_data_pos_ =
-            sizeof(CowHeaderV3) + header_.buffer_size + header_.op_count_max * sizeof(CowOperation);
+    next_data_pos_ = GetDataOffset();
+    return true;
+}
+
+bool CowWriterV3::OpenForAppend(uint64_t label) {
+    CowHeaderV3 header_v3;
+    if (!ReadCowHeader(fd_, &header_v3)) {
+        LOG(ERROR) << "Couldn't read Cow Header";
+        return false;
+    }
+
+    header_ = header_v3;
+
+    CHECK(label >= 0);
+    CowParserV3 parser;
+    if (!parser.Parse(fd_, header_, label)) {
+        PLOG(ERROR) << "unable to parse with given label: " << label;
+        return false;
+    }
+
+    resume_points_ = parser.resume_points();
+    options_.block_size = header_.block_size;
+    next_data_pos_ = GetDataOffset();
+
+    TranslatedCowOps ops;
+    parser.Translate(&ops);
+    header_.op_count = ops.ops->size();
+
+    for (const auto& op : *ops.ops) {
+        next_data_pos_ += op.data_length;
+    }
+
     return true;
 }
 
@@ -249,9 +284,32 @@ bool CowWriterV3::EmitZeroBlocks(uint64_t new_block_start, uint64_t num_blocks)
 }
 
 bool CowWriterV3::EmitLabel(uint64_t label) {
-    LOG(ERROR) << __LINE__ << " " << __FILE__ << " <- function here should never be called";
-    if (label) return false;
-    return false;
+    // remove all labels greater than this current one. we want to avoid the situation of adding
+    // in
+    // duplicate labels with differing op values
+    auto remove_if_callback = [&](const auto& resume_point) -> bool {
+        if (resume_point.label >= label) return true;
+        return false;
+    };
+    resume_points_->erase(
+            std::remove_if(resume_points_->begin(), resume_points_->end(), remove_if_callback),
+            resume_points_->end());
+
+    resume_points_->push_back({label, header_.op_count});
+    // remove the oldest resume point if resume_buffer is full
+    while (resume_points_->size() > header_.resume_buffer_size) {
+        resume_points_->erase(resume_points_->begin());
+    }
+
+    CHECK_LE(resume_points_->size(), header_.resume_buffer_size);
+
+    if (!android::base::WriteFullyAtOffset(fd_, resume_points_->data(),
+                                           resume_points_->size() * sizeof(ResumePoint),
+                                           GetResumeOffset())) {
+        PLOG(ERROR) << "writing resume buffer failed";
+        return false;
+    }
+    return Sync();
 }
 
 bool CowWriterV3::EmitSequenceData(size_t num_ops, const uint32_t* data) {

fs_mgr/libsnapshot/libsnapshot_cow/writer_v3.h

@@ -43,6 +43,7 @@ class CowWriterV3 : public CowWriterBase {
     void SetupHeaders();
     bool ParseOptions();
     bool OpenForWrite();
+    bool OpenForAppend(uint64_t label);
     bool WriteOperation(const CowOperationV3& op, const void* data = nullptr, size_t size = 0);
     bool EmitBlocks(uint64_t new_block_start, const void* data, size_t size, uint64_t old_block,
                     uint16_t offset, uint8_t type);
@@ -51,8 +52,15 @@ class CowWriterV3 : public CowWriterBase {
     off_t GetOpOffset(uint32_t op_index) const {
         CHECK_LT(op_index, header_.op_count_max);
         return header_.prefix.header_size + header_.buffer_size +
+               (header_.resume_buffer_size * sizeof(ResumePoint)) +
                (op_index * sizeof(CowOperationV3));
     }
+    off_t GetDataOffset() const {
+        return sizeof(CowHeaderV3) + header_.buffer_size +
+               (header_.resume_buffer_size * sizeof(ResumePoint)) +
+               header_.op_count_max * sizeof(CowOperation);
+    }
+    off_t GetResumeOffset() const { return sizeof(CowHeaderV3) + header_.buffer_size; }
 
   private:
     CowHeaderV3 header_{};
@@ -61,6 +69,8 @@ class CowWriterV3 : public CowWriterBase {
     // compressor
     std::unique_ptr<ICompressor> compressor_;
     std::vector<std::unique_ptr<CompressWorker>> compress_threads_;
+    // Resume points contain a laebl + cow_op_index.
+    std::shared_ptr<std::vector<ResumePoint>> resume_points_;
 
     uint64_t next_op_pos_ = 0;
     uint64_t next_data_pos_ = 0;