diff options
author | Henning Baldersheim <balder@yahoo-inc.com> | 2020-11-30 09:12:36 +0000 |
---|---|---|
committer | Henning Baldersheim <balder@yahoo-inc.com> | 2020-11-30 09:12:36 +0000 |
commit | b54e1aecb0317e7381fbba262bcb29d55af85923 (patch) | |
tree | 567cba69faa3c401a5bfba7b3094eccd44c2ce24 | |
parent | 8001f2c4cf42344178cc4ecb0c695ef701b2bbce (diff) |
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; + } } |