From 50dbb793fb8d8f2616a249e8166c4ed4a719826f Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Mon, 20 Jul 2026 10:48:11 +0200 Subject: [PATCH] Optimize client stats collection Replace stream grouping in client stats snapshots with explicit per-host counters to avoid redundant collector passes when metrics are scraped frequently. Count idle pooled channels with direct partition iteration while preserving existing host filtering and tombstone skipping behavior. Add coverage for idle per-host counts so claimed channels are not reported as leasable. Codex on behalf of Pavel Ptashyts Co-Authored-By: Codex --- .../netty/channel/ChannelManager.java | 58 +++++++++++++------ .../netty/channel/DefaultChannelPool.java | 28 +++++---- .../netty/channel/DefaultChannelPoolTest.java | 39 +++++++++++++ 3 files changed, 94 insertions(+), 31 deletions(-) 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");