A configurable limit for waking up fnet thread

6 files changed, 42 insertions, 9 deletions

diff --git a/container-messagebus/src/main/java/com/yahoo/container/jdisc/messagebus/SessionCache.java b/container-messagebus/src/main/java/com/yahoo/container/jdisc/messagebus/SessionCache.java
index 68b1f5aa5db..795278bca84 100644
--- a/container-messagebus/src/main/java/com/yahoo/container/jdisc/messagebus/SessionCache.java
+++ b/container-messagebus/src/main/java/com/yahoo/container/jdisc/messagebus/SessionCache.java
@@ -123,6 +123,7 @@ public final class SessionCache extends AbstractComponent {
                 .setListenPort(mbusConfig.port())
                 .setNumTargetsPerSpec(mbusConfig.numconnectionspertarget())
                 .setNumNetworkThreads(mbusConfig.numthreads())
+                .setWakeupTriggerCount(mbusConfig.wakeup_trigger_count())
                 .setOptimization(RPCNetworkParams.Optimization.valueOf(mbusConfig.optimize_for().name()));
         return SharedMessageBus.newInstance(mbusParams, netParams);
     }
diff --git a/container-messagebus/src/main/resources/configdefinitions/container.jdisc.container-mbus.def b/container-messagebus/src/main/resources/configdefinitions/container.jdisc.container-mbus.def
index 9aef2b32a66..40557f00dfb 100644
--- a/container-messagebus/src/main/resources/configdefinitions/container.jdisc.container-mbus.def
+++ b/container-messagebus/src/main/resources/configdefinitions/container.jdisc.container-mbus.def
@@ -15,6 +15,9 @@ numthreads int default=2
 # Optimize for latency, or throughput.
 optimize_for enum {LATENCY, THROUGHPUT} default=LATENCY
 
+# Number of writeevents before triggering wakeup of network thread.
+wakeup_trigger_count int default=1
+
 # Everying below is deprecated and will go away very soon.
 # Dynamic throttling is used, and works better than anything else.
 maxpendingcount int default=2048
