You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by ke...@apache.org on 2024/11/03 05:32:52 UTC
(tinkerpop) 43/48: Changed connection pooling to reflect having switched from long lived web socket connections to short lived http connections: (#2860)
This is an automated email from the ASF dual-hosted git repository.
kenhuuu pushed a commit to branch master-http-final
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit e07b23fc4109e8e9cf2bb24f8a410c7d44cb2400
Author: andreachild <an...@improving.com>
AuthorDate: Thu Oct 31 15:59:42 2024 -0700
Changed connection pooling to reflect having switched from long lived web socket connections to short lived http connections: (#2860)
-sped up obtaining a new connection by removing method-level synchronization of getAvailableConnection and introducing a separate set for available connections
-introduced cleanup of idle connections to release resources
-removed minSize since idle connections are now cleaned up and the pool can become empty if there is no traffic
-increased maxSize default to 128
-increased core pool size of ScheduledThreadPoolExecutor that is used for creating new connections to number of available processors
-added logging to help troubleshoot connection related integration tests failures
---
CHANGELOG.asciidoc | 2 +
docs/src/reference/gremlin-variants.asciidoc | 4 +-
.../tinkerpop/gremlin/driver/Channelizer.java | 26 ++-
.../apache/tinkerpop/gremlin/driver/Client.java | 2 +-
.../apache/tinkerpop/gremlin/driver/Cluster.java | 51 +++---
.../tinkerpop/gremlin/driver/Connection.java | 20 ++-
.../tinkerpop/gremlin/driver/ConnectionPool.java | 180 ++++++++-------------
.../apache/tinkerpop/gremlin/driver/Settings.java | 21 +--
.../driver/handler/HttpGremlinRequestEncoder.java | 2 +-
.../driver/handler/IdleConnectionHandler.java | 49 ++++++
...eckHandler.java => InactiveChannelHandler.java} | 13 +-
.../driver/util/ConfigurationEvaluator.java | 36 ++---
.../gremlin/driver/util/ProfilingApplication.java | 6 +-
.../gremlin/driver/ClusterBuilderTest.java | 6 +-
.../gremlin/driver/ConnectionPoolTest.java | 8 +-
.../tinkerpop/gremlin/driver/SettingsTest.java | 4 +-
.../driver/handler/IdleConnectionHandlerTest.java | 57 +++++++
.../driver/ClientConnectionIntegrateTest.java | 88 +++++++++-
.../gremlin/server/GremlinDriverIntegrateTest.java | 9 +-
19 files changed, 372 insertions(+), 212 deletions(-)
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index e049e260b6..5b791eedcf 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -80,6 +80,8 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima
* Updated `ElementIdStrategy.getConfiguration()` to help with serialization.
* Updated `TraversalStrategyProxy` to utilize strategy names instead of strategy classes
* Use `TraversalStrategyProxy` in Java, or `TraversalStrategy` in Python to pass custom strategies in traversals
+* Removed `minSize` setting for Gremlin Driver connection pool since connections are now short-lived HTTP connections
+* Added `idleConnectionTimeout` setting for Gremlin Driver and automatic closing of idle connections
== TinkerPop 3.7.0 (Gremfir Master of the Pan Flute)
diff --git a/docs/src/reference/gremlin-variants.asciidoc b/docs/src/reference/gremlin-variants.asciidoc
index 918becc487..881ce3a19e 100644
--- a/docs/src/reference/gremlin-variants.asciidoc
+++ b/docs/src/reference/gremlin-variants.asciidoc
@@ -800,12 +800,11 @@ The following table describes the various configuration options for the Gremlin
|connectionPool.maxResponseContentLength |The maximum length in bytes that a message can be received from the server. |2147483647
|connectionPool.maxInProcessPerConnection |The maximum number of in-flight requests that can occur on a connection. |4
|connectionPool.maxSimultaneousUsagePerConnection |The maximum number of times that a connection can be borrowed from the pool simultaneously. |16
-|connectionPool.maxSize |The maximum size of a connection pool for a host. |8
+|connectionPool.maxSize |The maximum size of a connection pool for a host. |128
|connectionPool.maxWaitForConnection |The amount of time in milliseconds to wait for a new connection before timing out. |3000
|connectionPool.maxWaitForClose |The amount of time in milliseconds to wait for pending messages to be returned from the server before closing the connection. |3000
|connectionPool.minInProcessPerConnection |The minimum number of in-flight requests that can occur on a connection. |1
|connectionPool.minSimultaneousUsagePerConnection |The maximum number of times that a connection can be borrowed from the pool simultaneously. |8
-|connectionPool.minSize |The minimum size of a connection pool for a host. |2
|connectionPool.reconnectInterval |The amount of time in milliseconds to wait before trying to reconnect to a dead host. |1000
|connectionPool.resultIterationBatchSize |The override value for the size of the result batches to be returned from the server. |64
|connectionPool.sslCipherSuites |The list of JSSE ciphers to support for SSL connections. If specified, only the ciphers that are listed and supported will be enabled. If not specified, the JVM default is used. |_none_
@@ -815,6 +814,7 @@ The following table describes the various configuration options for the Gremlin
|connectionPool.trustStorePassword |The password of the `trustStore` if it is password-protected |_none_
|connectionPool.validationRequest |A script that is used to test server connectivity. A good script to use is one that evaluates quickly and returns no data. The default simply returns an empty string, but if a graph is required by a particular provider, a good traversal might be `g.inject()`. |_''_
|connectionPool.connectionSetupTimeoutMillis | Duration of time in milliseconds provided for connection setup to complete which includes WebSocket protocol handshake and SSL handshake. |15000
+|connectionPool.idleConnectionTimeout | Duration of time in milliseconds that the driver will allow a channel to not receive read or writes before it automatically closes. |180000
|hosts |The list of hosts that the driver will connect to. |localhost
|jaasEntry |Sets the `AuthProperties.Property.JAAS_ENTRY` properties for authentication to Gremlin Server. |_none_
|nioPoolSize |Size of the pool for handling request/response operations. |available processors
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java
index b64f969838..1ee56a3aa7 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java
@@ -28,12 +28,14 @@ import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslHandler;
+import io.netty.handler.timeout.IdleStateHandler;
import org.apache.tinkerpop.gremlin.driver.exception.ConnectionException;
import org.apache.tinkerpop.gremlin.driver.handler.GremlinResponseHandler;
import org.apache.tinkerpop.gremlin.driver.handler.HttpContentDecompressionHandler;
import org.apache.tinkerpop.gremlin.driver.handler.HttpGremlinRequestEncoder;
import org.apache.tinkerpop.gremlin.driver.handler.HttpGremlinResponseDecoder;
-import org.apache.tinkerpop.gremlin.driver.handler.SslCheckHandler;
+import org.apache.tinkerpop.gremlin.driver.handler.IdleConnectionHandler;
+import org.apache.tinkerpop.gremlin.driver.handler.InactiveChannelHandler;
import org.apache.tinkerpop.gremlin.util.message.ResponseMessage;
import java.util.Collections;
@@ -93,8 +95,9 @@ public interface Channelizer extends ChannelHandler {
private AtomicReference<ResultQueue> pending;
protected static final String PIPELINE_GREMLIN_HANDLER = "gremlin-handler";
- public static final String PIPELINE_SSL_HANDLER = "gremlin-ssl-handler";
-
+ protected static final String PIPELINE_SSL_HANDLER = "gremlin-ssl-handler";
+ protected static final String PIPELINE_IDLE_STATE_HANDLER = "idle-state-handler";
+ protected static final String PIPELINE_IDLE_CONNECTION_HANDLER = "idle-connection-handler";
protected static final String PIPELINE_HTTP_CODEC = "http-codec";
protected static final String PIPELINE_HTTP_AGGREGATOR = "http-aggregator";
protected static final String PIPELINE_HTTP_ENCODER = "gremlin-encoder";
@@ -105,7 +108,7 @@ public interface Channelizer extends ChannelHandler {
"configured at both the client and the server. Ensure that client http handshake " +
"protocol matches the server. Ensure that the server is still reachable.";
- private static final SslCheckHandler sslCheckHandler = new SslCheckHandler();
+ private static final InactiveChannelHandler inactiveChannelHandler = new InactiveChannelHandler();
public boolean supportsSsl() {
return cluster.connectionPoolSettings().enableSsl;
@@ -151,7 +154,7 @@ public interface Channelizer extends ChannelHandler {
sslHandler.setHandshakeTimeoutMillis(0);
pipeline.addLast(PIPELINE_SSL_HANDLER, sslHandler);
} else {
- pipeline.addLast(PIPELINE_SSL_HANDLER, sslCheckHandler);
+ pipeline.addLast(PIPELINE_SSL_HANDLER, inactiveChannelHandler);
}
configure(pipeline);
@@ -187,6 +190,8 @@ public interface Channelizer extends ChannelHandler {
private HttpGremlinResponseDecoder gremlinResponseDecoder;
private HttpContentDecompressionHandler httpCompressionDecoder;
+ private IdleStateHandler idleStateHandler;
+ private IdleConnectionHandler idleConnectionHandler;
@Override
public void init(final Connection connection) {
@@ -196,6 +201,11 @@ public interface Channelizer extends ChannelHandler {
gremlinRequestEncoder = new HttpGremlinRequestEncoder(cluster.getSerializer(), cluster.getRequestInterceptors(),
cluster.isUserAgentOnConnectEnabled(), cluster.isBulkingEnabled(), connection.getUri());
gremlinResponseDecoder = new HttpGremlinResponseDecoder(cluster.getSerializer());
+ if (cluster.getIdleConnectionTimeout() > 0) {
+ final int idleConnectionTimeout = (int) (cluster.getIdleConnectionTimeout() / 1000);
+ idleStateHandler = new IdleStateHandler(idleConnectionTimeout, idleConnectionTimeout, 0);
+ idleConnectionHandler = new IdleConnectionHandler();
+ }
}
@Override
@@ -218,6 +228,12 @@ public interface Channelizer extends ChannelHandler {
if (!supportsSsl() && "https".equalsIgnoreCase(scheme))
throw new IllegalStateException("To use https scheme ensure that enableSsl is set to true in configuration");
+ if (cluster.getIdleConnectionTimeout() > 0) {
+ // idle connection handling is enabled
+ pipeline.addLast(PIPELINE_IDLE_STATE_HANDLER, idleStateHandler);
+ pipeline.addLast(PIPELINE_IDLE_CONNECTION_HANDLER, idleConnectionHandler);
+ }
+
final HttpClientCodec handler = new HttpClientCodec(DEFAULT_MAX_INITIAL_LINE_LENGTH, DEFAULT_MAX_HEADER_SIZE,
1024 * 1024, DEFAULT_FAIL_ON_MISSING_RESPONSE,
DEFAULT_VALIDATE_HEADERS, DEFAULT_INITIAL_BUFFER_SIZE, DEFAULT_PARSE_HTTP_AFTER_CONNECT_REQUEST,
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java
index ebd03ed359..c843fd49c5 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java
@@ -56,7 +56,7 @@ import java.util.stream.Collectors;
public abstract class Client {
private static final Logger logger = LoggerFactory.getLogger(Client.class);
- public static final String TOO_MANY_IN_FLIGHT_REQUESTS = "Number of active requests exceeds pool size. " +
+ public static final String TOO_MANY_IN_FLIGHT_REQUESTS = "Number of active requests (%s) exceeds pool size (%s). " +
"Consider increasing the value for maxConnectionPoolSize.";
protected final Cluster cluster;
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java
index 8d14d9cf53..7e5d71cb1c 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java
@@ -186,8 +186,8 @@ public final class Cluster {
.maxResponseContentLength(settings.connectionPool.maxResponseContentLength)
.maxWaitForConnection(settings.connectionPool.maxWaitForConnection)
.maxConnectionPoolSize(settings.connectionPool.maxSize)
- .minConnectionPoolSize(settings.connectionPool.minSize)
.connectionSetupTimeoutMillis(settings.connectionPool.connectionSetupTimeoutMillis)
+ .idleConnectionTimeoutMillis(settings.connectionPool.idleConnectionTimeout)
.enableUserAgentOnConnect(settings.enableUserAgentOnConnect)
.enableBulkedResult(settings.enableBulkedResult)
.validationRequest(settings.connectionPool.validationRequest);
@@ -309,13 +309,6 @@ public final class Cluster {
return manager.connectionPoolSettings.maxSize;
}
- /**
- * Gets the minimum size of the {@link ConnectionPool}.
- */
- public int minConnectionPoolSize() {
- return manager.connectionPoolSettings.minSize;
- }
-
/**
* Gets the override for the server setting that determines how many results are returned per batch.
*/
@@ -344,6 +337,13 @@ public final class Cluster {
return manager.connectionPoolSettings.maxResponseContentLength;
}
+ /**
+ * Get time in milliseconds that the driver will allow a channel to not receive read or writes before it automatically closes.
+ */
+ public long getIdleConnectionTimeout() {
+ return manager.connectionPoolSettings.idleConnectionTimeout;
+ }
+
/**
* Specifies the load balancing strategy to use on the client side.
*/
@@ -490,7 +490,6 @@ public final class Cluster {
private MessageSerializer<?> serializer = null;
private int nioPoolSize = Runtime.getRuntime().availableProcessors();
private int workerPoolSize = Runtime.getRuntime().availableProcessors() * 2;
- private int minConnectionPoolSize = ConnectionPool.MIN_POOL_SIZE;
private int maxConnectionPoolSize = ConnectionPool.MAX_POOL_SIZE;
private int maxWaitForConnection = Connection.MAX_WAIT_FOR_CONNECTION;
private int maxWaitForClose = Connection.MAX_WAIT_FOR_CLOSE;
@@ -512,6 +511,7 @@ public final class Cluster {
private LoadBalancingStrategy loadBalancingStrategy = new LoadBalancingStrategy.RoundRobin();
private LinkedList<Pair<String, ? extends RequestInterceptor>> interceptors = new LinkedList<>();
private long connectionSetupTimeoutMillis = Connection.CONNECTION_SETUP_TIMEOUT_MILLIS;
+ private long idleConnectionTimeoutMillis = Connection.CONNECTION_IDLE_TIMEOUT_MILLIS;
private boolean enableUserAgentOnConnect = true;
private boolean enableBulkedResult = false;
@@ -688,15 +688,6 @@ public final class Cluster {
return this;
}
- /**
- * The minimum size of the {@link ConnectionPool}. When the {@link Client} is started, {@link Connection}
- * objects will be initially constructed to this size.
- */
- public Builder minConnectionPoolSize(final int minSize) {
- this.minConnectionPoolSize = minSize;
- return this;
- }
-
/**
* Override the server setting that determines how many results are returned per batch.
*/
@@ -880,6 +871,14 @@ public final class Cluster {
return this;
}
+ /**
+ * Sets the time in milliseconds that the driver will allow a channel to not receive read or writes before it automatically closes.
+ */
+ public Builder idleConnectionTimeoutMillis(final long idleConnectionTimeoutMillis) {
+ this.idleConnectionTimeoutMillis = idleConnectionTimeoutMillis;
+ return this;
+ }
+
/**
* Configures whether cluster will send a user agent during
* web socket handshakes
@@ -984,7 +983,6 @@ public final class Cluster {
connectionPoolSettings = new Settings.ConnectionPoolSettings();
connectionPoolSettings.maxSize = builder.maxConnectionPoolSize;
- connectionPoolSettings.minSize = builder.minConnectionPoolSize;
connectionPoolSettings.maxWaitForConnection = builder.maxWaitForConnection;
connectionPoolSettings.maxWaitForClose = builder.maxWaitForClose;
connectionPoolSettings.maxResponseContentLength = builder.maxResponseContentLength;
@@ -1002,6 +1000,7 @@ public final class Cluster {
connectionPoolSettings.sslSkipCertValidation = builder.sslSkipCertValidation;
connectionPoolSettings.validationRequest = builder.validationRequest;
connectionPoolSettings.connectionSetupTimeoutMillis = builder.connectionSetupTimeoutMillis;
+ connectionPoolSettings.idleConnectionTimeout = builder.idleConnectionTimeoutMillis;
sslContextOptional = Optional.ofNullable(builder.sslContext);
@@ -1026,22 +1025,16 @@ public final class Cluster {
// we distinguish between the hostScheduler and the connectionScheduler because you can end in deadlock
// if all the possible jobs the driver allows for go to a single thread pool.
- this.connectionScheduler = new ScheduledThreadPoolExecutor(contactPoints.size() + 1,
+ this.connectionScheduler = new ScheduledThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),
new BasicThreadFactory.Builder().namingPattern("gremlin-driver-conn-scheduler-%d").build());
validationRequest = () -> RequestMessage.build(builder.validationRequest);
}
private void validateBuilder(final Builder builder) {
- if (builder.minConnectionPoolSize < 0)
- throw new IllegalArgumentException("minConnectionPoolSize must be greater than or equal to zero");
-
if (builder.maxConnectionPoolSize < 1)
throw new IllegalArgumentException("maxConnectionPoolSize must be greater than zero");
- if (builder.minConnectionPoolSize > builder.maxConnectionPoolSize)
- throw new IllegalArgumentException("maxConnectionPoolSize cannot be less than minConnectionPoolSize");
-
if (builder.maxWaitForConnection < 1)
throw new IllegalArgumentException("maxWaitForConnection must be greater than zero");
@@ -1065,6 +1058,12 @@ public final class Cluster {
if (builder.connectionSetupTimeoutMillis < 1)
throw new IllegalArgumentException("connectionSetupTimeoutMillis must be greater than zero");
+
+ // zero value will disable idle connection detection
+ // non-zero will be converted to seconds so any value between 1 and 999 is invalid as it will be less than 1 second
+ if (builder.idleConnectionTimeoutMillis != 0 && builder.idleConnectionTimeoutMillis < 1000)
+ throw new IllegalArgumentException("idleConnectionTimeoutMillis must be zero or greater than or equal to 1000");
+
}
synchronized void init() {
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
index 78424179d9..f39ccb851e 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
@@ -18,6 +18,7 @@
*/
package org.apache.tinkerpop.gremlin.driver;
+import org.apache.tinkerpop.gremlin.driver.handler.IdleConnectionHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.tinkerpop.gremlin.driver.exception.ConnectionException;
@@ -50,6 +51,7 @@ final class Connection {
public static final int RECONNECT_INTERVAL = 1000;
public static final int RESULT_ITERATION_BATCH_SIZE = 64;
public static final long CONNECTION_SETUP_TIMEOUT_MILLIS = 15000;
+ public static final long CONNECTION_IDLE_TIMEOUT_MILLIS = 180000;
private static final Logger logger = LoggerFactory.getLogger(Connection.class);
private final Channel channel;
@@ -107,19 +109,21 @@ final class Connection {
channel.closeFuture().addListener((ChannelFutureListener) future -> {
logger.debug("OnChannelClose callback called for channel {}", channel);
- // if the closeFuture is not set, it means that closeAsync() wasn't called which means that the
- // close did not come from the client side. it means the server closed the channel for some reason.
- // it's important to distinguish that difference in debugging
+ // if the closeFuture is not set, it means that closeAsync() wasn't called
if (thisConnection.closeFuture.get() == null) {
- logger.error(String.format(
- "Server closed the Connection on channel %s - scheduling removal from %s",
- channel.id().asShortText(), thisConnection.pool.getPoolInfo(thisConnection)));
+ if (!channel.hasAttr(IdleConnectionHandler.IDLE_STATE_EVENT)) {
+ // if idle state event is not present, it means the server closed the channel for some reason.
+ // it's important to distinguish that difference in debugging
+ logger.error(String.format(
+ "Server closed the Connection on channel %s - scheduling removal from %s",
+ channel.id().asShortText(), thisConnection.pool.getPoolInfo(thisConnection)));
+ }
// delegate the task to scheduler thread and free up the event loop
- thisConnection.cluster.connectionScheduler().submit(() -> thisConnection.pool.definitelyDestroyConnection(thisConnection));
+ thisConnection.cluster.connectionScheduler().submit(() -> thisConnection.pool.destroyConnection(thisConnection));
}
});
- logger.info("Created new connection for {}", uri);
+ logger.debug("Created new connection for {}", uri);
} catch (Exception ex) {
throw new ConnectionException(uri, "Could not open " + getConnectionInfo(true), ex);
}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ConnectionPool.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ConnectionPool.java
index 6187c6e096..45ef031d63 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ConnectionPool.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ConnectionPool.java
@@ -18,6 +18,8 @@
*/
package org.apache.tinkerpop.gremlin.driver;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.tinkerpop.gremlin.driver.exception.ConnectionException;
@@ -49,8 +51,7 @@ import java.util.concurrent.locks.ReentrantLock;
final class ConnectionPool {
private static final Logger logger = LoggerFactory.getLogger(ConnectionPool.class);
- public static final int MIN_POOL_SIZE = 2;
- public static final int MAX_POOL_SIZE = 16;
+ public static final int MAX_POOL_SIZE = 128;
// A small buffer in millis used for comparing if a connection was created within a certain amount of time.
private static final int CONNECTION_SETUP_TIME_DELTA = 25;
@@ -59,8 +60,8 @@ final class ConnectionPool {
private final Client client;
private final List<Connection> connections;
private final AtomicInteger open;
+ private final Queue<Connection> availableConnections = new ConcurrentLinkedQueue<>();
private final Set<Connection> bin = new CopyOnWriteArraySet<>();
- private final int minPoolSize;
private final int maxPoolSize;
private final String poolLabel;
@@ -69,7 +70,7 @@ final class ConnectionPool {
private final AtomicReference<CompletableFuture<Void>> closeFuture = new AtomicReference<>();
- private volatile int waiter = 0;
+ private final AtomicInteger waiter = new AtomicInteger();
private final Lock waitLock = new ReentrantLock(true);
private final Condition hasAvailableConnection = waitLock.newCondition();
ConnectionFactory connectionFactory;
@@ -90,16 +91,14 @@ final class ConnectionPool {
}
public ConnectionPool(final Host host, final Client client) {
- this(host, client, Optional.empty(), Optional.empty());
+ this(host, client, Optional.empty());
}
- public ConnectionPool(final Host host, final Client client, final Optional<Integer> overrideMinPoolSize,
- final Optional<Integer> overrideMaxPoolSize) {
- this(host, client, overrideMinPoolSize, overrideMaxPoolSize, new ConnectionFactory.DefaultConnectionFactory());
+ public ConnectionPool(final Host host, final Client client, final Optional<Integer> overrideMaxPoolSize) {
+ this(host, client, overrideMaxPoolSize, new ConnectionFactory.DefaultConnectionFactory());
}
- ConnectionPool(final Host host, final Client client, final Optional<Integer> overrideMinPoolSize,
- final Optional<Integer> overrideMaxPoolSize, final ConnectionFactory connectionFactory) {
+ ConnectionPool(final Host host, final Client client, final Optional<Integer> overrideMaxPoolSize, final ConnectionFactory connectionFactory) {
this.host = host;
this.client = client;
this.cluster = client.cluster;
@@ -107,51 +106,43 @@ final class ConnectionPool {
poolLabel = "Connection Pool {host=" + host + "}";
final Settings.ConnectionPoolSettings settings = settings();
- this.minPoolSize = overrideMinPoolSize.orElse(settings.minSize);
this.maxPoolSize = overrideMaxPoolSize.orElse(settings.maxSize);
this.connections = new CopyOnWriteArrayList<>();
this.open = new AtomicInteger();
try {
- final List<CompletableFuture<Void>> connectionCreationFutures = new ArrayList<>();
- for (int i = 0; i < minPoolSize; i++) {
- connectionCreationFutures.add(CompletableFuture.runAsync(() -> {
- final ConnectionResult result = new ConnectionResult();
- try {
- this.connections.add(connectionFactory.create(this));
- this.open.incrementAndGet();
- } catch (ConnectionException e) {
- result.setFailureCause(e);
- throw new CompletionException(e);
- } finally {
- result.setTimeNow();
- if ((latestConnectionResult.get() == null) ||
- (latestConnectionResult.get().getTime() < result.getTime())) {
- latestConnectionResult.set(result);
- }
+ // attempt a single connection to ensure the host is viable
+ // if there is no traffic, it will eventually be closed by IdleConnectionHandler
+ CompletableFuture.runAsync(() -> {
+ final ConnectionResult result = new ConnectionResult();
+ try {
+ final Connection conn = connectionFactory.create(this);
+ this.connections.add(conn);
+ this.availableConnections.add(conn);
+ this.open.incrementAndGet();
+ } catch (ConnectionException e) {
+ result.setFailureCause(e);
+ throw new CompletionException(e);
+ } finally {
+ result.setTimeNow();
+ if ((latestConnectionResult.get() == null) ||
+ (latestConnectionResult.get().getTime() < result.getTime())) {
+ latestConnectionResult.set(result);
}
- }, cluster.connectionScheduler()));
- }
-
- CompletableFuture.allOf(connectionCreationFutures.toArray(new CompletableFuture[0])).join();
+ }
+ }, cluster.connectionScheduler()).join();
} catch (CancellationException ce) {
- logger.warn("Initialization of connections cancelled for {}", this.getPoolInfo(), ce);
+ logger.warn("Initialization of connection pool cancelled for {}", this.getPoolInfo(), ce);
throw ce;
} catch (CompletionException ce) {
- // Some connections might have been initialized. Close the connection pool gracefully to close them.
+ // Close the connection pool gracefully to close any open connections.
this.closeAsync();
-
- final String errMsg = "Could not initialize " + minPoolSize + " (minPoolSize) connections in pool." +
- " Successful connections=" + this.connections.size() +
- ". Closing the connection pool.";
-
final Throwable result = ce.getCause() == null ? ce : ce.getCause();
-
- throw new CompletionException(errMsg, result);
+ throw new CompletionException("Could not initialize connection in pool. Closing the connection pool.", result);
}
- logger.info("Opening connection pool on {} with core size of {}", host, minPoolSize);
+ logger.info("Opening connection pool on {}", host);
}
public Settings.ConnectionPoolSettings settings() {
@@ -159,36 +150,18 @@ final class ConnectionPool {
}
public Connection borrowConnection(final long timeout, final TimeUnit unit) throws TimeoutException, ConnectionException {
- logger.debug("Borrowing connection from pool on {} - timeout in {} {}", host, timeout, unit);
+ logger.debug("Borrowing connection from pool on {} - timeout in {} {} with {}", host, timeout, unit, Thread.currentThread());
if (isClosed()) throw new ConnectionException(host.getHostUri(), host.getAddress(), "Pool is shutdown");
- if (connections.isEmpty()) {
- logger.debug("Tried to borrow connection but the pool was empty for {} - scheduling pool creation and waiting for connection", host);
- for (int i = 0; i < minPoolSize; i++) {
- // If many connections are borrowed at the same time there needs to be a check to make sure no
- // additional ones get scheduled for creation
- if (scheduledForCreation.get() < minPoolSize) {
- scheduledForCreation.incrementAndGet();
- newConnection();
- }
- }
-
- return waitForConnection(timeout, unit);
- }
-
// Get valid connection
final Connection availableConnection = getAvailableConnection();
-
- if (null == availableConnection) {
- if (isClosed())
- throw new ConnectionException(host.getHostUri(), host.getAddress(), "Pool is shutdown");
- logger.debug("Pool was initialized but a connection could not be selected earlier - waiting for connection on {}", host);
+ if (availableConnection == null) {
return waitForConnection(timeout, unit);
}
if (logger.isDebugEnabled())
- logger.debug("Return least used {} on {}", availableConnection.getConnectionInfo(), host);
+ logger.debug("Borrowed {} on {} with {}", availableConnection.getConnectionInfo(), host, Thread.currentThread());
return availableConnection;
}
@@ -217,6 +190,8 @@ final class ConnectionPool {
logger.debug("destroy {}", connection.getConnectionInfo());
destroyConnection(connection);
} else {
+ logger.debug("Pool size is {} - returning connection to pool: {}", poolSize, connection);
+ availableConnections.add(connection);
announceAvailableConnection();
}
}
@@ -240,7 +215,7 @@ final class ConnectionPool {
public synchronized CompletableFuture<Void> closeAsync() {
if (closeFuture.get() != null) return closeFuture.get();
- logger.info("Signalled closing of connection pool on {} with core size of {}", host, minPoolSize);
+ logger.info("Signalled closing of connection pool on {} with current connection size of {}", host, connections.size());
announceAllAvailableConnection();
final CompletableFuture<Void> future = killAvailableConnections();
@@ -291,7 +266,7 @@ final class ConnectionPool {
}
considerNewConnection();
- definitelyDestroyConnection(connection);
+ destroyConnection(connection);
}
private void considerNewConnection() {
@@ -347,7 +322,10 @@ final class ConnectionPool {
final ConnectionResult result = new ConnectionResult();
try {
- connections.add(connectionFactory.create(this));
+ final Connection conn = connectionFactory.create(this);
+ logger.debug("Created connection {}", conn.getConnectionInfo());
+ connections.add(conn);
+ availableConnections.add(conn);
} catch (Exception ex) {
open.decrementAndGet();
logger.error(String.format(
@@ -368,25 +346,12 @@ final class ConnectionPool {
return true;
}
- private boolean destroyConnection(final Connection connection) {
- while (true) {
- final int opened = open.get();
- if (opened <= minPoolSize)
- return false;
-
- if (open.compareAndSet(opened, opened - 1))
- break;
- }
-
- definitelyDestroyConnection(connection);
- return true;
- }
-
- public void definitelyDestroyConnection(final Connection connection) {
+ public void destroyConnection(final Connection connection) {
// only add to the bin for future removal if its not already there.
if (!bin.contains(connection) && !connection.isClosing()) {
bin.add(connection);
connections.remove(connection);
+ availableConnections.remove(connection);
open.decrementAndGet();
}
@@ -408,6 +373,7 @@ final class ConnectionPool {
try {
awaitAvailableConnection(remaining, unit);
} catch (InterruptedException e) {
+ logger.warn("Wait was interrupted", e);
Thread.currentThread().interrupt();
to = 0;
}
@@ -415,16 +381,16 @@ final class ConnectionPool {
if (isClosed())
throw new ConnectionException(host.getHostUri(), host.getAddress(), "Pool is shutdown");
- final Connection leastUsed = getAvailableConnection();
+ final Connection conn = getAvailableConnection();
- if (leastUsed != null) {
+ if (conn != null) {
if (logger.isDebugEnabled())
- logger.debug("Return least used {} on {} after waiting", leastUsed.getConnectionInfo(), host);
- return leastUsed;
+ logger.debug("Return {} on {} after waiting on {}", conn.getConnectionInfo(), host, Thread.currentThread());
+ return conn;
}
remaining = to - TimeUtil.timeSince(start, unit);
- logger.debug("Continue to wait for connection on {} if {} > 0", host, remaining);
+ logger.debug("Continue to wait for connection on {} if {} > 0 on {}", host, remaining, Thread.currentThread());
} while (remaining > 0);
final StringBuilder cause = new StringBuilder("Potential Cause: ");
@@ -436,7 +402,7 @@ final class ConnectionPool {
// Assumes that the root cause will give better information about why the connection failed.
cause.append(ExceptionHelper.getRootCause(res.getFailureCause()).getMessage());
} else if (open.get() >= maxPoolSize) {
- cause.append(Client.TOO_MANY_IN_FLIGHT_REQUESTS);
+ cause.append(String.format(Client.TOO_MANY_IN_FLIGHT_REQUESTS, open.get(), maxPoolSize));
} else {
cause.setLength(0);
}
@@ -478,7 +444,7 @@ final class ConnectionPool {
// let the load-balancer know that the host is acting poorly
if (!host.isAvailable()) {
// if the host is unavailable then we should release the connections
- connections.forEach(this::definitelyDestroyConnection);
+ connections.forEach(this::destroyConnection);
this.cluster.loadBalancingStrategy().onUnavailable(host);
}
}
@@ -522,9 +488,6 @@ final class ConnectionPool {
private void announceAvailableConnection() {
logger.debug("Announce connection available on {}", host);
- if (waiter == 0)
- return;
-
waitLock.lock();
try {
hasAvailableConnection.signal();
@@ -534,27 +497,27 @@ final class ConnectionPool {
}
/**
- * Get the least-used connection from the pool. Also triggers consideration of a new connection if the least-used
+ * Get an available connection from the pool. Also triggers consideration of a new connection if the least-used
* connection has hit the usage maximum or no valid connection could be retrieved from the pool.
*
- * @return The least-used connection from the pool. Returns null if no valid connection could be retrieved from the
+ * @return an available connection from the pool. Returns null if no valid connection could be retrieved from the
* pool.
*/
- private synchronized Connection getAvailableConnection() {
+ private Connection getAvailableConnection() {
Connection available = null;
- int availableConnectionsCount = 0;
- for (Connection connection : connections) {
- if (!connection.isDead() && !connection.isBorrowed().get()) {
- // try to borrow connection
- if (available == null && connection.isBorrowed().compareAndSet(false, true)) {
- available = connection;
- }
- availableConnectionsCount++;
+
+ // remove head of queue until we can borrow a connection or the queue is empty
+ Connection head = availableConnections.poll();
+ while (head != null) {
+ // try to borrow connection
+ if (!head.isDead() && !head.isBorrowed().get() && head.isBorrowed().compareAndSet(false, true)) {
+ available = head;
+ break;
}
+ head = availableConnections.poll();
}
- // todo: may be add new connection when only 10-20% of pool is available?
- if (availableConnectionsCount == 0 && connections.size() < maxPoolSize) {
+ if (available == null && connections.size() < maxPoolSize) {
considerNewConnection();
}
@@ -565,18 +528,17 @@ final class ConnectionPool {
logger.debug("Wait {} {} for an available connection on {} with {}", timeout, unit, host, Thread.currentThread());
waitLock.lock();
- waiter++;
+ waiter.incrementAndGet();
try {
hasAvailableConnection.await(timeout, unit);
} finally {
- waiter--;
+ waiter.decrementAndGet();
waitLock.unlock();
}
}
private void announceAllAvailableConnection() {
- if (waiter == 0)
- return;
+ logger.debug("Announce to all connection available on {}", host);
waitLock.lock();
try {
@@ -608,8 +570,8 @@ final class ConnectionPool {
} else {
final int connectionCount = connections.size();
sb.append(System.lineSeparator());
- sb.append(String.format("Connection Pool Status (size=%s max=%s min=%s toCreate=%s bin=%s)",
- connectionCount, maxPoolSize, minPoolSize, this.scheduledForCreation.get(), bin.size()));
+ sb.append(String.format("Connection Pool Status (size=%s available=%s max=%s toCreate=%s bin=%s waiter=%s)",
+ connectionCount, availableConnections.size(), maxPoolSize, this.scheduledForCreation.get(), bin.size(), waiter.get()));
sb.append(System.lineSeparator());
appendConnections(sb, connectionToCallout, (CopyOnWriteArrayList<Connection>) connections);
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Settings.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Settings.java
index a71983a2be..3bf0e784b4 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Settings.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Settings.java
@@ -190,9 +190,6 @@ public final class Settings {
if (connectionPoolConf.containsKey("sslSkipCertValidation"))
cpSettings.sslSkipCertValidation = connectionPoolConf.getBoolean("sslSkipCertValidation");
- if (connectionPoolConf.containsKey("minSize"))
- cpSettings.minSize = connectionPoolConf.getInt("minSize");
-
if (connectionPoolConf.containsKey("maxSize"))
cpSettings.maxSize = connectionPoolConf.getInt("maxSize");
@@ -217,6 +214,9 @@ public final class Settings {
if (connectionPoolConf.containsKey("connectionSetupTimeoutMillis"))
cpSettings.connectionSetupTimeoutMillis = connectionPoolConf.getLong("connectionSetupTimeoutMillis");
+ if (connectionPoolConf.containsKey("idleConnectionTimeout"))
+ cpSettings.idleConnectionTimeout = connectionPoolConf.getLong("idleConnectionTimeout");
+
settings.connectionPool = cpSettings;
}
@@ -301,12 +301,7 @@ public final class Settings {
public boolean sslSkipCertValidation = false;
/**
- * The minimum size of a connection pool for a {@link Host}. By default this is set to 2.
- */
- public int minSize = ConnectionPool.MIN_POOL_SIZE;
-
- /**
- * The maximum size of a connection pool for a {@link Host}. By default this is set to 8.
+ * The maximum size of a connection pool for a {@link Host}. By default this is set to 500.
*/
public int maxSize = ConnectionPool.MAX_POOL_SIZE;
@@ -352,6 +347,14 @@ public final class Settings {
* complete by then.
*/
public long connectionSetupTimeoutMillis = Connection.CONNECTION_SETUP_TIMEOUT_MILLIS;
+
+ /**
+ * Time in milliseconds that the driver will allow a channel to not receive read or writes before it automatically closes.
+ * Note that while this value is to be provided as milliseconds it will resolve to
+ * second precision. Set this value to 0 to disable this feature.
+ */
+ public long idleConnectionTimeout = Connection.CONNECTION_IDLE_TIMEOUT_MILLIS;
+
}
public static class SerializerSettings {
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinRequestEncoder.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinRequestEncoder.java
index c1ea9ae98c..d5e49bcb46 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinRequestEncoder.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinRequestEncoder.java
@@ -46,7 +46,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import static org.apache.tinkerpop.gremlin.driver.handler.SslCheckHandler.REQUEST_SENT;
+import static org.apache.tinkerpop.gremlin.driver.handler.InactiveChannelHandler.REQUEST_SENT;
/**
* Converts {@link RequestMessage} to a {@code HttpRequest}.
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/IdleConnectionHandler.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/IdleConnectionHandler.java
new file mode 100644
index 0000000000..fb4253160b
--- /dev/null
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/IdleConnectionHandler.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.driver.handler;
+
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.timeout.IdleStateEvent;
+import io.netty.util.AttributeKey;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Detects {@link IdleStateEvent}s and closes the channel with the goal of releasing resources that haven't been used in a while.
+ */
+@ChannelHandler.Sharable
+public class IdleConnectionHandler extends ChannelDuplexHandler {
+ private static final Logger logger = LoggerFactory.getLogger(IdleConnectionHandler.class);
+ public static final AttributeKey<IdleStateEvent> IDLE_STATE_EVENT = AttributeKey.valueOf("idleStateEvent");
+
+ /**
+ * Detects if an {@link IdleStateEvent} has occurred and closes the channel.
+ */
+ @Override
+ public void userEventTriggered(final ChannelHandlerContext ctx, final Object evt) {
+ if (evt instanceof IdleStateEvent) {
+ final IdleStateEvent e = (IdleStateEvent) evt;
+ logger.info("Detected IdleStateEvent {} - closing channel {}", e, ctx.channel().id().asShortText());
+ ctx.channel().attr(IDLE_STATE_EVENT).set(e);
+ ctx.close();
+ }
+ }
+}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/SslCheckHandler.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/InactiveChannelHandler.java
similarity index 77%
rename from gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/SslCheckHandler.java
rename to gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/InactiveChannelHandler.java
index dc58186d2c..f38dd77305 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/SslCheckHandler.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/InactiveChannelHandler.java
@@ -24,10 +24,10 @@ import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.AttributeKey;
/**
- * Handler used to detect if the server requires TLS when the client is sending plain HTTP.
+ * Handler used to detect if the channel has become inactive and throw an error in certain scenarios.
*/
@ChannelHandler.Sharable
-public class SslCheckHandler extends ChannelInboundHandlerAdapter {
+public class InactiveChannelHandler extends ChannelInboundHandlerAdapter {
public static final AttributeKey<Integer> BYTES_READ = AttributeKey.valueOf("bytesRead");
public static final AttributeKey<Boolean> REQUEST_SENT = AttributeKey.valueOf("requestSent");
@@ -40,8 +40,13 @@ public class SslCheckHandler extends ChannelInboundHandlerAdapter {
if ((null == ctx.channel().attr(BYTES_READ).get()) && (null != ctx.channel().attr(REQUEST_SENT).get())) {
// BYTES_READ is set by the HttpGremlinResponseStreamDecoder when any data is received. If it isn't set,
// it can be assumed that no bytes were received when the connection closed.
- final String errMsg = "Connection to server closed unexpectedly. Ensure that the server is still" +
- " reachable. The server may be expecting SSL to be enabled.";
+ String errMsg;
+ if (ctx.channel().hasAttr(IdleConnectionHandler.IDLE_STATE_EVENT)) {
+ errMsg = "Idle timeout occurred before response could be received from server - consider increasing idleConnectionTimeout";
+ } else {
+ errMsg = "Connection to server closed unexpectedly. Ensure that the server is still" +
+ " reachable. The server may be expecting SSL to be enabled.";
+ }
ctx.fireExceptionCaught(new RuntimeException(errMsg));
} else {
super.channelInactive(ctx);
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/util/ConfigurationEvaluator.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/util/ConfigurationEvaluator.java
index 172b8cf04c..b055fb0e90 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/util/ConfigurationEvaluator.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/util/ConfigurationEvaluator.java
@@ -32,7 +32,6 @@ import java.util.stream.Stream;
*/
public class ConfigurationEvaluator {
- private final List<Integer> minConnectionPoolSizeRange = Arrays.asList(4,8,12,16,32,64,96,128,192,256,384,512);
private final List<Integer> maxConnectionPoolSizeRange = Arrays.asList(16,32,64,96,128,192,256,384,512);
private final List<Integer> workerPoolSizeRange = Arrays.asList(1,2,3,4,8,16,32);
private final List<Integer> nioPoolSizeRange = Arrays.asList(1,2,4);
@@ -45,28 +44,23 @@ public class ConfigurationEvaluator {
for (int ir = 0; ir < nioPoolSizeRange.size(); ir++) {
for (int is = 0; is < parallelismSizeRange.size(); is++) {
for (int it = 0; it < workerPoolSizeRange.size(); it++) {
- for (int iw = 0; iw < minConnectionPoolSizeRange.size(); iw++) {
- for (int ix = 0; ix < maxConnectionPoolSizeRange.size(); ix++) {
- if (minConnectionPoolSizeRange.get(iw) <= maxConnectionPoolSizeRange.get(ix)) {
- final String s = String.join(",", String.valueOf(ir), String.valueOf(is), String.valueOf(it), String.valueOf(iw), String.valueOf(ix));
- if (!configsTried.contains(s)) {
- final Object[] argsToProfiler =
- Stream.of("nioPoolSize", nioPoolSizeRange.get(ir).toString(),
- "parallelism", parallelismSizeRange.get(is).toString(),
- "workerPoolSize", workerPoolSizeRange.get(it).toString(),
- "minConnectionPoolSize", minConnectionPoolSizeRange.get(iw).toString(),
- "maxConnectionPoolSize", maxConnectionPoolSizeRange.get(ix).toString(),
- "noExit", Boolean.TRUE.toString()).toArray();
+ for (int ix = 0; ix < maxConnectionPoolSizeRange.size(); ix++) {
+ final String s = String.join(",", String.valueOf(ir), String.valueOf(is), String.valueOf(it), String.valueOf(ix));
+ if (!configsTried.contains(s)) {
+ final Object[] argsToProfiler =
+ Stream.of("nioPoolSize", nioPoolSizeRange.get(ir).toString(),
+ "parallelism", parallelismSizeRange.get(is).toString(),
+ "workerPoolSize", workerPoolSizeRange.get(it).toString(),
+ "maxConnectionPoolSize", maxConnectionPoolSizeRange.get(ix).toString(),
+ "noExit", Boolean.TRUE.toString()).toArray();
- final Object[] withExtraArgs = args.length > 0 ? Stream.concat(Stream.of(args), Stream.of(argsToProfiler)).toArray() : argsToProfiler;
+ final Object[] withExtraArgs = args.length > 0 ? Stream.concat(Stream.of(args), Stream.of(argsToProfiler)).toArray() : argsToProfiler;
- final String[] stringProfilerArgs = Arrays.copyOf(withExtraArgs, withExtraArgs.length, String[].class);
- System.out.println("Testing with: " + Arrays.toString(stringProfilerArgs));
- ProfilingApplication.main(stringProfilerArgs);
- TimeUnit.SECONDS.sleep(5);
- configsTried.add(s);
- }
- }
+ final String[] stringProfilerArgs = Arrays.copyOf(withExtraArgs, withExtraArgs.length, String[].class);
+ System.out.println("Testing with: " + Arrays.toString(stringProfilerArgs));
+ ProfilingApplication.main(stringProfilerArgs);
+ TimeUnit.SECONDS.sleep(5);
+ configsTried.add(s);
}
}
}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/util/ProfilingApplication.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/util/ProfilingApplication.java
index 1d718f430a..8518d7b64f 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/util/ProfilingApplication.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/util/ProfilingApplication.java
@@ -184,7 +184,6 @@ public class ProfilingApplication {
final int executions = Integer.parseInt(options.getOrDefault("executions", "10").toString());
final int nioPoolSize = Integer.parseInt(options.getOrDefault("nioPoolSize", "1").toString());
final int requests = Integer.parseInt(options.getOrDefault("requests", "10000").toString());
- final int minConnectionPoolSize = Integer.parseInt(options.getOrDefault("minConnectionPoolSize", "256").toString());
final int maxConnectionPoolSize = Integer.parseInt(options.getOrDefault("maxConnectionPoolSize", "256").toString());
final int maxWaitForConnection = Integer.parseInt(options.getOrDefault("maxWaitForConnection", "3000").toString());
final int workerPoolSize = Integer.parseInt(options.getOrDefault("workerPoolSize", Runtime.getRuntime().availableProcessors() * 2).toString());
@@ -195,7 +194,6 @@ public class ProfilingApplication {
final String script = options.getOrDefault("script", "1+1").toString();
final Cluster cluster = Cluster.build(host)
- .minConnectionPoolSize(minConnectionPoolSize)
.maxConnectionPoolSize(maxConnectionPoolSize)
.nioPoolSize(nioPoolSize)
.maxWaitForConnection(maxWaitForConnection)
@@ -226,7 +224,7 @@ public class ProfilingApplication {
final File f = null == fileName ? null : new File(fileName.toString());
if (f != null && f.length() == 0) {
try (final PrintWriter writer = new PrintWriter(new BufferedWriter(new FileWriter(f, true)))) {
- writer.println("parallelism\tnioPoolSize\tminConnectionPoolSize\tmaxConnectionPoolSize\tworkerPoolSize\trequestPerSecond");
+ writer.println("parallelism\tnioPoolSize\tmaxConnectionPoolSize\tworkerPoolSize\trequestPerSecond");
}
}
@@ -257,7 +255,7 @@ public class ProfilingApplication {
System.out.println(String.format("avg req/sec: %s", averageRequestPerSecond));
if (f != null) {
try (final PrintWriter writer = new PrintWriter(new BufferedWriter(new FileWriter(f, true)))) {
- writer.println(String.join("\t", String.valueOf(parallelism), String.valueOf(nioPoolSize), String.valueOf(minConnectionPoolSize), String.valueOf(maxConnectionPoolSize), String.valueOf(workerPoolSize), String.valueOf(averageRequestPerSecond)));
+ writer.println(String.join("\t", String.valueOf(parallelism), String.valueOf(nioPoolSize), String.valueOf(maxConnectionPoolSize), String.valueOf(workerPoolSize), String.valueOf(averageRequestPerSecond)));
}
}
} else if (TestType.LATENCY == testType) {
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ClusterBuilderTest.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ClusterBuilderTest.java
index 2ee546d0b0..dca023d4b4 100644
--- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ClusterBuilderTest.java
+++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ClusterBuilderTest.java
@@ -40,9 +40,6 @@ public class ClusterBuilderTest {
return Arrays.asList(new Object[][]{
{"maxConnectionPoolSize0", Cluster.build().maxConnectionPoolSize(0), "maxConnectionPoolSize must be greater than zero"},
{"maxConnectionPoolSizeNeg1", Cluster.build().maxConnectionPoolSize(-1), "maxConnectionPoolSize must be greater than zero"},
- {"minConnectionPoolSizeNeg1", Cluster.build().minConnectionPoolSize(-1), "minConnectionPoolSize must be greater than or equal to zero"},
- {"minConnectionPoolSizeLteMax", Cluster.build().minConnectionPoolSize(100).maxConnectionPoolSize(99), "maxConnectionPoolSize cannot be less than minConnectionPoolSize"},
- {"minConnectionPoolSizeLteMax", Cluster.build().minConnectionPoolSize(100).maxConnectionPoolSize(99), "maxConnectionPoolSize cannot be less than minConnectionPoolSize"},
{"maxConnectionPoolSize0", Cluster.build().maxWaitForConnection(0), "maxWaitForConnection must be greater than zero"},
{"maxWaitForClose0", Cluster.build().maxWaitForClose(0), "maxWaitForClose must be greater than zero"},
{"maxWaitForCloseNeg1", Cluster.build().maxWaitForClose(-1), "maxWaitForClose must be greater than zero"},
@@ -54,6 +51,9 @@ public class ClusterBuilderTest {
{"nioPoolSize0", Cluster.build().nioPoolSize(0), "nioPoolSize must be greater than zero"},
{"nioPoolSizeNeg1", Cluster.build().nioPoolSize(-1), "nioPoolSize must be greater than zero"},
{"connectionSetupTimeoutMillis0", Cluster.build().connectionSetupTimeoutMillis(0), "connectionSetupTimeoutMillis must be greater than zero"},
+ {"idleConnectionTimeoutMillisNeg1", Cluster.build().idleConnectionTimeoutMillis(-1), "idleConnectionTimeoutMillis must be zero or greater than or equal to 1000"},
+ {"idleConnectionTimeoutMillisOne", Cluster.build().idleConnectionTimeoutMillis(1), "idleConnectionTimeoutMillis must be zero or greater than or equal to 1000"},
+ {"idleConnectionTimeoutMillis999", Cluster.build().idleConnectionTimeoutMillis(999), "idleConnectionTimeoutMillis must be zero or greater than or equal to 1000"},
{"workerPoolSize0", Cluster.build().workerPoolSize(0), "workerPoolSize must be greater than zero"},
{"workerPoolSizeNeg1", Cluster.build().workerPoolSize(-1), "workerPoolSize must be greater than zero"}});
}
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ConnectionPoolTest.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ConnectionPoolTest.java
index 01aa1f695e..1c8941b908 100644
--- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ConnectionPoolTest.java
+++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ConnectionPoolTest.java
@@ -63,10 +63,9 @@ public class ConnectionPoolTest {
final Client client = new Client.ClusteredClient(cluster);
- // create pool with 1 connection
- final ConnectionPool connectionPool = new ConnectionPool(host, client,
- Optional.of(1), Optional.of(2), connectionFactory);
- // try to borrow this connection.
+ // create pool - starts with 1 connection
+ final ConnectionPool connectionPool = new ConnectionPool(host, client, Optional.of(2), connectionFactory);
+ // try to borrow a connection.
final Connection conn0 = connectionPool.borrowConnection(100, TimeUnit.MILLISECONDS);
assertNotNull(connectionPool);
@@ -90,6 +89,7 @@ public class ConnectionPoolTest {
}
// return conn0 to pool, can be borrowed again
+ connectionPool.returnConnection(conn0);
when(mockConn0.isBorrowed()).thenReturn(new AtomicBoolean(false));
final Connection conn00 = connectionPool.borrowConnection(100, TimeUnit.MILLISECONDS);
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/SettingsTest.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/SettingsTest.java
index a2d47ff1db..82b5c2fc99 100644
--- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/SettingsTest.java
+++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/SettingsTest.java
@@ -58,7 +58,6 @@ public class SettingsTest {
conf.setProperty("connectionPool.sslEnabledProtocols", Arrays.asList("TLSv1.1","TLSv1.2"));
conf.setProperty("connectionPool.sslCipherSuites", Arrays.asList("TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384", "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384"));
conf.setProperty("connectionPool.sslSkipCertValidation", true);
- conf.setProperty("connectionPool.minSize", 100);
conf.setProperty("connectionPool.maxSize", 200);
conf.setProperty("connectionPool.minSimultaneousUsagePerConnection", 300);
conf.setProperty("connectionPool.maxSimultaneousUsagePerConnection", 400);
@@ -71,6 +70,7 @@ public class SettingsTest {
conf.setProperty("connectionPool.channelizer", "channelizer0");
conf.setProperty("connectionPool.validationRequest", "g.inject()");
conf.setProperty("connectionPool.connectionSetupTimeoutMillis", 15000);
+ conf.setProperty("connectionPool.idleConnectionTimeout", 160000);
final Settings settings = Settings.from(conf);
@@ -95,12 +95,12 @@ public class SettingsTest {
assertEquals(Arrays.asList("TLSv1.1","TLSv1.2"), settings.connectionPool.sslEnabledProtocols);
assertEquals(Arrays.asList("TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384", "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384"), settings.connectionPool.sslCipherSuites);
assertThat(settings.connectionPool.sslSkipCertValidation, is(true));
- assertEquals(100, settings.connectionPool.minSize);
assertEquals(200, settings.connectionPool.maxSize);
assertEquals(700, settings.connectionPool.maxWaitForConnection);
assertEquals(800, settings.connectionPool.maxResponseContentLength);
assertEquals(900, settings.connectionPool.reconnectInterval);
assertEquals(15000, settings.connectionPool.connectionSetupTimeoutMillis);
+ assertEquals(160000, settings.connectionPool.idleConnectionTimeout);
assertEquals(1100, settings.connectionPool.resultIterationBatchSize);
assertEquals("g.inject()", settings.connectionPool.validationRequest);
}
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/handler/IdleConnectionHandlerTest.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/handler/IdleConnectionHandlerTest.java
new file mode 100644
index 0000000000..f59f4ad7bc
--- /dev/null
+++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/handler/IdleConnectionHandlerTest.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.tinkerpop.gremlin.driver.handler;
+
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.timeout.IdleStateEvent;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class IdleConnectionHandlerTest {
+ private EmbeddedChannel testChannel;
+
+ @Before
+ public void setup() {
+ testChannel = new EmbeddedChannel(new IdleConnectionHandler());
+ }
+
+ @After
+ public void teardown() {
+ // if any exceptions happened, throw them otherwise the test will only silently fail
+ testChannel.checkException();
+ }
+
+
+ @Test
+ public void userEventTriggered_setsIdleStateEventAttribute() {
+ testChannel.pipeline().fireUserEventTriggered(IdleStateEvent.WRITER_IDLE_STATE_EVENT);
+ assertTrue(testChannel.hasAttr(IdleConnectionHandler.IDLE_STATE_EVENT));
+ }
+
+ @Test
+ public void userEventTriggered_notIdleStateEvent_doesNotSetAttribute() {
+ testChannel.pipeline().fireUserEventTriggered("some other event");
+ assertFalse(testChannel.hasAttr(IdleConnectionHandler.IDLE_STATE_EVENT));
+ }
+}
\ No newline at end of file
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientConnectionIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientConnectionIntegrateTest.java
index f780bba564..34ebd67854 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientConnectionIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientConnectionIntegrateTest.java
@@ -30,6 +30,7 @@ import org.apache.tinkerpop.gremlin.server.TestClientFactory;
import org.hamcrest.core.Is;
import org.junit.After;
import org.junit.AfterClass;
+import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -62,7 +63,8 @@ import static org.junit.Assert.*;
public class ClientConnectionIntegrateTest extends AbstractGremlinServerIntegrationTest {
private static LogCaptor logCaptor;
- private Level previousLevel;
+ private Level previousConnectionLevel;
+ private Level previousConnectionPoolLevel;
@BeforeClass
public static void setupLogCaptor() {
@@ -77,8 +79,11 @@ public class ClientConnectionIntegrateTest extends AbstractGremlinServerIntegrat
@Before
public void setupForEachTest() {
final Logger lc = (Logger) LoggerFactory.getLogger(Connection.class);
- previousLevel = lc.getLevel();
+ previousConnectionLevel = lc.getLevel();
lc.setLevel(Level.DEBUG);
+ final Logger lcp = (Logger) LoggerFactory.getLogger(ConnectionPool.class);
+ previousConnectionPoolLevel = lcp.getLevel();
+ lcp.setLevel(Level.DEBUG);
logCaptor.clearLogs();
}
@@ -86,7 +91,9 @@ public class ClientConnectionIntegrateTest extends AbstractGremlinServerIntegrat
@After
public void afterEachTest() {
final Logger lc = (Logger) LoggerFactory.getLogger(Connection.class);
- lc.setLevel(previousLevel);
+ lc.setLevel(previousConnectionLevel);
+ final Logger lcp = (Logger) LoggerFactory.getLogger(ConnectionPool.class);
+ lcp.setLevel(previousConnectionPoolLevel);
}
/**
@@ -97,7 +104,6 @@ public class ClientConnectionIntegrateTest extends AbstractGremlinServerIntegrat
// Set a low value of maxResponseContentLength to intentionally trigger CorruptedFrameException
final Cluster cluster = TestClientFactory.build()
.maxResponseContentLength(64)
- .minConnectionPoolSize(1)
.maxConnectionPoolSize(2)
.create();
final Client.ClusteredClient client = cluster.connect();
@@ -140,7 +146,6 @@ public class ClientConnectionIntegrateTest extends AbstractGremlinServerIntegrat
public void shouldBalanceConcurrentRequestsAcrossConnections() throws InterruptedException {
final int connPoolSize = 16;
final Cluster cluster = TestClientFactory.build()
- .minConnectionPoolSize(connPoolSize)
.maxConnectionPoolSize(connPoolSize)
.create();
final Client.ClusteredClient client = cluster.connect();
@@ -183,10 +188,9 @@ public class ClientConnectionIntegrateTest extends AbstractGremlinServerIntegrat
}
@Test
- public void overLimitOperationsShouldCreateNewHttpConnectionPerRequestAsNeeded() throws InterruptedException {
+ public void shouldCreateNewHttpConnectionPerRequestAsNeeded() throws InterruptedException {
final int operations = 6;
final Cluster cluster = TestClientFactory.build()
- .minConnectionPoolSize(1)
.maxConnectionPoolSize(operations)
.create();
final Client.ClusteredClient client = cluster.connect();
@@ -236,7 +240,7 @@ public class ClientConnectionIntegrateTest extends AbstractGremlinServerIntegrat
*/
@Test
public void shouldSucceedWithJitteryConnection() throws Exception {
- final Cluster cluster = TestClientFactory.build().minConnectionPoolSize(1).maxConnectionPoolSize(128).
+ final Cluster cluster = TestClientFactory.build().maxConnectionPoolSize(128).
reconnectInterval(1000).
maxWaitForConnection(4000).validationRequest("g.inject()").create();
final Client.ClusteredClient client = cluster.connect();
@@ -292,6 +296,74 @@ public class ClientConnectionIntegrateTest extends AbstractGremlinServerIntegrat
cluster.close();
}
+ @Test
+ public void shouldCloseIdleConnectionsAndRecreateNewConnections() throws InterruptedException {
+ int idleMillis = 3000;
+ final Cluster cluster = TestClientFactory.build()
+ .idleConnectionTimeoutMillis(idleMillis)
+ .create();
+ final Client.ClusteredClient client = cluster.connect();
+ client.init();
+ final ExecutorService executorServiceForTesting = cluster.executor();
+
+ try {
+ // create or reuse some connections
+ chooseConnections(3, client, executorServiceForTesting);
+
+ // simulate downtime where there is no traffic so that all connections become idle
+ TimeUnit.MILLISECONDS.sleep(idleMillis * 3);
+ assertEquals(1, client.hostConnectionPools.size());
+ // all connections should have been closed due to idle timeout
+ assertTrue(client.hostConnectionPools.values().iterator().next().getPoolInfo().contains("no connections in pool"));
+
+ // create or reuse some more connections
+ chooseConnections(4, client, executorServiceForTesting);
+
+ } finally {
+ executorServiceForTesting.shutdown();
+ client.close();
+ cluster.close();
+ }
+ }
+
+ @Test
+ public void shouldThrowErrorWithHelpfulMessageWhenIdleTimeoutReachedBeforeResponseReceived() throws InterruptedException {
+ int idleMillis = 1000;
+ final Cluster cluster = TestClientFactory.build()
+ .idleConnectionTimeoutMillis(idleMillis)
+ .create();
+ final Client.ClusteredClient client = cluster.connect();
+
+ try {
+ client.submit("Thread.sleep(" + idleMillis * 3 + ")").all().get();
+ fail("Expected exception due to idle timeout");
+ } catch (Exception e) {
+ assertTrue(e.getMessage().contains("Idle timeout occurred before response could be received from server - consider increasing idleConnectionTimeout"));
+ } finally {
+ client.close();
+ cluster.close();
+ }
+ }
+
+ private static void chooseConnections(int operations, Client.ClusteredClient client, ExecutorService executorServiceForTesting) throws InterruptedException {
+ final RequestMessage.Builder request = client.buildMessage(RequestMessage.build("g.inject(1)"));
+ final Callable<Connection> sendQueryCallable = () -> client.chooseConnection(request.create());
+ final List<Callable<Connection>> listOfTasks = new ArrayList<>();
+ for (int i = 0; i < operations; i++) {
+ listOfTasks.add(sendQueryCallable);
+ }
+
+ final List<Future<Connection>> executorSubmitFutures = executorServiceForTesting.invokeAll(listOfTasks);
+ executorSubmitFutures.parallelStream().map(fut -> {
+ try {
+ return fut.get();
+ } catch (InterruptedException | ExecutionException e) {
+ fail(e.getMessage());
+ return null;
+ }
+ }).forEach(Assert::assertNotNull);
+ }
+
/**
* Introduces random failures when creating a {@link Connection} for the {@link ConnectionPool}.
*/
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverIntegrateTest.java
index 19d56491af..acb411a9d0 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverIntegrateTest.java
@@ -260,7 +260,7 @@ public class GremlinDriverIntegrateTest extends AbstractGremlinServerIntegration
final AtomicInteger handshakeRequests = new AtomicInteger(0);
final Cluster cluster = TestClientFactory.build().
- minConnectionPoolSize(1).maxConnectionPoolSize(1).
+ maxConnectionPoolSize(1).
addInterceptor("counter", r -> {
handshakeRequests.incrementAndGet();
return r;
@@ -964,8 +964,7 @@ public class GremlinDriverIntegrateTest extends AbstractGremlinServerIntegration
@Test
public void shouldBeThreadSafeToUseOneClient() throws Exception {
final Cluster cluster = TestClientFactory.build().workerPoolSize(2)
- .maxConnectionPoolSize(16)
- .minConnectionPoolSize(8).create();
+ .maxConnectionPoolSize(16).create();
final Client client = cluster.connect();
final Map<Integer, Integer> results = new ConcurrentHashMap<>();
@@ -1168,8 +1167,8 @@ public class GremlinDriverIntegrateTest extends AbstractGremlinServerIntegration
*/
@Test
public void shouldReturnClearExceptionCauseWhenClientIsTooBusyAndConnectionPoolIsFull() throws InterruptedException {
+ int maxSize = 1;
final Cluster cluster = TestClientFactory.build()
- .minConnectionPoolSize(1)
.maxConnectionPoolSize(1)
.connectionSetupTimeoutMillis(100)
.maxWaitForConnection(150)
@@ -1183,7 +1182,7 @@ public class GremlinDriverIntegrateTest extends AbstractGremlinServerIntegration
} catch (Exception e) {
final Throwable root = ExceptionHelper.getRootCause(e);
assertTrue(root instanceof TimeoutException);
- assertTrue(root.getMessage().contains(Client.TOO_MANY_IN_FLIGHT_REQUESTS));
+ assertTrue(root.getMessage().contains(String.format(Client.TOO_MANY_IN_FLIGHT_REQUESTS, maxSize, maxSize)));
}
}