bitflicker64 commented on code in PR #3130:
URL: https://github.com/apache/hugegraph/pull/3130#discussion_r3765125072
##########
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:
Validated and fixed in commit e852a8ff. Stub construction now completes
before acquiring the target read lock, then acquires that lock for the final
published-pool identity check and cache publication/return. Refresh publication
and retirement take the same target write lock, so retirement cannot begin
between the final check and handoff. The existing blocking, async, and QueryV2
interleaving tests, including testQueryV2StubFollowsPublishedPoolAcrossRefresh,
pass; the focused run is 28/28 with zero failures or errors.
##########
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:
Validated and fixed in commit e852a8ff. CHANNEL_MAINTENANCE_EXECUTOR is now
prestarted with 64 workers, while refreshes remain single-flight per target via
refreshTasks; two blocked InetAddress lookups cannot occupy the whole
maintenance pool. Added testBlockedResolutionsDoNotStarveAnotherTarget, which
holds two resolutions and proves a third target replaces its pool and keeps all
replacement channels live. Focused AbstractGrpcClientTest passes 28/28 with
zero failures or errors.
--
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]
