Add --mbus-distributor-node-max-pending-count option to vespa-redistribute-bm.

5 files changed, 15 insertions, 1 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 e1cb6df1252..b957226dd8a 100644
--- a/searchcore/src/apps/vespa-redistribute-bm/vespa_redistribute_bm.cpp
+++ b/searchcore/src/apps/vespa-redistribute-bm/vespa_redistribute_bm.cpp
@@ -481,6 +481,7 @@ App::usage()
         "[--max-merge-queue-size max-merge-queue-size]\n"
         "[--max-pending max-pending]\n"
         "[--max-pending-idealstate-operations max-pending-idealstate-operations]\n"
+        "[--mbus-distributor-node-max-pending-count] count\n"
         "[--mode [grow, shrink, perm-crash, temp-crash, replace]\n"
         "[--nodes-per-group nodes-per-group]\n"
         "[--redundancy redundancy]\n"
@@ -516,6 +517,7 @@ App::get_options()
         { "max-merge-queue-size", 1, nullptr, 0 },
         { "max-pending", 1, nullptr, 0 },
         { "max-pending-idealstate-operations", 1, nullptr, 0 },
+        { "mbus-distributor-node-max-pending-count", 1, nullptr, 0 },
         { "mode", 1, nullptr, 0 },
         { "nodes-per-group", 1, nullptr, 0 },
         { "redundancy", 1, nullptr, 0 },
@@ -545,6 +547,7 @@ App::get_options()
         LONGOPT_MAX_MERGE_QUEUE_SIZE,
         LONGOPT_MAX_PENDING,
         LONGOPT_MAX_PENDING_IDEALSTATE_OPERATIONS,
+        LONGOPT_MBUS_DISTRIBUTOR_NODE_MAX_PENDING_COUNT,
         LONGOPT_MODE,
         LONGOPT_NODES_PER_GROUP,
         LONGOPT_REDUNDANCY,
@@ -608,6 +611,9 @@ App::get_options()
             case LONGOPT_MAX_PENDING_IDEALSTATE_OPERATIONS:
                 _bm_params.set_max_pending_idealstate_operations(atoi(opt_argument));
                 break;
