If there are hickups in the feed chain preventing normal commit rate, we need to prevent lasting huge temporary buffers.

So cap buffers and start from scratch if peaks are detected.

3 files changed, 63 insertions, 16 deletions

diff --git a/searchlib/src/tests/attribute/changevector/changevector_test.cpp b/searchlib/src/tests/attribute/changevector/changevector_test.cpp
index 7bcf519bb18..d7d1a8d2699 100644
--- a/searchlib/src/tests/attribute/changevector/changevector_test.cpp
+++ b/searchlib/src/tests/attribute/changevector/changevector_test.cpp
@@ -5,6 +5,9 @@
 #include <vespa/vespalib/stllike/hash_set.h>
 using namespace search;
 
+using Change = ChangeTemplate<NumericChangeData<long>>;
+using CV = ChangeVectorT<Change>;
+
 template <typename T>
 void verifyStrictOrdering(const T & v) {
     vespalib::hash_set<uint32_t> complete;
@@ -37,8 +40,6 @@ private:
 
 TEST("require insert ordering is preserved for same doc")
 {
-    typedef ChangeTemplate<NumericChangeData<long>> Change;
-    typedef ChangeVectorT<Change> CV;
     CV a;
     a.push_back(Change(Change::NOOP, 7, 1));
     EXPECT_EQUAL(1u, a.size());
@@ -49,8 +50,6 @@ TEST("require insert ordering is preserved for same doc")
 
 TEST("require insert ordering is preserved ")
 {
-    typedef ChangeTemplate<NumericChangeData<long>> Change;
-    typedef ChangeVectorT<Change> CV;
     CV a;
     a.push_back(Change(Change::NOOP, 7, 1));
     EXPECT_EQUAL(1u, a.size());
@@ -63,8 +62,6 @@ TEST("require insert ordering is preserved ")
 
 TEST("require insert ordering is preserved with mix")
 {
-    typedef ChangeTemplate<NumericChangeData<long>> Change;
-    typedef ChangeVectorT<Change> CV;
     CV a;
     a.push_back(Change(Change::NOOP, 7, 1));
     EXPECT_EQUAL(1u, a.size());
@@ -84,8 +81,6 @@ TEST("require insert ordering is preserved with mix")
 }
 
 TEST("require that inserting empty vector does not affect the vector.") {
-    typedef ChangeTemplate<NumericChangeData<long>> Change;
-    typedef ChangeVectorT<Change> CV;
     CV a;
     std::vector<long> v;
     Accessor ac(v);
@@ -93,4 +88,42 @@ TEST("require that inserting empty vector does not affect the vector.") {
     EXPECT_EQUAL(0u, a.size());
 }
 
+TEST("require that we have control over buffer construction size") {
+    CV a;
+    EXPECT_EQUAL(0u, a.size());
+    EXPECT_EQUAL(256u, a.capacity());
+    a.clear();
+    EXPECT_EQUAL(0u, a.size());
+    EXPECT_EQUAL(256u, a.capacity());
+}
+
+TEST("require that buffer can grow some") {
+    CV a;
+    for (size_t i(0); i < 1024; i++) {
+        a.push_back(Change(Change::NOOP, i, i));
+    }
+    EXPECT_EQUAL(1024u, a.size());
+    EXPECT_EQUAL(1024u, a.capacity());
+    a.clear();
+    EXPECT_EQUAL(0u, a.size());
+    EXPECT_EQUAL(1024u, a.capacity());
+}
+
+TEST("require that buffer can grow some, but not unbound") {
+    CV a;
+    for (size_t i(0); i < 1025; i++) {
+        a.push_back(Change(Change::NOOP, i, i));
+    }
+    EXPECT_EQUAL(1025u, a.size());
+    EXPECT_EQUAL(2048u, a.capacity());
+    a.clear();
+    EXPECT_EQUAL(0u, a.size());
+    EXPECT_EQUAL(256u, a.capacity());
+}
+
+TEST("Control Change size") {
+    EXPECT_EQUAL(32u, sizeof(ChangeTemplate<NumericChangeData<long>>));
+    EXPECT_EQUAL(88u, sizeof(ChangeTemplate<StringChangeData>));
+}
+
 TEST_MAIN() { TEST_RUN_ALL(); }
diff --git a/searchlib/src/vespa/searchlib/attribute/changevector.h b/searchlib/src/vespa/searchlib/attribute/changevector.h
index df0c4d839d2..d63ef2e2b35 100644
--- a/searchlib/src/vespa/searchlib/attribute/changevector.h
+++ b/searchlib/src/vespa/searchlib/attribute/changevector.h
@@ -3,6 +3,7 @@
 #pragma once
 
 #include <vespa/searchcommon/common/undefinedvalues.h>
+#include <vespa/vespalib/stllike/allocator.h>
 #include <vector>
 
 namespace vespalib { class MemoryUsage; }
@@ -98,7 +99,7 @@ struct ChangeTemplate : public ChangeBase {
         ChangeBase(type, d, w), _data(v)
     { }
 
-    T          _data;
+    T _data;
 };
 
 template <>
@@ -129,7 +130,7 @@ NumericChangeData<double>::operator<(const NumericChangeData<double> &rhs) const
 template <typename T>
 class ChangeVectorT {
 private:
-    using Vector = std::vector<T>;
+    using Vector = std::vector<T, vespalib::allocator_large<T>>;
 public:
     using const_iterator = typename Vector::const_iterator;
     ChangeVectorT();
@@ -139,6 +140,7 @@ public:
     void push_back(uint32_t doc, Accessor & ac);
     T & back()                   { return _v.back(); }
     size_t size()          const { return _v.size(); }
+    size_t capacity()      const { return _v.capacity(); }
     bool empty()           const { return _v.empty(); }
     void clear();
     class InsertOrder {
@@ -150,7 +152,7 @@ public:
         const Vector &_v;
     };
     class DocIdInsertOrder {
-        using AdjacentDocIds = std::vector<uint64_t>;
+        using AdjacentDocIds = std::vector<uint64_t, vespalib::allocator_large<uint64_t>>;
     public:
         class const_iterator {
         public:
@@ -172,14 +174,14 @@ public:
         const_iterator begin() const { return const_iterator(_v, _adjacent, 0); }
         const_iterator end()   const { return const_iterator(_v, _adjacent, _v.size()); }
     private:
-        const Vector &_v;
-        AdjacentDocIds _adjacent;
+        const Vector   &_v;
+        AdjacentDocIds  _adjacent;
     };
     InsertOrder getInsertOrder() const { return InsertOrder(_v); }
     DocIdInsertOrder getDocIdInsertOrder() const { return DocIdInsertOrder(_v); }
     vespalib::MemoryUsage getMemoryUsage() const;
 private:
-    Vector   _v;
+    Vector _v;
 };
 
 } // namespace search
diff --git a/searchlib/src/vespa/searchlib/attribute/changevector.hpp b/searchlib/src/vespa/searchlib/attribute/changevector.hpp
index 24cb05adf91..f8e4b266cba 100644
--- a/searchlib/src/vespa/searchlib/attribute/changevector.hpp
+++ b/searchlib/src/vespa/searchlib/attribute/changevector.hpp
@@ -13,13 +13,18 @@ namespace {
 // This number is selected to be large enough to hold bursts between commits
 constexpr size_t NUM_ELEMS_TO_RESERVE = 200;
 
+template <typename T>
+constexpr size_t roundUp2inN(size_t elems) {
+    return vespalib::roundUp2inN(elems, sizeof(T));
+}
+
 }
 
 template <typename T>
 ChangeVectorT<T>::ChangeVectorT()
     : _v()
 {
-    _v.reserve(vespalib::roundUp2inN(NUM_ELEMS_TO_RESERVE, sizeof(T)));
+    _v.reserve(roundUp2inN<T>(NUM_ELEMS_TO_RESERVE));
 }
 
 template <typename T>
@@ -28,7 +33,14 @@ ChangeVectorT<T>::~ChangeVectorT() = default;
 template <typename T>
 void
 ChangeVectorT<T>::clear() {
-    _v.clear();
+    if (_v.capacity() > roundUp2inN<T>(NUM_ELEMS_TO_RESERVE * 5)) {
+        // Ensure we do not keep insanely large buffers over time, due to abnormal peaks
+        // caused by hickups else where.
+        _v = Vector();
+        _v.reserve(roundUp2inN<T>(NUM_ELEMS_TO_RESERVE));
+    } else {
+        _v.clear();
+    }
 }
 
 template <typename T>