bitflicker64 commented on code in PR #3130:
URL: https://github.com/apache/hugegraph/pull/3130#discussion_r3774890694
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClient.java:
##########
@@ -34,10 +46,59 @@
import io.grpc.stub.AbstractAsyncStub;
import io.grpc.stub.AbstractBlockingStub;
import io.grpc.stub.AbstractStub;
+import lombok.extern.slf4j.Slf4j;
+@Slf4j
public abstract class AbstractGrpcClient {
protected static Map<String, ManagedChannel[]> channels = new
ConcurrentHashMap<>();
+ private static final Map<String, String> resolvedTargets = new
ConcurrentHashMap<>();
+ // A null deadline is the explicit "never scheduled" state; every long is
a valid clock value.
+ private static final Map<String, AtomicReference<Long>> nextResolutions =
+ new ConcurrentHashMap<>();
+ private static final Map<String, CompletableFuture<Void>> refreshTasks =
+ new ConcurrentHashMap<>();
+ private static final Map<String, ReentrantReadWriteLock> channelLocks =
+ new ConcurrentHashMap<>();
+ /*
+ * Refresh runs here rather than on a request thread: a caller of
getChannels() may hold a
+ * Gremlin worker stack, which HugeSecurityManager denies socket access
to. Creating the very
+ * first pool for a target is still done by the caller, so that path stays
exposed.
+ */
+ private static final ScheduledThreadPoolExecutor
CHANNEL_MAINTENANCE_EXECUTOR =
Review Comment:
Fixed in exact head `bdd8df11`. The trusted SecurityManager boundary is
preserved with bounded prestarted pools: 4 maintenance threads, 4
cold-initialization threads, and 1 retirement thread (9 total rather than 129).
The focused Java 11 `AbstractGrpcClientTest` suite now passes 37/37 with zero
failures/errors, including executor footprint and restricted-caller coverage.
Three independent reviews and the final Design Audit found no blocking issue on
this head.
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClient.java:
##########
@@ -34,10 +46,59 @@
import io.grpc.stub.AbstractAsyncStub;
import io.grpc.stub.AbstractBlockingStub;
import io.grpc.stub.AbstractStub;
+import lombok.extern.slf4j.Slf4j;
+@Slf4j
public abstract class AbstractGrpcClient {
protected static Map<String, ManagedChannel[]> channels = new
ConcurrentHashMap<>();
+ private static final Map<String, String> resolvedTargets = new
ConcurrentHashMap<>();
Review Comment:
Fixed in exact head `bdd8df11`. `closeChannel(target)` and client `close()`
now remove channel/stub pools plus all target-keyed resolution, deadline,
refresh, and lock state; terminal Store notices invoke exact-node eviction, and
stale sessions cannot recreate evicted state. Same-address replacements and
duplicate stale notices are identity-safe. The Java 11 focused suite passes
37/37; independent lifecycle/concurrency review and the Design Audit found no
blocker.
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClient.java:
##########
@@ -169,6 +268,323 @@ 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);
+ }
+
+ void submitChannelRefresh(Runnable task) {
+ CHANNEL_MAINTENANCE_EXECUTOR.execute(task);
+ }
+
+ void submitChannelInitialization(Runnable task) {
+ CHANNEL_INITIALIZATION_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;
+ }
+
+ ReentrantReadWriteLock.WriteLock writeLock =
channelLock(target).writeLock();
+ writeLock.lock();
+ try {
+ 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);
+ } finally {
+ writeLock.unlock();
+ }
+ }
+
+ private boolean shouldRefreshChannels(String target) {
+ AtomicReference<Long> nextResolution =
+ nextResolutions.computeIfAbsent(target, key -> new
AtomicReference<>());
+ Long deadline = nextResolution.get();
+ return deadline == null || this.nanoTime() - deadline >= 0L;
+ }
+
+ private void postponeNextRefresh(String target) {
+ long interval = Math.max(0L, this.channelRefreshIntervalNanos());
+ nextResolutions.computeIfAbsent(target, key -> new AtomicReference<>())
+ .set(this.nanoTime() + interval);
+ }
+
+ protected long nanoTime() {
+ return System.nanoTime();
+ }
+
+ protected long channelRefreshIntervalNanos() {
+ return DEFAULT_CHANNEL_REFRESH_INTERVAL_NANOS;
+ }
+
+ private long initialResolutionTimeoutNanos() {
+ return DEFAULT_INITIAL_RESOLUTION_TIMEOUT_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;
+ try {
+ this.submitChannelCreation(() -> {
+ try {
+ value[fi] = createChannel(target);
+ } catch (Exception e) {
Review Comment:
Fixed in exact head `bdd8df11`. Channel tasks capture `Throwable`, all
submitted tasks converge, every slot is validated before publication, and
partial channels are force-terminated on failure. Fatal `Error` is rethrown
after cleanup, null pools cannot publish, creation submission/shutdown races
cannot strand the latch, and interrupted callers regain interrupt status. The
focused Java 11 suite passes 37/37, including `Error`, null-channel,
partial-cleanup, submission, shutdown, and interrupt regressions; independent
failure-path review and the Design Audit found no blocker.
--
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]
