bitflicker64 commented on code in PR #3130:
URL: https://github.com/apache/hugegraph/pull/3130#discussion_r3689220724
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClient.java:
##########
@@ -91,31 +100,44 @@ public ManagedChannel[] getChannels(String target) {
public abstract AbstractBlockingStub getBlockingStub(ManagedChannel
channel);
public AbstractBlockingStub getBlockingStub(String target) {
- ManagedChannel[] channels = getChannels(target);
- HgPair<ManagedChannel, AbstractBlockingStub>[] pairs =
blockingStubs.get(target);
- long l = counter.getAndIncrement();
- if (l >= limit) {
- counter.set(0);
- }
- int index = (int) (l & (concurrency - 1));
- if (pairs == null) {
- synchronized (blockingStubs) {
- pairs = blockingStubs.get(target);
- if (pairs == null) {
- HgPair<ManagedChannel, AbstractBlockingStub>[] value = new
HgPair[concurrency];
- IntStream.range(0, concurrency).forEach(i -> {
- ManagedChannel channel = channels[index];
- AbstractBlockingStub stub = getBlockingStub(channel);
- value[i] = new HgPair<>(channel, stub);
- // log.info("create channel for {}",target);
- });
- blockingStubs.put(target, value);
- AbstractBlockingStub stub = value[index].getValue();
- return (AbstractBlockingStub) setBlockingStubOption(stub);
+ while (true) {
+ ManagedChannel[] targetChannels = getChannels(target);
+ HgPair<ManagedChannel, AbstractBlockingStub>[] pairs =
blockingStubs.get(target);
+ long l = counter.getAndIncrement();
+ if (l >= limit) {
+ counter.set(0);
+ }
+ int index = (int) (l & (concurrency - 1));
+ if (!usesChannels(pairs, targetChannels)) {
+ synchronized (blockingStubs) {
+ pairs = blockingStubs.get(target);
+ if (!usesChannels(pairs, targetChannels)) {
+ HgPair<ManagedChannel, AbstractBlockingStub>[] value =
+ new HgPair[concurrency];
+ IntStream.range(0, concurrency).forEach(i -> {
+ ManagedChannel channel = targetChannels[index];
+ AbstractBlockingStub stub =
getBlockingStub(channel);
+ value[i] = new HgPair<>(channel, stub);
+ // log.info("create channel for {}",target);
+ });
Review Comment:
Validated as a real issue: the pool build loop used the selected request
`index`, so every cached slot could bind to the same channel. Fixed in
`95489135` by using the loop slot `i` for both blocking and async stub pools,
and added spread assertions that the rebuilt pools cover every channel.
Verified with the refreshed 9-test suite on Java 11.
##########
hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClientTest.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * 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.hugegraph.store.client.grpc;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+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.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+import io.grpc.CallOptions;
+import io.grpc.Channel;
+import io.grpc.ClientCall;
+import io.grpc.ManagedChannel;
+import io.grpc.MethodDescriptor;
+import io.grpc.stub.AbstractAsyncStub;
+import io.grpc.stub.AbstractBlockingStub;
+
+/**
+ * Verifies that Store address changes replace channels and their cached stubs
safely.
+ */
+public class AbstractGrpcClientTest {
+
+ private static final AtomicInteger TARGET_SEQ = new AtomicInteger();
+
+ private static String uniqueTarget(String prefix) {
+ return prefix + "-" + TARGET_SEQ.incrementAndGet() + ":8500";
+ }
+
+ private static boolean belongsToPool(Channel channel,
+ ManagedChannel[] channels) {
+ return Arrays.stream(channels).anyMatch(current -> current == channel);
+ }
+
+ @Test
+ public void testAddressChangeReplacesChannelAndStubPools() {
+ String target = uniqueTarget("address-change");
+ RecordingGrpcClient client = new RecordingGrpcClient();
+ ManagedChannel[] oldChannels = client.getChannels(target);
+ assertNotNull(client.getBlockingStub(target));
+ assertNotNull(client.getAsyncStub(target));
+
+ client.resolvedTarget = "10.0.0.2";
+ ManagedChannel[] newChannels = client.getChannels(target);
+ assertNotSame("an address change must replace the channel pool",
+ oldChannels, newChannels);
+ assertTrue("every stale channel must be shut down",
+
Arrays.stream(oldChannels).allMatch(ManagedChannel::isShutdown));
+
+ client.blockingStubChannels.clear();
+ client.asyncStubChannels.clear();
+ assertNotNull(client.getBlockingStub(target));
+ assertNotNull(client.getAsyncStub(target));
+ assertEquals("the blocking stub pool must be rebuilt",
+ newChannels.length, client.blockingStubChannels.size());
+ assertTrue("replacement blocking stubs must use the new channel pool",
+ client.blockingStubChannels.stream()
+ .allMatch(channel ->
+ belongsToPool(channel,
newChannels)));
+ assertEquals("the async stub pool must be rebuilt",
+ newChannels.length, client.asyncStubChannels.size());
+ assertTrue("replacement async stubs must use the new channel pool",
+ client.asyncStubChannels.stream()
+ .allMatch(channel ->
+ belongsToPool(channel,
newChannels)));
+ }
+
+ @Test
+ public void testFirstSuccessfulResolutionReplacesUnknownChannels() {
+ String target = uniqueTarget("first-successful-resolution");
+ RecordingGrpcClient client = new RecordingGrpcClient();
+ client.resolvedTarget = "";
+ ManagedChannel[] unknownChannels = client.getChannels(target);
+
+ client.resolvedTarget = "10.0.0.1";
+ ManagedChannel[] resolvedChannels = client.getChannels(target);
+ assertNotSame("a pool with unknown addresses must be replaced",
+ unknownChannels, resolvedChannels);
+ assertTrue("every channel from the unknown pool must be shut down",
+
Arrays.stream(unknownChannels).allMatch(ManagedChannel::isShutdown));
+ assertTrue("the resolved channel pool must remain live",
+
Arrays.stream(resolvedChannels).noneMatch(ManagedChannel::isShutdown));
+ }
+
+ @Test
+ public void testOlderResolutionCannotReplaceNewerChannels() throws
Exception {
+ String target = uniqueTarget("concurrent-address-change");
+ OutOfOrderResolverGrpcClient client = new
OutOfOrderResolverGrpcClient();
+ ManagedChannel[] oldChannels = client.getChannels(target);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+
+ try {
+ Future<ManagedChannel[]> staleResolution =
+ executor.submit(() -> client.getChannels(target));
+ assertTrue("the stale resolution must be in flight",
+ client.staleResolutionStarted.await(5,
TimeUnit.SECONDS));
+ Future<ManagedChannel[]> freshResolution =
+ executor.submit(() -> client.getChannels(target));
+ ManagedChannel[] freshChannels = freshResolution.get(5,
TimeUnit.SECONDS);
+
+ assertNotSame("the newer address must replace the old channel
pool",
+ oldChannels, freshChannels);
+ client.releaseStaleResolution.countDown();
+ assertSame("the late stale result must retain the newer channel
pool",
+ freshChannels, staleResolution.get(5,
TimeUnit.SECONDS));
+ assertTrue("the replaced channel pool must be shut down",
+
Arrays.stream(oldChannels).allMatch(ManagedChannel::isShutdown));
+ assertTrue("the newer channel pool must remain live",
+
Arrays.stream(freshChannels).noneMatch(ManagedChannel::isShutdown));
+ } finally {
+ client.releaseStaleResolution.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testStubBuildRetriesAfterChannelRefresh() throws Exception {
Review Comment:
Fixed in `95489135`. Added `testAsyncStubBuildRetriesAfterChannelRefresh()`
with the same latch-controlled stale-build interleaving as the blocking path,
and it asserts both returned async stubs plus the final async cache only
reference current live channels.
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClient.java:
##########
@@ -169,6 +205,70 @@ protected AbstractStub setStubOption(AbstractStub value) {
config.getGrpcMaxOutboundMessageSize());
}
+ private static boolean usesChannels(HgPair<ManagedChannel, ?>[] pairs,
+ ManagedChannel[] channels) {
+ if (pairs == null || pairs.length != channels.length) {
+ return false;
+ }
+ for (HgPair<ManagedChannel, ?> pair : pairs) {
+ if (pair == null || pair.getKey() == null ||
+ !containsChannel(channels, pair.getKey())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean containsChannel(ManagedChannel[] channels,
+ ManagedChannel expected) {
+ return Arrays.stream(channels).anyMatch(channel -> channel ==
expected);
+ }
+
+ private void refreshChannelsIfAddressChanged(String target) {
+ long resolutionRequest = resolutionRequests.computeIfAbsent(target,
+ key -> new
AtomicLong())
+ .incrementAndGet();
+ String resolvedTarget = this.resolveTarget(target);
+ if (resolvedTarget.isEmpty()) {
+ return;
+ }
+ synchronized (channels) {
+ Long appliedResolution = appliedResolutions.get(target);
+ if (appliedResolution != null && appliedResolution >=
resolutionRequest) {
+ return;
+ }
+ appliedResolutions.put(target, resolutionRequest);
+ String previousTarget = resolvedTargets.put(target,
resolvedTarget);
+ if (previousTarget == null && !channels.containsKey(target)) {
+ return;
+ }
+ if (resolvedTarget.equals(previousTarget)) {
+ return;
+ }
+ ManagedChannel[] staleChannels = channels.remove(target);
+ if (staleChannels != null) {
+ Arrays.stream(staleChannels)
+ .filter(channel -> channel != null &&
!channel.isShutdown())
+ .forEach(ManagedChannel::shutdownNow);
+ }
+ }
+ }
+
+ protected String resolveTarget(String target) {
+ try {
+ String host = URI.create("dns://" + target).getHost();
Review Comment:
Fixed in `95489135` for the target formats this client can safely
fingerprint: raw `host:port`, `dns:///host:port`, and bracketed IPv6.
Unsupported resolver schemes now return an empty fingerprint so the client does
not silently monitor the wrong host while gRPC handles the original target.
Broader non-DNS resolver fingerprinting is out of scope for this
channel-lifecycle fix. Added URI-form, IPv6, and unsupported-scheme tests.
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClient.java:
##########
@@ -169,6 +205,70 @@ protected AbstractStub setStubOption(AbstractStub value) {
config.getGrpcMaxOutboundMessageSize());
}
+ private static boolean usesChannels(HgPair<ManagedChannel, ?>[] pairs,
+ ManagedChannel[] channels) {
+ if (pairs == null || pairs.length != channels.length) {
+ return false;
+ }
+ for (HgPair<ManagedChannel, ?> pair : pairs) {
+ if (pair == null || pair.getKey() == null ||
+ !containsChannel(channels, pair.getKey())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean containsChannel(ManagedChannel[] channels,
+ ManagedChannel expected) {
+ return Arrays.stream(channels).anyMatch(channel -> channel ==
expected);
+ }
+
+ private void refreshChannelsIfAddressChanged(String target) {
+ long resolutionRequest = resolutionRequests.computeIfAbsent(target,
+ key -> new
AtomicLong())
+ .incrementAndGet();
+ String resolvedTarget = this.resolveTarget(target);
+ if (resolvedTarget.isEmpty()) {
+ return;
+ }
+ synchronized (channels) {
+ Long appliedResolution = appliedResolutions.get(target);
+ if (appliedResolution != null && appliedResolution >=
resolutionRequest) {
+ return;
+ }
+ appliedResolutions.put(target, resolutionRequest);
+ String previousTarget = resolvedTargets.put(target,
resolvedTarget);
+ if (previousTarget == null && !channels.containsKey(target)) {
+ return;
+ }
+ if (resolvedTarget.equals(previousTarget)) {
+ return;
+ }
+ ManagedChannel[] staleChannels = channels.remove(target);
+ if (staleChannels != null) {
+ Arrays.stream(staleChannels)
+ .filter(channel -> channel != null &&
!channel.isShutdown())
+ .forEach(ManagedChannel::shutdownNow);
Review Comment:
Fixed in `95489135`. The refresh path now creates and publishes the
replacement pool before retiring the old pool. Retired channels receive
graceful `shutdown()` first, and a bounded drain task only calls
`shutdownNow()` if termination does not complete. Added
`testRefreshGracefullyRetiresActiveStreamChannels()` to cover a held
async/stream stub during refresh.
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClient.java:
##########
@@ -169,6 +205,70 @@ protected AbstractStub setStubOption(AbstractStub value) {
config.getGrpcMaxOutboundMessageSize());
}
+ private static boolean usesChannels(HgPair<ManagedChannel, ?>[] pairs,
+ ManagedChannel[] channels) {
+ if (pairs == null || pairs.length != channels.length) {
+ return false;
+ }
+ for (HgPair<ManagedChannel, ?> pair : pairs) {
+ if (pair == null || pair.getKey() == null ||
+ !containsChannel(channels, pair.getKey())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean containsChannel(ManagedChannel[] channels,
+ ManagedChannel expected) {
+ return Arrays.stream(channels).anyMatch(channel -> channel ==
expected);
+ }
+
+ private void refreshChannelsIfAddressChanged(String target) {
+ long resolutionRequest = resolutionRequests.computeIfAbsent(target,
+ key -> new
AtomicLong())
+ .incrementAndGet();
+ String resolvedTarget = this.resolveTarget(target);
Review Comment:
Fixed in `95489135`. Refresh checks are now throttled by a per-target TTL
and guarded with a try-lock, so only one caller performs DNS resolution while
concurrent callers keep using the last cached healthy pool. Added repeated
acquisition and delayed concurrent resolver coverage to verify this path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