+            case LONGOPT_MBUS_DISTRIBUTOR_NODE_MAX_PENDING_COUNT:
+                _bm_params.set_mbus_distributor_node_max_pending_count(atoi(opt_argument));
+                break;
             case LONGOPT_MODE:
                 _bm_params.set_mode(get_mode(opt_argument));
                 if (_bm_params.get_mode() == Mode::BAD) {
diff --git a/searchcore/src/vespa/searchcore/bmcluster/bm_cluster_params.cpp b/searchcore/src/vespa/searchcore/bmcluster/bm_cluster_params.cpp
index b75541d0e0d..3ff10b19164 100644
--- a/searchcore/src/vespa/searchcore/bmcluster/bm_cluster_params.cpp
+++ b/searchcore/src/vespa/searchcore/bmcluster/bm_cluster_params.cpp
@@ -19,6 +19,7 @@ BmClusterParams::BmClusterParams()
       _max_merges_per_node(16),     // Same default as in stor-server.def
       _max_merge_queue_size(1024),  // Same default as in stor-server.def
       _max_pending_idealstate_operations(100), // Same default as in stor-distributormanager.def
+      _mbus_distributor_node_max_pending_count(),
       _num_nodes(1),
       _nodes_per_group(1),
       _redundancy(1),
diff --git a/searchcore/src/vespa/searchcore/bmcluster/bm_cluster_params.h b/searchcore/src/vespa/searchcore/bmcluster/bm_cluster_params.h
index 0918ce41478..d365a28b0b6 100644
--- a/searchcore/src/vespa/searchcore/bmcluster/bm_cluster_params.h
+++ b/searchcore/src/vespa/searchcore/bmcluster/bm_cluster_params.h
@@ -4,6 +4,7 @@
 
 #include <cstdint>
 #include <vespa/vespalib/stllike/string.h>
+#include <optional>
 
 namespace search::bmcluster {
 
@@ -25,6 +26,7 @@ class BmClusterParams
     uint32_t _max_merges_per_node;
     uint32_t _max_merge_queue_size;
     uint32_t _max_pending_idealstate_operations;
+    std::optional<uint32_t> _mbus_distributor_node_max_pending_count;
     uint32_t _num_nodes;
     uint32_t _nodes_per_group;
     uint32_t _redundancy;
@@ -54,6 +56,7 @@ public:
     uint32_t get_max_merges_per_node() const noexcept { return _max_merges_per_node; }
     uint32_t get_max_merge_queue_size() const noexcept { return _max_merge_queue_size; }
     uint32_t get_max_pending_idealstate_operations() const noexcept { return _max_pending_idealstate_operations; }
+    const std::optional<uint32_t>& get_mbus_distributor_node_max_pending_count() const noexcept { return _mbus_distributor_node_max_pending_count; }
     uint32_t get_nodes_per_group() const noexcept { return _nodes_per_group; }
     uint32_t get_num_nodes() const { return _num_nodes; }
     uint32_t get_redundancy() const { return _redundancy; }
@@ -83,6 +86,7 @@ public:
     void set_max_merges_per_node(uint32_t value) { _max_merges_per_node = value; }
     void set_max_merge_queue_size(uint32_t value) { _max_merge_queue_size = value; }
     void set_max_pending_idealstate_operations(uint32_t value) { _max_pending_idealstate_operations = value; }
+    void set_mbus_distributor_node_max_pending_count(int32_t value) { _mbus_distributor_node_max_pending_count = value; }
     void set_nodes_per_group(uint32_t value);
     void set_redundancy(uint32_t value) { _redundancy = value; }
     void set_response_threads(uint32_t threads_in) { _response_threads = threads_in; }
diff --git a/searchcore/src/vespa/searchcore/bmcluster/bm_feed_params.cpp b/searchcore/src/vespa/searchcore/bmcluster/bm_feed_params.cpp
index 773232556e7..e3d178e250c 100644
--- a/searchcore/src/vespa/searchcore/bmcluster/bm_feed_params.cpp
+++ b/searchcore/src/vespa/searchcore/bmcluster/bm_feed_params.cpp
@@ -28,7 +28,7 @@ BmFeedParams::check() const
         std::cerr << "Too few client threads: " << _client_threads << std::endl;
         return false;
     }
-    if (_client_threads > 256) {
+    if (_client_threads > 1024) {
         std::cerr << "Too many client threads: " << _client_threads << std::endl;
         return false;
     }
diff --git a/searchcore/src/vespa/searchcore/bmcluster/bm_node.cpp b/searchcore/src/vespa/searchcore/bmcluster/bm_node.cpp
index be56ff91719..545fb08c762 100644
--- a/searchcore/src/vespa/searchcore/bmcluster/bm_node.cpp
+++ b/searchcore/src/vespa/searchcore/bmcluster/bm_node.cpp
@@ -335,6 +335,9 @@ struct StorageConfigSet
         stor_communicationmanager.rpc.numNetworkThreads = params.get_rpc_network_threads();
         stor_communicationmanager.rpc.eventsBeforeWakeup = params.get_rpc_events_before_wakeup();
         stor_communicationmanager.rpc.numTargetsPerNode = params.get_rpc_targets_per_node();
+        if (params.get_mbus_distributor_node_max_pending_count().has_value()) {
+            stor_communicationmanager.mbusDistributorNodeMaxPendingCount = params.get_mbus_distributor_node_max_pending_count().value();
+        }
         stor_communicationmanager.mbusport = mbus_port;
         stor_communicationmanager.rpcport = rpc_port;
         stor_communicationmanager.skipThread = params.get_skip_communicationmanager_thread();