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`. Resolution no longer runs on the caller of
`getChannels()`.
`resolveTarget()`, replacement creation and retirement now run on a
`channel-maintenance` executor; `getChannels()` only triggers the refresh and
keeps serving the last healthy pool while it runs. `callFromGremlin()` goes
through `callFromWorkerWithClass()`, which short-circuits on the current
thread's name (`gremlin-server-exec`, `task-worker`) before walking the stack,
so a maintenance thread sits outside every one of those checks.
I did not take the policy-exemption option. `getChannels(String)` was
public, so exempting this path from `checkConnect` would let any script that
gets it onto the stack reach an arbitrary host, and it would add a fourth entry
to the set already marked `// TODO: remove this unsafe entrance`.
Moving it surfaced the same problem a second time on this path:
`retireChannels()` calls `schedule()` from the caller, and the first
`schedule()` creates its worker lazily with `new Thread(...)`, which trips
`checkAccess(ThreadGroup)` under the same predicate. A `checkConnect`-only
exemption would have left that live. The maintenance threads are now prestarted
at class init, guarded so a denied prestart cannot leave the class
uninitializable, and a denied submission is logged and cleaned up rather than
silently wedging refresh for that target.
Also changed while here:
- The first pool for a target is built once its address is known — the cold
call waits a bounded time for the in-flight resolution — so a cold start no
longer creates 32 channels and immediately retires them.
- Targets are parsed with `URI`. This also fixes `unix:/var/run/store.sock`:
the previous `contains("://")` guard missed the single-slash spelling, so that
target resolved a literal host named `unix` every refresh interval.
- The refresh path now logs. It had no logging at all, and moving to a
scheduled executor made it worse, since that executor discards whatever a task
throws — a failing refresh was invisible. There is now a warning on resolution,
creation and submission failure, and an info line on each pool replacement with
the old and new address.
Coverage: `testRefreshSucceedsWhenTheCallerThreadIsDeniedSocketAccess()`
installs a security manager that denies `checkConnect` and
`checkAccess(ThreadGroup)` for `gremlin-server-exec-*` threads, as
`HugeSecurityManager` does, then asserts a caller on such a thread still
receives a stub, the refresh still publishes a replacement pool, and every
resolution ran on a maintenance thread.
Two things this does not cover, which I think belong in their own change:
1. The *first* pool for a cold target is still created on the calling
thread, and `createChannels()` submits to the shared `common` executor, whose
workers are also created lazily — so a Gremlin worker that is the first caller
for a new target can still trip `checkAccess(ThreadGroup)` there. That path is
unchanged from master and predates this PR, but it means the fix is incomplete
for exactly the case where a *new* store address appears. Happy to file it, or
fold it in here if you would rather it ship together.
2. If the resolved address set for a hostname legitimately rotates (a
headless service scaling up and down), the fingerprint changes every refresh
interval and the pool is rebuilt each time. Worth deciding whether refresh
should require N consecutive observations before acting.
--
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]
