Add initial multi stripe support to access guard

Still missing functionality for:
 - Merging bucket entries across stripes
 - Aggregating pending operation stats across stripes

4 files changed, 103 insertions, 34 deletions

diff --git a/storage/src/tests/distributor/CMakeLists.txt b/storage/src/tests/distributor/CMakeLists.txt
index d808acc0d3a..f43280a5b44 100644
--- a/storage/src/tests/distributor/CMakeLists.txt
+++ b/storage/src/tests/distributor/CMakeLists.txt
@@ -25,6 +25,7 @@ vespa_add_executable(storage_distributor_gtest_runner_app TEST
     maintenanceschedulertest.cpp
     mergelimitertest.cpp
     mergeoperationtest.cpp
+    multi_thread_stripe_access_guard_test.cpp
     nodeinfotest.cpp
     nodemaintenancestatstrackertest.cpp
     operation_sequencer_test.cpp
diff --git a/storage/src/tests/distributor/distributor_stripe_pool_test.cpp b/storage/src/tests/distributor/distributor_stripe_pool_test.cpp
index 00f6349b218..fb7c446a781 100644
--- a/storage/src/tests/distributor/distributor_stripe_pool_test.cpp
+++ b/storage/src/tests/distributor/distributor_stripe_pool_test.cpp
@@ -1,6 +1,6 @@
 // Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+#include "mock_tickable_stripe.h"
 #include <vespa/storage/distributor/distributor_stripe_pool.h>
-#include <vespa/storage/distributor/tickable_stripe.h>
 #include <vespa/vespalib/gtest/gtest.h>
 #include <vespa/vespalib/util/time.h>
 #include <atomic>
@@ -48,39 +48,6 @@ struct DistributorStripePoolThreadingTest : Test {
     }
 };
 
