[java] Deprecate max-threads flag. Add an alternate flag in the distr…

…ibutor for new session thread pool size. (#10995) * [java] Make new session creation threadpool size configurable * [java] Deprecate max-threads flag Co-authored-by: Diego Molina <[email protected]>
SeleniumHQ · Oct 17, 2022 · 01bf373 · 01bf373
1 parent 02b23e0
commit 01bf373
Show file tree

Hide file tree

Showing 20 changed files with 207 additions and 60 deletions.
diff --git a/java/src/org/openqa/selenium/grid/commands/EventBusCommand.java b/java/src/org/openqa/selenium/grid/commands/EventBusCommand.java
@@ -45,6 +45,7 @@
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import static java.net.HttpURLConnection.HTTP_NO_CONTENT;
@@ -137,6 +138,13 @@ public Server<?> asServer(Config initialConfig) {
   protected void execute(Config config) {
     Require.nonNull("Config", config);
 
+    config.get("server", "max-threads")
+      .ifPresent(value -> LOG.log(Level.WARNING,
+                                  () ->
+                                    "Support for max-threads flag is deprecated. " +
+                                    "The intent of the flag is to set the thread pool size in the Distributor. " +
+                                    "Please use newsession-threadpool-size flag instead."));
+
     Server<?> server = asServer(config);
     server.start();
 

diff --git a/java/src/org/openqa/selenium/grid/commands/Hub.java b/java/src/org/openqa/selenium/grid/commands/Hub.java
@@ -62,6 +62,7 @@
 import java.net.URL;
 import java.util.Collections;
 import java.util.Set;
+import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import static java.net.HttpURLConnection.HTTP_OK;
@@ -163,7 +164,8 @@ protected Handlers createHandlers(Config config) {
       secret,
       distributorOptions.getHealthCheckInterval(),
       distributorOptions.shouldRejectUnsupportedCaps(),
-      newSessionRequestOptions.getSessionRequestRetryInterval());
+      newSessionRequestOptions.getSessionRequestRetryInterval(),
+      distributorOptions.getNewSessionThreadPoolSize());
     handler.addHandler(distributor);
 
     Router router = new Router(tracer, clientFactory, sessions, queue, distributor);
@@ -207,6 +209,13 @@ protected Handlers createHandlers(Config config) {
   protected void execute(Config config) {
     Require.nonNull("Config", config);
 
+    config.get("server", "max-threads")
+      .ifPresent(value -> LOG.log(Level.WARNING,
+                                  () ->
+                                    "Support for max-threads flag is deprecated. " +
+                                    "The intent of the flag is to set the thread pool size in the Distributor. " +
+                                    "Please use newsession-threadpool-size flag instead."));
+
     Server<?> server = asServer(config).start();
 
     LOG.info(String.format("Started Selenium Hub %s: %s", getServerVersion(), server.getUrl()));

diff --git a/java/src/org/openqa/selenium/grid/commands/Standalone.java b/java/src/org/openqa/selenium/grid/commands/Standalone.java
@@ -65,7 +65,9 @@
 import java.net.URI;
 import java.net.URL;
 import java.util.Collections;
+import java.util.Optional;
 import java.util.Set;
+import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import static java.net.HttpURLConnection.HTTP_OK;
@@ -162,7 +164,8 @@ protected Handlers createHandlers(Config config) {
       registrationSecret,
       distributorOptions.getHealthCheckInterval(),
       distributorOptions.shouldRejectUnsupportedCaps(),
-      newSessionRequestOptions.getSessionRequestRetryInterval());
+      newSessionRequestOptions.getSessionRequestRetryInterval(),
+      distributorOptions.getNewSessionThreadPoolSize());
     combinedHandler.addHandler(distributor);
 
     Routable router = new Router(tracer, clientFactory, sessions, queue, distributor)
@@ -232,6 +235,13 @@ protected Handlers createHandlers(Config config) {
   protected void execute(Config config) {
     Require.nonNull("Config", config);
 
+    config.get("server", "max-threads")
+      .ifPresent(value -> LOG.log(Level.WARNING,
+                                  () ->
+                                    "Support for max-threads flag is deprecated. " +
+                                    "The intent of the flag is to set the thread pool size in the Distributor. " +
+                                    "Please use newsession-threadpool-size flag instead."));
+
     Server<?> server = asServer(config).start();
 
     LOG.info(String.format(

diff --git a/java/src/org/openqa/selenium/grid/distributor/config/DistributorFlags.java b/java/src/org/openqa/selenium/grid/distributor/config/DistributorFlags.java
@@ -33,6 +33,7 @@
 import static org.openqa.selenium.grid.config.StandardGridRoles.DISTRIBUTOR_ROLE;
 import static org.openqa.selenium.grid.distributor.config.DistributorOptions.DEFAULT_DISTRIBUTOR_IMPLEMENTATION;
 import static org.openqa.selenium.grid.distributor.config.DistributorOptions.DEFAULT_HEALTHCHECK_INTERVAL;
+import static org.openqa.selenium.grid.distributor.config.DistributorOptions.DEFAULT_NEWSESSION_THREADPOOL_SIZE;
 import static org.openqa.selenium.grid.distributor.config.DistributorOptions.DEFAULT_REJECT_UNSUPPORTED_CAPS;
 import static org.openqa.selenium.grid.distributor.config.DistributorOptions.DEFAULT_SLOT_MATCHER;
 import static org.openqa.selenium.grid.distributor.config.DistributorOptions.DEFAULT_SLOT_SELECTOR_IMPLEMENTATION;
@@ -92,6 +93,15 @@ public class DistributorFlags implements HasRoles {
   @ConfigValue(section = DISTRIBUTOR_SECTION, name = "reject-unsupported-caps", example = "true")
   private boolean rejectUnsupportedCaps = DEFAULT_REJECT_UNSUPPORTED_CAPS;
 
+  @Parameter(
+    names = {"--newsession-threadpool-size"},
+    description = "The Distributor uses a fixed-sized thread pool to create new sessions as it consumes new session requests from the queue."
+                  + "This allows configuring the size of the thread pool. The default value is no. of available processors * 3. "
+                  + "Note: If the no. of threads is way greater than the available processors it will not always increase the performance. "
+                  + "A high number of threads causes more context switching which is an expensive operation. ")
+  @ConfigValue(section = DISTRIBUTOR_SECTION, name = "newsession-threadpool-size", example = "4")
+  public int newSessionThreadPoolSize = DEFAULT_NEWSESSION_THREADPOOL_SIZE;
+
   @Override
   public Set<Role> getRoles() {
     return Collections.singleton(DISTRIBUTOR_ROLE);

diff --git a/java/src/org/openqa/selenium/grid/distributor/config/DistributorOptions.java b/java/src/org/openqa/selenium/grid/distributor/config/DistributorOptions.java
@@ -38,6 +38,8 @@ public class DistributorOptions {
   static final String DEFAULT_SLOT_SELECTOR_IMPLEMENTATION =
     "org.openqa.selenium.grid.distributor.selector.DefaultSlotSelector";
   static final boolean DEFAULT_REJECT_UNSUPPORTED_CAPS = false;
+  static final int DEFAULT_NEWSESSION_THREADPOOL_SIZE =
+    Runtime.getRuntime().availableProcessors() * 3;
   private final Config config;
 
   public DistributorOptions(Config config) {
@@ -117,6 +119,14 @@ public SlotSelector getSlotSelector() {
       DEFAULT_SLOT_SELECTOR_IMPLEMENTATION);
   }
 
+  public int getNewSessionThreadPoolSize() {
+    // If the user sets 0 or less, we default to 1 to ensure Grid is running.
+    return Math.max(
+      config.getInt(DISTRIBUTOR_SECTION, "newsession-threadpool-size")
+        .orElse(DEFAULT_NEWSESSION_THREADPOOL_SIZE),
+      1);
+  }
+
   public boolean shouldRejectUnsupportedCaps() {
     return config.getBool(DISTRIBUTOR_SECTION,
       "reject-unsupported-caps").orElse(DEFAULT_REJECT_UNSUPPORTED_CAPS);

diff --git a/java/src/org/openqa/selenium/grid/distributor/httpd/DistributorServer.java b/java/src/org/openqa/selenium/grid/distributor/httpd/DistributorServer.java
@@ -38,6 +38,7 @@
 
 import java.util.Collections;
 import java.util.Set;
+import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import static java.net.HttpURLConnection.HTTP_OK;
@@ -116,6 +117,14 @@ protected Handlers createHandlers(Config config) {
   protected void execute(Config config) {
     Require.nonNull("Config", config);
 
+    config.get("server", "max-threads")
+      .ifPresent(value -> LOG.log(Level.WARNING,
+                                  () ->
+                                    "Support for max-threads flag is deprecated. " +
+                                    "The intent of the flag is to set the thread pool size in the Distributor. " +
+                                    "Please use newsession-threadpool-size flag instead."));
+
+
     Server<?> server = asServer(config).start();
 
     BuildInfo info = new BuildInfo();

diff --git a/java/src/org/openqa/selenium/grid/distributor/local/LocalDistributor.java b/java/src/org/openqa/selenium/grid/distributor/local/LocalDistributor.java
@@ -168,15 +168,8 @@ public class LocalDistributor extends Distributor implements Closeable {
         return thread;
       });
 
-  private final Executor sessionCreatorExecutor = Executors.newFixedThreadPool(
-    Runtime.getRuntime().availableProcessors(),
-    r -> {
-      Thread thread = new Thread(r);
-      thread.setName("Local Distributor - Session Creation");
-      thread.setDaemon(true);
-      return thread;
-    }
-  );
+  private final Executor sessionCreatorExecutor;
+
   private final NewSessionQueue sessionQueue;
 
   private final boolean rejectUnsupportedCaps;
@@ -191,7 +184,8 @@ public LocalDistributor(
     Secret registrationSecret,
     Duration healthcheckInterval,
     boolean rejectUnsupportedCaps,
-    Duration sessionRequestRetryInterval) {
+    Duration sessionRequestRetryInterval,
+    int newSessionThreadPoolSize) {
     super(tracer, clientFactory, registrationSecret);
     this.tracer = Require.nonNull("Tracer", tracer);
     this.bus = Require.nonNull("Event bus", bus);
@@ -217,6 +211,16 @@ public LocalDistributor(
       }
     }));
 
+    sessionCreatorExecutor =  Executors.newFixedThreadPool(
+      newSessionThreadPoolSize,
+      r -> {
+        Thread thread = new Thread(r);
+        thread.setName("Local Distributor - Session Creation");
+        thread.setDaemon(true);
+        return thread;
+      }
+    );
+
     NewSessionRunnable newSessionRunnable = new NewSessionRunnable();
     bus.addListener(NodeDrainComplete.listener(this::remove));
 
@@ -262,7 +266,8 @@ public static Distributor create(Config config) {
       secretOptions.getRegistrationSecret(),
       distributorOptions.getHealthCheckInterval(),
       distributorOptions.shouldRejectUnsupportedCaps(),
-      newSessionQueueOptions.getSessionRequestRetryInterval());
+      newSessionQueueOptions.getSessionRequestRetryInterval(),
+      distributorOptions.getNewSessionThreadPoolSize());
   }
 
   @Override

diff --git a/java/src/org/openqa/selenium/grid/node/httpd/NodeServer.java b/java/src/org/openqa/selenium/grid/node/httpd/NodeServer.java
@@ -58,6 +58,7 @@
 import java.util.Set;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import static java.net.HttpURLConnection.HTTP_NO_CONTENT;
@@ -227,6 +228,13 @@ public NettyServer start() {
   protected void execute(Config config) {
     Require.nonNull("Config", config);
 
+    config.get("server", "max-threads")
+      .ifPresent(value -> LOG.log(Level.WARNING,
+                                  () ->
+                                    "Support for max-threads flag is deprecated. " +
+                                    "The intent of the flag is to set the thread pool size in the Distributor. " +
+                                    "Please use newsession-threadpool-size flag instead."));
+
     Runtime.getRuntime().addShutdownHook(shutdownHook);
     Server<?> server = asServer(config).start();
 

diff --git a/java/src/org/openqa/selenium/grid/router/httpd/RouterServer.java b/java/src/org/openqa/selenium/grid/router/httpd/RouterServer.java
@@ -61,6 +61,7 @@
 import java.time.Duration;
 import java.util.Collections;
 import java.util.Set;
+import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import static java.net.HttpURLConnection.HTTP_OK;
@@ -192,6 +193,14 @@ protected Handlers createHandlers(Config config) {
   protected void execute(Config config) {
     Require.nonNull("Config", config);
 
+    config.get("server", "max-threads")
+      .ifPresent(value -> LOG.log(Level.WARNING,
+                                  () ->
+                                    "Support for max-threads flag is deprecated. " +
+                                    "The intent of the flag is to set the thread pool size in the Distributor. " +
+                                    "Please use newsession-threadpool-size flag instead."));
+
+
     Server<?> server = asServer(config).start();
 
     LOG.info(String.format(

diff --git a/java/src/org/openqa/selenium/grid/sessionmap/httpd/SessionMapServer.java b/java/src/org/openqa/selenium/grid/sessionmap/httpd/SessionMapServer.java
@@ -35,6 +35,7 @@
 
 import java.util.Collections;
 import java.util.Set;
+import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import static java.net.HttpURLConnection.HTTP_NO_CONTENT;
@@ -103,6 +104,14 @@ protected Handlers createHandlers(Config config) {
 
   @Override
   protected void execute(Config config) {
+
+    config.get("server", "max-threads")
+      .ifPresent(value -> LOG.log(Level.WARNING,
+                                  () ->
+                                    "Support for max-threads flag is deprecated. " +
+                                    "The intent of the flag is to set the thread pool size in the Distributor. " +
+                                    "Please use newsession-threadpool-size flag instead."));
+
     Server<?> server = asServer(config);
     server.start();
 

diff --git a/java/src/org/openqa/selenium/grid/sessionqueue/httpd/NewSessionQueueServer.java b/java/src/org/openqa/selenium/grid/sessionqueue/httpd/NewSessionQueueServer.java
@@ -35,6 +35,7 @@
 
 import java.util.Collections;
 import java.util.Set;
+import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import static java.net.HttpURLConnection.HTTP_NO_CONTENT;
@@ -105,6 +106,13 @@ protected Handlers createHandlers(Config config) {
   protected void execute(Config config) {
     Require.nonNull("Config", config);
 
+    config.get("server", "max-threads")
+      .ifPresent(value -> LOG.log(Level.WARNING,
+                                  () ->
+                                    "Support for max-threads flag is deprecated. " +
+                                    "The intent of the flag is to set the thread pool size in the Distributor. " +
+                                    "Please use newsession-threadpool-size flag instead."));
+
     Server<?> server = asServer(config);
     server.start();
 

diff --git a/java/test/org/openqa/selenium/grid/distributor/AddingNodesTest.java b/java/test/org/openqa/selenium/grid/distributor/AddingNodesTest.java
@@ -84,6 +84,7 @@ class AddingNodesTest {
 
   private static final Capabilities CAPS = new ImmutableCapabilities("cheese", "gouda");
   private static final Secret registrationSecret = new Secret("caerphilly");
+  private static final int newSessionThreadPoolSize = Runtime.getRuntime().availableProcessors();
 
   private Distributor distributor;
   private Tracer tracer;
@@ -138,7 +139,8 @@ void shouldBeAbleToRegisterALocalNode() throws URISyntaxException {
       registrationSecret,
       Duration.ofMinutes(5),
       false,
-      Duration.ofSeconds(5));
+      Duration.ofSeconds(5),
+      newSessionThreadPoolSize);
 
     distributor = new RemoteDistributor(tracer, new PassthroughHttpClient.Factory(local), externalUrl, registrationSecret);
 
@@ -170,7 +172,8 @@ void shouldBeAbleToRegisterACustomNode() throws URISyntaxException {
       registrationSecret,
       Duration.ofMinutes(5),
       false,
-      Duration.ofSeconds(5));
+      Duration.ofSeconds(5),
+      newSessionThreadPoolSize);
 
     distributor = new RemoteDistributor(tracer, new PassthroughHttpClient.Factory(local), externalUrl, registrationSecret);
 
@@ -203,7 +206,8 @@ void shouldBeAbleToRegisterNodesByListeningForEvents() throws URISyntaxException
       registrationSecret,
       Duration.ofMinutes(5),
       false,
-      Duration.ofSeconds(5));
+      Duration.ofSeconds(5),
+      newSessionThreadPoolSize);
 
     distributor = new RemoteDistributor(tracer, new PassthroughHttpClient.Factory(local), externalUrl, registrationSecret);
 
@@ -246,7 +250,8 @@ void shouldKeepOnlyOneNodeWhenTwoRegistrationsHaveTheSameUriByListeningForEvents
       registrationSecret,
       Duration.ofMinutes(5),
       false,
-      Duration.ofSeconds(5));
+      Duration.ofSeconds(5),
+      newSessionThreadPoolSize);
 
     distributor = new RemoteDistributor(tracer, new PassthroughHttpClient.Factory(local), externalUrl, registrationSecret);
 
@@ -282,7 +287,8 @@ void distributorShouldUpdateStateOfExistingNodeWhenNodePublishesStateChange()
       registrationSecret,
       Duration.ofMinutes(5),
       false,
-      Duration.ofSeconds(5));
+      Duration.ofSeconds(5),
+      newSessionThreadPoolSize);
 
     distributor = new RemoteDistributor(tracer, new PassthroughHttpClient.Factory(local), externalUrl, registrationSecret);