Merge pull request #652 from zhenlineo/4.0-event-loop-thread

Config option to allow config netty event loop threads

Author: zhenlineo

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/main/java/org/neo4j/driver/internal/async/connection/BootstrapFactory.java

@@ -28,11 +28,6 @@ private BootstrapFactory()
     {
     }
 
-    public static Bootstrap newBootstrap()
-    {
-        return newBootstrap( EventLoopGroupFactory.newEventLoopGroup() );
-    }
-
     public static Bootstrap newBootstrap( int threadCount )
     {
         return newBootstrap( EventLoopGroupFactory.newEventLoopGroup( threadCount ) );

driver/src/main/java/org/neo4j/driver/internal/async/connection/EventLoopGroupFactory.java

@@ -49,7 +49,7 @@ private EventLoopGroupFactory()
      * Get class of {@link Channel} for {@link Bootstrap#channel(Class)} method.
      *
      * @return class of the channel, which should be consistent with {@link EventLoopGroup}s returned by
-     * {@link #newEventLoopGroup()} and {@link #newEventLoopGroup(int)}.
+     * {@link #newEventLoopGroup(int)}.
      */
     public static Class<? extends Channel> channelClass()
     {
@@ -68,17 +68,6 @@ public static EventLoopGroup newEventLoopGroup( int threadCount )
         return new DriverEventLoopGroup( threadCount );
     }
 
-    /**
-     * Create new {@link EventLoopGroup} with default thread count. Returned group should by given to
-     * {@link Bootstrap#group(EventLoopGroup)}.
-     *
-     * @return new group consistent with channel class returned by {@link #channelClass()}.
-     */
-    public static EventLoopGroup newEventLoopGroup()
-    {
-        return new DriverEventLoopGroup();
-    }
-
     /**
      * Assert that current thread is not an event loop used for async IO operations. This check is needed because
      * blocking API methods like {@link Session#run(String)} are implemented on top of corresponding async API methods

driver/src/test/java/org/neo4j/driver/ConfigTest.java

@@ -300,4 +300,19 @@ void shouldErrorWithIllegalFetchSize( long value ) throws Throwable
     {
         assertThrows( IllegalArgumentException.class, () -> Config.builder().withFetchSize( value ).build() );
     }
+
+    @ParameterizedTest
+    @ValueSource( ints = {100, 1, 1000, Integer.MAX_VALUE} )
+    void shouldChangeEventLoopThreads( int value ) throws Throwable
+    {
+        Config config = Config.builder().withEventLoopThreads( value ).build();
+        assertThat( config.eventLoopThreads(), equalTo( value ) );
+    }
+
+    @ParameterizedTest
+    @ValueSource( ints = {0, -100, -2} )
+    void shouldErrorWithIllegalEventLoopThreadsSize( int value ) throws Throwable
+    {
+        assertThrows( IllegalArgumentException.class, () -> Config.builder().withEventLoopThreads( value ).build() );
+    }
 }

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

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/async/connection/EventLoopGroupFactoryTest.java

@@ -63,13 +63,6 @@ void shouldCreateEventLoopGroupWithSpecifiedThreadCount()
         assertThat( eventLoopGroup, instanceOf( NioEventLoopGroup.class ) );
     }
 
-    @Test
-    void shouldCreateEventLoopGroup()
-    {
-        eventLoopGroup = EventLoopGroupFactory.newEventLoopGroup();
-        assertThat( eventLoopGroup, instanceOf( NioEventLoopGroup.class ) );
-    }
-
     @Test
     void shouldAssertNotInEventLoopThread() throws Exception
     {

driver/src/test/java/org/neo4j/driver/internal/cluster/loadbalancing/RoutingTableAndConnectionPoolTest.java

@@ -311,7 +311,7 @@ private ConnectionPool newConnectionPool()
     {
         InternalAbstractMetrics metrics = DEV_NULL_METRICS;
         PoolSettings poolSettings = new PoolSettings( 10, 5000, -1, -1 );
-        Bootstrap bootstrap = BootstrapFactory.newBootstrap();
+        Bootstrap bootstrap = BootstrapFactory.newBootstrap( 1 );
         NettyChannelTracker channelTracker = new NettyChannelTracker( metrics, bootstrap.config().group().next(), logging );
 
         return new TestConnectionPool( bootstrap, channelTracker, poolSettings, metrics, logging, clock, true );

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