Optimize client stats collection

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<String, Long> 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<String, Long> idleConnectionsPerHost = channelPool.getIdleChannelCountPerHost();
-
-        Map<String, HostStats> 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<String, ConnectionCounts> 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<String, Long> entry : channelPool.getIdleChannelCountPerHost().entrySet()) {
+            ConnectionCounts counts = connectionsPerHost.get(entry.getKey());
+            if (counts != null) {
+                counts.idleConnectionCount = entry.getValue();
+            }
+        }
+
+        Map<String, HostStats> statsPerHost = new HashMap<>(connectionsPerHost.size());
+        for (Entry<String, ConnectionCounts> 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();
     }

client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java

@@ -27,18 +27,18 @@
 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;
 import java.util.concurrent.ConcurrentLinkedDeque;
 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<Object> predicate) {
 
     @Override
     public Map<String, Long> getIdleChannelCountPerHost() {
-        return partitions
-                .values()
-                .stream()
-                .flatMap(ConcurrentLinkedDeque::stream)
+        Map<String, Long> idleChannelsPerHost = new HashMap<>();
+        for (ConcurrentLinkedDeque<Channel> 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) {

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<String, Long> 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");