imbajin commented on code in PR #3130:
URL: https://github.com/apache/hugegraph/pull/3130#discussion_r3696134172
##########
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:
‼️ This synchronous lookup runs on the caller of `getChannels()`, but
`InetAddress.getAllByName()` invokes `SecurityManager.checkConnect(host, -1)`.
The default launcher enables `HugeSecurityManager`, whose `checkConnect()`
rejects Gremlin worker stacks and has no HStore/gRPC exemption; only
`UnknownHostException` is caught here. An HStore-backed Gremlin request can
therefore throw `SecurityException` on initial resolution or a periodic refresh
instead of using the healthy pool. Please move resolution to a trusted bounded
background path (preserving the current pool on failure), or add the narrowest
safe HStore policy exemption, and cover a real Gremlin-worker +
`HugeSecurityManager` lookup regression.
##########
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:
⚠️ A caller that loses `tryLock()` can still receive `staleChannels`; after
this swap, retirement immediately calls `shutdown()` on that pool.
`QueryV2Client#getManagedChannel()` uses `getChannels()` directly and creates
its async stub only afterwards, bypassing the identity/retry loops added to
`getAsyncStub()` and `getBlockingStub()`, so a refresh race can hand QueryV2 a
channel that rejects the new RPC. Please route QueryV2 through the guarded
async-stub path or provide an atomic channel/stub lease, and add an
interleaving test for this production consumer.
--
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]
