Reflect concurrency all the way to the documedbs.

3 files changed, 92 insertions, 6 deletions

diff --git a/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java b/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java
index f4754b4f958..3e87b4ba627 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java
@@ -306,6 +306,8 @@ public class ContentSearchCluster extends AbstractConfigProducer implements Prot
     @Override
     public void getConfig(ProtonConfig.Builder builder) {
         double visibilityDelay = hasIndexedCluster() ? getIndexed().getVisibilityDelay() : 0.0;
+        builder.feeding.concurrency(0.2);
+        boolean hasAnyNonIndexedCluster = false;
         for (NewDocumentType type : TopologicalDocumentTypeSorter.sort(documentDefinitions.values())) {
             ProtonConfig.Documentdb.Builder ddbB = new ProtonConfig.Documentdb.Builder();
             String docTypeName = type.getFullName().getName();
@@ -316,9 +318,18 @@ public class ContentSearchCluster extends AbstractConfigProducer implements Prot
                 .global(globalDocType);
             Optional<StreamingSearchCluster> ssc = findStreamingCluster(docTypeName);
             if (ssc.isPresent()) {
-                ddbB.inputdoctypename(type.getFullName().getName()).configid(ssc.get().getDocumentDBConfigId());
+                hasAnyNonIndexedCluster = true;
+                ddbB.inputdoctypename(type.getFullName().getName())
+                    .configid(ssc.get().getDocumentDBConfigId())
+                    .feeding.concurrency(0.0);
             } else if (hasIndexedCluster()) {
                 getIndexed().fillDocumentDBConfig(type.getFullName().getName(), ddbB);
+                if (tuning != null && tuning.searchNode != null && tuning.searchNode.feeding != null) {
+                    ddbB.feeding.concurrency(tuning.searchNode.feeding.concurrency/2);
+                }
+            } else {
+                hasAnyNonIndexedCluster = true;
+                ddbB.feeding.concurrency(0.0);
             }
             if (globalDocType) {
                 ddbB.visibilitydelay(0.0);
@@ -339,6 +350,9 @@ public class ContentSearchCluster extends AbstractConfigProducer implements Prot
         if (redundancy != null) {
             redundancy.getConfig(builder);
         }
+        if (hasAnyNonIndexedCluster) {
+            builder.feeding.concurrency(builder.feeding.build().concurrency() * 2);
+        }
     }
 
     private boolean isGloballyDistributed(NewDocumentType docType) {
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/search/Tuning.java b/config-model/src/main/java/com/yahoo/vespa/model/search/Tuning.java
index 8e8192b74fa..16b9680a0b2 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/search/Tuning.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/search/Tuning.java
@@ -134,6 +134,9 @@ public class Tuning extends AbstractConfigProducer implements PartitionsConfig.P
             public void getConfig(ProtonConfig.Builder builder) {
                 if (initialDocumentCount!=null) {
                     builder.grow.initial(initialDocumentCount);
+                    for (ProtonConfig.Documentdb.Builder db : builder.documentdb) {
+                        db.allocation.initialnumdocs(initialDocumentCount);
+                    }
                 }
             }
 
@@ -338,7 +341,7 @@ public class Tuning extends AbstractConfigProducer implements PartitionsConfig.P
             @Override
             public void getConfig(ProtonConfig.Builder builder) {
                 if (concurrency != null) {
-                    builder.feeding.concurrency(concurrency);
+                    builder.feeding.concurrency(concurrency/2);
                 }
             }
         }
diff --git a/config-model/src/test/java/com/yahoo/vespa/model/search/test/DocumentDatabaseTestCase.java b/config-model/src/test/java/com/yahoo/vespa/model/search/test/DocumentDatabaseTestCase.java
index bd27f381e4e..36a0c4647b5 100644
--- a/config-model/src/test/java/com/yahoo/vespa/model/search/test/DocumentDatabaseTestCase.java
+++ b/config-model/src/test/java/com/yahoo/vespa/model/search/test/DocumentDatabaseTestCase.java
@@ -2,6 +2,7 @@
 package com.yahoo.vespa.model.search.test;
 
 import com.google.common.collect.ImmutableMap;
+import com.yahoo.collections.Pair;
 import com.yahoo.vespa.config.search.IndexschemaConfig;
 import com.yahoo.vespa.config.search.core.ProtonConfig;
 import com.yahoo.vespa.config.search.RankProfilesConfig;
@@ -18,7 +19,7 @@ import com.yahoo.vespa.model.test.utils.VespaModelCreatorWithMockPkg;
 import org.junit.Test;
 import org.xml.sax.SAXException;
 import java.io.IOException;
-import java.util.HashMap;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Arrays;
 import java.util.Map;
@@ -31,6 +32,8 @@ import static org.junit.Assert.assertEquals;
  */
 public class DocumentDatabaseTestCase {
 
+    private static final double SMALL = 0.00000000000001;
+
     private String vespaHosts = "<?xml version='1.0' encoding='utf-8' ?>" +
             "<hosts>  " +
             "  <host name='foo'>" +
@@ -38,7 +41,14 @@ public class DocumentDatabaseTestCase {
             "  </host>" +
             "</hosts>";
 
-    private String createVespaServices(List<String> sdNames, String mode) {
+    private String createVespaServices(List<String> sds, String mode) {
+        List<Pair<String, String>> nameAndModes = new ArrayList<>(sds.size());
+        for (String sd : sds) {
+            nameAndModes.add(new Pair<>(sd, mode));
+        }
+        return createVespaServicesXml(nameAndModes, "");
+    }
+    private String createVespaServicesXml(List<Pair<String, String>> nameAndModes, String xmlTuning) {
         StringBuilder retval = new StringBuilder();
         retval.append("" +
                       "<?xml version='1.0' encoding='utf-8' ?>\n" +
@@ -55,12 +65,21 @@ public class DocumentDatabaseTestCase {
                       "<content version='1.0' id='test'>\n" +
                       "   <redundancy>1</redundancy>\n");
         retval.append("   <documents>\n");
-        for (String sdName : sdNames) {
-            retval.append("").append("      <document type='").append(sdName).append("' mode='").append(mode).append("'");
+        for (Pair<String, String> nameAndMode : nameAndModes) {
+            retval.append("      <document type='").append(nameAndMode.getFirst()).append("' mode='").append(nameAndMode.getSecond()).append("'");
             retval.append("/>\n");
         }
         retval.append("   </documents>\n");
         retval.append("" +
+                      "<engine>\n" +
+                "<proton>\n" +
+                "<tuning>\n" +
+                "<searchnode>\n" +
+                xmlTuning +
+                "</searchnode>\n" +
+                "</tuning\n>" +
+                "</proton\n>" +
+                "</engine\n>" +
                       "   <nodes>\n" +
                       "      <node hostalias='node0' distribution-key='0'/>\n" +
                       "   </nodes>\n" +
@@ -93,6 +112,56 @@ public class DocumentDatabaseTestCase {
         assertSingleSD("index");
     }
 
+    private VespaModel createModel(List<Pair<String, String>> nameAndModes, String xmlTuning) {
+        List<String> sds = new ArrayList<>(nameAndModes.size());
+        for (Pair<String, String> nameAndMode : nameAndModes) {
+            sds.add(nameAndMode.getFirst());
+        }
+        return new VespaModelCreatorWithMockPkg(vespaHosts, createVespaServicesXml(nameAndModes, xmlTuning),
+                ApplicationPackageUtils.generateSearchDefinitions(sds)).create();
+    }
+
+    @Test
+    public void requireThatConcurrencyIsReflectedCorrectlyForDefault() {
+        verifyDefaultConcurrency("index", "", 0.2, 0.2);
+        verifyDefaultConcurrency("streaming", "", 0.4, 0.0);
+        verifyDefaultConcurrency("store-only", "", 0.4, 0.0);
+    }
+    @Test
+    public void requireThatMixedModeConcurrencyIsReflectedCorrectlyForDefault() {
+        verifyDefaultConcurrency(Arrays.asList(new Pair("a", "index"), new Pair("b", "streaming")), "", 0.4, Arrays.asList(0.2, 0.0));
+    }
+    @Test
+    public void requireThatMixedModeConcurrencyIsReflected() {
+        String feedTuning = "<feeding>" +
+                "  <concurrency>0.7</concurrency>" +
+                "</feeding>\n";
+        verifyDefaultConcurrency(Arrays.asList(new Pair("a", "index"), new Pair("b", "streaming")), feedTuning, 0.7, Arrays.asList(0.35, 0.0));
+    }
+    @Test
+    public void requireThatConcurrencyIsReflected() {
+        String feedTuning = "<feeding>" +
+                            "  <concurrency>0.7</concurrency>" +
+                            "</feeding>\n";
+        verifyDefaultConcurrency("index", feedTuning, 0.35, 0.35);
+        verifyDefaultConcurrency("streaming", feedTuning, 0.7, 0.0);
+        verifyDefaultConcurrency("store-only", feedTuning, 0.7, 0.0);
+    }
+    private void verifyDefaultConcurrency(String mode, String xmlTuning, double global, double local) {
+        verifyDefaultConcurrency(Arrays.asList(new Pair<>("a", mode)), xmlTuning, global, Arrays.asList(local));
+    }
+    private void verifyDefaultConcurrency(List<Pair<String, String>> nameAndModes, String xmlTuning, double global, List<Double> local) {
+        assertEquals(nameAndModes.size(), local.size());
+        VespaModel model = createModel(nameAndModes, xmlTuning);
+        ContentSearchCluster contentSearchCluster = model.getContentClusters().get("test").getSearch();
+        ProtonConfig proton = getProtonCfg(contentSearchCluster);
+        assertEquals(global, proton.feeding().concurrency(), SMALL);
+        assertEquals(local.size(), proton.documentdb().size());
+        for (int i = 0; i < local.size(); i++) {
+            assertEquals(local.get(i), proton.documentdb(i).feeding().concurrency(), SMALL);
+        }
+    }
+
     private void assertDocTypeConfig(VespaModel model, String configId, String indexField, String attributeField) {
         IndexschemaConfig icfg = model.getConfig(IndexschemaConfig.class, configId);
         assertEquals(1, icfg.indexfield().size());