From 61f466e96a22c3bb87814660ca726728f95f1de0 Mon Sep 17 00:00:00 2001 From: Kushagra Thapar Date: Tue, 17 Dec 2019 22:09:40 -0800 Subject: [PATCH 1/4] Added some logging around open connections --- .../internal/http/ReactorNettyClient.java | 42 ++++++++++++++++++- 1 file changed, 41 insertions(+), 1 deletion(-) diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/ReactorNettyClient.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/ReactorNettyClient.java index fc318844ecbd..ac2510e1b9de 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/ReactorNettyClient.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/ReactorNettyClient.java @@ -23,6 +23,9 @@ import java.nio.charset.Charset; import java.util.Objects; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiFunction; import static com.azure.data.cosmos.internal.http.HttpClientConfig.REACTOR_NETWORK_LOG_CATEGORY; @@ -32,12 +35,15 @@ */ class ReactorNettyClient implements HttpClient { - private static final Logger logger = LoggerFactory.getLogger(ReactorNettyClient.class.getSimpleName()); + private static final Logger logger = LoggerFactory.getLogger(ReactorNettyClient.class); private HttpClientConfig httpClientConfig; private reactor.netty.http.client.HttpClient httpClient; private ConnectionProvider connectionProvider; + private Set openChannels; + private AtomicInteger maxConnectionsSoFar; + private ReactorNettyClient() {} /** @@ -46,6 +52,8 @@ private ReactorNettyClient() {} public static ReactorNettyClient create(HttpClientConfig httpClientConfig) { ReactorNettyClient reactorNettyClient = new ReactorNettyClient(); reactorNettyClient.httpClientConfig = httpClientConfig; + reactorNettyClient.openChannels = ConcurrentHashMap.newKeySet(); + reactorNettyClient.maxConnectionsSoFar = new AtomicInteger(); reactorNettyClient.httpClient = reactor.netty.http.client.HttpClient.newConnection(); reactorNettyClient.configureChannelPipelineHandlers(); return reactorNettyClient; @@ -58,6 +66,8 @@ public static ReactorNettyClient createWithConnectionProvider(ConnectionProvider ReactorNettyClient reactorNettyClient = new ReactorNettyClient(); reactorNettyClient.connectionProvider = connectionProvider; reactorNettyClient.httpClientConfig = httpClientConfig; + reactorNettyClient.openChannels = ConcurrentHashMap.newKeySet(); + reactorNettyClient.maxConnectionsSoFar = new AtomicInteger(); reactorNettyClient.httpClient = reactor.netty.http.client.HttpClient.create(connectionProvider); reactorNettyClient.configureChannelPipelineHandlers(); return reactorNettyClient; @@ -80,6 +90,35 @@ private void configureChannelPipelineHandlers() { tcpClient = tcpClient.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, configs.getConnectionAcquireTimeoutInMillis()); + tcpClient = tcpClient.doOnConnect(connection -> { + logger.info("About to connect {}", connection.hashCode()); + }); + + tcpClient = tcpClient.doOnConnected(connection -> { + int connectionHashCode = connection.hashCode(); + openChannels.add(connectionHashCode); + + // Update maxConnectionsSoFar with maximum open connection so far + synchronized (this) { + if (openChannels.size() > maxConnectionsSoFar.get()) { + maxConnectionsSoFar.set(openChannels.size()); + } + } + logger.info("Connection established {}", connection.hashCode()); + logger.info("Current connected channels {}", openChannels.size()); + logger.info("Max connections so far {}", maxConnectionsSoFar); + }); + + tcpClient = tcpClient.doOnDisconnected(connection -> { + int connectionHashCode = connection.hashCode(); + openChannels.remove(connectionHashCode); + logger.info("Connection disconnected {}", connectionHashCode); + logger.info("Current connected channels {}", openChannels.size()); + connection.onDispose(() -> { + logger.info("Connection disposed {}", connection.hashCode()); + }); + }); + return tcpClient; }).httpResponseDecoder(httpResponseDecoderSpec -> { httpResponseDecoderSpec.maxInitialLineLength(configs.getMaxHttpInitialLineLength()); @@ -192,6 +231,7 @@ public Mono bodyAsString(Charset charset) { @Override public void close() { + logger.info("Disposing connection {}", reactorNettyConnection.hashCode()); if (reactorNettyConnection.channel().eventLoop().inEventLoop()) { reactorNettyConnection.dispose(); } else { From a3ac4b5cc9737ad1a1f06afba7db2a9ad246e8ca Mon Sep 17 00:00:00 2001 From: Kushagra Thapar Date: Wed, 18 Dec 2019 17:52:06 -0800 Subject: [PATCH 2/4] Removed extra logging --- .../cosmos/internal/http/ReactorNettyClient.java | 12 ------------ 1 file changed, 12 deletions(-) diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/ReactorNettyClient.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/ReactorNettyClient.java index ac2510e1b9de..5268eeb65efe 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/ReactorNettyClient.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/ReactorNettyClient.java @@ -90,10 +90,6 @@ private void configureChannelPipelineHandlers() { tcpClient = tcpClient.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, configs.getConnectionAcquireTimeoutInMillis()); - tcpClient = tcpClient.doOnConnect(connection -> { - logger.info("About to connect {}", connection.hashCode()); - }); - tcpClient = tcpClient.doOnConnected(connection -> { int connectionHashCode = connection.hashCode(); openChannels.add(connectionHashCode); @@ -104,19 +100,12 @@ private void configureChannelPipelineHandlers() { maxConnectionsSoFar.set(openChannels.size()); } } - logger.info("Connection established {}", connection.hashCode()); - logger.info("Current connected channels {}", openChannels.size()); logger.info("Max connections so far {}", maxConnectionsSoFar); }); tcpClient = tcpClient.doOnDisconnected(connection -> { int connectionHashCode = connection.hashCode(); openChannels.remove(connectionHashCode); - logger.info("Connection disconnected {}", connectionHashCode); - logger.info("Current connected channels {}", openChannels.size()); - connection.onDispose(() -> { - logger.info("Connection disposed {}", connection.hashCode()); - }); }); return tcpClient; @@ -231,7 +220,6 @@ public Mono bodyAsString(Charset charset) { @Override public void close() { - logger.info("Disposing connection {}", reactorNettyConnection.hashCode()); if (reactorNettyConnection.channel().eventLoop().inEventLoop()) { reactorNettyConnection.dispose(); } else { From a29cabd22d2334ef2ad90424fcd6dc9f76403428 Mon Sep 17 00:00:00 2001 From: Kushagra Thapar Date: Sun, 5 Jan 2020 02:04:11 -0800 Subject: [PATCH 3/4] Removed dispose calls to re-use connection --- .../internal/http/ReactorNettyClient.java | 43 +++---------------- 1 file changed, 5 insertions(+), 38 deletions(-) diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/ReactorNettyClient.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/ReactorNettyClient.java index 5268eeb65efe..f44b4d15b8ec 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/ReactorNettyClient.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/ReactorNettyClient.java @@ -23,9 +23,6 @@ import java.nio.charset.Charset; import java.util.Objects; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiFunction; import static com.azure.data.cosmos.internal.http.HttpClientConfig.REACTOR_NETWORK_LOG_CATEGORY; @@ -35,15 +32,12 @@ */ class ReactorNettyClient implements HttpClient { - private static final Logger logger = LoggerFactory.getLogger(ReactorNettyClient.class); + private static final Logger logger = LoggerFactory.getLogger(ReactorNettyClient.class.getSimpleName()); private HttpClientConfig httpClientConfig; private reactor.netty.http.client.HttpClient httpClient; private ConnectionProvider connectionProvider; - private Set openChannels; - private AtomicInteger maxConnectionsSoFar; - private ReactorNettyClient() {} /** @@ -52,8 +46,6 @@ private ReactorNettyClient() {} public static ReactorNettyClient create(HttpClientConfig httpClientConfig) { ReactorNettyClient reactorNettyClient = new ReactorNettyClient(); reactorNettyClient.httpClientConfig = httpClientConfig; - reactorNettyClient.openChannels = ConcurrentHashMap.newKeySet(); - reactorNettyClient.maxConnectionsSoFar = new AtomicInteger(); reactorNettyClient.httpClient = reactor.netty.http.client.HttpClient.newConnection(); reactorNettyClient.configureChannelPipelineHandlers(); return reactorNettyClient; @@ -66,8 +58,6 @@ public static ReactorNettyClient createWithConnectionProvider(ConnectionProvider ReactorNettyClient reactorNettyClient = new ReactorNettyClient(); reactorNettyClient.connectionProvider = connectionProvider; reactorNettyClient.httpClientConfig = httpClientConfig; - reactorNettyClient.openChannels = ConcurrentHashMap.newKeySet(); - reactorNettyClient.maxConnectionsSoFar = new AtomicInteger(); reactorNettyClient.httpClient = reactor.netty.http.client.HttpClient.create(connectionProvider); reactorNettyClient.configureChannelPipelineHandlers(); return reactorNettyClient; @@ -90,24 +80,6 @@ private void configureChannelPipelineHandlers() { tcpClient = tcpClient.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, configs.getConnectionAcquireTimeoutInMillis()); - tcpClient = tcpClient.doOnConnected(connection -> { - int connectionHashCode = connection.hashCode(); - openChannels.add(connectionHashCode); - - // Update maxConnectionsSoFar with maximum open connection so far - synchronized (this) { - if (openChannels.size() > maxConnectionsSoFar.get()) { - maxConnectionsSoFar.set(openChannels.size()); - } - } - logger.info("Max connections so far {}", maxConnectionsSoFar); - }); - - tcpClient = tcpClient.doOnDisconnected(connection -> { - int connectionHashCode = connection.hashCode(); - openChannels.remove(connectionHashCode); - }); - return tcpClient; }).httpResponseDecoder(httpResponseDecoderSpec -> { httpResponseDecoderSpec.maxInitialLineLength(configs.getMaxHttpInitialLineLength()); @@ -200,31 +172,26 @@ public HttpHeaders headers() { @Override public Flux body() { - return bodyIntern().doFinally(s -> this.close()); + return bodyIntern(); } @Override public Mono bodyAsByteArray() { - return bodyIntern().aggregate().asByteArray().doFinally(s -> this.close()); + return bodyIntern().aggregate().asByteArray(); } @Override public Mono bodyAsString() { - return bodyIntern().aggregate().asString().doFinally(s -> this.close()); + return bodyIntern().aggregate().asString(); } @Override public Mono bodyAsString(Charset charset) { - return bodyIntern().aggregate().asString(charset).doFinally(s -> this.close()); + return bodyIntern().aggregate().asString(charset); } @Override public void close() { - if (reactorNettyConnection.channel().eventLoop().inEventLoop()) { - reactorNettyConnection.dispose(); - } else { - reactorNettyConnection.channel().eventLoop().execute(reactorNettyConnection::dispose); - } } private ByteBufFlux bodyIntern() { From b0b9b05fd266949d96d1949904a3e38c6e9011f7 Mon Sep 17 00:00:00 2001 From: Kushagra Thapar Date: Sun, 5 Jan 2020 10:10:11 -0800 Subject: [PATCH 4/4] Removed empty close method --- .../azure/data/cosmos/internal/http/ReactorNettyClient.java | 4 ---- 1 file changed, 4 deletions(-) diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/ReactorNettyClient.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/ReactorNettyClient.java index f44b4d15b8ec..424f4886b779 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/ReactorNettyClient.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/http/ReactorNettyClient.java @@ -190,10 +190,6 @@ public Mono bodyAsString(Charset charset) { return bodyIntern().aggregate().asString(charset); } - @Override - public void close() { - } - private ByteBufFlux bodyIntern() { return reactorNettyConnection.inbound().receive(); }