Let bucket maintenance priority queue be FIFO ordered within priority class

This avoids a potential starvation issue caused by the existing
implementation, which is bucket ID ordered within a given priority
class. The latter has the unfortunate effect that frequently reinserting
buckets that sort before buckets that are already in the queue may
starve these from being popped from the queue.

Move to a composite key that first sorts on priority, then on a strictly
increasing sequence number. Add a secondary index into this structure
that allows for lookups on bucket IDs as before.

4 files changed, 126 insertions, 163 deletions

diff --git a/storage/src/tests/distributor/simplebucketprioritydatabasetest.cpp b/storage/src/tests/distributor/simplebucketprioritydatabasetest.cpp
index 129695f3e9e..10454ae9850 100644
--- a/storage/src/tests/distributor/simplebucketprioritydatabasetest.cpp
+++ b/storage/src/tests/distributor/simplebucketprioritydatabasetest.cpp
@@ -6,101 +6,109 @@
 #include <string>
 
 using document::test::makeDocumentBucket;
+using namespace ::testing;
 
 namespace storage::distributor {
 
 using document::BucketId;
 using Priority = MaintenancePriority;
 
-TEST(SimpleBucketPriorityDatabaseTest, iterator_range_is_equal_on_empty_database) {
-    SimpleBucketPriorityDatabase queue;
-    auto begin = queue.begin();
-    auto end = queue.end();
+struct SimpleBucketPriorityDatabaseTest : Test {
+    SimpleBucketPriorityDatabase _queue;
+};
+
+TEST_F(SimpleBucketPriorityDatabaseTest, iterator_range_is_equal_on_empty_database) {
+    auto begin = _queue.begin();
+    auto end = _queue.end();
 
     EXPECT_TRUE(begin == end);
     EXPECT_TRUE(begin == begin);
     EXPECT_TRUE(end == end);
 }
 
-TEST(SimpleBucketPriorityDatabaseTest, can_get_prioritized_bucket) {
-    SimpleBucketPriorityDatabase queue;
-
+TEST_F(SimpleBucketPriorityDatabaseTest, can_get_prioritized_bucket) {
     PrioritizedBucket lowPriBucket(makeDocumentBucket(BucketId(16, 1234)), Priority::VERY_LOW);
-    queue.setPriority(lowPriBucket);
+    _queue.setPriority(lowPriBucket);
 
-    PrioritizedBucket highest(*queue.begin());
+    PrioritizedBucket highest(*_queue.begin());
     EXPECT_EQ(lowPriBucket, highest);
 }
 
-TEST(SimpleBucketPriorityDatabaseTest, iterate_over_multiple_priorities) {
-    SimpleBucketPriorityDatabase queue;
-
+TEST_F(SimpleBucketPriorityDatabaseTest, iterate_over_multiple_priorities) {
     PrioritizedBucket lowPriBucket(makeDocumentBucket(BucketId(16, 1234)), Priority::LOW);
     PrioritizedBucket highPriBucket(makeDocumentBucket(BucketId(16, 4321)), Priority::HIGH);
-    queue.setPriority(lowPriBucket);
-    queue.setPriority(highPriBucket);
+    _queue.setPriority(lowPriBucket);
+    _queue.setPriority(highPriBucket);
 
-    auto iter = queue.begin();
+    auto iter = _queue.begin();
     ASSERT_EQ(highPriBucket, *iter);
     ++iter;
-    ASSERT_TRUE(iter != queue.end());
+    ASSERT_TRUE(iter != _queue.end());
     ASSERT_EQ(lowPriBucket, *iter);
     ++iter;
-    ASSERT_TRUE(iter == queue.end());
+    ASSERT_TRUE(iter == _queue.end());
 }
 
-TEST(SimpleBucketPriorityDatabaseTest, multiple_set_priority_for_one_bucket) {
-    SimpleBucketPriorityDatabase queue;
-
+TEST_F(SimpleBucketPriorityDatabaseTest, multiple_set_priority_for_one_bucket) {
     PrioritizedBucket lowPriBucket(makeDocumentBucket(BucketId(16, 1234)), Priority::LOW);
     PrioritizedBucket highPriBucket(makeDocumentBucket(BucketId(16, 1234)), Priority::HIGH);
 
-    queue.setPriority(lowPriBucket);
-    queue.setPriority(highPriBucket);
+    _queue.setPriority(lowPriBucket);
+    _queue.setPriority(highPriBucket);
 
-    auto iter = queue.begin();
+    auto iter = _queue.begin();
     ASSERT_EQ(highPriBucket, *iter);
     ++iter;
-    ASSERT_TRUE(iter == queue.end());
+    ASSERT_TRUE(iter == _queue.end());
 }
 
-TEST(SimpleBucketPriorityDatabaseTest, no_maintenance_needed_clears_bucket_from_database) {
-    SimpleBucketPriorityDatabase queue;
-
+TEST_F(SimpleBucketPriorityDatabaseTest, no_maintenance_needed_clears_bucket_from_database) {
     PrioritizedBucket highPriBucket(makeDocumentBucket(BucketId(16, 1234)), Priority::HIGH);
     PrioritizedBucket noPriBucket(makeDocumentBucket(BucketId(16, 1234)),
                                   Priority::NO_MAINTENANCE_NEEDED);
-    queue.setPriority(highPriBucket);
-    queue.setPriority(noPriBucket);
+    _queue.setPriority(highPriBucket);
+    _queue.setPriority(noPriBucket);
 
-    auto iter = queue.begin();
-    ASSERT_TRUE(iter == queue.end());
+    auto iter = _queue.begin();
+    ASSERT_TRUE(iter == _queue.end());
 }
 
-TEST(SimpleBucketPriorityDatabaseTest, iterate_over_multiple_buckets_with_multiple_priorities) {
-    SimpleBucketPriorityDatabase queue;
-
+TEST_F(SimpleBucketPriorityDatabaseTest, iterate_over_multiple_buckets_with_multiple_priorities) {
     PrioritizedBucket lowPriBucket1(makeDocumentBucket(BucketId(16, 1)), Priority::LOW);
     PrioritizedBucket lowPriBucket2(makeDocumentBucket(BucketId(16, 2)), Priority::LOW);
     PrioritizedBucket mediumPriBucket(makeDocumentBucket(BucketId(16, 3)), Priority::MEDIUM);
     PrioritizedBucket highPriBucket1(makeDocumentBucket(BucketId(16, 4)), Priority::HIGH);
     PrioritizedBucket highPriBucket2(makeDocumentBucket(BucketId(16, 5)), Priority::HIGH);
 
-    queue.setPriority(highPriBucket1);
-    queue.setPriority(lowPriBucket2);
-    queue.setPriority(mediumPriBucket);
-    queue.setPriority(highPriBucket2);
-    queue.setPriority(lowPriBucket1);
+    _queue.setPriority(highPriBucket1);
+    _queue.setPriority(lowPriBucket2);
+    _queue.setPriority(mediumPriBucket);
+    _queue.setPriority(highPriBucket2);
+    _queue.setPriority(lowPriBucket1);
 
-    auto iter = queue.begin();
+    auto iter = _queue.begin();
     PrioritizedBucket lastBucket(makeDocumentBucket(BucketId()), Priority::PRIORITY_LIMIT);
     for (int i = 0; i < 5; ++i) {
-        ASSERT_TRUE(iter != queue.end());
+        ASSERT_TRUE(iter != _queue.end());
         ASSERT_FALSE(iter->moreImportantThan(lastBucket));
         lastBucket = *iter;
         ++iter;
     }
-    ASSERT_TRUE(iter == queue.end());
+    ASSERT_TRUE(iter == _queue.end());
+}
+
+TEST_F(SimpleBucketPriorityDatabaseTest, buckets_within_same_priority_class_are_fifo_ordered) {
+    // We want FIFO order (2, 1) within the same priority class, not bucket ID order (1, 2).
+    PrioritizedBucket first_bucket(makeDocumentBucket(BucketId(16, 2)), Priority::LOW);
+    PrioritizedBucket second_bucket(makeDocumentBucket(BucketId(16, 1)), Priority::LOW);
+
+    _queue.setPriority(first_bucket);
+    _queue.setPriority(second_bucket);
+
+    auto iter = _queue.begin();
+    EXPECT_EQ(first_bucket, *iter);
+    ++iter;
+    EXPECT_EQ(second_bucket, *iter);
 }
 
 }
diff --git a/storage/src/vespa/storage/distributor/maintenance/bucketprioritydatabase.h b/storage/src/vespa/storage/distributor/maintenance/bucketprioritydatabase.h
index c19268d461a..4c193492f43 100644
--- a/storage/src/vespa/storage/distributor/maintenance/bucketprioritydatabase.h
+++ b/storage/src/vespa/storage/distributor/maintenance/bucketprioritydatabase.h
@@ -13,11 +13,9 @@ protected:
     class ConstIteratorImpl
     {
     public:
-        virtual ~ConstIteratorImpl() { }
+        virtual ~ConstIteratorImpl() = default;
         virtual void increment() = 0;
-
         virtual bool equal(const ConstIteratorImpl& other) const = 0;
-
         virtual PrioritizedBucket dereference() const = 0;
     };
 
@@ -33,13 +31,13 @@ public:
     {
         ConstIteratorImplPtr _impl;
     public:
-        ConstIterator(ConstIteratorImplPtr impl)
+        explicit ConstIterator(ConstIteratorImplPtr impl)
             : _impl(std::move(impl))
         {}
         ConstIterator(const ConstIterator &) = delete;
         ConstIterator(ConstIterator &&) = default;
 
-        virtual ~ConstIterator() {}
+        virtual ~ConstIterator() = default;
     private:
         friend class boost::iterator_core_access;
 
@@ -61,9 +59,7 @@ public:
     virtual ~BucketPriorityDatabase() = default;
     
     virtual const_iterator begin() const = 0;
-
     virtual const_iterator end() const  = 0;
-
     virtual void setPriority(const PrioritizedBucket&) = 0;
 };
 
diff --git a/storage/src/vespa/storage/distributor/maintenance/simplebucketprioritydatabase.cpp b/storage/src/vespa/storage/distributor/maintenance/simplebucketprioritydatabase.cpp
index 547cf26583a..1dc0ff15e81 100644
--- a/storage/src/vespa/storage/distributor/maintenance/simplebucketprioritydatabase.cpp
+++ b/storage/src/vespa/storage/distributor/maintenance/simplebucketprioritydatabase.cpp
@@ -1,22 +1,28 @@
 // Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 
 #include "simplebucketprioritydatabase.h"
-#include <iostream>
+#include <vespa/vespalib/stllike/hash_map.hpp>
+#include <ostream>
 #include <sstream>
 
 namespace storage::distributor {
 
+SimpleBucketPriorityDatabase::SimpleBucketPriorityDatabase()
+    : _pri_fifo_buckets(),
+      _bucket_to_pri_iterators(),
+      _fifo_seq_num(0)
+{
+}
+
 SimpleBucketPriorityDatabase::~SimpleBucketPriorityDatabase() = default;
 
 void
 SimpleBucketPriorityDatabase::clearAllEntriesForBucket(const document::Bucket &bucket)
 {
-    for (PriorityMap::iterator priIter(_prioritizedBuckets.begin()),
-             priEnd(_prioritizedBuckets.end());
-         priIter != priEnd;
-         ++priIter)
-    {
-        priIter->second.erase(bucket);
+    auto maybe_iter = _bucket_to_pri_iterators.find(bucket);
+    if (maybe_iter != _bucket_to_pri_iterators.end()) {
+        _pri_fifo_buckets.erase(maybe_iter->second);
+        _bucket_to_pri_iterators.erase(maybe_iter);
     }
 }
 
@@ -25,112 +31,57 @@ SimpleBucketPriorityDatabase::setPriority(const PrioritizedBucket& bucket)
 {
     clearAllEntriesForBucket(bucket.getBucket());
     if (bucket.requiresMaintenance()) {
-        _prioritizedBuckets[bucket.getPriority()].insert(bucket.getBucket());
-    }
-}
-
-void
-SimpleBucketPriorityDatabase::SimpleConstIteratorImpl::initializeBucketIterToFirstAvailableEntry()
-{
-    _bucketIter = _priorityIter->second.begin();
-    if (currentPriorityAtEnd()) {
-        increment();
-    }
-}
-
-bool
-SimpleBucketPriorityDatabase::SimpleConstIteratorImpl::atEnd() const
-{
-    return _priorityIter == _priorityEnd;
-}
-
-void
-SimpleBucketPriorityDatabase::SimpleConstIteratorImpl::stepWithinCurrentPriority()
-{
-    ++_bucketIter;
-}
-
-bool
-SimpleBucketPriorityDatabase::SimpleConstIteratorImpl::currentPriorityAtEnd() const
-{
-    return _bucketIter == _priorityIter->second.end();
-}
-
-void
-SimpleBucketPriorityDatabase::SimpleConstIteratorImpl::stepToNextPriority()
-{
-    ++_priorityIter;
-    if (atEnd()) {
-        return;
-    }
-    _bucketIter = _priorityIter->second.begin();
-}
-
-void
-SimpleBucketPriorityDatabase::SimpleConstIteratorImpl::step()
-{
-    if (currentPriorityAtEnd()) {
-        stepToNextPriority();
-    } else {
-        stepWithinCurrentPriority();
+        auto pri_insert_res = _pri_fifo_buckets.emplace(PriFifoCompositeKey(bucket.getPriority(), _fifo_seq_num),
+                                                        bucket.getBucket());
+        assert(pri_insert_res.second);
+        ++_fifo_seq_num;
+        auto inv_insert_res = _bucket_to_pri_iterators.insert(std::make_pair(bucket.getBucket(), pri_insert_res.first));
+        assert(inv_insert_res.second);
     }
 }
 
 void
-SimpleBucketPriorityDatabase::SimpleConstIteratorImpl::increment()
+SimpleBucketPriorityDatabase::PriFifoMappingConstIteratorImpl::increment()
 {
-    while (!atEnd()) {
-        step();
-        if (atEnd() || !currentPriorityAtEnd()) {
-            break;
-        }
+    if (_pri_fifo_iter != _pri_fifo_end) {
+        ++_pri_fifo_iter;
     }
 }
 
 bool
-SimpleBucketPriorityDatabase::SimpleConstIteratorImpl::equal(const ConstIteratorImpl& otherBase) const
+SimpleBucketPriorityDatabase::PriFifoMappingConstIteratorImpl::equal(const ConstIteratorImpl& other) const
 {
-    const SimpleConstIteratorImpl& other(
-            static_cast<const SimpleConstIteratorImpl&>(otherBase));
-    if (_priorityIter != other._priorityIter) {
-        return false;
-    }
-    if (atEnd()) {
-        return true;
-    }
-    return _bucketIter == other._bucketIter;
+    auto& typed_other = dynamic_cast<const PriFifoMappingConstIteratorImpl&>(other);
+    return (_pri_fifo_iter == typed_other._pri_fifo_iter);
 }
 
 PrioritizedBucket
-SimpleBucketPriorityDatabase::SimpleConstIteratorImpl::dereference() const
+SimpleBucketPriorityDatabase::PriFifoMappingConstIteratorImpl::dereference() const
 {
-    return PrioritizedBucket(*_bucketIter, _priorityIter->first);
+    assert(_pri_fifo_iter != _pri_fifo_end);
+    return {_pri_fifo_iter->second, _pri_fifo_iter->first._pri};
 }
 
 SimpleBucketPriorityDatabase::const_iterator
 SimpleBucketPriorityDatabase::begin() const
 {
-    return const_iterator(ConstIteratorImplPtr(new SimpleConstIteratorImpl(
-                            _prioritizedBuckets.rbegin(),
-                            _prioritizedBuckets.rend())));
+    return const_iterator(std::make_unique<PriFifoMappingConstIteratorImpl>(
+            _pri_fifo_buckets.begin(), _pri_fifo_buckets.end()));
 }
 
 SimpleBucketPriorityDatabase::const_iterator
 SimpleBucketPriorityDatabase::end() const
 {
-    return const_iterator(ConstIteratorImplPtr(new SimpleConstIteratorImpl(
-                            _prioritizedBuckets.rend(),
-                            _prioritizedBuckets.rend())));
+    return const_iterator(std::make_unique<PriFifoMappingConstIteratorImpl>(
+            _pri_fifo_buckets.end(), _pri_fifo_buckets.end()));
 }
 
 std::string
 SimpleBucketPriorityDatabase::toString() const
 {
     std::ostringstream ss;
-    const_iterator i(begin());
-    const_iterator e(end());
-    for (; i != e; ++i) {
-        ss << *i << '\n';
+    for (const auto& e : *this) {
+        ss << e << '\n';
     }
     return ss.str();
 }
diff --git a/storage/src/vespa/storage/distributor/maintenance/simplebucketprioritydatabase.h b/storage/src/vespa/storage/distributor/maintenance/simplebucketprioritydatabase.h
index ff5ebd9c6bb..aa2605e94ca 100644
--- a/storage/src/vespa/storage/distributor/maintenance/simplebucketprioritydatabase.h
+++ b/storage/src/vespa/storage/distributor/maintenance/simplebucketprioritydatabase.h
@@ -2,6 +2,7 @@
 #pragma once
 
 #include "bucketprioritydatabase.h"
+#include <vespa/vespalib/stllike/hash_map.h>
 #include <set>
 #include <map>
 
@@ -10,45 +11,50 @@ namespace storage::distributor {
 class SimpleBucketPriorityDatabase : public BucketPriorityDatabase
 {
 public:
-    virtual ~SimpleBucketPriorityDatabase();
+    SimpleBucketPriorityDatabase();
+    ~SimpleBucketPriorityDatabase() override;
     using Priority = PrioritizedBucket::Priority;
 
-    virtual void setPriority(const PrioritizedBucket&) override;
-    virtual const_iterator begin() const override;
-    virtual const_iterator end() const override;
+    void setPriority(const PrioritizedBucket&) override;
+    const_iterator begin() const override;
+    const_iterator end() const override;
 
     std::string toString() const;
 
 private:
-    using BucketSet   = std::set<document::Bucket>;
-    using PriorityMap = std::map<Priority, BucketSet>;
-
-    class SimpleConstIteratorImpl : public ConstIteratorImpl
-    {
-        PriorityMap::const_reverse_iterator _priorityIter;
-        PriorityMap::const_reverse_iterator _priorityEnd;
-        BucketSet::const_iterator _bucketIter;
-    public:
-        SimpleConstIteratorImpl(PriorityMap::const_reverse_iterator first,
-                                PriorityMap::const_reverse_iterator end)
-            : _priorityIter(first),
-              _priorityEnd(end),
-              _bucketIter()
-        {
-            if (!atEnd()) {
-                initializeBucketIterToFirstAvailableEntry();
+    struct PriFifoCompositeKey {
+        Priority _pri;
+        uint64_t _seq_num;
+
+        constexpr PriFifoCompositeKey() noexcept : _pri(Priority::VERY_LOW), _seq_num(0) {}
+        constexpr PriFifoCompositeKey(Priority pri, uint64_t seq_num) noexcept
+            : _pri(pri),
+              _seq_num(seq_num)
+        {}
+
+        constexpr bool operator<(const PriFifoCompositeKey& rhs) const noexcept {
+            if (_pri != rhs._pri) {
+                // Unlike StorageAPI priorities, MaintenancePriority is higher value == higher priority
+                return (_pri > rhs._pri);
             }
+            return _seq_num < rhs._seq_num;
         }
-        SimpleConstIteratorImpl(const SimpleConstIteratorImpl&) = delete;
-        SimpleConstIteratorImpl& operator=(const SimpleConstIteratorImpl&) = delete;
-    private:
-        void initializeBucketIterToFirstAvailableEntry();
+    };
+
+    using PriFifoBucketMap = std::map<PriFifoCompositeKey, document::Bucket>;
+    // Important: the map iterator instances MUST be stable in the presence of other inserts/erases!
+    using BucketToPriIteratorMap = vespalib::hash_map<document::Bucket, PriFifoBucketMap::iterator, document::Bucket::hash>;
 
-        bool atEnd() const;
-        void stepWithinCurrentPriority();
-        bool currentPriorityAtEnd() const;
-        void stepToNextPriority();
-        void step();
+    class PriFifoMappingConstIteratorImpl final : public ConstIteratorImpl {
+        PriFifoBucketMap::const_iterator _pri_fifo_iter;
+        PriFifoBucketMap::const_iterator _pri_fifo_end;
+    public:
+        PriFifoMappingConstIteratorImpl(PriFifoBucketMap::const_iterator pri_fifo_iter,
+                                        PriFifoBucketMap::const_iterator pri_fifo_end)
+            : _pri_fifo_iter(pri_fifo_iter),
+              _pri_fifo_end(pri_fifo_end)
+        {}
+        ~PriFifoMappingConstIteratorImpl() override = default;
 
         void increment() override;
         bool equal(const ConstIteratorImpl& other) const override;
@@ -57,7 +63,9 @@ private:
 
     void clearAllEntriesForBucket(const document::Bucket &bucket);
 
-    PriorityMap _prioritizedBuckets;
+    PriFifoBucketMap       _pri_fifo_buckets;
+    BucketToPriIteratorMap _bucket_to_pri_iterators;
+    uint64_t               _fifo_seq_num;
 };
 
 }