From 71354c405ee87632caa6bb229f064ba0bf1cd3b7 Mon Sep 17 00:00:00 2001 From: Henning Baldersheim Date: Wed, 6 Jul 2022 12:15:30 +0200 Subject: - Control number of rpc targets. - Control number of write events before waking up thread. - For mbus in java, mbus in c++ and the rpc connection towards content nodes. --- .../com/yahoo/config/model/api/ModelContext.java | 12 +++++-- .../yahoo/config/model/deploy/TestProperties.java | 37 ++++++++++++++++++++++ .../container/ApplicationContainerCluster.java | 6 ++++ .../com/yahoo/vespa/model/content/ContentNode.java | 12 +++++++ .../vespa/model/content/StorageClusterTest.java | 14 +++++++- .../config/server/deploy/ModelContextImpl.java | 18 +++++++++++ .../src/main/java/com/yahoo/vespa/flags/Flags.java | 37 ++++++++++++++++++++++ 7 files changed, 132 insertions(+), 4 deletions(-) diff --git a/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java b/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java index 5e9f1f52a0c..c01bf3c234b 100644 --- a/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java +++ b/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java @@ -77,13 +77,19 @@ public interface ModelContext { @ModelFeatureFlag(owners = {"baldersheim"}, comment = "Select sequencer type use while feeding") default String feedSequencerType() { return "THROUGHPUT"; } @ModelFeatureFlag(owners = {"baldersheim"}) default String responseSequencerType() { throw new UnsupportedOperationException("TODO specify default value"); } @ModelFeatureFlag(owners = {"baldersheim"}) default int defaultNumResponseThreads() { return 2; } - @ModelFeatureFlag(owners = {"baldersheim"}) default boolean skipCommunicationManagerThread() { return true; } - @ModelFeatureFlag(owners = {"baldersheim"}) default boolean skipMbusRequestThread() { return true; } - @ModelFeatureFlag(owners = {"baldersheim"}) default boolean skipMbusReplyThread() { return true; } + @ModelFeatureFlag(owners = {"baldersheim"}, removeAfter="7.last") default boolean skipCommunicationManagerThread() { return true; } + @ModelFeatureFlag(owners = {"baldersheim"}, removeAfter="7.last") default boolean skipMbusRequestThread() { return true; } + @ModelFeatureFlag(owners = {"baldersheim"}, removeAfter="7.last") default boolean skipMbusReplyThread() { return true; } @ModelFeatureFlag(owners = {"baldersheim"}) default boolean mbusDispatchOnDecode() { return true; } @ModelFeatureFlag(owners = {"baldersheim"}) default boolean mbusDispatchOnEncode() { return true; } @ModelFeatureFlag(owners = {"baldersheim"}) default int mbusNetworkThreads() { return 1; } @ModelFeatureFlag(owners = {"baldersheim"}) default int mbusThreads() { return 4; } + @ModelFeatureFlag(owners = {"baldersheim"}) default int mbusJavaRpcNumTargets() { return 1; } + @ModelFeatureFlag(owners = {"baldersheim"}) default int mbusJavaEventsBeforeWakeup() { return 1; } + @ModelFeatureFlag(owners = {"baldersheim"}) default int mbusCppRpcNumTargets() { return 1; } + @ModelFeatureFlag(owners = {"baldersheim"}) default int mbusCppEventsBeforeWakeup() { return 1; } + @ModelFeatureFlag(owners = {"baldersheim"}) default int rpcNumTargets() { return 1; } + @ModelFeatureFlag(owners = {"baldersheim"}) default int rpcEventsBeforeWakeup() { return 1; } @ModelFeatureFlag(owners = {"baldersheim"}) default boolean useAsyncMessageHandlingOnSchedule() { throw new UnsupportedOperationException("TODO specify default value"); } @ModelFeatureFlag(owners = {"baldersheim"}) default double feedConcurrency() { throw new UnsupportedOperationException("TODO specify default value"); } @ModelFeatureFlag(owners = {"baldersheim"}) default double feedNiceness() { return 0.0; } diff --git a/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java b/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java index 5f189a63701..5d64dfe7041 100644 --- a/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java +++ b/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java @@ -80,6 +80,12 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea private boolean sharedStringRepoNoReclaim = false; private boolean mbus_dispatch_on_decode = true; private boolean mbus_dispatch_on_encode = true; + private int mbus_java_num_targets = 1; + private int mbus_java_events_before_wakeup = 1; + private int mbus_cpp_num_targets = 1; + private int mbus_cpp_events_before_wakeup = 1; + private int rpc_num_targets = 1; + private int rpc_events_before_wakeup = 1; private int mbus_threads = 4; private int mbus_network_threads = 1; private Architecture adminClusterNodeResourcesArchitecture = Architecture.getDefault(); @@ -141,6 +147,12 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea @Override public boolean mbusDispatchOnEncode() { return mbus_dispatch_on_encode; } @Override public int mbusNetworkThreads() { return mbus_network_threads; } @Override public int mbusThreads() { return mbus_threads; } + @Override public int mbusJavaRpcNumTargets() { return mbus_java_num_targets; } + @Override public int mbusJavaEventsBeforeWakeup() { return mbus_java_events_before_wakeup; } + @Override public int mbusCppRpcNumTargets() { return mbus_cpp_num_targets; } + @Override public int mbusCppEventsBeforeWakeup() { return mbus_cpp_events_before_wakeup; } + @Override public int rpcNumTargets() { return rpc_num_targets; } + @Override public int rpcEventsBeforeWakeup() { return rpc_events_before_wakeup; } public TestProperties sharedStringRepoNoReclaim(boolean sharedStringRepoNoReclaim) { @@ -385,6 +397,31 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea this.mbus_network_threads = value; return this; } + public TestProperties setMbusJavaRpcNumTargets(int value) { + this.mbus_java_num_targets = value; + return this; + } + public TestProperties setMbusJavaEventsBeforeWakeup(int value) { + this.mbus_java_events_before_wakeup = value; + return this; + } + public TestProperties setMbusCppEventsBeforeWakeup(int value) { + this.mbus_cpp_events_before_wakeup = value; + return this; + } + public TestProperties setMbusCppRpcNumTargets(int value) { + this.mbus_cpp_num_targets = value; + return this; + } + public TestProperties setRpcNumTargets(int value) { + this.rpc_num_targets = value; + return this; + } + public TestProperties setRpcEventsBeforeWakeup(int value) { + this.rpc_events_before_wakeup = value; + return this; + } + public TestProperties setAdminClusterNodeResourcesArchitecture(Architecture architecture) { this.adminClusterNodeResourcesArchitecture = architecture; return this; diff --git a/config-model/src/main/java/com/yahoo/vespa/model/container/ApplicationContainerCluster.java b/config-model/src/main/java/com/yahoo/vespa/model/container/ApplicationContainerCluster.java index cb4fe8f67ca..9785858b338 100644 --- a/config-model/src/main/java/com/yahoo/vespa/model/container/ApplicationContainerCluster.java +++ b/config-model/src/main/java/com/yahoo/vespa/model/container/ApplicationContainerCluster.java @@ -89,6 +89,8 @@ public final class ApplicationContainerCluster extends ContainerCluster parent, String clusterName, String rootDirectory, int distributionKey) { super(parent, "" + distributionKey); @@ -34,6 +38,10 @@ public abstract class ContentNode extends AbstractService dispatch_on_encode = featureFlags.mbusDispatchOnEncode(); mbus_threads = featureFlags.mbusThreads(); mbus_network_threads = featureFlags.mbusNetworkThreads(); + mbus_cpp_rpc_targets = featureFlags.mbusCppRpcNumTargets(); + mbus_events_before_wakeup = featureFlags.mbusCppEventsBeforeWakeup(); + rpc_num_targets = featureFlags.rpcNumTargets(); + rpc_events_before_wakeup = featureFlags.rpcEventsBeforeWakeup(); initialize(); setProp("clustertype", "content"); @@ -81,6 +89,10 @@ public abstract class ContentNode extends AbstractService builder.mbus.dispatch_on_encode(dispatch_on_encode); builder.mbus.num_threads(mbus_threads); builder.mbus.num_network_threads(mbus_network_threads); + builder.mbus.num_rpc_targets(mbus_cpp_rpc_targets); + builder.mbus.events_before_wakeup(mbus_events_before_wakeup); + builder.rpc.num_targets_per_node(rpc_num_targets); + builder.rpc.events_before_wakeup(rpc_events_before_wakeup); } @Override diff --git a/config-model/src/test/java/com/yahoo/vespa/model/content/StorageClusterTest.java b/config-model/src/test/java/com/yahoo/vespa/model/content/StorageClusterTest.java index 5748f260bf8..b74fc763462 100644 --- a/config-model/src/test/java/com/yahoo/vespa/model/content/StorageClusterTest.java +++ b/config-model/src/test/java/com/yahoo/vespa/model/content/StorageClusterTest.java @@ -158,6 +158,10 @@ public class StorageClusterTest { assertTrue(confg.mbus().dispatch_on_encode()); assertEquals(4, confg.mbus().num_threads()); assertEquals(1, confg.mbus().num_network_threads()); + assertEquals(1, confg.mbus().num_rpc_targets()); + assertEquals(1, confg.mbus().events_before_wakeup()); + assertEquals(1, confg.rpc().num_targets_per_node()); + assertEquals(1, confg.rpc().events_before_wakeup()); } @Test @@ -166,11 +170,19 @@ public class StorageClusterTest { .setMbusDispatchOnDecode(false) .setMbusDispatchOnEncode(false) .setMbusThreads(3) - .setMbusNetworkThreads(7)); + .setMbusNetworkThreads(7) + .setRpcNumTargets(11) + .setRpcEventsBeforeWakeup(12) + .setMbusCppRpcNumTargets(8) + .setMbusCppEventsBeforeWakeup(9)); assertFalse(confg.mbus().dispatch_on_decode()); assertFalse(confg.mbus().dispatch_on_encode()); assertEquals(3, confg.mbus().num_threads()); assertEquals(7, confg.mbus().num_network_threads()); + assertEquals(8, confg.mbus().num_rpc_targets()); + assertEquals(9, confg.mbus().events_before_wakeup()); + assertEquals(11, confg.rpc().num_targets_per_node()); + assertEquals(12, confg.rpc().events_before_wakeup()); } @Test diff --git a/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java b/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java index 47d1193cd4c..177f7c5b823 100644 --- a/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java +++ b/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java @@ -215,6 +215,12 @@ public class ModelContextImpl implements ModelContext { private final boolean mbus_dispatch_on_encode; private final int mbus_threads; private final int mbus_network_threads; + private int mbus_java_num_targets; + private int mbus_java_events_before_wakeup; + private int mbus_cpp_num_targets; + private int mbus_cpp_events_before_wakeup; + private int rpc_num_targets; + private int rpc_events_before_wakeup; public FeatureFlags(FlagSource source, ApplicationId appId, Version version) { this.defaultTermwiseLimit = flagValue(source, appId, version, Flags.DEFAULT_TERM_WISE_LIMIT); @@ -266,6 +272,12 @@ public class ModelContextImpl implements ModelContext { this.enableProxyProtocolMixedMode = flagValue(source, appId, version, Flags.ENABLE_PROXY_PROTOCOL_MIXED_MODE); this.sharedStringRepoNoReclaim = flagValue(source, appId, version, Flags.SHARED_STRING_REPO_NO_RECLAIM); this.logFileCompressionAlgorithm = flagValue(source, appId, version, Flags.LOG_FILE_COMPRESSION_ALGORITHM); + this.mbus_java_num_targets = flagValue(source, appId, version, Flags.MBUS_JAVA_NUM_TARGETS); + this.mbus_java_events_before_wakeup = flagValue(source, appId, version, Flags.MBUS_JAVA_EVENTS_BEFORE_WAKEUP); + this.mbus_cpp_num_targets = flagValue(source, appId, version, Flags.MBUS_CPP_NUM_TARGETS); + this.mbus_cpp_events_before_wakeup = flagValue(source, appId, version, Flags.MBUS_CPP_EVENTS_BEFORE_WAKEUP); + this.rpc_num_targets = flagValue(source, appId, version, Flags.RPC_NUM_TARGETS); + this.rpc_events_before_wakeup = flagValue(source, appId, version, Flags.RPC_EVENTS_BEFORE_WAKEUP); } @Override public double defaultTermwiseLimit() { return defaultTermwiseLimit; } @@ -318,6 +330,12 @@ public class ModelContextImpl implements ModelContext { @Override public Architecture adminClusterArchitecture() { return adminClusterArchitecture; } @Override public boolean enableProxyProtocolMixedMode() { return enableProxyProtocolMixedMode; } @Override public boolean sharedStringRepoNoReclaim() { return sharedStringRepoNoReclaim; } + @Override public int mbusJavaRpcNumTargets() { return mbus_java_num_targets; } + @Override public int mbusJavaEventsBeforeWakeup() { return mbus_java_events_before_wakeup; } + @Override public int mbusCppRpcNumTargets() { return mbus_cpp_num_targets; } + @Override public int mbusCppEventsBeforeWakeup() { return mbus_cpp_events_before_wakeup; } + @Override public int rpcNumTargets() { return rpc_num_targets; } + @Override public int rpcEventsBeforeWakeup() { return rpc_events_before_wakeup; } @Override public String logFileCompressionAlgorithm(String defVal) { var fflag = this.logFileCompressionAlgorithm; if (fflag != null && ! fflag.equals("")) { diff --git a/flags/src/main/java/com/yahoo/vespa/flags/Flags.java b/flags/src/main/java/com/yahoo/vespa/flags/Flags.java index f30cfbecc53..00ca552e2c8 100644 --- a/flags/src/main/java/com/yahoo/vespa/flags/Flags.java +++ b/flags/src/main/java/com/yahoo/vespa/flags/Flags.java @@ -143,6 +143,43 @@ public class Flags { "Takes effect at redeployment", ZONE_ID, APPLICATION_ID); + public static final UnboundIntFlag MBUS_JAVA_NUM_TARGETS = defineIntFlag( + "mbus-java-num-targets", 1, + List.of("baldersheim"), "2022-07-05", "2023-01-01", + "Number of rpc targets per service", + "Takes effect at redeployment", + ZONE_ID, APPLICATION_ID); + public static final UnboundIntFlag MBUS_CPP_NUM_TARGETS = defineIntFlag( + "mbus-cpp-num-targets", 1, + List.of("baldersheim"), "2022-07-05", "2023-01-01", + "Number of rpc targets per service", + "Takes effect at redeployment", + ZONE_ID, APPLICATION_ID); + public static final UnboundIntFlag RPC_NUM_TARGETS = defineIntFlag( + "rpc-num-targets", 1, + List.of("baldersheim"), "2022-07-05", "2023-01-01", + "Number of rpc targets per content node", + "Takes effect at redeployment", + ZONE_ID, APPLICATION_ID); + public static final UnboundIntFlag MBUS_JAVA_EVENTS_BEFORE_WAKEUP = defineIntFlag( + "mbus-java-events-before-wakeup", 1, + List.of("baldersheim"), "2022-07-05", "2023-01-01", + "Number write events before waking up transport thread", + "Takes effect at redeployment", + ZONE_ID, APPLICATION_ID); + public static final UnboundIntFlag MBUS_CPP_EVENTS_BEFORE_WAKEUP = defineIntFlag( + "mbus-cpp-events-before-wakeup", 1, + List.of("baldersheim"), "2022-07-05", "2023-01-01", + "Number write events before waking up transport thread", + "Takes effect at redeployment", + ZONE_ID, APPLICATION_ID); + public static final UnboundIntFlag RPC_EVENTS_BEFORE_WAKEUP = defineIntFlag( + "rpc-events-before-wakeup", 1, + List.of("baldersheim"), "2022-07-05", "2023-01-01", + "Number write events before waking up transport thread", + "Takes effect at redeployment", + ZONE_ID, APPLICATION_ID); + public static final UnboundIntFlag MBUS_NUM_THREADS = defineIntFlag( "mbus-num-threads", 4, List.of("baldersheim"), "2022-07-01", "2023-01-01", -- cgit v1.2.3 From c3196c5567fa7b3b5d09bde4b2530cf6e9668f10 Mon Sep 17 00:00:00 2001 From: Henning Baldersheim Date: Wed, 6 Jul 2022 14:48:32 +0200 Subject: Unify communicationmanager config member naming --- .../src/main/java/com/yahoo/vespa/model/content/ContentNode.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/config-model/src/main/java/com/yahoo/vespa/model/content/ContentNode.java b/config-model/src/main/java/com/yahoo/vespa/model/content/ContentNode.java index 776da1c8a95..6ff9329ce7e 100644 --- a/config-model/src/main/java/com/yahoo/vespa/model/content/ContentNode.java +++ b/config-model/src/main/java/com/yahoo/vespa/model/content/ContentNode.java @@ -25,7 +25,7 @@ public abstract class ContentNode extends AbstractService private final boolean dispatch_on_decode; private final int mbus_threads; private final int mbus_network_threads; - private final int mbus_cpp_rpc_targets; + private final int mbus_rpc_targets; private final int mbus_events_before_wakeup; private final int rpc_num_targets; private final int rpc_events_before_wakeup; @@ -38,7 +38,7 @@ public abstract class ContentNode extends AbstractService dispatch_on_encode = featureFlags.mbusDispatchOnEncode(); mbus_threads = featureFlags.mbusThreads(); mbus_network_threads = featureFlags.mbusNetworkThreads(); - mbus_cpp_rpc_targets = featureFlags.mbusCppRpcNumTargets(); + mbus_rpc_targets = featureFlags.mbusCppRpcNumTargets(); mbus_events_before_wakeup = featureFlags.mbusCppEventsBeforeWakeup(); rpc_num_targets = featureFlags.rpcNumTargets(); rpc_events_before_wakeup = featureFlags.rpcEventsBeforeWakeup(); @@ -89,7 +89,7 @@ public abstract class ContentNode extends AbstractService builder.mbus.dispatch_on_encode(dispatch_on_encode); builder.mbus.num_threads(mbus_threads); builder.mbus.num_network_threads(mbus_network_threads); - builder.mbus.num_rpc_targets(mbus_cpp_rpc_targets); + builder.mbus.num_rpc_targets(mbus_rpc_targets); builder.mbus.events_before_wakeup(mbus_events_before_wakeup); builder.rpc.num_targets_per_node(rpc_num_targets); builder.rpc.events_before_wakeup(rpc_events_before_wakeup); -- cgit v1.2.3