Cache super bucket ownership decisions when processing bucket DB

Distributor bucket ownership is assigned on a per superbucket basis,
so all buckets with the same superbucket can use the same decision.
The bucket DB is explicitly ordered in such a way that all buckets belonging
to the same superbucket are ordered after each other, so we need only
maintain O(1) extra state for this.

2 files changed, 46 insertions, 12 deletions

diff --git a/storage/src/vespa/storage/distributor/bucketdbupdater.cpp b/storage/src/vespa/storage/distributor/bucketdbupdater.cpp
index 7b2b2561f91..08c8cc59933 100644
--- a/storage/src/vespa/storage/distributor/bucketdbupdater.cpp
+++ b/storage/src/vespa/storage/distributor/bucketdbupdater.cpp
@@ -730,6 +730,23 @@ BucketDBUpdater::BucketListGenerator::process(BucketDatabase::Entry& e)
     return true;
 }
 
+BucketDBUpdater::NodeRemover::NodeRemover(
+        const lib::ClusterState& oldState,
+        const lib::ClusterState& s,
+        uint16_t localIndex,
+        const lib::Distribution& distribution,
+        const char* upStates)
+    : _oldState(oldState),
+      _state(s),
+      _nonOwnedBuckets(),
+      _removedBuckets(),
+      _localIndex(localIndex),
+      _distribution(distribution),
+      _upStates(upStates),
+      _cachedDecisionSuperbucket(UINT64_MAX),
+      _cachedOwned(false)
+{}
+
 void
 BucketDBUpdater::NodeRemover::logRemove(const document::BucketId& bucketId, const char* msg) const
 {
@@ -737,14 +754,35 @@ BucketDBUpdater::NodeRemover::logRemove(const document::BucketId& bucketId, cons
     LOG_BUCKET_OPERATION_NO_LOCK(bucketId, msg);    
 }
 
+namespace {
+
+uint64_t superbucket_from_id(const document::BucketId& id, uint16_t distribution_bits) noexcept {
+    // The n LSBs of the bucket ID contain the superbucket number. Mask off the rest.
+    return id.getRawId() & ~(UINT64_MAX << distribution_bits);
+}
+
+}
+
 bool
 BucketDBUpdater::NodeRemover::distributorOwnsBucket(
         const document::BucketId& bucketId) const
 {
+    // TODO "no distributors available" case is the same for _all_ buckets; cache once in constructor.
+    // TODO "too few bits used" case can be cheaply checked without needing exception
     try {
-        uint16_t distributor(
-                _distribution.getIdealDistributorNode(_state, bucketId, "uim"));
-        if (distributor != _localIndex) {
+        const auto bits = _state.getDistributionBitCount();
+        const auto this_superbucket = superbucket_from_id(bucketId, bits);
+        if (_cachedDecisionSuperbucket == this_superbucket) {
+            if (!_cachedOwned) {
+                logRemove(bucketId, "bucket now owned by another distributor (cached)");
+            }
+            return _cachedOwned;
+        }
+
+        uint16_t distributor = _distribution.getIdealDistributorNode(_state, bucketId, "uim");
+        _cachedDecisionSuperbucket = this_superbucket;
+        _cachedOwned = (distributor == _localIndex);
+        if (!_cachedOwned) {
             logRemove(bucketId, "bucket now owned by another distributor");
             return false;
         }
diff --git a/storage/src/vespa/storage/distributor/bucketdbupdater.h b/storage/src/vespa/storage/distributor/bucketdbupdater.h
index 393e1e2524e..9ff732a09f5 100644
--- a/storage/src/vespa/storage/distributor/bucketdbupdater.h
+++ b/storage/src/vespa/storage/distributor/bucketdbupdater.h
@@ -201,16 +201,9 @@ private:
                     const lib::ClusterState& s,
                     uint16_t localIndex,
                     const lib::Distribution& distribution,
-                    const char* upStates)
-            : _oldState(oldState),
-              _state(s),
-              _nonOwnedBuckets(),
-              _removedBuckets(),
-              _localIndex(localIndex),
-              _distribution(distribution),
-              _upStates(upStates) {}
-
+                    const char* upStates);
         ~NodeRemover() override;
+
         bool process(BucketDatabase::Entry& e) override;
         void logRemove(const document::BucketId& bucketId, const char* msg) const;
         bool distributorOwnsBucket(const document::BucketId&) const;
@@ -233,6 +226,9 @@ private:
         uint16_t _localIndex;
         const lib::Distribution& _distribution;
         const char* _upStates;
+
+        mutable uint64_t _cachedDecisionSuperbucket;
+        mutable bool _cachedOwned;
     };
 
     std::deque<std::pair<framework::MilliSecTime, BucketRequest> > _delayedRequests;