From 3869b5ab51003cd57615e7a18d3d439d6e61eb09 Mon Sep 17 00:00:00 2001
From: Avi Kivity <avi@scylladb.com>
Date: Mon, 4 Mar 2019 14:14:32 +0200
Subject: [PATCH] =?UTF-8?q?Merge=20"Fix=20commitlog=20chunks=20overwriting?=
 =?UTF-8?q?=20each=20other"=20from=20Pawe=C5=82?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

"
This series fixes a problem in the commitlog cycle() function that
confused in-memory and on-disk size of chunks it wrote to disk. The
former was used to decide how much data needs to be actually written,
and the latter was used to compute the offset of the next chunk. If two
chunk writes happened concurrently one the one positioned earlier in
the file could corrupt the header of the next one.

Fixes #4231.

Tests: unit(dev), dtest(commitlog_test.py:TestCommitLog.test_commitlog_replay_on_startup,test_commitlog_replay_with_alter_table)
"

* tag 'fix-commitlog-cycle/v1' of https://github.com/pdziepak/scylla:
  commitlog: write the correct buffer size
  utils/fragmented_temporary_buffer_view: add remove suffix

(cherry picked from commit d95dec22d9ec59c9c13d0c376a68a5f515e8c3a4)
---
 db/commitlog/commitlog.cc                 |  2 ++
 tests/fragmented_temporary_buffer_test.cc | 33 +++++++++++++++++++++++
 utils/fragmented_temporary_buffer.hh      | 13 ++++++---
 3 files changed, 45 insertions(+), 3 deletions(-)

diff --git a/db/commitlog/commitlog.cc b/db/commitlog/commitlog.cc
index 6fbed2ae56..4662107213 100644
--- a/db/commitlog/commitlog.cc
+++ b/db/commitlog/commitlog.cc
@@ -689,6 +689,8 @@ public:
         // but all previous write/flush pairs.
         return _pending_ops.run_with_ordered_post_op(rp, [this, size, off, buf = std::move(buf)]() mutable { ///////////////////////////////////////////////////
             auto view = fragmented_temporary_buffer::view(buf);
+            view.remove_suffix(buf.size_bytes() - size);
+            assert(size == view.size_bytes());
             return do_with(off, view, [&] (uint64_t& off, fragmented_temporary_buffer::view& view) {
                 if (view.empty()) {
                     return make_ready_future<>();
diff --git a/tests/fragmented_temporary_buffer_test.cc b/tests/fragmented_temporary_buffer_test.cc
index aa80e5bb42..4710abf338 100644
--- a/tests/fragmented_temporary_buffer_test.cc
+++ b/tests/fragmented_temporary_buffer_test.cc
@@ -133,6 +133,39 @@ SEASTAR_THREAD_TEST_CASE(test_view) {
         data_view.remove_prefix(data_view.size());
         test(frag_view);
 
+        data_view = bytes_view(data);
+        frag_view = fragmented_temporary_buffer::view(frag_buffer);
+
+        frag_view.remove_suffix(sizeof(value2) - 1);
+        data_view.remove_suffix(sizeof(value2) - 1);
+        test(frag_view);
+
+        frag_view.remove_suffix(data_view.size());
+        data_view.remove_suffix(data_view.size());
+        test(frag_view);
+
+        data_view = bytes_view(data);
+        frag_view = fragmented_temporary_buffer::view(frag_buffer);
+
+        frag_view.remove_suffix(sizeof(value2) - 1);
+        data_view.remove_suffix(sizeof(value2) - 1);
+        test(frag_view);
+
+        frag_view.remove_prefix(data_view.size());
+        data_view.remove_prefix(data_view.size());
+        test(frag_view);
+
+        data_view = bytes_view(data);
+        frag_view = fragmented_temporary_buffer::view(frag_buffer);
+
+        frag_view.remove_prefix(sizeof(value2) - 1);
+        data_view.remove_prefix(sizeof(value2) - 1);
+        test(frag_view);
+
+        frag_view.remove_suffix(data_view.size());
+        data_view.remove_suffix(data_view.size());
+        test(frag_view);
+
         data_view = bytes_view(data);
     }
 
diff --git a/utils/fragmented_temporary_buffer.hh b/utils/fragmented_temporary_buffer.hh
index 5a6283458b..7a36f5fa9c 100644
--- a/utils/fragmented_temporary_buffer.hh
+++ b/utils/fragmented_temporary_buffer.hh
@@ -166,21 +166,28 @@ public:
         if (!_total_size) {
             return;
         }
-        _total_size -= n;
         while (n > _current_size) {
+            _total_size -= _current_size;
             n -= _current_size;
             ++_current;
-            _current_size = _current->size();
+            _current_size = std::min(_current->size(), _total_size);
         }
+        _total_size -= n;
         _current_size -= n;
         _current_position = _current->get() + n;
         if (!_current_size && _total_size) {
             ++_current;
-            _current_size = _current->size();
+            _current_size = std::min(_current->size(), _total_size);
             _current_position = _current->get();
         }
     }
 
+    // Invalidates iterators
+    void remove_suffix(size_t n) noexcept {
+        _total_size -= n;
+        _current_size = std::min(_current_size, _total_size);
+    }
+
     bool operator==(const fragmented_temporary_buffer::view& other) const noexcept {
         auto this_it = begin();
         auto other_it = other.begin();