Let feeder control throttle parameters

3 files changed, 75 insertions, 4 deletions

diff --git a/vespa_feed_perf/src/main/java/com/yahoo/vespa/feed/perf/FeederParams.java b/vespa_feed_perf/src/main/java/com/yahoo/vespa/feed/perf/FeederParams.java
index c1e164f7fe8..0ee1db5dbff 100644
--- a/vespa_feed_perf/src/main/java/com/yahoo/vespa/feed/perf/FeederParams.java
+++ b/vespa_feed_perf/src/main/java/com/yahoo/vespa/feed/perf/FeederParams.java
@@ -34,6 +34,12 @@ class FeederParams {
     private boolean benchmarkMode = false;
     private int numDispatchThreads = 1;
     private int maxPending = 0;
+
+    private double windowSizeBackOff = 0.95;
+    private double windowDecrementFactor = 1.2;
+    private double windowResizeRate = 3;
+    private int windowIncrementSize = 20;
+
     private int numConnectionsPerTarget = 1;
     private long numMessagesToSend = Long.MAX_VALUE;
     private List<InputStream> inputStreams = new ArrayList<>();
@@ -83,6 +89,21 @@ class FeederParams {
         this.configId = configId;
         return this;
     }
+    public double getWindowSizeBackOff() {
+        return windowSizeBackOff;
+    }
+
+    public double getWindowDecrementFactor() {
+        return windowDecrementFactor;
+    }
+
+    public double getWindowResizeRate() {
+        return windowResizeRate;
+    }
+
+    public int getWindowIncrementSize() {
+        return windowIncrementSize;
+    }
 
     int getNumConnectionsPerTarget() { return numConnectionsPerTarget; }
 
@@ -117,6 +138,10 @@ class FeederParams {
         opts.addOption("o", "output", true, "File to write to. Extensions gives format (.xml, .json, .vespa) json will be produced if no extension.");
         opts.addOption("c", "numconnections", true, "Number of connections per host.");
         opts.addOption("l", "nummessages", true, "Number of messages to send (all is default).");
+        opts.addOption("wi", "window_incrementsize", true, "Dynamic window increment step size. default = " + windowIncrementSize);
+        opts.addOption("wd", "window_decrementfactor", true, "Dynamic window decrement step size factor. default = " + windowDecrementFactor);
+        opts.addOption("wb", "window_backoffactor", true, "Dynamic window backoff factor. default = " + windowSizeBackOff);
+        opts.addOption("wr", "window_resizerate", true, "Dynamic window resize rate. default = " + windowResizeRate);
 
         CommandLine cmd = new DefaultParser().parse(opts, args);
 
@@ -129,6 +154,18 @@ class FeederParams {
         if (cmd.hasOption('c')) {
             numConnectionsPerTarget = Integer.valueOf(cmd.getOptionValue('c').trim());
         }
+        if (cmd.hasOption("wi")) {
+            windowIncrementSize = Integer.valueOf(cmd.getOptionValue("wi").trim());
+        }
+        if (cmd.hasOption("wi")) {
+            windowDecrementFactor = Double.valueOf(cmd.getOptionValue("wi").trim());
+        }
+        if (cmd.hasOption("wb")) {
+            windowSizeBackOff = Double.valueOf(cmd.getOptionValue("wb").trim());
+        }
+        if (cmd.hasOption("wr")) {
+            windowResizeRate = Double.valueOf(cmd.getOptionValue("wr").trim());
+        }
         if (cmd.hasOption('r')) {
             route = Route.parse(cmd.getOptionValue('r').trim());
         }
diff --git a/vespa_feed_perf/src/main/java/com/yahoo/vespa/feed/perf/SimpleFeeder.java b/vespa_feed_perf/src/main/java/com/yahoo/vespa/feed/perf/SimpleFeeder.java
index 556d9bd60c7..18c0de339e6 100644
--- a/vespa_feed_perf/src/main/java/com/yahoo/vespa/feed/perf/SimpleFeeder.java
+++ b/vespa_feed_perf/src/main/java/com/yahoo/vespa/feed/perf/SimpleFeeder.java
@@ -20,6 +20,7 @@ import com.yahoo.documentapi.messagebus.protocol.PutDocumentMessage;
 import com.yahoo.documentapi.messagebus.protocol.RemoveDocumentMessage;
 import com.yahoo.documentapi.messagebus.protocol.UpdateDocumentMessage;
 import com.yahoo.io.GrowableByteBuffer;
+import com.yahoo.messagebus.DynamicThrottlePolicy;
 import com.yahoo.messagebus.Error;
 import com.yahoo.messagebus.Message;
 import com.yahoo.messagebus.MessageBusParams;
@@ -346,7 +347,7 @@ public class SimpleFeeder implements ReplyHandler {
         numThreads = params.getNumDispatchThreads();
         numMessagesToSend = params.getNumMessagesToSend();
         mbus = newMessageBus(docTypeMgr, params);
-        session = newSession(mbus, this, params.getMaxPending());
+        session = newSession(mbus, this, params);
         docTypeMgr.configure(params.getConfigId());
         benchmarkMode = params.isBenchmarkMode();
         destination = (params.getDumpStream() != null)
@@ -473,11 +474,18 @@ public class SimpleFeeder implements ReplyHandler {
                                  params.getConfigId());
     }
 
-    private static SourceSession newSession(RPCMessageBus mbus, ReplyHandler replyHandler, int maxPending) {
+    private static SourceSession newSession(RPCMessageBus mbus, ReplyHandler replyHandler, FeederParams feederParams ) {
         SourceSessionParams params = new SourceSessionParams();
         params.setReplyHandler(replyHandler);
-        if (maxPending > 0) {
-            params.setThrottlePolicy(new StaticThrottlePolicy().setMaxPendingCount(maxPending));
+        if (feederParams.getMaxPending() > 0) {
+            params.setThrottlePolicy(new StaticThrottlePolicy().setMaxPendingCount(feederParams.getMaxPending()));
+        } else {
+            DynamicThrottlePolicy throttlePolicy = new DynamicThrottlePolicy()
+                    .setWindowSizeIncrement(feederParams.getWindowIncrementSize())
+                    .setResizeRate(feederParams.getWindowResizeRate())
+                    .setWindowSizeDecrementFactor(feederParams.getWindowDecrementFactor())
+                    .setWindowSizeBackOff(feederParams.getWindowSizeBackOff());
+            params.setThrottlePolicy(throttlePolicy);
         }
         return mbus.getMessageBus().createSourceSession(params);
     }
diff --git a/vespa_feed_perf/src/test/java/com/yahoo/vespa/feed/perf/FeederParamsTest.java b/vespa_feed_perf/src/test/java/com/yahoo/vespa/feed/perf/FeederParamsTest.java
index 13e307d9973..bbc27cfa084 100644
--- a/vespa_feed_perf/src/test/java/com/yahoo/vespa/feed/perf/FeederParamsTest.java
+++ b/vespa_feed_perf/src/test/java/com/yahoo/vespa/feed/perf/FeederParamsTest.java
@@ -101,6 +101,32 @@ public class FeederParamsTest {
     }
 
     @Test
+    public void requireThatWindowSizeIncrementIsParsed() throws ParseException, FileNotFoundException {
+        assertEquals(20, new FeederParams().getWindowIncrementSize());
+        assertEquals(17, new FeederParams().parseArgs("--window_incrementsize", "17").getWindowIncrementSize());
+    }
+
+    static final double SMALL_NUMBER = 0.000000000001;
+
+    @Test
+    public void requireThatWindowSizeDecrementFactorIsParsed() throws ParseException, FileNotFoundException {
+        assertEquals(1.2, new FeederParams().getWindowDecrementFactor(), SMALL_NUMBER);
+        assertEquals(1.3, new FeederParams().parseArgs("--window_decrementfactor", "1.3").getWindowIncrementSize(), SMALL_NUMBER);
+    }
+
+    @Test
+    public void requireThatWindowResizeRateIsParsed() throws ParseException, FileNotFoundException {
+        assertEquals(3.0, new FeederParams().getWindowResizeRate(), SMALL_NUMBER);
+        assertEquals(5.5, new FeederParams().parseArgs("--window_resizerate", "5.5").getWindowResizeRate(), SMALL_NUMBER);
+    }
+
+    @Test
+    public void requireThatWindowBackOffIsParsed() throws ParseException, FileNotFoundException {
+        assertEquals(0.95, new FeederParams().getWindowSizeBackOff(), SMALL_NUMBER);
+        assertEquals(0.97, new FeederParams().parseArgs("--window_backoff", "0.97").getWindowSizeBackOff(), SMALL_NUMBER);
+    }
+
+    @Test
     public void requireThatDumpStreamAreParsed() throws ParseException, IOException {
         assertNull(new FeederParams().getDumpStream());