diff --git a/jrt/src/com/yahoo/jrt/Transport.java b/jrt/src/com/yahoo/jrt/Transport.java
index 003e40b8aa9..71784703fc2 100644
--- a/jrt/src/com/yahoo/jrt/Transport.java
+++ b/jrt/src/com/yahoo/jrt/Transport.java
@@ -27,6 +27,7 @@ public class Transport {
     private final Worker            worker;
     private final AtomicInteger     runCnt;
     private final boolean tcpNoDelay;
+    private final int wakeupTriggerCount;
 
     private final TransportMetrics metrics = TransportMetrics.getInstance();
     private final ArrayList<TransportThread> threads = new ArrayList<>();
@@ -42,12 +43,14 @@ public class Transport {
      * @param fatalHandler fatal error handler
      * @param cryptoEngine crypto engine to use
      * @param numThreads number of {@link TransportThread}s.
+     * @param wakeupTriggerCount number write events in Q before waking thread up
      **/
-    public Transport(String name, FatalErrorHandler fatalHandler, CryptoEngine cryptoEngine, int numThreads, boolean tcpNoDelay) {
+    public Transport(String name, FatalErrorHandler fatalHandler, CryptoEngine cryptoEngine, int numThreads, boolean tcpNoDelay, int wakeupTriggerCount) {
         this.name = name;
         this.fatalHandler = fatalHandler; // NB: this must be set first
         this.cryptoEngine = cryptoEngine;
         this.tcpNoDelay = tcpNoDelay;
+        this.wakeupTriggerCount = Math.max(1, wakeupTriggerCount);
         connector = new Connector();
         worker = new Worker(this);
         runCnt = new AtomicInteger(numThreads);
@@ -55,10 +58,23 @@ public class Transport {
             threads.add(new TransportThread(this, i));
         }
     }
-    public Transport(String name, CryptoEngine cryptoEngine, int numThreads) { this(name, null, cryptoEngine, numThreads, true); }
-    public Transport(String name, int numThreads) { this(name, null, CryptoEngine.createDefault(), numThreads, true); }
-    public Transport(String name, int numThreads, boolean tcpNoDelay) { this(name, null, CryptoEngine.createDefault(), numThreads, tcpNoDelay); }
-    public Transport(String name) { this(name, null, CryptoEngine.createDefault(), 1, true); }
+    public Transport(String name, CryptoEngine cryptoEngine, int numThreads, int wakeupTriggerCount) {
+        this(name, null, cryptoEngine, numThreads, true, wakeupTriggerCount);
+    }
+    public Transport(String name, CryptoEngine cryptoEngine, int numThreads) {
+        this(name, null, cryptoEngine, numThreads, true, 1);
+    }
+    public Transport(String name, int numThreads, int wakeupTriggerCount) {
+        this(name, null, CryptoEngine.createDefault(), numThreads, true, wakeupTriggerCount);
+    }
+    public Transport(String name, int numThreads, boolean tcpNoDelay, int wakeupTriggerCount) {
+        this(name, null, CryptoEngine.createDefault(), numThreads, tcpNoDelay, wakeupTriggerCount); }
+    public Transport(String name, int numThreads) {
+        this(name, null, CryptoEngine.createDefault(), numThreads, true, 1);
+    }
+    public Transport(String name) {
+        this(name, null, CryptoEngine.createDefault(), 1, true, 1);
+    }
     // Only for testing
     public Transport() { this("default"); }
 
@@ -72,6 +88,7 @@ public class Transport {
     }
 
     boolean getTcpNoDelay() { return tcpNoDelay; }
+    int getWakeupTriggerCount() { return wakeupTriggerCount; }
 
     String getName() { return name; }
 
diff --git a/jrt/src/com/yahoo/jrt/TransportThread.java b/jrt/src/com/yahoo/jrt/TransportThread.java
index 8f158161888..e42534d12fe 100644
--- a/jrt/src/com/yahoo/jrt/TransportThread.java
+++ b/jrt/src/com/yahoo/jrt/TransportThread.java
@@ -120,15 +120,15 @@ public class TransportThread {
     }
 
     private boolean postCommand(Runnable cmd) {
-        boolean wakeup;
+        int qlen;
         synchronized (this) {
             if (state == CLOSED) {
                 return false;
             }
-            wakeup = queue.isEmpty();
             queue.enqueue(cmd);
+            qlen = queue.size();
         }
-        if (wakeup) {
+        if (qlen == parent.getWakeupTriggerCount()) {
             selector.wakeup();
         }
         return true;
diff --git a/messagebus/src/main/java/com/yahoo/messagebus/network/rpc/RPCNetwork.java b/messagebus/src/main/java/com/yahoo/messagebus/network/rpc/RPCNetwork.java
index b0724ad6029..0f2bc07a366 100644
--- a/messagebus/src/main/java/com/yahoo/messagebus/network/rpc/RPCNetwork.java
+++ b/messagebus/src/main/java/com/yahoo/messagebus/network/rpc/RPCNetwork.java
@@ -87,7 +87,8 @@ public class RPCNetwork implements Network, MethodHandler {
     public RPCNetwork(RPCNetworkParams params, SlobrokConfigSubscriber slobrokConfig) {
         this.slobroksConfig = slobrokConfig;
         identity = params.getIdentity();
-        orb = new Supervisor(new Transport("mbus-rpc-" + identity.getServicePrefix(), params.getNumNetworkThreads(), shouldEnableTcpNodelay(params.getOptimization())));
+        orb = new Supervisor(new Transport("mbus-rpc-" + identity.getServicePrefix(), params.getNumNetworkThreads(),
+                shouldEnableTcpNodelay(params.getOptimization()), params.getWakeupTriggerCount()));
         orb.setMaxInputBufferSize(params.getMaxInputBufferSize());
         orb.setMaxOutputBufferSize(params.getMaxOutputBufferSize());
         targetPool = new RPCTargetPool(params.getConnectionExpireSecs(), params.getNumTargetsPerSpec());
diff --git a/messagebus/src/main/java/com/yahoo/messagebus/network/rpc/RPCNetworkParams.java b/messagebus/src/main/java/com/yahoo/messagebus/network/rpc/RPCNetworkParams.java
index e77cddd8b06..86f426a0a7f 100755
--- a/messagebus/src/main/java/com/yahoo/messagebus/network/rpc/RPCNetworkParams.java
+++ b/messagebus/src/main/java/com/yahoo/messagebus/network/rpc/RPCNetworkParams.java
@@ -21,6 +21,8 @@ public class RPCNetworkParams {
     private double connectionExpireSecs = 30;
     private int numTargetsPerSpec = 1;
     private int numNetworkThreads = 2;
+
+    private int wakeupTriggerCount = 1;
     public enum Optimization {LATENCY, THROUGHPUT}
     Optimization optimization = Optimization.LATENCY;
 
@@ -216,4 +218,13 @@ public class RPCNetworkParams {
         this.maxOutputBufferSize = maxOutputBufferSize;
         return this;
     }
+
+    public int getWakeupTriggerCount() {
+        return wakeupTriggerCount;
+    }
+
+    public RPCNetworkParams setWakeupTriggerCount(int wakeupTriggerCount) {
+        this.wakeupTriggerCount = wakeupTriggerCount;
+        return this;
+    }
 }