utils/chunked_managed_vector: Fix corruption in case there is more than one chunk

If reserve() allocates more than one chunk, push_back() should not
work with the last chunk. This can result in items being pushed to the
wrong chunk, breaking internal invariants.

Also, pop_back() should not work with the last chunk. This breaks when
there is more than one chunk.

Currently, the container is only used in the sstable partition index
cache.

Manifests by crashes in sstable reader which touch sstables which have
partition index pages with more than 1638 partition entries.

Introduced in 78e5b9fd85 (4.6.0)

Fixes #10290

Message-Id: <20220407174023.527059-1-tgrabiec@scylladb.com>

test/boost/chunked_managed_vector_test.cc

@@ -12,6 +12,8 @@
 #include <deque>
 #include <random>
 #include "utils/lsa/chunked_managed_vector.hh"
+#include "utils/managed_ref.hh"
+#include "test/lib/log.hh"
 
 #include <boost/range/algorithm/sort.hpp>
 #include <boost/range/algorithm/equal.hpp>
@@ -203,3 +205,64 @@ SEASTAR_TEST_CASE(tests_reserve_partial) {
   });
   return make_ready_future<>();
 }
+
+SEASTAR_TEST_CASE(test_clear_and_release) {
+    region region;
+    allocating_section as;
+
+    with_allocator(region.allocator(), [&] {
+        lsa::chunked_managed_vector<managed_ref<uint64_t>> v;
+
+        for (uint64_t i = 1; i < 4000; ++i) {
+            as(region, [&] {
+                v.emplace_back(make_managed<uint64_t>(i));
+            });
+        }
+
+        v.clear_and_release();
+    });
+
+    return make_ready_future<>();
+}
+
+SEASTAR_TEST_CASE(test_chunk_reserve) {
+    region region;
+    allocating_section as;
+
+    for (auto conf :
+            { // std::make_pair(reserve size, push count)
+                std::make_pair(0, 4000),
+                std::make_pair(100, 4000),
+                std::make_pair(200, 4000),
+                std::make_pair(1000, 4000),
+                std::make_pair(2000, 4000),
+                std::make_pair(3000, 4000),
+                std::make_pair(5000, 4000),
+                std::make_pair(500, 8000),
+                std::make_pair(1000, 8000),
+                std::make_pair(2000, 8000),
+                std::make_pair(8000, 500),
+            })
+    {
+        with_allocator(region.allocator(), [&] {
+            auto [reserve_size, push_count] = conf;
+            testlog.info("Testing reserve({}), {}x emplace_back()", reserve_size, push_count);
+            lsa::chunked_managed_vector<managed_ref<uint64_t>> v;
+            v.reserve(reserve_size);
+            uint64_t seed = rand();
+            for (uint64_t i = 0; i < push_count; ++i) {
+                as(region, [&] {
+                    v.emplace_back(make_managed<uint64_t>(seed + i));
+                    BOOST_REQUIRE(**v.begin() == seed);
+                });
+            }
+            auto v_it = v.begin();
+            for (uint64_t i = 0; i < push_count; ++i) {
+                BOOST_REQUIRE(**v_it++ == seed + i);
+            }
+            v.clear_and_release();
+        });
+    }
+
+    return make_ready_future<>();
+}

utils/lsa/chunked_managed_vector.hh

@@ -60,6 +60,9 @@ private:
             throw std::out_of_range("chunked_managed_vector out of range access");
         }
     }
+    chunk_ptr& back_chunk() {
+        return _chunks[_size / max_chunk_capacity()];
+    }
     static void migrate(T* begin, T* end, managed_vector<T>& result);
 public:
     using value_type = T;
@@ -106,24 +109,24 @@ public:
 
     void push_back(const T& x) {
         reserve_for_push_back();
-        _chunks.back().emplace_back(x);
+        back_chunk().emplace_back(x);
         ++_size;
     }
     void push_back(T&& x) {
         reserve_for_push_back();
-        _chunks.back().emplace_back(std::move(x));
+        back_chunk().emplace_back(std::move(x));
         ++_size;
     }
     template <typename... Args>
     T& emplace_back(Args&&... args) {
         reserve_for_push_back();
-        auto& ret = _chunks.back().emplace_back(std::forward<Args>(args)...);
+        auto& ret = back_chunk().emplace_back(std::forward<Args>(args)...);
         ++_size;
         return ret;
     }
     void pop_back() {
         --_size;
-        _chunks.back().pop_back();
+        back_chunk().pop_back();
     }
     const T& back() const {
         return *addr(_size - 1);