[DE-1016] Async connection pool

core/pom.xml

@@ -8,7 +8,7 @@
         <relativePath>../release-parent</relativePath>
         <groupId>com.arangodb</groupId>
         <artifactId>release-parent</artifactId>
-        <version>7.18.0</version>
+        <version>7.19.0-SNAPSHOT</version>
     </parent>
 
     <name>core</name>

core/src/main/java/com/arangodb/ArangoDB.java

@@ -517,6 +517,17 @@ public Builder chunkSize(final Integer chunkSize) {
             return this;
         }
 
+        /**
+         * Set whether to use requests pipelining in HTTP/1.1 ({@link Protocol#HTTP_JSON} or {@link Protocol#HTTP_VPACK}).
+         *
+         * @param pipelining {@code true} if enabled
+         * @return {@link ArangoDB.Builder}
+         */
+        public Builder pipelining(final Boolean pipelining) {
+            config.setPipelining(pipelining);
+            return this;
+        }
+
         /**
          * Sets the maximum number of connections the built in connection pool will open per host.
          *

core/src/main/java/com/arangodb/config/ArangoConfigProperties.java

@@ -21,6 +21,7 @@ public interface ArangoConfigProperties {
     String KEY_USE_SSL = "useSsl";
     String KEY_VERIFY_HOST = "verifyHost";
     String KEY_CHUNK_SIZE = "chunkSize";
+    String KEY_PIPELINING = "pipelining";
     String KEY_MAX_CONNECTIONS = "maxConnections";
     String KEY_CONNECTION_TTL = "connectionTtl";
     String KEY_KEEP_ALIVE_INTERVAL = "keepAliveInterval";
@@ -110,6 +111,10 @@ default Optional<Integer> getChunkSize() {
         return Optional.empty();
     }
 
+    default Optional<Boolean> getPipelining() {
+        return Optional.empty();
+    }
+
     default Optional<Integer> getMaxConnections() {
         return Optional.empty();
     }

core/src/main/java/com/arangodb/internal/ArangoDefaults.java

@@ -50,6 +50,7 @@ public final class ArangoDefaults {
     public static final Boolean DEFAULT_USE_SSL = false;
     public static final Boolean DEFAULT_VERIFY_HOST = true;
     public static final Integer DEFAULT_CHUNK_SIZE = 30_000;
+    public static final Boolean DEFAULT_PIPELINING = false;
     public static final Boolean DEFAULT_ACQUIRE_HOST_LIST = false;
     public static final Integer DEFAULT_ACQUIRE_HOST_LIST_INTERVAL = 60 * 60 * 1000; // hour
     public static final LoadBalancingStrategy DEFAULT_LOAD_BALANCING_STRATEGY = LoadBalancingStrategy.NONE;

core/src/main/java/com/arangodb/internal/config/ArangoConfig.java

@@ -33,6 +33,7 @@ public class ArangoConfig {
     private SSLContext sslContext;
     private Boolean verifyHost;
     private Integer chunkSize;
+    private Boolean pipelining;
     private Integer maxConnections;
     private Long connectionTtl;
     private Integer keepAliveInterval;
@@ -70,6 +71,7 @@ public void loadProperties(final ArangoConfigProperties properties) {
         useSsl = properties.getUseSsl().orElse(ArangoDefaults.DEFAULT_USE_SSL);
         verifyHost = properties.getVerifyHost().orElse(ArangoDefaults.DEFAULT_VERIFY_HOST);
         chunkSize = properties.getChunkSize().orElse(ArangoDefaults.DEFAULT_CHUNK_SIZE);
+        pipelining = properties.getPipelining().orElse(ArangoDefaults.DEFAULT_PIPELINING);
         // FIXME: make maxConnections field Optional
         maxConnections = properties.getMaxConnections().orElse(null);
         // FIXME: make connectionTtl field Optional
@@ -173,6 +175,14 @@ public void setChunkSize(Integer chunkSize) {
         this.chunkSize = chunkSize;
     }
 
+    public Boolean getPipelining() {
+        return pipelining;
+    }
+
+    public void setPipelining(Boolean pipelining) {
+        this.pipelining = pipelining;
+    }
+
     public Integer getMaxConnections() {
         if (maxConnections == null) {
             maxConnections = getDefaultMaxConnections();

core/src/main/java/com/arangodb/internal/config/ArangoConfigPropertiesImpl.java

@@ -119,6 +119,11 @@ public Optional<Integer> getChunkSize() {
         return Optional.ofNullable(getProperty(KEY_CHUNK_SIZE)).map(Integer::valueOf);
     }
 
+    @Override
+    public Optional<Boolean> getPipelining() {
+        return Optional.ofNullable(getProperty(KEY_PIPELINING)).map(Boolean::valueOf);
+    }
+
     @Override
     public Optional<Integer> getMaxConnections() {
         return Optional.ofNullable(getProperty(KEY_MAX_CONNECTIONS)).map(Integer::valueOf);

core/src/main/java/com/arangodb/internal/net/Communication.java

@@ -50,7 +50,9 @@ public CompletableFuture<InternalResponse> executeAsync(final InternalRequest re
 
     private CompletableFuture<InternalResponse> executeAsync(final InternalRequest request, final HostHandle hostHandle, final Host host, final int attemptCount) {
         long reqId = reqCount.getAndIncrement();
-        return doExecuteAsync(request, hostHandle, host, attemptCount, host.connection(), reqId);
+        return host.connection().thenCompose(c ->
+                doExecuteAsync(request, hostHandle, host, attemptCount, c, reqId)
+                        .whenComplete((r, t) -> host.release(c)));
     }
 
     private CompletableFuture<InternalResponse> doExecuteAsync(

core/src/main/java/com/arangodb/internal/net/ConnectionPool.java

@@ -23,6 +23,7 @@
 import com.arangodb.config.HostDescription;
 
 import java.io.Closeable;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * @author Mark Vollmary
@@ -31,7 +32,9 @@ public interface ConnectionPool extends Closeable {
 
     Connection createConnection(final HostDescription host);
 
-    Connection connection();
+    CompletableFuture<Connection> connection();
+
+    void release(final Connection connection);
 
     void setJwt(String jwt);
 

core/src/main/java/com/arangodb/internal/net/ConnectionPoolImpl.java

@@ -23,22 +23,28 @@
 import com.arangodb.ArangoDBException;
 import com.arangodb.config.HostDescription;
 import com.arangodb.internal.config.ArangoConfig;
+import com.arangodb.internal.util.AsyncQueue;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * @author Mark Vollmary
  */
 public class ConnectionPoolImpl implements ConnectionPool {
 
+    public static final int HTTP1_PIPELINING_LIMIT = 10;
+    public static final int HTTP2_STREAMS = 32;    // hard-coded, see BTS-2049
+
+    private final AsyncQueue<Connection> slots = new AsyncQueue<>();
     private final HostDescription host;
     private final ArangoConfig config;
     private final int maxConnections;
     private final List<Connection> connections;
     private final ConnectionFactory factory;
-    private int current;
+    private final int maxSlots;
     private volatile String jwt = null;
     private boolean closed = false;
 
@@ -49,7 +55,14 @@ public ConnectionPoolImpl(final HostDescription host, final ArangoConfig config,
         this.maxConnections = config.getMaxConnections();
         this.factory = factory;
         connections = new ArrayList<>();
-        current = 0;
+        switch (config.getProtocol()) {
+            case HTTP_JSON:
+            case HTTP_VPACK:
+                maxSlots = config.getPipelining() ? HTTP1_PIPELINING_LIMIT : 1;
+                break;
+            default:
+                maxSlots = HTTP2_STREAMS;
+        }
     }
 
     @Override
@@ -60,23 +73,25 @@ public Connection createConnection(final HostDescription host) {
     }
 
     @Override
-    public synchronized Connection connection() {
+    public synchronized CompletableFuture<Connection> connection() {
         if (closed) {
             throw new ArangoDBException("Connection pool already closed!");
         }
 
-        final Connection connection;
-
         if (connections.size() < maxConnections) {
-            connection = createConnection(host);
+            Connection connection = createConnection(host);
             connections.add(connection);
-            current++;
-        } else {
-            final int index = Math.floorMod(current++, connections.size());
-            connection = connections.get(index);
+            for (int i = 0; i < maxSlots; i++) {
+                slots.offer((connection));
+            }
         }
 
-        return connection;
+        return slots.poll();
+    }
+
+    @Override
+    public void release(Connection connection) {
+        slots.offer(connection);
     }
 
     @Override
@@ -101,7 +116,7 @@ public synchronized void close() throws IOException {
     @Override
     public String toString() {
         return "ConnectionPoolImpl [host=" + host + ", maxConnections=" + maxConnections + ", connections="
-                + connections.size() + ", current=" + current + ", factory=" + factory.getClass().getSimpleName() + "]";
+                + connections.size() + ", factory=" + factory.getClass().getSimpleName() + "]";
     }
 
 }

core/src/main/java/com/arangodb/internal/net/Host.java

@@ -24,6 +24,7 @@
 import com.arangodb.config.HostDescription;
 
 import java.io.IOException;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * @author Mark Vollmary
@@ -33,7 +34,9 @@ public interface Host {
 
     HostDescription getDescription();
 
-    Connection connection();
+    CompletableFuture<Connection> connection();
+
+    void release(Connection c);
 
     void closeOnError();
 
@@ -44,5 +47,4 @@ public interface Host {
     void setMarkforDeletion(boolean markforDeletion);
 
     void setJwt(String jwt);
-
 }

core/src/main/java/com/arangodb/internal/net/HostImpl.java

@@ -24,6 +24,7 @@
 import com.arangodb.config.HostDescription;
 
 import java.io.IOException;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * @author Mark Vollmary
@@ -51,10 +52,15 @@ public HostDescription getDescription() {
     }
 
     @Override
-    public Connection connection() {
+    public CompletableFuture<Connection> connection() {
         return connectionPool.connection();
     }
 
+    @Override
+    public void release(Connection c) {
+        connectionPool.release(c);
+    }
+
     @Override
     public void closeOnError() {
         try {

core/src/main/java/com/arangodb/internal/util/AsyncQueue.java

@@ -0,0 +1,30 @@
+package com.arangodb.internal.util;
+
+import java.util.ArrayDeque;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
+
+public class AsyncQueue<T> {
+    private final Queue<CompletableFuture<T>> requests = new ArrayDeque<>();
+    private final Queue<T> offers = new ArrayDeque<>();
+
+    public synchronized CompletableFuture<T> poll() {
+        CompletableFuture<T> r = new CompletableFuture<>();
+        T o = offers.poll();
+        if (o != null) {
+            r.complete(o);
+        } else {
+            requests.add(r);
+        }
+        return r;
+    }
+
+    public synchronized void offer(T o) {
+        CompletableFuture<T> r = requests.poll();
+        if (r != null) {
+            r.complete(o);
+        } else {
+            offers.add(o);
+        }
+    }
+}

docker/start_db.sh

@@ -66,7 +66,7 @@ docker run -d \
     --starter.address="${GW}" \
     --docker.image="${DOCKER_IMAGE}" \
     --starter.local --starter.mode=${STARTER_MODE} --all.log.level=debug --all.log.output=+ --log.verbose \
-    --all.server.descriptors-minimum=1024 --all.javascript.allow-admin-execute=true
+    --all.server.descriptors-minimum=1024 --all.javascript.allow-admin-execute=true --all.server.maximal-threads=128
 
 
 wait_server() {

driver/pom.xml

@@ -8,7 +8,7 @@
         <relativePath>../release-parent</relativePath>
         <groupId>com.arangodb</groupId>
         <artifactId>release-parent</artifactId>
-        <version>7.18.0</version>
+        <version>7.19.0-SNAPSHOT</version>
     </parent>
 
     <name>arangodb-java-driver</name>

http-protocol/pom.xml

@@ -8,7 +8,7 @@
         <relativePath>../release-parent</relativePath>
         <groupId>com.arangodb</groupId>
         <artifactId>release-parent</artifactId>
-        <version>7.18.0</version>
+        <version>7.19.0-SNAPSHOT</version>
     </parent>
 
     <name>http-protocol</name>

http-protocol/src/main/java/com/arangodb/http/HttpConnection.java

@@ -63,6 +63,9 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static com.arangodb.internal.net.ConnectionPoolImpl.HTTP1_PIPELINING_LIMIT;
+import static com.arangodb.internal.net.ConnectionPoolImpl.HTTP2_STREAMS;
+
 
 /**
  * @author Mark Vollmary
@@ -88,7 +91,6 @@ private static String getUserAgent() {
     }
 
     HttpConnection(final ArangoConfig config, final HostDescription host, final HttpProtocolConfig protocolConfig) {
-        super();
         Protocol protocol = config.getProtocol();
         ContentType contentType = ContentTypeFactory.of(protocol);
         if (contentType == ContentType.VPACK) {
@@ -148,7 +150,9 @@ private static String getUserAgent() {
                 .setLogActivity(true)
                 .setKeepAlive(true)
                 .setTcpKeepAlive(true)
-                .setPipelining(true)
+                .setPipelining(config.getPipelining())
+                .setPipeliningLimit(HTTP1_PIPELINING_LIMIT)
+                .setHttp2MultiplexingLimit(HTTP2_STREAMS)
                 .setReuseAddress(true)
                 .setReusePort(true)
                 .setHttp2ClearTextUpgrade(false)
@@ -273,7 +277,7 @@ public CompletableFuture<InternalResponse> executeAsync(@UnstableApi final Inter
         return rfuture;
     }
 
-    public void doExecute(@UnstableApi final InternalRequest request, @UnstableApi final CompletableFuture<InternalResponse> rfuture) {
+    private void doExecute(@UnstableApi final InternalRequest request, @UnstableApi final CompletableFuture<InternalResponse> rfuture) {
         String path = buildUrl(request);
         HttpRequest<Buffer> httpRequest = client
                 .request(requestTypeToHttpMethod(request.getRequestType()), path)

jackson-serde-json/pom.xml

@@ -8,7 +8,7 @@
         <relativePath>../release-parent</relativePath>
         <groupId>com.arangodb</groupId>
         <artifactId>release-parent</artifactId>
-        <version>7.18.0</version>
+        <version>7.19.0-SNAPSHOT</version>
     </parent>
 
     <name>jackson-serde-json</name>

jackson-serde-vpack/pom.xml

@@ -8,7 +8,7 @@
         <relativePath>../release-parent</relativePath>
         <groupId>com.arangodb</groupId>
         <artifactId>release-parent</artifactId>
-        <version>7.18.0</version>
+        <version>7.19.0-SNAPSHOT</version>
     </parent>
 
     <name>jackson-serde-vpack</name>

jsonb-serde/pom.xml

@@ -8,7 +8,7 @@
         <relativePath>../release-parent</relativePath>
         <groupId>com.arangodb</groupId>
         <artifactId>release-parent</artifactId>
-        <version>7.18.0</version>
+        <version>7.19.0-SNAPSHOT</version>
     </parent>
 
     <name>jsonb-serde</name>

pom.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.arangodb</groupId>
     <artifactId>arangodb-java-driver-parent</artifactId>
-    <version>7.18.0</version>
+    <version>7.19.0-SNAPSHOT</version>
     <inceptionYear>2016</inceptionYear>
     <modules>
         <module>release-parent</module>