Control threadpool and dispatch of encode/decode by config.

2 files changed, 20 insertions, 4 deletions

diff --git a/storage/src/vespa/storage/config/stor-communicationmanager.def b/storage/src/vespa/storage/config/stor-communicationmanager.def
index e29540de064..2a2a840dd4e 100644
--- a/storage/src/vespa/storage/config/stor-communicationmanager.def
+++ b/storage/src/vespa/storage/config/stor-communicationmanager.def
@@ -28,3 +28,16 @@ mbus.compress.type enum {NONE, LZ4, ZSTD} default=LZ4
 
 ## TTL for rpc target cache
 mbus.rpctargetcache.ttl double default = 600
+
+## Number of threads for mbus threadpool
+## Any value below 1 will be 1.
+mbus.num_threads int default=4
+
+## Enable to use above thread pool for encoding replies
+## False will use network(fnet) thread
+mbus.dispatch_on_encode bool default=true
+
+## Enable to use above thread pool for decoding replies
+## False will use network(fnet) thread
+## Todo: Change default once verified in large scale deployment.
+mbus.dispatch_on_decode bool default=false
diff --git a/storage/src/vespa/storage/storageserver/communicationmanager.cpp b/storage/src/vespa/storage/storageserver/communicationmanager.cpp
index 94a151bcdc1..1704b97ccc7 100644
--- a/storage/src/vespa/storage/storageserver/communicationmanager.cpp
+++ b/storage/src/vespa/storage/storageserver/communicationmanager.cpp
@@ -79,7 +79,7 @@ StorageTransportContext::StorageTransportContext(std::unique_ptr<RPCRequestWrapp
     : _request(std::move(request))
 { }
 
-StorageTransportContext::~StorageTransportContext() { }
+StorageTransportContext::~StorageTransportContext() = default;
 
 void
 CommunicationManager::receiveStorageReply(const std::shared_ptr<api::StorageReply>& reply)
@@ -278,13 +278,13 @@ void CommunicationManager::fail_with_unresolvable_bucket_space(
 namespace {
 
 struct PlaceHolderBucketResolver : public BucketResolver {
-    virtual document::Bucket bucketFromId(const document::DocumentId &) const override {
+    document::Bucket bucketFromId(const document::DocumentId &) const override {
         return document::Bucket(FixedBucketSpaces::default_space(), document::BucketId(0));
     }
-    virtual document::BucketSpace bucketSpaceFromName(const vespalib::string &) const override {
+    document::BucketSpace bucketSpaceFromName(const vespalib::string &) const override {
         return FixedBucketSpaces::default_space();
     }
-    virtual vespalib::string nameFromBucketSpace(const document::BucketSpace &bucketSpace) const override {
+    vespalib::string nameFromBucketSpace(const document::BucketSpace &bucketSpace) const override {
         assert(bucketSpace == FixedBucketSpaces::default_space());
         return FixedBucketSpaces::to_string(bucketSpace);
     }
@@ -423,6 +423,9 @@ void CommunicationManager::configure(std::unique_ptr<CommunicationManagerConfig>
 
         params.setSlobrokConfig(_configUri);
         params.setConnectionExpireSecs(config->mbus.rpctargetcache.ttl);
+        params.setNumThreads(config->mbus.numThreads);
+        params.setDispatchOnDecode(config->mbus.dispatchOnDecode);
+        params.setDispatchOnEncode(config->mbus.dispatchOnEncode);
 
         params.setIdentity(mbus::Identity(_component.getIdentity()));
         if (config->mbusport != -1) {