From 1dddf57aa6c130b3e6e526dc237cfd57d89d933d Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Mon, 13 Jul 2026 20:49:05 -0400 Subject: [PATCH 1/2] HDDS-15849 reverted HDDS-14571 that caused performance regression --- .../hadoop/hdds/scm/XceiverClientGrpc.java | 38 ++--- ...stXceiverClientManagerCloseContention.java | 161 ++++++++++++++++++ 2 files changed, 172 insertions(+), 27 deletions(-) create mode 100644 hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientManagerCloseContention.java diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java index 8c52f339027..66a28edb067 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java @@ -39,7 +39,6 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.LockSupport; -import java.util.stream.Collectors; import org.apache.hadoop.hdds.HddsConfigKeys; import org.apache.hadoop.hdds.HddsUtils; import org.apache.hadoop.hdds.client.BlockID; @@ -92,7 +91,6 @@ */ public class XceiverClientGrpc extends XceiverClientSpi { private static final Logger LOG = LoggerFactory.getLogger(XceiverClientGrpc.class); - private static final int SHUTDOWN_WAIT_INTERVAL_MILLIS = 100; private static final int SHUTDOWN_WAIT_MAX_SECONDS = 5; private final Pipeline pipeline; private final ConfigurationSource config; @@ -267,39 +265,25 @@ public void close() { return; } + // Use shutdownNow() (not the graceful shutdown()) so in-flight RPCs are + // cancelled and the channel terminates immediately. close() is frequently + // invoked from cache eviction while the XceiverClientManager clientCache + // monitor is held (HDDS-15849); a blocking graceful drain there serializes + // every concurrent acquireClient()/releaseClient() call. for (ChannelInfo channelInfo : dnChannelInfoMap.values()) { - channelInfo.getChannel().shutdown(); - } - - final long maxWaitNanos = TimeUnit.SECONDS.toNanos(SHUTDOWN_WAIT_MAX_SECONDS); - long deadline = System.nanoTime() + maxWaitNanos; - List nonTerminatedChannels = dnChannelInfoMap.values() - .stream() - .map(ChannelInfo::getChannel) - .filter(Objects::nonNull) - .collect(Collectors.toList()); - - while (!nonTerminatedChannels.isEmpty() && System.nanoTime() < deadline) { - nonTerminatedChannels.removeIf(ManagedChannel::isTerminated); + ManagedChannel channel = channelInfo.getChannel(); + channel.shutdownNow(); try { - Thread.sleep(SHUTDOWN_WAIT_INTERVAL_MILLIS); + if (!channel.awaitTermination(SHUTDOWN_WAIT_MAX_SECONDS, TimeUnit.SECONDS)) { + LOG.warn("Channel {} did not terminate within {}s.", channel, SHUTDOWN_WAIT_MAX_SECONDS); + } } catch (InterruptedException e) { - LOG.error("Interrupted while waiting for channels to terminate", e); + LOG.error("Interrupted while waiting for channel termination", e); Thread.currentThread().interrupt(); break; } } - List failedChannels = dnChannelInfoMap.entrySet() - .stream() - .filter(e -> !e.getValue().getChannel().isTerminated()) - .map(Map.Entry::getKey) - .collect(Collectors.toList()); - - if (!failedChannels.isEmpty()) { - LOG.warn("Channels {} did not terminate within timeout.", failedChannels); - } - dnChannelInfoMap.clear(); } diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientManagerCloseContention.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientManagerCloseContention.java new file mode 100644 index 00000000000..b817f73cb76 --- /dev/null +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientManagerCloseContention.java @@ -0,0 +1,161 @@ +/* + * 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.hadoop.hdds.scm; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import java.util.stream.IntStream; +import org.apache.hadoop.hdds.conf.ConfigurationSource; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.scm.XceiverClientManager.ScmClientConfig; +import org.apache.hadoop.hdds.scm.XceiverClientManager.XceiverClientManagerConfigBuilder; +import org.apache.hadoop.hdds.scm.client.ClientTrustManager; +import org.apache.hadoop.hdds.scm.pipeline.MockPipeline; +import org.apache.hadoop.hdds.scm.pipeline.Pipeline; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Regression test for HDDS-15849. + * + *

HDDS-14571 changed {@link XceiverClientGrpc#close()} from a forced + * {@code shutdownNow()} (which cancels in-flight RPCs and terminates in ~1ms) + * to a graceful {@code shutdown()} followed by a polling loop that always + * sleeps at least {@code SHUTDOWN_WAIT_INTERVAL_MILLIS} (100ms) before it can + * observe channel termination. Every {@code close()} therefore costs ≥100ms + * instead of ~1ms. + * + *

The damage is amplified by where {@code close()} runs. Clients + * are torn down through {@link XceiverClientManager}'s cache eviction, and the + * whole acquire / evict / close sequence executes under the {@code clientCache} + * monitor: + * 

+ *   acquireClient()                    // synchronized (clientCache)
+ *     -> getClient()
+ *       -> clientCache.get(key, newClient)
+ *          -> (Guava evicts an entry on the calling thread)
+ *             -> RemovalListener.onRemoval()   // synchronized (clientCache)
+ *                -> XceiverClientSpi.setEvicted() -> cleanup() -> close()
+ *
+ * So the blocking graceful-shutdown wait runs while the {@code clientCache} + * monitor is held, and every other thread in {@code acquireClient()} / + * {@code releaseClient()} serializes behind it. + * + *

This test exercises the real {@code XceiverClientManager} and + * {@code XceiverClientGrpc.close()} — nothing is emulated. Real (lazily + * connected, plaintext) gRPC channels are created against random local + * datanode addresses; no live datanode is required because + * {@code close()} only shuts the channels down. {@code maxCacheSize == 1} plus + * a distinct pipeline per iteration forces the previously cached client to be + * evicted and closed on essentially every acquisition — mirroring the per-file + * client churn of EC checksum collection. + * + *

Run it against the pre-fix {@code close()} and the wall-clock is dominated + * by serialized 100ms sleeps ({@code ~ evictions * 100ms}); run it against the + * fix (immediate {@code shutdownNow()}) and per-close cost collapses. The + * assertion below encodes the fixed expectation, so this test FAILS on the + * regression and PASSES once HDDS-15849 restores immediate termination. + */ +class TestXceiverClientManagerCloseContention { + + private static final Logger LOG = + LoggerFactory.getLogger(TestXceiverClientManagerCloseContention.class); + + private static final int THREADS = 8; + private static final int CYCLES_PER_THREAD = 8; + + /** + * Upper bound on acceptable average wall-clock cost per eviction-driven + * close(). The pre-fix code has a hard ≥100ms floor per close (the + * mandatory Thread.sleep in the termination polling loop); immediate + * shutdownNow() is well under a millisecond. 40ms sits safely between the + * two so the assertion is not timing-flaky. + */ + private static final long MAX_MILLIS_PER_CLOSE = 40; + + @Test + void evictionDrivenCloseDoesNotSerializeAcquisition() throws Exception { + ConfigurationSource conf = new OzoneConfiguration(); + // maxCacheSize == 1 => each acquisition of a new pipeline evicts (and thus + // closes) the previously cached client. + ScmClientConfig clientConf = new XceiverClientManagerConfigBuilder() + .setMaxCacheSize(1) + .setStaleThresholdMs(TimeUnit.SECONDS.toMillis(30)) + .build(); + + try (XceiverClientManager manager = + new XceiverClientManager(conf, clientConf, (ClientTrustManager) null)) { + + ExecutorService pool = Executors.newFixedThreadPool(THREADS); + CountDownLatch start = new CountDownLatch(1); + long wallStartNanos; + try { + List> futures = IntStream.range(0, THREADS) + .mapToObj(t -> pool.submit(() -> { + start.await(); + for (int i = 0; i < CYCLES_PER_THREAD; i++) { + // Distinct pipeline per cycle => distinct cache key => the + // previously cached client is evicted and closed. + Pipeline pipeline = MockPipeline.createPipeline(1); + XceiverClientSpi client = manager.acquireClient(pipeline); + manager.releaseClient(client, false); + } + return null; + })) + .collect(Collectors.toList()); + + wallStartNanos = System.nanoTime(); + start.countDown(); + for (Future f : futures) { + f.get(120, TimeUnit.SECONDS); + } + } finally { + pool.shutdownNow(); + } + long wallMillis = + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - wallStartNanos); + + // recordStats() is enabled on the cache; each eviction drives one close(). + long evictions = manager.getClientCache().stats().evictionCount(); + long millisPerClose = evictions == 0 ? 0 : wallMillis / evictions; + + LOG.info("threads={} cyclesPerThread={} acquisitions={} evictions(closes)={}", + THREADS, CYCLES_PER_THREAD, THREADS * CYCLES_PER_THREAD, evictions); + LOG.info("wall={}ms ms-per-close={}ms (regression floor is ~100ms/close)", + wallMillis, millisPerClose); + + assertThat(evictions) + .as("cache eviction should have driven close() calls") + .isPositive(); + + assertThat(millisPerClose) + .as("HDDS-15849: eviction-driven close() must not serialize " + + "acquisition under the clientCache monitor; the pre-fix " + + "graceful-shutdown wait forces a ~100ms floor per close") + .isLessThan(MAX_MILLIS_PER_CLOSE); + } + } +} From e0fa4ee837d1a0a1203f28294617f08bb092a25d Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Tue, 14 Jul 2026 11:01:49 -0400 Subject: [PATCH 2/2] fixed javadoc to match to imlementation --- .../main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java index 66a28edb067..6882595928d 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java @@ -256,7 +256,7 @@ public boolean isConnected(DatanodeDetails details) { /** * Closes all the communication channels of the client one-by-one. * When a channel is closed, no further requests can be sent via the channel, - * and the method waits to finish all ongoing communication. + * and any in-flight RPCs are cancelled immediately (shutdownNow semantics). */ @Override public void close() {