Config option to allow config netty event loop threads

Added configuration option to allow users to config how many netty event loop threads they want to use.

Author: Zhen Li

driver/src/main/java/org/neo4j/driver/Config.java

@@ -94,13 +94,14 @@ public class Config
     private final int routingFailureLimit;
     private final long routingRetryDelayMillis;
     private final long fetchSize;
-    private long routingTablePurgeDelayMillis;
+    private final long routingTablePurgeDelayMillis;
 
     private final int connectionTimeoutMillis;
     private final RetrySettings retrySettings;
     private final ServerAddressResolver resolver;
 
     private final boolean isMetricsEnabled;
+    private final int eventLoopThreads;
 
     private Config( ConfigBuilder builder )
     {
@@ -122,6 +123,7 @@ private Config( ConfigBuilder builder )
         this.resolver = builder.resolver;
         this.fetchSize = builder.fetchSize;
 
+        this.eventLoopThreads = builder.eventLoopThreads;
         this.isMetricsEnabled = builder.isMetricsEnabled;
     }
 
@@ -242,6 +244,11 @@ public long fetchSize()
         return fetchSize;
     }
 
+    public int eventLoopThreads()
+    {
+        return eventLoopThreads;
+    }
+
     /**
      * @return if the metrics is enabled or not on this driver.
      */
@@ -271,6 +278,7 @@ public static class ConfigBuilder
         private ServerAddressResolver resolver;
         private boolean isMetricsEnabled = false;
         private long fetchSize = FetchSizeUtil.DEFAULT_FETCH_SIZE;
+        private int eventLoopThreads = 0;
 
         private ConfigBuilder() {}
 
@@ -699,6 +707,23 @@ public ConfigBuilder withoutDriverMetrics()
             return this;
         }
 
+        /**
+         * Configure the event loop thread count. This specifies how many threads the driver can use to handle network I/O events
+         * and user's events in driver's I/O threads. By default, 2 * NumberOfProcessors amount of threads will be used instead.
+         * @param size the thread count.
+         * @return this builder.
+         * @throws IllegalArgumentException if the value of the size is set to a number that is less than 1.
+         */
+        public ConfigBuilder withEventLoopThreads( int size )
+        {
+            if ( size < 1 )
+            {
+                throw new IllegalArgumentException( String.format( "The event loop thread may not be smaller than 1, but was %d.", size ) );
+            }
+            this.eventLoopThreads = size;
+            return this;
+        }
+
         /**
          * Create a config instance from this builder.
          *

driver/src/main/java/org/neo4j/driver/internal/DriverFactory.java

@@ -82,7 +82,7 @@ public final Driver newInstance ( URI uri, AuthToken authToken, RoutingSettings
         boolean ownsEventLoopGroup;
         if ( eventLoopGroup == null )
         {
-            bootstrap = createBootstrap();
+            bootstrap = createBootstrap( config.eventLoopThreads() );
             ownsEventLoopGroup = true;
         }
         else
@@ -272,9 +272,9 @@ protected RetryLogic createRetryLogic( RetrySettings settings, EventExecutorGrou
      * <p>
      * <b>This method is protected only for testing</b>
      */
-    protected Bootstrap createBootstrap()
+    protected Bootstrap createBootstrap( int size )
     {
-        return BootstrapFactory.newBootstrap();
+        return BootstrapFactory.newBootstrap( size );
     }
 
     /**

driver/src/test/java/org/neo4j/driver/integration/DirectDriverIT.java

@@ -25,10 +25,11 @@
 
 import java.net.URI;
 
+import org.neo4j.driver.Config;
 import org.neo4j.driver.Driver;
 import org.neo4j.driver.GraphDatabase;
-import org.neo4j.driver.Session;
 import org.neo4j.driver.Result;
+import org.neo4j.driver.Session;
 import org.neo4j.driver.internal.BoltServerAddress;
 import org.neo4j.driver.util.DatabaseExtension;
 import org.neo4j.driver.util.ParallelizableIT;
@@ -37,6 +38,7 @@
 import static org.hamcrest.core.IsEqual.equalTo;
 import static org.hamcrest.junit.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.neo4j.driver.Values.parameters;
 import static org.neo4j.driver.internal.util.Matchers.directDriverWithAddress;
 
 @ParallelizableIT
@@ -109,4 +111,28 @@ void shouldConnectIPv6Uri()
             assertThat( result.single().get( 0 ).asInt(), CoreMatchers.equalTo( 1 ) );
         }
     }
+
+    @Test
+    void shouldRespectEventLoopThreadsConfig()
+    {
+        // Given
+        URI uri = neo4j.uri();
+        Config config = Config.builder().withEventLoopThreads( 2 ).withoutEncryption().build();
+        int start = Thread.activeCount();
+
+        // When
+        driver = GraphDatabase.driver( uri, neo4j.authToken(), config );
+
+        for ( int i = 0; i < 1000; i++ )
+        {
+            try ( Session session = driver.session() )
+            {
+                int size = i;
+                session.readTransaction( tx -> tx.run( "UNWIND range(1, $size) AS x RETURN count(x)", parameters( "size", size ) ).consume() );
+            }
+        }
+
+        // Then only two more threads are added
+        assertThat( Thread.activeCount() - start, equalTo( 2 ) );
+    }
 }

driver/src/test/java/org/neo4j/driver/internal/DriverFactoryTest.java

@@ -262,7 +262,7 @@ private static class DriverFactoryWithSessions extends DriverFactory
         }
 
         @Override
-        protected Bootstrap createBootstrap()
+        protected Bootstrap createBootstrap( int ignored )
         {
             return BootstrapFactory.newBootstrap( 1 );
         }

driver/src/test/java/org/neo4j/driver/internal/util/DriverFactoryWithOneEventLoopThread.java

@@ -38,7 +38,7 @@ public Driver newInstance( URI uri, AuthToken authToken, Config config )
     }
 
     @Override
-    protected Bootstrap createBootstrap()
+    protected Bootstrap createBootstrap( int ignored )
     {
         return BootstrapFactory.newBootstrap( 1 );
     }

driver/src/test/java/org/neo4j/driver/internal/util/io/ChannelTrackingDriverFactory.java

@@ -60,9 +60,9 @@ public ChannelTrackingDriverFactory( int eventLoopThreads, Clock clock )
     }
 
     @Override
-    protected Bootstrap createBootstrap()
+    protected Bootstrap createBootstrap( int size )
     {
-        return eventLoopThreads == 0 ? super.createBootstrap() : BootstrapFactory.newBootstrap( eventLoopThreads );
+        return BootstrapFactory.newBootstrap( eventLoopThreads );
     }
 
     @Override