This is an automated email from the ASF dual-hosted git repository. dmsolr pushed a commit to branch perf/upstream-snapshot-publish in repository https://gitbox.apache.org/repos/asf/shenyu.git
commit 54d43f31eb134f67d3b09eee0905f16868650998 Author: Luke Haochao Zhuang <[email protected]> AuthorDate: Tue Oct 20 17:26:05 2026 +0800 perf(loadbalancer): publish read-only healthy upstream snapshots to avoid per-request list copies findUpstreamListBySelectorId previously created a defensive ArrayList copy of the healthy CopyOnWriteArrayList on every routing lookup. Move the copy cost to the write side: UpstreamCheckTask now re-publishes an immutable List.copyOf snapshot inside its lock whenever the healthy set structurally changes (config submit, health transition, removal), and the request path takes the snapshot reference directly with zero allocation. - add healthySnapshot map + publishHealthySnapshot/getHealthySnapshot to UpstreamCheckTask - findUpstreamListBySelectorId returns the published snapshot (null when no entry) - add UpstreamSnapshotPublishTest: read-only snapshot, removal visibility, health transition republish, concurrent readers/writers regression, and a ThreadMXBean allocation comparison against the legacy per-request copy --- .../loadbalancer/cache/UpstreamCacheManager.java | 11 +- .../loadbalancer/cache/UpstreamCheckTask.java | 41 ++- .../cache/UpstreamSnapshotPublishTest.java | 288 +++++++++++++++++++++ 3 files changed, 334 insertions(+), 6 deletions(-) diff --git a/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCacheManager.java b/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCacheManager.java index 2c80c83fa4..9ccd654c73 100644 --- a/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCacheManager.java +++ b/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCacheManager.java @@ -26,7 +26,6 @@ import org.apache.shenyu.common.utils.MapUtils; import org.apache.shenyu.common.utils.Singleton; import org.apache.shenyu.loadbalancer.entity.Upstream; -import java.util.ArrayList; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -122,12 +121,16 @@ public final class UpstreamCacheManager { /** * Find upstream list by selector id list. * + * <p>Returns the read-only snapshot published by the health-check task on + * configuration and health-state updates, so the request path does not copy + * the healthy list per lookup. Callers must not mutate the returned list; + * elements keep the same shared-instance semantics as before.</p> + * * @param selectorId the selector id - * @return the list + * @return the immutable snapshot list, or null when the selector has no entry */ public List<Upstream> findUpstreamListBySelectorId(final String selectorId) { - List<Upstream> upstreamList = task.getHealthyUpstream().get(selectorId); - return Objects.isNull(upstreamList) ? null : new ArrayList<>(upstreamList); + return task.getHealthySnapshot(selectorId); } /** diff --git a/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCheckTask.java b/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCheckTask.java index 114aff74c0..5e24024535 100644 --- a/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCheckTask.java +++ b/shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/cache/UpstreamCheckTask.java @@ -55,6 +55,14 @@ public final class UpstreamCheckTask implements Runnable { private final Map<String, List<Upstream>> unhealthyUpstream = Maps.newConcurrentMap(); + /** + * Read-only snapshots of {@link #healthyUpstream}, re-published inside {@link #lock} + * on every structural or health-state change. Request-path readers take the snapshot + * reference directly, so routing lookups no longer copy the whole healthy list per + * request while writers own the (bounded, low-frequency) copy cost. + */ + private final Map<String, List<Upstream>> healthySnapshot = Maps.newConcurrentMap(); + private final Object lock = new Object(); private final AtomicBoolean checkStarted = new AtomicBoolean(false); @@ -290,6 +298,9 @@ public final class UpstreamCheckTask implements Runnable { List<Upstream> list = MapUtils.computeIfAbsent(map, selectorId, k -> new CopyOnWriteArrayList<>()); if (!list.contains(upstream)) { list.add(upstream); + if (healthyUpstream == map) { + publishHealthySnapshot(selectorId); + } } } } @@ -304,8 +315,8 @@ public final class UpstreamCheckTask implements Runnable { public void removeFromMap(final Map<String, List<Upstream>> map, final String selectorId, final Upstream upstream) { synchronized (lock) { List<Upstream> list = map.get(selectorId); - if (CollectionUtils.isNotEmpty(list)) { - list.remove(upstream); + if (CollectionUtils.isNotEmpty(list) && list.remove(upstream) && healthyUpstream == map) { + publishHealthySnapshot(selectorId); } } } @@ -319,8 +330,21 @@ public final class UpstreamCheckTask implements Runnable { synchronized (lock) { healthyUpstream.remove(selectorId); unhealthyUpstream.remove(selectorId); + healthySnapshot.remove(selectorId); } } + + /** + * Re-publish an immutable snapshot of the healthy list for the selector. + * Must be called while holding {@link #lock} so readers never observe a + * snapshot torn across concurrent structural changes. + * + * @param selectorId selectorId + */ + private void publishHealthySnapshot(final String selectorId) { + List<Upstream> live = healthyUpstream.get(selectorId); + healthySnapshot.put(selectorId, Objects.isNull(live) ? List.of() : List.copyOf(live)); + } /** * Print healthy and unhealthy check log. @@ -365,6 +389,19 @@ public final class UpstreamCheckTask implements Runnable { public List<Upstream> getHealthyUpstreamListBySelectorId(final String selectorId) { return getHealthyUpstream().get(selectorId); } + + /** + * Get the published read-only healthy upstream snapshot of the selector. + * The snapshot is re-published on configuration and health-state updates and + * must not be mutated by callers; it may be empty but never contains a + * removed target after the removal has been published. + * + * @param selectorId selectorId + * @return immutable snapshot list, or null when the selector has no entry + */ + public List<Upstream> getHealthySnapshot(final String selectorId) { + return healthySnapshot.get(selectorId); + } /** * Get unhealthy upstream map. diff --git a/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/cache/UpstreamSnapshotPublishTest.java b/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/cache/UpstreamSnapshotPublishTest.java new file mode 100644 index 0000000000..28b71953ef --- /dev/null +++ b/shenyu-loadbalancer/src/test/java/org/apache/shenyu/loadbalancer/cache/UpstreamSnapshotPublishTest.java @@ -0,0 +1,288 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.shenyu.loadbalancer.cache; + +import org.apache.shenyu.loadbalancer.entity.Upstream; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Assumptions; +import org.junit.jupiter.api.Test; + +import java.lang.management.ManagementFactory; +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * Tests for the read-only healthy-upstream snapshot publication that replaces + * the per-request defensive copy in {@code findUpstreamListBySelectorId}. + */ +public class UpstreamSnapshotPublishTest { + + @Test + public void testReturnedListIsReadOnlySnapshot() { + final UpstreamCacheManager cacheManager = UpstreamCacheManager.getInstance(); + final String selectorId = "SNAPSHOT_READONLY_TEST"; + cacheManager.submit(selectorId, buildUpstreams(2)); + try { + List<Upstream> snapshot = cacheManager.findUpstreamListBySelectorId(selectorId); + Assertions.assertNotNull(snapshot); + Assertions.assertEquals(2, snapshot.size()); + Assertions.assertThrows(UnsupportedOperationException.class, + () -> snapshot.add(Upstream.builder().url("intruder:8080").status(true).healthCheckEnabled(false).build())); + Assertions.assertThrows(UnsupportedOperationException.class, () -> snapshot.remove(0)); + } finally { + cacheManager.removeByKey(selectorId); + } + } + + @Test + public void testRemovedTargetIsNotARoutingCandidateInNewSnapshot() { + final UpstreamCacheManager cacheManager = UpstreamCacheManager.getInstance(); + final String selectorId = "SNAPSHOT_REMOVAL_TEST"; + List<Upstream> upstreams = buildUpstreams(3); + cacheManager.submit(selectorId, upstreams); + try { + List<Upstream> before = cacheManager.findUpstreamListBySelectorId(selectorId); + Assertions.assertEquals(3, Objects.requireNonNull(before).size()); + + UpstreamCheckTask task = getUpstreamCheckTask(cacheManager); + Assertions.assertNotNull(task); + task.triggerRemoveOne(selectorId, upstreams.get(1)); + + List<Upstream> after = cacheManager.findUpstreamListBySelectorId(selectorId); + Assertions.assertEquals(2, Objects.requireNonNull(after).size()); + Assertions.assertTrue(after.stream().noneMatch(u -> u.getUrl().equals(upstreams.get(1).getUrl())), + "removed target must not remain a routing candidate"); + // an already handed-out snapshot stays coherent, it is not mutated in place + Assertions.assertEquals(3, before.size()); + } finally { + cacheManager.removeByKey(selectorId); + } + Assertions.assertNull(cacheManager.findUpstreamListBySelectorId(selectorId), + "removeByKey must clear the published snapshot"); + } + + @Test + public void testHealthTransitionRepublishesSnapshot() { + final UpstreamCacheManager cacheManager = UpstreamCacheManager.getInstance(); + final String selectorId = "SNAPSHOT_HEALTH_TRANSITION_TEST"; + List<Upstream> upstreams = buildUpstreams(2); + cacheManager.submit(selectorId, upstreams); + try { + UpstreamCheckTask task = getUpstreamCheckTask(cacheManager); + Assertions.assertNotNull(task); + // simulate a health transition: healthy -> unhealthy + upstreams.get(0).setHealthy(false); + task.putToMap(task.getUnhealthyUpstream(), selectorId, upstreams.get(0)); + task.removeFromMap(task.getHealthyUpstream(), selectorId, upstreams.get(0)); + List<Upstream> afterDown = cacheManager.findUpstreamListBySelectorId(selectorId); + Assertions.assertEquals(1, Objects.requireNonNull(afterDown).size()); + // and back: unhealthy -> healthy + upstreams.get(0).setHealthy(true); + task.putToMap(task.getHealthyUpstream(), selectorId, upstreams.get(0)); + task.removeFromMap(task.getUnhealthyUpstream(), selectorId, upstreams.get(0)); + List<Upstream> afterUp = cacheManager.findUpstreamListBySelectorId(selectorId); + Assertions.assertEquals(2, Objects.requireNonNull(afterUp).size()); + } finally { + cacheManager.removeByKey(selectorId); + } + } + + @Test + public void testConcurrentReadersAndWritersNeverSeeTornState() throws InterruptedException { + final UpstreamCacheManager cacheManager = UpstreamCacheManager.getInstance(); + final String selectorId = "SNAPSHOT_CONCURRENCY_TEST"; + final int baseSize = 20; + cacheManager.submit(selectorId, buildUpstreams(baseSize)); + UpstreamCheckTask task = getUpstreamCheckTask(cacheManager); + Assertions.assertNotNull(task); + final List<Upstream> churnUpstreams = buildUpstreams(5, "churn"); + final List<Throwable> failures = new CopyOnWriteArrayList<>(); + final AtomicBoolean stop = new AtomicBoolean(false); + final CountDownLatch started = new CountDownLatch(6); + ExecutorService pool = Executors.newFixedThreadPool(6); + try { + for (int r = 0; r < 4; r++) { + pool.execute(newReaderLoop(cacheManager, selectorId, failures, stop, started)); + } + pool.execute(newChurnWriterLoop(task, selectorId, churnUpstreams, failures, stop, started)); + pool.execute(newConfigChurnWriterLoop(cacheManager, selectorId, baseSize, failures, stop, started)); + Assertions.assertTrue(started.await(5, TimeUnit.SECONDS)); + TimeUnit.MILLISECONDS.sleep(1500); + stop.set(true); + } finally { + pool.shutdown(); + Assertions.assertTrue(pool.awaitTermination(10, TimeUnit.SECONDS)); + } + Assertions.assertTrue(failures.isEmpty(), "concurrent readers/writers must not fail: " + failures); + // after churn ends, removed targets must not remain routing candidates + for (Upstream churn : churnUpstreams) { + task.triggerRemoveOne(selectorId, churn); + } + List<Upstream> finalSnapshot = cacheManager.findUpstreamListBySelectorId(selectorId); + Assertions.assertNotNull(finalSnapshot); + for (Upstream churn : churnUpstreams) { + Assertions.assertTrue(finalSnapshot.stream().noneMatch(u -> u.getUrl().equals(churn.getUrl())), + "removed churn target must not remain a routing candidate"); + } + cacheManager.removeByKey(selectorId); + } + + @Test + @SuppressWarnings("deprecation") + public void testSnapshotLookupAllocatesFarLessThanPerRequestCopy() { + final UpstreamCacheManager cacheManager = UpstreamCacheManager.getInstance(); + final String selectorId = "SNAPSHOT_ALLOCATION_TEST"; + final int upstreamCount = 100; + final int iterations = 50_000; + cacheManager.submit(selectorId, buildUpstreams(upstreamCount)); + try { + UpstreamCheckTask task = getUpstreamCheckTask(cacheManager); + Assertions.assertNotNull(task); + java.lang.management.ThreadMXBean bean = ManagementFactory.getThreadMXBean(); + Assumptions.assumeTrue(bean instanceof com.sun.management.ThreadMXBean, + "thread allocation measurement not available on this JVM"); + com.sun.management.ThreadMXBean sunBean = (com.sun.management.ThreadMXBean) bean; + Assumptions.assumeTrue(sunBean.isThreadAllocatedMemorySupported()); + sunBean.setThreadAllocatedMemoryEnabled(true); + long threadId = Thread.currentThread().getId(); + + long sink = 0; + // warm up both paths + for (int i = 0; i < 5_000; i++) { + sink += cacheManager.findUpstreamListBySelectorId(selectorId).size(); + sink += new ArrayList<>(task.getHealthyUpstreamListBySelectorId(selectorId)).size(); + } + + long before = sunBean.getThreadAllocatedBytes(threadId); + for (int i = 0; i < iterations; i++) { + sink += cacheManager.findUpstreamListBySelectorId(selectorId).size(); + } + final long snapshotAllocation = sunBean.getThreadAllocatedBytes(threadId) - before; + + before = sunBean.getThreadAllocatedBytes(threadId); + for (int i = 0; i < iterations; i++) { + sink += new ArrayList<>(task.getHealthyUpstreamListBySelectorId(selectorId)).size(); + } + final long copyAllocation = sunBean.getThreadAllocatedBytes(threadId) - before; + + Assertions.assertTrue(sink > 0); + Assertions.assertTrue(copyAllocation > 0, "the legacy per-request copy must allocate"); + Assertions.assertTrue(snapshotAllocation * 10 < copyAllocation, + "snapshot lookup should be allocation-free compared to the copy, snapshot=" + + snapshotAllocation + " bytes, copy=" + copyAllocation + " bytes"); + } finally { + cacheManager.removeByKey(selectorId); + } + } + + private Runnable newReaderLoop(final UpstreamCacheManager cacheManager, final String selectorId, + final List<Throwable> failures, final AtomicBoolean stop, final CountDownLatch started) { + return () -> { + started.countDown(); + long sink = 0; + while (!stop.get()) { + try { + List<Upstream> snapshot = cacheManager.findUpstreamListBySelectorId(selectorId); + if (Objects.nonNull(snapshot)) { + for (Upstream upstream : snapshot) { + sink += upstream.getUrl().length(); + } + } + } catch (Throwable t) { + failures.add(t); + return; + } + } + if (sink < 0) { + failures.add(new IllegalStateException("unreachable")); + } + }; + } + + private Runnable newChurnWriterLoop(final UpstreamCheckTask task, final String selectorId, + final List<Upstream> churnUpstreams, final List<Throwable> failures, + final AtomicBoolean stop, final CountDownLatch started) { + return () -> { + started.countDown(); + while (!stop.get()) { + try { + for (Upstream churn : churnUpstreams) { + task.triggerAddOne(selectorId, churn); + task.triggerRemoveOne(selectorId, churn); + } + } catch (Throwable t) { + failures.add(t); + return; + } + } + }; + } + + private Runnable newConfigChurnWriterLoop(final UpstreamCacheManager cacheManager, final String selectorId, + final int baseSize, final List<Throwable> failures, + final AtomicBoolean stop, final CountDownLatch started) { + return () -> { + started.countDown(); + while (!stop.get()) { + try { + // configuration churn: full replacement and re-submission + cacheManager.submit(selectorId, buildUpstreams(baseSize)); + TimeUnit.MILLISECONDS.sleep(5); + } catch (Throwable t) { + failures.add(t); + return; + } + } + }; + } + + private List<Upstream> buildUpstreams(final int count) { + return buildUpstreams(count, "upstream"); + } + + private List<Upstream> buildUpstreams(final int count, final String prefix) { + List<Upstream> upstreams = new ArrayList<>(count); + for (int i = 0; i < count; i++) { + upstreams.add(Upstream.builder() + .protocol("http://";) + .url(prefix + i + ":8080") + .status(true) + .healthCheckEnabled(false) + .build()); + } + return upstreams; + } + + private UpstreamCheckTask getUpstreamCheckTask(final UpstreamCacheManager manager) { + try { + Field field = UpstreamCacheManager.class.getDeclaredField("task"); + field.setAccessible(true); + return (UpstreamCheckTask) field.get(manager); + } catch (Exception e) { + return null; + } + } +}
