bitflicker64 commented on code in PR #3130:
URL: https://github.com/apache/hugegraph/pull/3130#discussion_r3754617246
##########
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:
Final-head confirmation on `6beea6f46`: per-channel spreading remains
intact. The blocking and async cache builders bind slot `i` to channel `i`, and
the guarded pool-identity check retries only if publication changed while the
stubs were built. The Java 11 suite exercises first-build and post-refresh
spreading across all 32 channels and passes 27/27 in both focused and
Store-profile runs.
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClient.java:
##########
@@ -169,6 +193,224 @@ 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 (int i = 0; i < pairs.length; i++) {
+ HgPair<ManagedChannel, ?> pair = pairs[i];
+ if (pair == null || pair.getKey() != channels[i]) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private void refreshChannelsIfAddressChanged(String target) {
+ if (!this.shouldRefreshChannels(target)) {
+ return;
+ }
+
+ ReentrantLock refreshLock = refreshLocks.computeIfAbsent(target,
+ key -> new
ReentrantLock());
+ if (!refreshLock.tryLock()) {
+ return;
+ }
+
+ try {
+ if (!this.shouldRefreshChannels(target)) {
+ return;
+ }
+
+ String resolvedTarget = this.resolveTarget(target);
+ this.postponeNextRefresh(target);
+ if (resolvedTarget.isEmpty()) {
+ return;
+ }
+
+ ManagedChannel[] staleChannels = channels.get(target);
+ String previousTarget = resolvedTargets.get(target);
+ if (previousTarget == null && staleChannels == null) {
+ resolvedTargets.put(target, resolvedTarget);
+ return;
+ }
+ if (resolvedTarget.equals(previousTarget)) {
+ return;
+ }
+ if (staleChannels == null) {
+ resolvedTargets.put(target, resolvedTarget);
+ return;
+ }
+
+ ManagedChannel[] replacementChannels;
+ try {
+ replacementChannels = this.createChannels(target);
+ } catch (RuntimeException ignored) {
+ return;
+ }
+
+ boolean replaced = false;
+ synchronized (channels) {
+ if (channels.get(target) == staleChannels) {
+ channels.put(target, replacementChannels);
+ resolvedTargets.put(target, resolvedTarget);
+ replaced = true;
+ }
+ }
+
+ if (replaced) {
+ this.retireChannels(staleChannels);
Review Comment:
Final-head confirmation on `6beea6f46`: QueryV2 continues to call the
guarded `getAsyncStub(target)` path, which validates the configured stub pool
against the currently published channel pool before returning. The final
candidate also prevents a directly injected test channel from entering DNS
refresh retirement while proving the normal no-injection branch still delegates
to inherited DNS resolution. Both interleavings have focused regressions; the
Java 11 suite passes 27/27, JaCoCo line 47 is `mi=0`, `ci=5`, and the lifecycle
guard has `mb=0`, `cb=2`.
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClient.java:
##########
@@ -169,6 +201,289 @@ 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 (int i = 0; i < pairs.length; i++) {
+ HgPair<ManagedChannel, ?> pair = pairs[i];
+ if (pair == null || pair.getKey() != channels[i]) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Submits a refresh for the target unless one is already in flight or the
refresh interval
+ * has not elapsed. Returns the in-flight refresh, or null when none is
running.
+ */
+ private CompletableFuture<Void> triggerChannelRefresh(String target) {
+ CompletableFuture<Void> inFlight = refreshTasks.get(target);
+ if (inFlight != null) {
+ return inFlight;
+ }
+ if (!this.shouldRefreshChannels(target)) {
+ return null;
+ }
+
+ CompletableFuture<Void> refresh = new CompletableFuture<>();
+ CompletableFuture<Void> running = refreshTasks.putIfAbsent(target,
refresh);
+ if (running != null) {
+ return running;
+ }
+
+ // Throttle before submitting, so that a failing resolver cannot be
retried in a loop.
+ this.postponeNextRefresh(target);
+ try {
+ this.submitChannelRefresh(() -> {
+ try {
+ this.refreshChannelsIfAddressChanged(target);
+ } catch (Throwable e) {
+ // The executor discards what a task throws, so report it
here.
+ log.warn("Failed to refresh channels of target {}",
target, e);
+ } finally {
+ this.completeRefresh(target, refresh);
+ }
+ });
+ } catch (Throwable e) {
+ // Includes a thread creation denied on this thread; never leave
the entry behind.
+ log.warn("Failed to submit a channel refresh for target {}",
target, e);
+ this.completeRefresh(target, refresh);
+ }
+ return refresh;
+ }
+
+ private void completeRefresh(String target, CompletableFuture<Void>
refresh) {
+ /*
+ * Throttle from completion as well as from submission: a resolver
that is slow rather
+ * than failing can outlast its own interval, which would let every
later call queue
+ * another lookup behind it.
+ */
+ this.postponeNextRefresh(target);
+ refreshTasks.remove(target, refresh);
+ refresh.complete(null);
+ }
+
+ private void submitChannelRefresh(Runnable task) {
+ CHANNEL_MAINTENANCE_EXECUTOR.execute(task);
+ }
+
+ private void awaitInitialResolution(CompletableFuture<Void> refresh) {
+ if (refresh == null) {
+ return;
+ }
+ try {
+ refresh.get(Math.max(0L, this.initialResolutionTimeoutNanos()),
+ TimeUnit.NANOSECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } catch (Exception ignored) {
+ // A slow or failing resolver must not delay the first pool any
further.
+ }
+ }
+
+ /**
+ * Runs on a maintenance thread, never on a request thread. At most one
runs per target at a
+ * time — that comes from the refreshTasks entry, not from the size of the
executor. Replaces
+ * the target's pool when its resolved address set has changed, publishing
the replacement
+ * before retiring the previous pool.
+ */
+ private void refreshChannelsIfAddressChanged(String target) {
+ String resolvedTarget = this.resolveTarget(target);
+ if (resolvedTarget.isEmpty()) {
+ return;
+ }
+
+ ManagedChannel[] staleChannels = channels.get(target);
+ String previousTarget = resolvedTargets.get(target);
+ if (resolvedTarget.equals(previousTarget)) {
+ return;
+ }
+ if (staleChannels == null) {
+ /*
+ * Nothing to replace yet. Recording the address here is what lets
the common path
+ * build its first pool already knowing the address, instead of
rebuilding it.
+ */
+ resolvedTargets.put(target, resolvedTarget);
+ return;
+ }
+
+ ManagedChannel[] replacementChannels;
+ try {
+ replacementChannels = this.createChannels(target);
+ } catch (RuntimeException e) {
+ // Keep serving from the last healthy pool.
+ log.warn("Failed to create replacement channels of target {}, " +
+ "keeping the current pool", target, e);
+ return;
+ }
+
+ boolean replaced = false;
+ synchronized (channels) {
+ if (channels.get(target) == staleChannels) {
+ channels.put(target, replacementChannels);
+ resolvedTargets.put(target, resolvedTarget);
+ replaced = true;
+ }
+ }
+ if (replaced) {
+ log.info("Replaced the channel pool of target {}, address changed
from {} to {}",
+ target, previousTarget, resolvedTarget);
+ }
+
+ this.retireChannels(replaced ? staleChannels : replacementChannels);
+ }
+
+ private boolean shouldRefreshChannels(String target) {
+ AtomicLong nextResolution = nextResolutions.computeIfAbsent(target,
+ key -> new
AtomicLong());
+ return System.nanoTime() - nextResolution.get() >= 0L;
Review Comment:
Final-head confirmation on `6beea6f46`: first refresh state is still
explicit (`AtomicReference<Long>` with `null` meaning uninitialized), so no
`nanoTime()` value is a sentinel. Deadline checks use overflow-safe signed
subtraction (`now - deadline >= 0`), and focused regressions inject both
negative initial time and wraparound across `Long.MAX_VALUE`/`Long.MIN_VALUE`.
The final Java 11 focused and Store-profile runs pass all 27 tests.
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClient.java:
##########
@@ -169,6 +193,224 @@ 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 (int i = 0; i < pairs.length; i++) {
+ HgPair<ManagedChannel, ?> pair = pairs[i];
+ if (pair == null || pair.getKey() != channels[i]) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private void refreshChannelsIfAddressChanged(String target) {
+ if (!this.shouldRefreshChannels(target)) {
+ return;
+ }
+
+ ReentrantLock refreshLock = refreshLocks.computeIfAbsent(target,
+ key -> new
ReentrantLock());
+ if (!refreshLock.tryLock()) {
+ return;
+ }
+
+ try {
+ if (!this.shouldRefreshChannels(target)) {
+ return;
+ }
+
+ String resolvedTarget = this.resolveTarget(target);
+ this.postponeNextRefresh(target);
+ if (resolvedTarget.isEmpty()) {
+ return;
+ }
+
+ ManagedChannel[] staleChannels = channels.get(target);
+ String previousTarget = resolvedTargets.get(target);
+ if (previousTarget == null && staleChannels == null) {
+ resolvedTargets.put(target, resolvedTarget);
+ return;
+ }
+ if (resolvedTarget.equals(previousTarget)) {
+ return;
+ }
+ if (staleChannels == null) {
+ resolvedTargets.put(target, resolvedTarget);
+ return;
+ }
+
+ ManagedChannel[] replacementChannels;
+ try {
+ replacementChannels = this.createChannels(target);
+ } catch (RuntimeException ignored) {
+ return;
+ }
+
+ boolean replaced = false;
+ synchronized (channels) {
+ if (channels.get(target) == staleChannels) {
+ channels.put(target, replacementChannels);
+ resolvedTargets.put(target, resolvedTarget);
+ replaced = true;
+ }
+ }
+
+ if (replaced) {
+ this.retireChannels(staleChannels);
+ } else {
+ this.retireChannels(replacementChannels);
+ }
+ } finally {
+ refreshLock.unlock();
+ }
+ }
+
+ private boolean shouldRefreshChannels(String target) {
+ AtomicLong nextResolution = nextResolutions.computeIfAbsent(target,
+ key -> new
AtomicLong());
+ return System.nanoTime() - nextResolution.get() >= 0L;
+ }
+
+ private void postponeNextRefresh(String target) {
+ long interval = Math.max(0L, this.channelRefreshIntervalNanos());
+ nextResolutions.computeIfAbsent(target, key -> new AtomicLong())
+ .set(System.nanoTime() + interval);
+ }
+
+ protected long channelRefreshIntervalNanos() {
+ return DEFAULT_CHANNEL_REFRESH_INTERVAL_NANOS;
+ }
+
+ protected long channelDrainTimeoutNanos() {
+ return TimeUnit.SECONDS.toNanos(config.getGrpcTimeoutSeconds());
+ }
+
+ private ManagedChannel[] createChannels(String target) {
+ ManagedChannel[] value = new ManagedChannel[concurrency];
+ CountDownLatch latch = new CountDownLatch(concurrency);
+ AtomicReference<RuntimeException> failure = new AtomicReference<>();
+ for (int i = 0; i < concurrency; i++) {
+ int fi = i;
+ executor.execute(() -> {
+ try {
+ value[fi] = createChannel(target);
+ } catch (Exception e) {
+ failure.compareAndSet(null, new RuntimeException(e));
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+
+ InterruptedException interruption = null;
+ while (latch.getCount() > 0L) {
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ interruption = e;
+ }
+ }
+
+ if (failure.get() != null || interruption != null) {
+ forceTerminateChannels(value);
+ }
+ if (interruption != null) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(interruption);
+ }
+ if (failure.get() != null) {
+ throw failure.get();
+ }
+ return value;
+ }
+
+ private void retireChannels(ManagedChannel[] retiredChannels) {
+ Arrays.stream(retiredChannels)
+ .filter(channel -> channel != null && !channel.isShutdown())
+ .forEach(ManagedChannel::shutdown);
+
+ long timeout = Math.max(0L, this.channelDrainTimeoutNanos());
+ CHANNEL_CLEANUP_EXECUTOR.schedule(
+ () -> forceTerminateChannels(retiredChannels), timeout,
+ TimeUnit.NANOSECONDS);
+ }
+
+ private void forceTerminateChannels(ManagedChannel[] retiredChannels) {
+ for (ManagedChannel channel : retiredChannels) {
+ if (channel != null && !channel.isTerminated()) {
+ channel.shutdownNow();
+ }
+ }
+ }
+
+ private static String targetHost(String target) {
+ if (target == null || target.isEmpty()) {
+ return "";
+ }
+
+ String endpoint = target;
+ if (target.startsWith("dns://")) {
+ endpoint = target.substring("dns://".length());
+ while (endpoint.startsWith("/")) {
+ endpoint = endpoint.substring(1);
+ }
+ int pathStart = endpoint.indexOf('/');
+ if (pathStart >= 0) {
+ endpoint = endpoint.substring(pathStart + 1);
+ }
+ } else if (target.contains("://")) {
+ return "";
+ }
+
+ return endpointHost(endpoint);
+ }
+
+ private static String endpointHost(String endpoint) {
+ if (endpoint == null || endpoint.isEmpty()) {
+ return "";
+ }
+
+ if (endpoint.charAt(0) == '[') {
+ int hostEnd = endpoint.indexOf(']');
+ if (hostEnd <= 1) {
+ return "";
+ }
+ return endpoint.substring(1, hostEnd);
+ }
+
+ int lastColon = endpoint.lastIndexOf(':');
+ if (lastColon < 0) {
+ return endpoint;
+ }
+ if (endpoint.indexOf(':') != lastColon) {
+ return endpoint;
+ }
+ return endpoint.substring(0, lastColon);
+ }
+
+ protected InetAddress[] resolveHost(String host) throws
UnknownHostException {
+ return InetAddress.getAllByName(host);
+ }
+
+ protected String resolveTarget(String target) {
+ String host = targetHost(target);
+ if (host.isEmpty()) {
+ return "";
+ }
+ try {
+ return Arrays.stream(this.resolveHost(host))
Review Comment:
Final-head confirmation on `6beea6f46`: refresh DNS resolution and
replacement/retirement coordination still run on prestarted
`channel-maintenance-*` threads, channel construction uses the permitted common
executor, and delayed forced retirement uses the independent
`channel-retirement-*` scheduler. The Java 11 SecurityManager regression denies
socket and thread access on the Gremlin caller while proving resolution,
creation, graceful retirement, and forced retirement occur elsewhere; all 27
tests pass with zero skips. Resolution, submission, partial creation, and
retirement-scheduling failures retain the healthy published pool and allow
retry.
--
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]
