Skip to content

Config option to allow config netty event loop threads #652

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Nov 27, 2019
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 26 additions & 1 deletion driver/src/main/java/org/neo4j/driver/Config.java
Original file line number Diff line number Diff line change
Expand Up @@ -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 )
{
Expand All @@ -122,6 +123,7 @@ private Config( ConfigBuilder builder )
this.resolver = builder.resolver;
this.fetchSize = builder.fetchSize;

this.eventLoopThreads = builder.eventLoopThreads;
this.isMetricsEnabled = builder.isMetricsEnabled;
}

Expand Down Expand Up @@ -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.
*/
Expand Down Expand Up @@ -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() {}

Expand Down Expand Up @@ -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.
*
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 );
}

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 ) );
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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()
{
Expand All @@ -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
Expand Down
15 changes: 15 additions & 0 deletions driver/src/test/java/org/neo4j/driver/ConfigTest.java
Original file line number Diff line number Diff line change
Expand Up @@ -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() );
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -262,7 +262,7 @@ private static class DriverFactoryWithSessions extends DriverFactory
}

@Override
protected Bootstrap createBootstrap()
protected Bootstrap createBootstrap( int ignored )
{
return BootstrapFactory.newBootstrap( 1 );
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
{
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 );
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 );
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down