Count moved documents in vespa-redistribute-bm.

8 files changed, 247 insertions, 8 deletions

diff --git a/searchcore/src/apps/vespa-redistribute-bm/vespa_redistribute_bm.cpp b/searchcore/src/apps/vespa-redistribute-bm/vespa_redistribute_bm.cpp
index 0227d9539d2..43346c366c0 100644
--- a/searchcore/src/apps/vespa-redistribute-bm/vespa_redistribute_bm.cpp
+++ b/searchcore/src/apps/vespa-redistribute-bm/vespa_redistribute_bm.cpp
@@ -16,6 +16,7 @@
 #include <vespa/searchcore/bmcluster/bm_node_stats.h>
 #include <vespa/searchcore/bmcluster/bm_node_stats_reporter.h>
 #include <vespa/searchcore/bmcluster/bm_range.h>
+#include <vespa/searchcore/bmcluster/bucket_db_snapshot_vector.h>
 #include <vespa/searchcore/bmcluster/bucket_selector.h>
 #include <vespa/searchcore/bmcluster/calculate_moved_docs_ratio.h>
 #include <vespa/searchcore/bmcluster/estimate_moved_docs_ratio.h>
@@ -178,7 +179,7 @@ class Benchmark {
     void adjust_cluster_state_before_feed();
     void adjust_cluster_state_after_feed();
     void adjust_cluster_state_after_first_redistribution();
-    double estimate_lost_docs();
+    double estimate_lost_unique_docs();
     double estimate_moved_docs();
     void feed();
     std::chrono::duration<double> redistribute();
@@ -318,14 +319,13 @@ Benchmark::redistribute()
 }
 
 double
-Benchmark::estimate_lost_docs()
+Benchmark::estimate_lost_unique_docs()
 {
     switch (_params.get_mode()) {
     case Mode::PERM_CRASH:
     case Mode::TEMP_CRASH:
     {
-        double new_redundancy = std::min(_params.get_redundancy(), _params.get_num_nodes() - _params.get_flip_nodes());
-        auto lost_docs_ratio = EstimateMovedDocsRatio().estimate_lost_docs_base_ratio(_params.get_redundancy(), _params.get_flip_nodes(), _params.get_num_nodes()) * new_redundancy;
+        auto lost_docs_ratio = EstimateMovedDocsRatio().estimate_lost_docs_base_ratio(_params.get_redundancy(), _params.get_flip_nodes(), _params.get_num_nodes());
         return _params.get_documents() * lost_docs_ratio;
     }
     default:
@@ -365,11 +365,15 @@ Benchmark::run()
     _cluster->start(_feed);
     feed();
     LOG(info, "--------------------------------");
+    auto old_snapshot = _cluster->get_bucket_db_snapshots();
     adjust_cluster_state_after_feed();
     auto elapsed = redistribute();
-    double moved_docs = estimate_moved_docs();
-    double lost_docs = estimate_lost_docs();
-    LOG(info, "Redistributed estimated %4.2f docs in %5.3f seconds, %4.2f docs/s, estimated %4.2f lost docs", moved_docs, elapsed.count(), moved_docs / elapsed.count(), lost_docs);
+    double estimated_moved_docs = estimate_moved_docs();
+    double estimated_lost_unique_docs = estimate_lost_unique_docs();
+    auto new_snapshot = _cluster->get_bucket_db_snapshots();
+    uint32_t moved_docs = new_snapshot.count_moved_documents(old_snapshot);
+    uint32_t lost_unique_docs = new_snapshot.count_lost_unique_documents(old_snapshot);
+    LOG(info, "Redistributed (estimated %4.2f) %u docs in %5.3f seconds, %4.2f docs/s, (estimated %4.2f) %u lost unique docs", estimated_moved_docs, moved_docs, elapsed.count(), moved_docs / elapsed.count(), estimated_lost_unique_docs, lost_unique_docs);
     if (_params.get_mode() == Mode::TEMP_CRASH) {
         if (_params.get_use_feed_settle()) {
             LOG(info, "Settling redistribution");
@@ -377,7 +381,7 @@ Benchmark::run()
         }
         adjust_cluster_state_after_first_redistribution();
         elapsed = redistribute();
-        LOG(info, "Cleanup of %4.2f docs in %5.3f seconds, %4.2f docs/s, estimated %4.2f refound docs", moved_docs, elapsed.count(), moved_docs / elapsed.count(), lost_docs);
+        LOG(info, "Cleanup of (estimated %4.2f) %u docs in %5.3f seconds, %4.2f docs/s, (estimated %4.2f) %u refound unique docs", estimated_moved_docs, moved_docs, elapsed.count(), moved_docs / elapsed.count(), estimated_lost_unique_docs, lost_unique_docs);
     }
     _cluster->stop();
 }
diff --git a/searchcore/src/vespa/searchcore/bmcluster/CMakeLists.txt b/searchcore/src/vespa/searchcore/bmcluster/CMakeLists.txt
index 0de021a9514..501c4af447b 100644
--- a/searchcore/src/vespa/searchcore/bmcluster/CMakeLists.txt
+++ b/searchcore/src/vespa/searchcore/bmcluster/CMakeLists.txt
@@ -18,6 +18,8 @@ vespa_add_library(searchcore_bmcluster STATIC
     bm_storage_chain_builder.cpp
     bm_storage_link.cpp
     bm_storage_message_addresses.cpp
+    bucket_db_snapshot.cpp
+    bucket_db_snapshot_vector.cpp
     bucket_info_queue.cpp
     calculate_moved_docs_ratio.cpp
     document_api_message_bus_bm_feed_handler.cpp
diff --git a/searchcore/src/vespa/searchcore/bmcluster/bm_cluster.cpp b/searchcore/src/vespa/searchcore/bmcluster/bm_cluster.cpp
index 9fc0b26fff5..4042b809294 100644
--- a/searchcore/src/vespa/searchcore/bmcluster/bm_cluster.cpp
+++ b/searchcore/src/vespa/searchcore/bmcluster/bm_cluster.cpp
@@ -7,6 +7,7 @@
 #include "bm_message_bus.h"
 #include "bm_node.h"
 #include "bm_node_stats.h"
+#include "bucket_db_snapshot_vector.h"
 #include "document_api_message_bus_bm_feed_handler.h"
 #include "spi_bm_feed_handler.h"
 #include "storage_api_chain_bm_feed_handler.h"
@@ -445,4 +446,11 @@ BmCluster::propagate_cluster_state()
     _cluster_controller->propagate_cluster_state();
 }
 
+BucketDbSnapshotVector
+BmCluster::get_bucket_db_snapshots()
+{
+    auto providers = collect_persistence_providers(_nodes);
+    return BucketDbSnapshotVector(providers, _distribution->get_cluster_state_bundle());
+}
+
 }
