bitflicker64 commented on code in PR #3157:
URL: https://github.com/apache/hugegraph/pull/3157#discussion_r3901065262
##########
hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java:
##########
@@ -75,6 +131,16 @@ protected AbstractBlockingStub createBlockingStub() {
return KvServiceGrpc.newBlockingStub(channel);
}
+ @Override
+ protected long stubResetTimeoutMillis() {
+ return Math.max(1L, Math.min(config.getGrpcTimeOut(),
WATCH_START_TIMEOUT_MS));
Review Comment:
⚠️ Important. The watch-start budget is applied client-wide, so it also
shrinks the PD reconnect budget for every blocking KV and lock call.
`stubResetTimeoutMillis()` is consumed by `AbstractClient.resetStub()` as
the total deadline for the whole peer loop (AbstractClient.java:156-159), and
`resetStub()` is reached from `getBlockingStub()` (AbstractClient.java:107-116)
— not just from the watch path. That method backs `put`, `get`, `delete`,
`deletePrefix`, `scanPrefix`, `putTTL`, `keepTTLAlive`, `lock`,
`lockWithoutReentrant`, `isLocked`, `unlock` and `keepAlive`. With the default
`PDConfig.grpcTimeOut = 60000` (PDConfig.java:34) this override returns 5000,
and the derived per-peer deadline becomes `min(60000, remaining /
remainingHosts)` — about 1.7 s each across three PD peers instead of 60 s.
Measured against three unreachable peers (`10.255.255.1-3:8686`), same
`PDConfig`, same `getBlockingStub()` path:
| build | `KvClient.get()` | `MetaClient.getGraphs()` |
|---|---|---|
| base `7bb624b99` | 21328 / 35403 / 6511 ms | 37053 ms |
| head `f99b6bdd` | 4984 / 5213 / 5212 ms | 25071 ms |
At this head `KvClient` pins to ~5.0 s on every run — exactly
`WATCH_START_TIMEOUT_MS` — while a sibling client on identical configuration
keeps its multi-peer budget. So a configured `pd.grpc_timeout` of 60 s is
silently honoured for `MetaClient` but capped at 5 s for `KvClient`. During a
slow PD leader election or a GC pause longer than the per-peer slice, `lock()`
and `keepAlive()` now fail with `PD_UNREACHABLE` where they previously waited.
The same cap applies to store-node startup config reads
(`PdConfigureListener.onApplicationEvent` calls `scanPrefix`/`put`) and to
`PdMetaDriver`/`SchemaDriver`.
Requested change: scope the 5 s readiness budget to the watch-start path
only — for example thread it through the async stub reset that `startWatch()`
triggers, or gate the override on the async stub — and leave
`getBlockingStub()` on the inherited `grpcTimeOut * hostCount` budget so
`pd.grpc_timeout` keeps governing lock and KV traffic.
##########
hugegraph-struct/src/main/java/org/apache/hugegraph/SchemaDriver.java:
##########
@@ -98,24 +117,73 @@ public static void init(PDConfig pdConfig) {
}
public static void init(PDConfig pdConfig, int cacheSize, long expiration)
{
- SchemaDriver instance = INSTANCE.get();
- if (instance != null) {
- throw new NotAllowException(
- "The SchemaDriver [cacheSize=%s, expiration=%s, " +
- "client=%s] has already been initialized and is not " +
- "allowed to be initialized again", instance.caches.limit(),
- instance.caches.expiration(), instance.client);
+ synchronized (LIFECYCLE_LOCK) {
+ SchemaDriver instance = INSTANCE.get();
+ if (instance != null) {
+ throw new NotAllowException(
+ "The SchemaDriver [cacheSize=%s, expiration=%s, " +
+ "client=%s] has already been initialized and is not " +
+ "allowed to be initialized again",
instance.caches.limit(),
+ instance.caches.expiration(), instance.client);
+ }
+ INSTANCE.set(new SchemaDriver(pdConfig, cacheSize, expiration));
}
- INSTANCE.compareAndSet(null, new SchemaDriver(pdConfig, cacheSize,
- expiration));
}
public static void destroy() {
- SchemaDriver instance = INSTANCE.get();
- if (instance != null) {
- instance.caches.cancelScheduleCacheClean();
- instance.caches.destroyAll();
- INSTANCE.set(null);
+ SchemaDriver instance = null;
+ CountDownLatch completion;
+ boolean closeResources = false;
+ synchronized (LIFECYCLE_LOCK) {
Review Comment:
⚠️ Important. `destroy()` keeps `INSTANCE` published while the driver's
`KvClient` is being closed, so callers can be handed a driver whose PD client
is already shut down.
`INSTANCE` is only cleared in the `finally` at line 158-161, after
`instance.closeResources()` (line 156) has already run `client.close()`. The
new `testDestroyKeepsInstanceUntilResourcesAreClosed` asserts exactly this
window. Meanwhile `SchemaGraph.schemaDriverInit()` (SchemaGraph.java:56-64)
checks `SchemaDriver.getInstance() == null`, sees the still-published instance,
skips `init()`, and returns it — and `SchemaGraph`'s constructor immediately
calls `loadConfig()` -> `graphConfig()` -> `client.get()` on that driver.
By then `KvClient.close()` has set `closed = true`, shut down both reconnect
executors and called `shutdownNow()` on the channel, so
`AbstractClient.resetStub()` breaks out on its first `isShutdown()` check
(AbstractClient.java:159) and `getBlockingStub()` throws `PD_UNREACHABLE`. On
the base commit `destroy()` never closed the client, so a driver observed in
this window still worked. `init()` is unusable in the window too: it takes
`LIFECYCLE_LOCK`, sees the non-null `INSTANCE`, and throws `NotAllowException`
(asserted by `testInitDoesNotWaitForDestroyCleanup`).
This is the mirror image of the earlier review note about clearing
`INSTANCE` too early — the fix moved the hazard rather than removing it,
because publication and usability are still two separate states.
Requested change: make the instance unreachable before its resources are
torn down while still preventing a premature re-`init()` — for example clear
`INSTANCE` inside the first `LIFECYCLE_LOCK` block and let the `destroying`
flag alone block `init()` until cleanup finishes, so `getInstance()` never
returns a driver with a closed client. Please add a test that calls
`getInstance()` during the cleanup window and asserts the returned driver is
either absent or still usable.
##########
hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/AbstractClient.java:
##########
@@ -249,39 +326,70 @@ protected <ReqT, RespT> KVPair<Boolean, RespT>
concurrentBlockingUnaryCall(
protected <ReqT, RespT> void streamingCall(MethodDescriptor<ReqT, RespT>
method, ReqT request,
StreamObserver<RespT>
responseObserver,
int retry) throws PDException {
- AbstractStub stub = getStub();
+ AbstractStub stub;
+ Channel attemptChannel;
+ synchronized (this) {
+ stub = getStub();
+ AbstractStub currentStub = proxy.getStub();
+ attemptChannel = currentStub == null ? stub.getChannel() :
currentStub.getChannel();
+ Consumer<Channel> attemptConsumer =
this.streamingAttemptConsumer.get();
+ if (attemptConsumer != null) {
+ attemptConsumer.accept(attemptChannel);
+ }
+ }
try {
ClientCall<ReqT, RespT> call = stub.getChannel().newCall(method,
stub.getCallOptions());
ClientCalls.asyncServerStreamingCall(call, request,
responseObserver);
} catch (Exception e) {
log.error("rpc call with exception :", e);
if (e instanceof StatusRuntimeException) {
if (retry < proxy.getHostCount()) {
- synchronized (this) {
- proxy.setStub(null);
- }
+ invalidateAsyncStub(attemptChannel);
streamingCall(method, request, responseObserver, ++retry);
+ return;
}
}
+ throw new PDException(Pdpb.ErrorType.PD_UNREACHABLE_VALUE,
+ "RPC streaming call failed", e);
+ }
+ }
+
+ protected <ReqT, RespT> void streamingCall(MethodDescriptor<ReqT, RespT>
method, ReqT request,
+ StreamObserver<RespT>
responseObserver,
+ int retry,
+ Consumer<Channel>
attemptConsumer)
+ throws PDException {
+ Consumer<Channel> previous = this.streamingAttemptConsumer.get();
Review Comment:
🧹 Minor. Routing `attemptConsumer` through a `ThreadLocal` makes it
invisible to the documented 4-arg override point, which silently disables
transport rotation.
This overload stores the consumer in `streamingAttemptConsumer` (line 63)
and relies on the 4-arg `streamingCall` reading it back (line 335-338). A
subclass that overrides the 4-arg method — which
`testLegacyStreamingCallOverrideRemainsDispatchTarget` (KvClientTest.java:893,
`LegacyStreamingOverrideKvClient` at KvClientTest.java:1502-1518) deliberately
establishes as supported — never invokes the consumer.
`KvClient.recordAttemptChannel()` then never runs, so
`subscription.attemptChannel` stays null. On `Leader_Changed`,
`requestReconnect(..., rotateTransport = true)` copies that null into
`reconnectChannel` (KvClient.java:429-432), `reconnect()` skips
`invalidateAttemptStub()` (KvClient.java:556-558), the cached async stub
survives, and the watch re-attaches to the same follower — the exact failure
this PR is fixing, reintroduced for any subclass that keeps the legacy override.
The coupling is also implicit: the two calls only work because they happen
on the same thread, and the value is keyed on a per-instance `ThreadLocal`
whose entry is only reclaimed when the `finally` runs or the client is
collected.
Requested change: pass the consumer explicitly rather than through thread
state — have both public overloads delegate to one private implementation that
takes `Consumer<Channel> attemptConsumer` as a parameter (null for the legacy
signature), and drop the `ThreadLocal` field. That keeps the 4-arg override
working as a dispatch target while making it obvious at the call site that an
overrider must forward the consumer.
--
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]