-struct MockTickableStripe : TickableStripe {
-    bool tick() override { abort(); }
-    void flush_and_close() override { abort(); }
-    void update_total_distributor_config(std::shared_ptr<const DistributorConfiguration>) override { abort(); }
-    void update_distribution_config(const BucketSpaceDistributionConfigs&) override { abort(); }
-    void set_pending_cluster_state_bundle(const lib::ClusterStateBundle&) override { abort(); }
-    void clear_pending_cluster_state_bundle() override { abort(); }
-    void enable_cluster_state_bundle(const lib::ClusterStateBundle&) override { abort(); }
-    void notify_distribution_change_enabled() override { abort(); }
-    PotentialDataLossReport remove_superfluous_buckets(document::BucketSpace, const lib::ClusterState&, bool) override {
-        abort();
-    }
-    void merge_entries_into_db(document::BucketSpace,
-                               api::Timestamp,
-                               const lib::Distribution&,
-                               const lib::ClusterState&,
-                               const char*,
-                               const std::unordered_set<uint16_t>&,
-                               const std::vector<dbtransition::Entry>&) override
-    {
-        abort();
-    }
-    void update_read_snapshot_before_db_pruning() override { abort(); }
-    void update_read_snapshot_after_db_pruning(const lib::ClusterStateBundle&) override { abort(); }
-    void update_read_snapshot_after_activation(const lib::ClusterStateBundle&) override { abort(); }
-    void clear_read_only_bucket_repo_databases() override { abort(); }
-
-    void report_bucket_db_status(document::BucketSpace, std::ostream&) const override { abort(); }
-    StripeAccessGuard::PendingOperationStats pending_operation_stats() const override { abort(); }
-    void report_single_bucket_requests(vespalib::xml::XmlOutputStream&) const override { abort(); }
-    void report_delayed_single_bucket_requests(vespalib::xml::XmlOutputStream&) const override { abort(); }
-};
-
 // Optimistic invariant checker that cannot prove correctness, but will hopefully
 // make tests scream if something is obviously incorrect.
 struct ParkingInvariantCheckingMockStripe : MockTickableStripe {
diff --git a/storage/src/tests/distributor/mock_tickable_stripe.h b/storage/src/tests/distributor/mock_tickable_stripe.h
new file mode 100644
index 00000000000..532bef5e50d
--- /dev/null
+++ b/storage/src/tests/distributor/mock_tickable_stripe.h
@@ -0,0 +1,42 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+#pragma once
+
+#include <vespa/storage/distributor/tickable_stripe.h>
+#include <cstdlib>
+
+namespace storage::distributor {
+
+struct MockTickableStripe : TickableStripe {
+    bool tick() override { abort(); }
+    void flush_and_close() override { abort(); }
+    void update_total_distributor_config(std::shared_ptr<const DistributorConfiguration>) override { abort(); }
+    void update_distribution_config(const BucketSpaceDistributionConfigs&) override { abort(); }
+    void set_pending_cluster_state_bundle(const lib::ClusterStateBundle&) override { abort(); }
+    void clear_pending_cluster_state_bundle() override { abort(); }
+    void enable_cluster_state_bundle(const lib::ClusterStateBundle&) override { abort(); }
+    void notify_distribution_change_enabled() override { abort(); }
+    PotentialDataLossReport remove_superfluous_buckets(document::BucketSpace, const lib::ClusterState&, bool) override {
+        abort();
+    }
+    void merge_entries_into_db(document::BucketSpace,
+                               api::Timestamp,
+                               const lib::Distribution&,
+                               const lib::ClusterState&,
+                               const char*,
+                               const std::unordered_set<uint16_t>&,
+                               const std::vector<dbtransition::Entry>&) override
+    {
+        abort();
+    }
+    void update_read_snapshot_before_db_pruning() override { abort(); }
+    void update_read_snapshot_after_db_pruning(const lib::ClusterStateBundle&) override { abort(); }
+    void update_read_snapshot_after_activation(const lib::ClusterStateBundle&) override { abort(); }
+    void clear_read_only_bucket_repo_databases() override { abort(); }
+
+    void report_bucket_db_status(document::BucketSpace, std::ostream&) const override { abort(); }
+    StripeAccessGuard::PendingOperationStats pending_operation_stats() const override { abort(); }
+    void report_single_bucket_requests(vespalib::xml::XmlOutputStream&) const override { abort(); }
+    void report_delayed_single_bucket_requests(vespalib::xml::XmlOutputStream&) const override { abort(); }
+};
+
+}
diff --git a/storage/src/tests/distributor/multi_thread_stripe_access_guard_test.cpp b/storage/src/tests/distributor/multi_thread_stripe_access_guard_test.cpp
new file mode 100644
index 00000000000..8513186d1e1
--- /dev/null
+++ b/storage/src/tests/distributor/multi_thread_stripe_access_guard_test.cpp
@@ -0,0 +1,59 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+#include "mock_tickable_stripe.h"
+#include <vespa/document/bucket/fixed_bucket_spaces.h>
+#include <vespa/storage/distributor/distributor_stripe_pool.h>
+#include <vespa/storage/distributor/multi_threaded_stripe_access_guard.h>
+#include <vespa/vdslib/state/clusterstate.h>
+#include <vespa/vespalib/gtest/gtest.h>
+
+using namespace ::testing;
+
+namespace storage::distributor {
+
+struct AggregationTestingMockTickableStripe : MockTickableStripe {
+    PotentialDataLossReport report;
+
+    PotentialDataLossReport remove_superfluous_buckets(document::BucketSpace, const lib::ClusterState&, bool) override {
+        return report;
+    }
+
+    bool tick() override {
+        return false;
+    }
+};
+
+struct MultiThreadedStripeAccessGuardTest : Test {
+    DistributorStripePool                _pool;
+    MultiThreadedStripeAccessor          _accessor;
+    AggregationTestingMockTickableStripe _stripe1;
+    AggregationTestingMockTickableStripe _stripe2;
+    AggregationTestingMockTickableStripe _stripe3;
+
+    MultiThreadedStripeAccessGuardTest()
+        : _pool(),
+          _accessor(_pool)
+    {}
+
+    ~MultiThreadedStripeAccessGuardTest() {
+        _pool.stop_and_join();
+    }
+
+    void start_pool_with_stripes() {
+        _pool.start({{&_stripe1, &_stripe2, &_stripe3}});
+    }
+};
+
+TEST_F(MultiThreadedStripeAccessGuardTest, remove_superfluous_buckets_aggregates_reports_across_stripes) {
+    _stripe1.report = PotentialDataLossReport(20, 100);
+    _stripe2.report = PotentialDataLossReport(5,  200);
+    _stripe3.report = PotentialDataLossReport(7,  350);
+    start_pool_with_stripes();
+
+    auto guard = _accessor.rendezvous_and_hold_all();
+    auto report = guard->remove_superfluous_buckets(document::FixedBucketSpaces::default_space(),
+                                                    lib::ClusterState(), false);
+    EXPECT_EQ(report.buckets, 32);
+    EXPECT_EQ(report.documents, 650);
+}
+
+}