diff --git a/searchcore/src/vespa/searchcore/bmcluster/bm_cluster.h b/searchcore/src/vespa/searchcore/bmcluster/bm_cluster.h
index 62ec710c296..9fd1743321e 100644
--- a/searchcore/src/vespa/searchcore/bmcluster/bm_cluster.h
+++ b/searchcore/src/vespa/searchcore/bmcluster/bm_cluster.h
@@ -31,6 +31,7 @@ class BmFeed;
 class BmMessageBus;
 class BmNode;
 class BmNodeStats;
+class BucketDbSnapshotVector;
 class IBmDistribution;
 class IBmFeedHandler;
 
@@ -94,6 +95,7 @@ public:
     std::vector<BmNodeStats> get_node_stats();
     BmDistribution& get_real_distribution() { return *_real_distribution; }
     void propagate_cluster_state();
+    BucketDbSnapshotVector get_bucket_db_snapshots();
 };
 
 }
diff --git a/searchcore/src/vespa/searchcore/bmcluster/bucket_db_snapshot.cpp b/searchcore/src/vespa/searchcore/bmcluster/bucket_db_snapshot.cpp
new file mode 100644
index 00000000000..3ddb1afdb35
--- /dev/null
+++ b/searchcore/src/vespa/searchcore/bmcluster/bucket_db_snapshot.cpp
@@ -0,0 +1,74 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "bucket_db_snapshot.h"
+#include <vespa/persistence/spi/persistenceprovider.h>
+#include <vespa/vespalib/stllike/hash_set.hpp>
+#include <vespa/vespalib/stllike/hash_map.hpp>
+#include <cassert>
+
+using document::BucketId;
+using document::BucketSpace;
+using storage::spi::BucketInfo;
+using storage::spi::PersistenceProvider;
+
+namespace search::bmcluster {
+
+BucketDbSnapshot::BucketDbSnapshot()
+    : _buckets()
+{
+}
+
+BucketDbSnapshot::~BucketDbSnapshot() = default;
+
+void
+BucketDbSnapshot::populate(BucketSpace bucket_space, PersistenceProvider& provider)
+{
+    auto bucket_list = provider.listBuckets(bucket_space);
+    assert(!bucket_list.hasError());
+    for (auto& id : bucket_list.getList()) {
+        auto info = provider.getBucketInfo(storage::spi::Bucket(document::Bucket(bucket_space, id)));
+        assert(!info.hasError());
+        _buckets.insert(std::make_pair(id, info.getBucketInfo()));
+    }
+}
+
+uint32_t
+BucketDbSnapshot::count_new_documents(const BucketDbSnapshot &old) const
+{
+    uint32_t result = 0;
+    for (auto& bucket_id_and_info : _buckets) {
+        auto old_buckets_itr = old._buckets.find(bucket_id_and_info.first);
+        const BucketInfo* old_info = (old_buckets_itr != old._buckets.end()) ? &old_buckets_itr->second : nullptr;
+        auto& new_info = bucket_id_and_info.second;
+        uint32_t new_doc_cnt = new_info.getDocumentCount();
+        uint32_t old_doc_cnt = (old_info != nullptr) ? old_info->getDocumentCount() : 0u;
+        if (new_doc_cnt > old_doc_cnt) {
+            result += (new_doc_cnt - old_doc_cnt);
+        }
+    }
+    return result;
+}
+
+void
+BucketDbSnapshot::populate_bucket_id_set(BucketIdSet& buckets) const
+{
+    for (auto& id_and_info : _buckets) {
+        buckets.insert(id_and_info.first);
+    }
+}
+
+const BucketInfo*
+BucketDbSnapshot::try_get_bucket_info(BucketId bucket_id) const
+{
+    auto buckets_itr = _buckets.find(bucket_id);
+    return (buckets_itr != _buckets.end()) ? &buckets_itr->second : nullptr;
+}
+
+}
+
+namespace vespalib {
+
+template class hash_map<BucketId, BucketInfo, BucketId::hash>;
+template class hash_set<BucketId, BucketId::hash>;
+
+}
diff --git a/searchcore/src/vespa/searchcore/bmcluster/bucket_db_snapshot.h b/searchcore/src/vespa/searchcore/bmcluster/bucket_db_snapshot.h
new file mode 100644
index 00000000000..d750a73f207
--- /dev/null
+++ b/searchcore/src/vespa/searchcore/bmcluster/bucket_db_snapshot.h
@@ -0,0 +1,32 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include <vespa/document/bucket/bucketid.h>
+#include <vespa/document/bucket/bucketspace.h>
+#include <vespa/persistence/spi/bucketinfo.h>
+#include <vespa/vespalib/stllike/hash_map.h>
+#include <vespa/vespalib/stllike/hash_set.h>
+
+namespace storage::spi { struct PersistenceProvider; }
+
+namespace search::bmcluster {
+
+/*
+ * Class representing a snapshot of bucket db below SPI for a single node and a single bucket space
+ */
+class BucketDbSnapshot
+{
+    using BucketInfoMap = vespalib::hash_map<document::BucketId, storage::spi::BucketInfo, document::BucketId::hash>;
+    BucketInfoMap _buckets;
+public:
+    using BucketIdSet = vespalib::hash_set<document::BucketId, document::BucketId::hash>;
+    BucketDbSnapshot();
+    ~BucketDbSnapshot();
+    void populate(document::BucketSpace bucket_space, storage::spi::PersistenceProvider& provider);
+    uint32_t count_new_documents(const BucketDbSnapshot &old) const;
+    void populate_bucket_id_set(BucketIdSet& buckets) const;
+    const storage::spi::BucketInfo* try_get_bucket_info(document::BucketId bucket_id) const;
+};
+
+}
diff --git a/searchcore/src/vespa/searchcore/bmcluster/bucket_db_snapshot_vector.cpp b/searchcore/src/vespa/searchcore/bmcluster/bucket_db_snapshot_vector.cpp
new file mode 100644
index 00000000000..33a5707a763
--- /dev/null
+++ b/searchcore/src/vespa/searchcore/bmcluster/bucket_db_snapshot_vector.cpp
@@ -0,0 +1,91 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "bucket_db_snapshot_vector.h"
+#include <vespa/document/bucket/fixed_bucket_spaces.h>
+#include <vespa/vespalib/stllike/hash_map.hpp>
+#include <vespa/vdslib/state/clusterstate.h>
+#include <vespa/vdslib/state/cluster_state_bundle.h>
+#include <cassert>
+
+using document::BucketSpace;
+using document::FixedBucketSpaces;
+using storage::lib::ClusterStateBundle;
+using storage::lib::Node;
+using storage::lib::NodeType;
+
+namespace search::bmcluster {
+
+namespace {
+
+std::vector<BucketSpace> bucket_spaces = { FixedBucketSpaces::default_space(), FixedBucketSpaces::global_space() };
+
+}
+
+
+BucketDbSnapshotVector::BucketDbSnapshotVector(const std::vector<storage::spi::PersistenceProvider *>& providers, const ClusterStateBundle &cluster_state_bundle)
+    : _snapshots()
+{
+    for (const auto bucket_space : bucket_spaces) {
+        auto &bs_snapshots = _snapshots[bucket_space];
+        bs_snapshots.resize(providers.size());
+        auto cluster_state = *cluster_state_bundle.getDerivedClusterState(bucket_space);
+        uint32_t node_idx = 0;
+        for (const auto &provider : providers) {
+            auto node_state = cluster_state.getNodeState(Node(NodeType::STORAGE, node_idx));
+            if (provider && node_state.getState().oneOf("ur")) {
+                bs_snapshots[node_idx].populate(bucket_space, *provider);
+            }
+            ++node_idx;
+        }
+    }
+}
+
+BucketDbSnapshotVector::~BucketDbSnapshotVector() = default;
+
+uint32_t
+BucketDbSnapshotVector::count_moved_documents(const BucketDbSnapshotVector &old) const
+{
+    uint32_t moved_documents = 0;
+    for (const auto bucket_space : bucket_spaces) {
+        auto& bs_snapshots = _snapshots.find(bucket_space)->second;
+        auto& old_bs_snapshots = old._snapshots.find(bucket_space)->second;
+        assert(bs_snapshots.size() == old_bs_snapshots.size());
+        for (uint32_t node_idx = 0; node_idx < bs_snapshots.size(); ++node_idx) {
+            moved_documents += bs_snapshots[node_idx].count_new_documents(old_bs_snapshots[node_idx]);
+        }
+    }
+    return moved_documents;
+}
+
+uint32_t
+BucketDbSnapshotVector::count_lost_unique_documents(const BucketDbSnapshotVector &old) const
+{
+    uint32_t lost_documents = 0;
+    for (const auto bucket_space : bucket_spaces) {
+        auto& bs_snapshots = _snapshots.find(bucket_space)->second;
+        auto& old_bs_snapshots = old._snapshots.find(bucket_space)->second;
+        BucketIdSet old_buckets;
+        BucketIdSet new_buckets;
+        for (auto &snapshot : old_bs_snapshots) {
+            snapshot.populate_bucket_id_set(old_buckets);
+        }
+        for (auto &snapshot : bs_snapshots) {
+            snapshot.populate_bucket_id_set(new_buckets);
+        }
+        for (auto &old_bucket : old_buckets) {
+            if (new_buckets.find(old_bucket) != new_buckets.end()) {
+                continue;
+            }
+            for (auto &snapshot : old_bs_snapshots) {
+                auto info = snapshot.try_get_bucket_info(old_bucket);
+                if (info != nullptr) {
+                    lost_documents += info->getDocumentCount();
+                    break;
+                }
+            }
+        }
+    }
+    return lost_documents;
+}
+
+}
diff --git a/searchcore/src/vespa/searchcore/bmcluster/bucket_db_snapshot_vector.h b/searchcore/src/vespa/searchcore/bmcluster/bucket_db_snapshot_vector.h
new file mode 100644
index 00000000000..7799ed8fc58
--- /dev/null
+++ b/searchcore/src/vespa/searchcore/bmcluster/bucket_db_snapshot_vector.h
@@ -0,0 +1,26 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include "bucket_db_snapshot.h"
+
+namespace storage::lib { class ClusterStateBundle; }
+
+namespace search::bmcluster {
+
+/*
+ * Class representing snapshots of bucket db below SPI for multiple nodes and bucket spaces.
+ */
+class BucketDbSnapshotVector
+{
+    vespalib::hash_map<document::BucketSpace, std::vector<BucketDbSnapshot>, document::BucketSpace::hash> _snapshots;
+    using BucketIdSet = BucketDbSnapshot::BucketIdSet;
+public:
+    
+    BucketDbSnapshotVector(const std::vector<storage::spi::PersistenceProvider *>& providers, const storage::lib::ClusterStateBundle &cluster_state_bundle);
+    ~BucketDbSnapshotVector();
+    uint32_t count_moved_documents(const BucketDbSnapshotVector &old) const;
+    uint32_t count_lost_unique_documents(const BucketDbSnapshotVector &old) const;
+};
+
+}