diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java index d5db73d54..677518e4e 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java @@ -94,6 +94,8 @@ import java.io.IOException; import java.net.InetAddress; import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.util.HashMap; import java.util.LinkedHashMap; import java.util.Map; import java.util.Map.Entry; @@ -103,8 +105,6 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; -import java.util.function.Function; -import java.util.stream.Collectors; public class ChannelManager { @@ -1129,27 +1129,47 @@ public EventLoopGroup getEventLoopGroup() { return addressResolverGroup; } + /** + * Builds a point-in-time stats snapshot in O(open channels + idle pooled channels). + */ public ClientStats getClientStats() { - Map totalConnectionsPerHost = openChannels.stream() - .map(Channel::remoteAddress) - .filter(a -> a instanceof InetSocketAddress) - .map(a -> (InetSocketAddress) a) - .map(InetSocketAddress::getHostString) - .collect(Collectors.groupingBy(Function.identity(), Collectors.counting())); - - Map idleConnectionsPerHost = channelPool.getIdleChannelCountPerHost(); - - Map statsPerHost = totalConnectionsPerHost.entrySet() - .stream() - .collect(Collectors.toMap(Entry::getKey, entry -> { - final long totalConnectionCount = entry.getValue(); - final long idleConnectionCount = idleConnectionsPerHost.getOrDefault(entry.getKey(), 0L); - final long activeConnectionCount = totalConnectionCount - idleConnectionCount; - return new HostStats(activeConnectionCount, idleConnectionCount); - })); + Map connectionsPerHost = new HashMap<>(); + for (Channel channel : openChannels) { + SocketAddress remoteAddress = channel.remoteAddress(); + if (remoteAddress instanceof InetSocketAddress) { + String host = ((InetSocketAddress) remoteAddress).getHostString(); + ConnectionCounts counts = connectionsPerHost.get(host); + if (counts == null) { + counts = new ConnectionCounts(); + connectionsPerHost.put(host, counts); + } + counts.totalConnectionCount++; + } + } + + for (Entry entry : channelPool.getIdleChannelCountPerHost().entrySet()) { + ConnectionCounts counts = connectionsPerHost.get(entry.getKey()); + if (counts != null) { + counts.idleConnectionCount = entry.getValue(); + } + } + + Map statsPerHost = new HashMap<>(connectionsPerHost.size()); + for (Entry entry : connectionsPerHost.entrySet()) { + ConnectionCounts counts = entry.getValue(); + statsPerHost.put(entry.getKey(), new HostStats( + counts.totalConnectionCount - counts.idleConnectionCount, + counts.idleConnectionCount)); + } return new ClientStats(statsPerHost); } + private static final class ConnectionCounts { + + private long totalConnectionCount; + private long idleConnectionCount; + } + public boolean isOpen() { return channelPool.isOpen(); } diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java b/client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java index f59d427f9..d0e89a809 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java @@ -27,8 +27,10 @@ import org.slf4j.LoggerFactory; import java.net.InetSocketAddress; +import java.net.SocketAddress; import java.time.Duration; import java.util.Deque; +import java.util.HashMap; import java.util.Iterator; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -36,9 +38,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; -import java.util.function.Function; import java.util.function.Predicate; -import java.util.stream.Collectors; import static org.asynchttpclient.util.DateUtils.unpreciseMillisTime; @@ -238,18 +238,22 @@ public void flushPartitions(Predicate predicate) { @Override public Map getIdleChannelCountPerHost() { - return partitions - .values() - .stream() - .flatMap(ConcurrentLinkedDeque::stream) + Map idleChannelsPerHost = new HashMap<>(); + for (ConcurrentLinkedDeque partition : partitions.values()) { + for (Channel channel : partition) { // Skip channels that have been claimed (removeAll tombstone, or a node a concurrent // poll already leased) but not yet unlinked, so the count reflects leasable channels. - .filter(DefaultChannelPool::isLeasable) - .map(Channel::remoteAddress) - .filter(a -> a.getClass() == InetSocketAddress.class) - .map(a -> (InetSocketAddress) a) - .map(InetSocketAddress::getHostString) - .collect(Collectors.groupingBy(Function.identity(), Collectors.counting())); + if (isLeasable(channel)) { + SocketAddress remoteAddress = channel.remoteAddress(); + if (remoteAddress.getClass() == InetSocketAddress.class) { + String host = ((InetSocketAddress) remoteAddress).getHostString(); + Long currentCount = idleChannelsPerHost.get(host); + idleChannelsPerHost.put(host, currentCount == null ? 1L : currentCount + 1L); + } + } + } + } + return idleChannelsPerHost; } private static boolean isLeasable(Channel channel) { diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/DefaultChannelPoolTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/DefaultChannelPoolTest.java index da256149f..9795b4155 100644 --- a/client/src/test/java/org/asynchttpclient/netty/channel/DefaultChannelPoolTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/channel/DefaultChannelPoolTest.java @@ -25,8 +25,11 @@ import org.junit.jupiter.api.Test; import java.lang.reflect.Field; +import java.net.InetSocketAddress; +import java.net.SocketAddress; import java.time.Duration; import java.util.Collections; +import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedDeque; @@ -384,6 +387,30 @@ public void cleanerUnlinksManyTombstonesInOneTick() throws Exception { pool.destroy(); } + @Test + public void idleCountPerHostCountsOnlyLeasableChannels() { + CapturingTimer timer = new CapturingTimer(); + DefaultChannelPool pool = ttlPool(timer); + Channel first = channelWithRemoteAddress("example.com"); + Channel second = channelWithRemoteAddress("example.com"); + Channel otherHost = channelWithRemoteAddress("example.org"); + Channel claimed = channelWithRemoteAddress("example.com"); + + pool.offer(first, KEY); + pool.offer(second, KEY); + pool.offer(otherHost, KEY); + pool.offer(claimed, KEY); + assertTrue(pool.removeAll(claimed)); + + Map idleCounts = pool.getIdleChannelCountPerHost(); + + assertEquals(2, idleCounts.size()); + assertEquals(Long.valueOf(2), idleCounts.get("example.com")); + assertEquals(Long.valueOf(1), idleCounts.get("example.org")); + + pool.destroy(); + } + // ---- concurrency: no leaked tombstones, never leases a claimed channel ---- @Test @@ -471,6 +498,18 @@ private static Object idleState(Channel channel) throws Exception { return channel.attr(key).get(); } + private static Channel channelWithRemoteAddress(String host) { + return new EmbeddedChannel() { + + private final InetSocketAddress remoteAddress = InetSocketAddress.createUnresolved(host, 443); + + @Override + protected SocketAddress remoteAddress0() { + return remoteAddress; + } + }; + } + @SuppressWarnings("unchecked") private static int partitionSize(DefaultChannelPool pool, Object key) throws Exception { Field partitionsField = DefaultChannelPool.class.getDeclaredField("partitions");