bitflicker64 commented on code in PR #3130:
URL: https://github.com/apache/hugegraph/pull/3130#discussion_r3699643098
##########
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)`, so QueryV2 goes through the same identity and retry
loop as every other consumer, and `getManagedChannel()` is gone.
The commented-out line beside it was worth checking the history for: as
imported (`6c72ca5f`), the async pool bound all 32 entries to `channels[index]`
instead of `channels[i]`, so routing QueryV2 through `getAsyncStub()` back then
would have funnelled every stream through a single connection. #3128
(`b026a90a`) fixed that initializer, so the reason for the hand-rolled
round-robin no longer exists. Spread is preserved: `getAsyncStub(String)`
indexes with `counter.getAndIncrement() & (concurrency - 1)` across all 32
channels, and `setStubOption` is applied identically.
I did not add a channel/stub lease. QueryV2 holds a stream for the life of a
scan (`QueryExecutor#getIterator`), so a lease would have to be released
through every iterator termination path, and any long or leaked scan would pin
a retired pool open — which is the failure mode in #3124.
`getChannels(String)` is now `protected`. It has no caller outside the class
after this change, and handing a raw `ManagedChannel[]` to outside code is no
longer safe, since those channels are shut down once a refresh replaces them.
Coverage: `testQueryV2StubFollowsPublishedPoolAcrossRefresh()` blocks a real
`QueryV2Client` inside its stub build, refreshes the pool underneath it, and
then asserts the returned stub is bound to a channel in the published pool and
not shut down. I checked it is load-bearing by reverting
`getQueryServiceStub()` to the old body and re-running: it fails. An earlier
version of this test passed against the reverted code, because by the time it
asked for a stub the replacement had already been published — the interleaving
is what makes it meaningful.
While unifying the two acquisition loops I also dropped the `synchronized
(channels)` from the cached stub path. Since a replacement pool is always
published before the previous one is retired, reading the current pool from the
map already orders the check, so the monitor added nothing — and `channels` is
static, so it was serialising every store RPC in the JVM, with the
`withDeadlineAfter`/`withMaxMessageSize` stub derivation happening inside it.
The identity re-check and the `synchronized (stubCache)` build guard are
unchanged. Related: unifying the loops removed the `.parallel()` from the async
pool build, which was building 32 stub wrappers on the common pool while
holding the `asyncStubs` monitor, and on a Gremlin worker the pool ramp-up
would itself trip `checkAccess(ThreadGroup)`.
To be explicit about what this does not fix: routing narrows the window, it
does not close it. The guarded path re-checks pool identity before returning,
so it can never hand back a stub from an already-swapped pool — but a refresh
landing between `return` and `stub.query(observer)` still reaches a channel on
which `shutdown()` has been called, and gRPC rejects new calls there even
during the drain window. Closing it needs either retry on `UNAVAILABLE` or
deferring `shutdown()` until the drain deadline. I would rather do that as a
follow-up than widen this PR, unless you want it here.
One related note: `QueryV2Client.setTestChannel()` has no callers anywhere
in the tree, and with refresh enabled it is now actively unsafe — every pool
entry would be the same channel, so retiring the stale pool would shut down the
replacement. Worth deleting; I left it alone to keep this diff to the review
points.
--
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]
