bitflicker64 commented on code in PR #3130:
URL: https://github.com/apache/hugegraph/pull/3130#discussion_r3699641804
##########
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:
Fixed in `198de19e`. `resolveTarget()`, replacement creation and retirement
now run on a prestarted `channel-maintenance` executor; `getChannels()` only
triggers a refresh and keeps serving the last healthy pool. I avoided a
`HugeSecurityManager` exemption: `getChannels(String)` was public, and the lazy
thread creation in `retireChannels()`'s `schedule()` would trip
`checkAccess(ThreadGroup)` under the same predicate anyway. Covered by
`testRefreshSucceedsWhenTheCallerThreadIsDeniedSocketAccess()`, which denies
`checkConnect`/`checkAccess` on `gremlin-server-exec-*` threads and asserts the
caller still gets a stub and all resolution ran on maintenance threads.
One pre-existing gap I'd file separately: the very first pool for a cold
target is still built via the caller, so a Gremlin worker that is the first
caller for a new address can still be denied there.
##########
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:
Fixed in `198de19e`. `getQueryServiceStub()` now returns
`getAsyncStub(target)` and `getManagedChannel()` is gone, so QueryV2 uses the
same identity/retry loop as every other consumer. The hand-rolled round-robin
only existed because the async pool once bound every entry to `channels[index]`
(fixed in #3128), so nothing is lost. `getChannels(String)` is now `protected`
— handing out a raw `ManagedChannel[]` is unsafe once refresh can retire it.
Covered by `testQueryV2StubFollowsPublishedPoolAcrossRefresh()`, which
refreshes the pool while a stub build is blocked mid-flight; it fails against
the old body.
Note: routing narrows the race but a refresh landing between stub return and
first call can still hit a just-shut-down channel — retry-on-`UNAVAILABLE`
would close it, better as a follow-up.
--
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]
