Make "events before wakeup" configurable for SharedRpcResources.

1 files changed, 14 insertions, 2 deletions

diff --git a/searchcore/src/apps/vespa-feed-bm/vespa_feed_bm.cpp b/searchcore/src/apps/vespa-feed-bm/vespa_feed_bm.cpp
index 5a5f5cb46d6..170885d1d99 100644
--- a/searchcore/src/apps/vespa-feed-bm/vespa_feed_bm.cpp
+++ b/searchcore/src/apps/vespa-feed-bm/vespa_feed_bm.cpp
@@ -269,6 +269,7 @@ class BMParams {
     uint32_t _update_passes;
     uint32_t _remove_passes;
     uint32_t _rpc_network_threads;
+    uint32_t _rpc_events_before_wakeup;
     uint32_t _rpc_targets_per_node;
     uint32_t _response_threads;
     uint32_t _max_pending;
@@ -292,7 +293,8 @@ public:
           _put_passes(2),
           _update_passes(1),
           _remove_passes(2),
-          _rpc_network_threads(1), // Same default as in stor-communicationmanager.def
+          _rpc_network_threads(1), // Same default as previous in stor-communicationmanager.def
+          _rpc_events_before_wakeup(1), // Same default as in stor-communicationmanager.def
           _rpc_targets_per_node(1), // Same default as in stor-communicationmanager.def
           _response_threads(2), // Same default as in stor-filestor.def
           _max_pending(1000),
@@ -318,6 +320,7 @@ public:
     uint32_t get_update_passes() const { return _update_passes; }
     uint32_t get_remove_passes() const { return _remove_passes; }
     uint32_t get_rpc_network_threads() const { return _rpc_network_threads; }
+    uint32_t get_rpc_events_before_wakup() const { return _rpc_events_before_wakeup; }
     uint32_t get_rpc_targets_per_node() const { return _rpc_targets_per_node; }
     uint32_t get_response_threads() const { return _response_threads; }
     bool get_enable_distributor() const { return _enable_distributor; }
@@ -336,6 +339,7 @@ public:
     void set_update_passes(uint32_t update_passes_in) { _update_passes = update_passes_in; }
     void set_remove_passes(uint32_t remove_passes_in) { _remove_passes = remove_passes_in; }
     void set_rpc_network_threads(uint32_t threads_in) { _rpc_network_threads = threads_in; }
+    void set_rpc_events_before_wakeup(uint32_t value) { _rpc_events_before_wakeup = value; }
     void set_rpc_targets_per_node(uint32_t targets_in) { _rpc_targets_per_node = targets_in; }
     void set_response_threads(uint32_t threads_in) { _response_threads = threads_in; }
     void set_enable_distributor(bool value) { _enable_distributor = value; }
@@ -491,6 +495,7 @@ struct MyStorageConfig
         make_slobroks_config(slobroks, slobrok_port);
         stor_communicationmanager.useDirectStorageapiRpc = true;
         stor_communicationmanager.rpc.numNetworkThreads = params.get_rpc_network_threads();
+        stor_communicationmanager.rpc.eventsBeforeWakeup = params.get_rpc_events_before_wakup();
         stor_communicationmanager.rpc.numTargetsPerNode = params.get_rpc_targets_per_node();
         stor_communicationmanager.mbusport = mbus_port;
         stor_communicationmanager.rpcport = rpc_port;
@@ -878,7 +883,8 @@ PersistenceProviderFixture::start_service_layer(const BMParams& params)
     _service_layer->getNode().waitUntilInitialized();
     LOG(info, "start rpc client shared resources");
     config::ConfigUri client_config_uri("bm-rpc-client", _config_context);
-    _rpc_client_shared_rpc_resources = std::make_unique<SharedRpcResources>(client_config_uri, _rpc_client_port, 100);
+    _rpc_client_shared_rpc_resources = std::make_unique<SharedRpcResources>
+            (client_config_uri, _rpc_client_port, 100, params.get_rpc_events_before_wakup());
     _rpc_client_shared_rpc_resources->start_server_and_register_slobrok("bm-rpc-client");
     wait_slobrok("storage/cluster.storage/storage/0/default");
     wait_slobrok("storage/cluster.storage/storage/0");
@@ -1369,6 +1375,7 @@ App::usage()
         "[--put-passes put-passes]\n"
         "[--update-passes update-passes]\n"
         "[--remove-passes remove-passes]\n"
+        "[--rpc-events-before-wakeup events]\n"
         "[--rpc-network-threads threads]\n"
         "[--rpc-targets-per-node targets]\n"
         "[--response-threads threads]\n"
@@ -1399,6 +1406,7 @@ App::get_options()
         { "put-passes", 1, nullptr, 0 },
         { "remove-passes", 1, nullptr, 0 },
         { "response-threads", 1, nullptr, 0 },
+        { "rpc-events-before-wakeup", 1, nullptr, 0 },
         { "rpc-network-threads", 1, nullptr, 0 },
         { "rpc-targets-per-node", 1, nullptr, 0 },
         { "skip-get-spi-bucket-info", 0, nullptr, 0 },
@@ -1420,6 +1428,7 @@ App::get_options()
         LONGOPT_PUT_PASSES,
         LONGOPT_REMOVE_PASSES,
         LONGOPT_RESPONSE_THREADS,
+        LONGOPT_RPC_EVENTS_BEFORE_WAKEUP,
         LONGOPT_RPC_NETWORK_THREADS,
         LONGOPT_RPC_TARGETS_PER_NODE,
         LONGOPT_SKIP_GET_SPI_BUCKET_INFO,
@@ -1471,6 +1480,9 @@ App::get_options()
             case LONGOPT_RESPONSE_THREADS:
                 _bm_params.set_response_threads(atoi(opt_argument));
                 break;
+            case LONGOPT_RPC_EVENTS_BEFORE_WAKEUP:
+                _bm_params.set_rpc_events_before_wakeup(atoi(opt_argument));
+                break;
             case LONGOPT_RPC_NETWORK_THREADS:
                 _bm_params.set_rpc_network_threads(atoi(opt_argument));
                 break;