imbajin commented on code in PR #3130:
URL: https://github.com/apache/hugegraph/pull/3130#discussion_r3760142892
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClient.java:
##########
@@ -34,10 +44,53 @@
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<>();
+ /*
+ * 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:
⚠️ Two blocked DNS lookups can stop address refresh for every target.
Evidence: CHANNEL_MAINTENANCE_EXECUTOR has only two threads, and each refresh
calls resolveTarget at :307, which reaches InetAddress.getAllByName without a
client deadline; refreshTasks is cleared only after that call returns. If two
targets hang, later targets queue behind them and keep stale channel pools.
Isolate resolution per target or use a resolver with a real
timeout/cancellation, and add a regression where two blocked lookups do not
prevent a third target from refreshing.
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClient.java:
##########
@@ -91,31 +134,60 @@ 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);
+ return this.acquireStub(target, this.blockingStubs,
this::getBlockingStub,
+ stub -> (AbstractBlockingStub)
this.setBlockingStubOption(stub));
+ }
+
+ /**
+ * Returns a cached stub bound to a channel of the target's current pool,
rebuilding the
+ * cache when the pool has been replaced. The stub comes from the pool
that was published at
+ * the last check; a refresh landing immediately afterwards can still
retire that pool, so
+ * callers are not shielded from an in-flight replacement.
+ *
+ * <p>The pool check needs no lock: the pool is published before the
previous one is retired,
+ * so reading the current pool from the map is enough to know retirement
has not started.
+ */
+ @SuppressWarnings("unchecked")
+ private <S> S acquireStub(String target,
+ Map<String, HgPair<ManagedChannel, S>[]>
stubCache,
+ Function<ManagedChannel, S> stubFactory,
+ Function<S, S> stubOption) {
+ while (true) {
+ ManagedChannel[] targetChannels = this.getChannels(target);
+ HgPair<ManagedChannel, S>[] pairs = stubCache.get(target);
+ int index = nextStubIndex();
+ if (!usesChannels(pairs, targetChannels)) {
+ synchronized (stubCache) {
+ pairs = stubCache.get(target);
+ if (!usesChannels(pairs, targetChannels)) {
+ HgPair<ManagedChannel, S>[] value = new
HgPair[concurrency];
+ IntStream.range(0, concurrency).forEach(i -> {
+ ManagedChannel channel = targetChannels[i];
+ value[i] = new HgPair<>(channel,
stubFactory.apply(channel));
+ });
+ S configuredStub =
stubOption.apply(value[index].getValue());
+ if (channels.get(target) != targetChannels) {
+ continue;
+ }
+ stubCache.put(target, value);
+ return configuredStub;
+ }
+ }
+ }
+ S configuredStub = stubOption.apply(pairs[index].getValue());
+ if (channels.get(target) != targetChannels) {
Review Comment:
‼️ The identity check does not protect the handoff from returning a stub
bound to a channel that is being retired. Evidence: acquireStub checks
channels.get(target) at :178 and then returns; a concurrent refresh publishes
the replacement at :339 and calls retireChannels at :349, whose shutdown
immediately rejects new RPCs on the old pool. A caller can pass the check, be
preempted, and receive the stub after retirement begins. Protect stub
acquisition through first-call creation with a pool lease or read/write lock,
or retry through a wrapper that binds to the current pool.
--
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]
