Use the concurrency in the document DB.

4 files changed, 24 insertions, 17 deletions

diff --git a/searchcore/src/tests/proton/documentdb/threading_service_config/threading_service_config_test.cpp b/searchcore/src/tests/proton/documentdb/threading_service_config/threading_service_config_test.cpp
index 658ebe818eb..340619f09bd 100644
--- a/searchcore/src/tests/proton/documentdb/threading_service_config/threading_service_config_test.cpp
+++ b/searchcore/src/tests/proton/documentdb/threading_service_config/threading_service_config_test.cpp
@@ -22,11 +22,10 @@ struct Fixture {
         builder.indexing.threads = baseLineIndexingThreads;
         builder.indexing.tasklimit = 500;
         builder.indexing.semiunboundtasklimit = 50000;
-        builder.feeding.concurrency = 0.5;
         return builder;
     }
     ThreadingServiceConfig make(uint32_t cpuCores) {
-        return ThreadingServiceConfig::make(cfg, HwInfo::Cpu(cpuCores));
+        return ThreadingServiceConfig::make(cfg, 0.5, HwInfo::Cpu(cpuCores));
     }
     void assertIndexingThreads(uint32_t expIndexingThreads, uint32_t cpuCores) {
         EXPECT_EQUAL(expIndexingThreads, make(cpuCores).indexingThreads());
diff --git a/searchcore/src/vespa/searchcore/proton/server/documentdb.cpp b/searchcore/src/vespa/searchcore/proton/server/documentdb.cpp
index 6a1d053745f..045ae1c9bc6 100644
--- a/searchcore/src/vespa/searchcore/proton/server/documentdb.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/documentdb.cpp
@@ -89,6 +89,16 @@ makeIndexConfig(const ProtonConfig::Index & cfg) {
     return index::IndexConfig(WarmupConfig(cfg.warmup.time, cfg.warmup.unpack), cfg.maxflushed, cfg.cache.size);
 }
 
+const ProtonConfig::Documentdb *
+findDocumentDB(const ProtonConfig::DocumentdbVector & documentDBs, const vespalib::string & docType) {
+    for (const auto & dbCfg : documentDBs) {
+        if (dbCfg.inputdoctypename == docType) {
+            return & dbCfg;
+        }
+    }
+    return nullptr;
+}
+
 }
 
 template <typename FunctionType>
@@ -124,7 +134,10 @@ DocumentDB::DocumentDB(const vespalib::string &baseDir,
       _bucketSpace(bucketSpace),
       _baseDir(baseDir + "/" + _docTypeName.toString()),
       // Only one thread per executor, or performDropFeedView() will fail.
-      _writeServiceConfig(ThreadingServiceConfig::make(protonCfg, hwInfo.cpu())),
+      _writeServiceConfig(
+              ThreadingServiceConfig::make(protonCfg,
+                      findDocumentDB(protonCfg.documentdb, docTypeName.getName())->feeding.concurrency,
+                      hwInfo.cpu())),
       _writeService(_writeServiceConfig.indexingThreads(),
                     indexing_thread_stack_size,
                     _writeServiceConfig.defaultTaskLimit()),
@@ -165,8 +178,7 @@ DocumentDB::DocumentDB(const vespalib::string &baseDir,
 {
     assert(configSnapshot);
 
-    LOG(debug, "DocumentDB(%s): Creating database in directory '%s'",
-        _docTypeName.toString().c_str(), _baseDir.c_str());
+    LOG(debug, "DocumentDB(%s): Creating database in directory '%s'", _docTypeName.toString().c_str(), _baseDir.c_str());
 
     _feedHandler.init(_config_store->getOldestSerialNum());
     _feedHandler.setBucketDBHandler(&_subDBs.getBucketDBHandler());
diff --git a/searchcore/src/vespa/searchcore/proton/server/threading_service_config.cpp b/searchcore/src/vespa/searchcore/proton/server/threading_service_config.cpp
index cd1608feadb..8f1c3560e9b 100644
--- a/searchcore/src/vespa/searchcore/proton/server/threading_service_config.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/threading_service_config.cpp
@@ -20,21 +20,20 @@ ThreadingServiceConfig::ThreadingServiceConfig(uint32_t indexingThreads_,
 namespace {
 
 uint32_t
-calculateIndexingThreads(const ProtonConfig &cfg, const HwInfo::Cpu &cpuInfo)
+calculateIndexingThreads(uint32_t cfgIndexingThreads, double concurrency, const HwInfo::Cpu &cpuInfo)
 {
-    double scaledCores = cpuInfo.cores() * cfg.feeding.concurrency;
-    uint32_t indexingThreads = std::max((uint32_t)std::ceil(scaledCores / 3), (uint32_t)cfg.indexing.threads);
+    double scaledCores = cpuInfo.cores() * concurrency;
+    uint32_t indexingThreads = std::max((uint32_t)std::ceil(scaledCores / 3), cfgIndexingThreads);
     return std::max(indexingThreads, 1u);
 }
 
 }
 
 ThreadingServiceConfig
-ThreadingServiceConfig::make(const ProtonConfig &cfg, const HwInfo::Cpu &cpuInfo)
+ThreadingServiceConfig::make(const ProtonConfig &cfg, double concurrency, const HwInfo::Cpu &cpuInfo)
 {
-    uint32_t indexingThreads = calculateIndexingThreads(cfg, cpuInfo);
-    return ThreadingServiceConfig(indexingThreads,
-                                  cfg.indexing.tasklimit,
+    uint32_t indexingThreads = calculateIndexingThreads(cfg.indexing.threads, concurrency, cpuInfo);
+    return ThreadingServiceConfig(indexingThreads, cfg.indexing.tasklimit,
                                   (cfg.indexing.semiunboundtasklimit / indexingThreads));
 }
 
diff --git a/searchcore/src/vespa/searchcore/proton/server/threading_service_config.h b/searchcore/src/vespa/searchcore/proton/server/threading_service_config.h
index 03c0c4d1842..be39f516598 100644
--- a/searchcore/src/vespa/searchcore/proton/server/threading_service_config.h
+++ b/searchcore/src/vespa/searchcore/proton/server/threading_service_config.h
@@ -20,13 +20,10 @@ private:
     uint32_t _semiUnboundTaskLimit;
 
 private:
-    ThreadingServiceConfig(uint32_t indexingThreads_,
-                           uint32_t defaultTaskLimit_,
-                           uint32_t semiUnboundTaskLimit_);
+    ThreadingServiceConfig(uint32_t indexingThreads_, uint32_t defaultTaskLimit_, uint32_t semiUnboundTaskLimit_);
 
 public:
-    static ThreadingServiceConfig make(const ProtonConfig &cfg,
-                                       const HwInfo::Cpu &cpuInfo);
+    static ThreadingServiceConfig make(const ProtonConfig &cfg, double concurrency, const HwInfo::Cpu &cpuInfo);
 
     uint32_t indexingThreads() const { return _indexingThreads; }
     uint32_t defaultTaskLimit() const { return _defaultTaskLimit; }