bitflicker64 commented on code in PR #3157: URL: https://github.com/apache/hugegraph/pull/3157#discussion_r3896390667
########## hugegraph-struct/src/test/java/org/apache/hugegraph/SchemaDriverTest.java: ########## @@ -0,0 +1,149 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hugegraph; + +import java.lang.reflect.Field; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; + +import org.apache.hugegraph.exception.HugeException; +import org.apache.hugegraph.pd.client.KvClient; +import org.apache.hugegraph.pd.client.PDConfig; +import org.apache.hugegraph.pd.common.PDException; +import org.apache.hugegraph.pd.grpc.kv.WatchResponse; +import org.junit.Assert; +import org.junit.Test; + +public class SchemaDriverTest { + + @Test + public void testDestroyClosesOwnedKvClient() throws Exception { + TrackingKvClient client = new TrackingKvClient(); + SchemaDriver driver = new SchemaDriver(client, 10, 60_000L); + instanceReference().set(driver); + + try { + SchemaDriver.destroy(); + + Assert.assertTrue(client.closed); + Assert.assertNull(SchemaDriver.getInstance()); + } finally { + instanceReference().set(null); + client.close(); + } + } + + @Test(timeout = 3000L) Review Comment: ‼️ These two new tests fail CI at this head, so the PR is currently red on code it adds. `build-server-macos-rocksdb (macos-15-intel)`, job `99501401894`: ``` [ERROR] Tests run: 3, Failures: 1, Errors: 1 - in org.apache.hugegraph.SchemaDriverTest [ERROR] SchemaDriverTest.testDestroyClosesOwnedKvClient:46 expected null, but was:<org.apache.hugegraph.SchemaDriver@29176cc1> [ERROR] SchemaDriverTest.testDestroyKeepsInstanceUntilResourcesAreClosed:55 » TestTimedOut [INFO] BUILD FAILURE ``` The file is new here, so this is not pre-existing, and the same class passes on the ubuntu `pd` and `struct` runners at this head, so it is timing dependent rather than deterministic. The two failures look like one problem. All three tests mutate the `SchemaDriver.INSTANCE` static by reflection (lines 40, 48, 57-58, 92), and `@Test(timeout = 3000L)` only interrupts the test thread; JUnit's `FailOnTimeout` does not stop it. The reported frame is line 55, `new BlockingCloseKvClient()`, so first-touch loading of the `KvClient` and gRPC stub classes plausibly blew the 3 s budget on a slow Intel runner. The abandoned thread would then still publish a driver into `INSTANCE` and start `destroyThread`, which holds the `static synchronized destroy()` monitor. That fits both symptoms, including the sibling test's 2.587 s runtime, but I inferred the interleaving from the logs rather than reproducing it. Requested change: snapshot and restore `INSTANCE` in `@Before`/`@After` instead of inside each test body, and replace the wall clock `@Test(timeout = 3000L)` and `await(1L, TimeUnit.SECONDS)` handshake with latches the test fully controls, so a slow runner cannot leave a thread mutating global state after the timeout fires. ########## hugegraph-struct/src/main/java/org/apache/hugegraph/SchemaDriver.java: ########## @@ -106,16 +121,26 @@ public static void init(PDConfig pdConfig, int cacheSize, long expiration) { "allowed to be initialized again", instance.caches.limit(), instance.caches.expiration(), instance.client); } - INSTANCE.compareAndSet(null, new SchemaDriver(pdConfig, cacheSize, - expiration)); + INSTANCE.set(new SchemaDriver(pdConfig, cacheSize, expiration)); } - public static void destroy() { + public static synchronized void destroy() { Review Comment: ⚠️ `destroy()` can pin the `SchemaDriver.class` monitor and block `init()` indefinitely. `init()` (line 115) and `destroy()` share that class monitor, and `destroy()` now blocks inside `closeResources()`, then `this.client.close()` (line 140), and finally `AbstractClient.closeChannel()`, which spins until the channel terminates: ```java while (channel != null && !channel.shutdownNow().awaitTermination(100, TimeUnit.MILLISECONDS)) { continue; } ``` A channel that does not terminate holds the monitor for good, and every later `init()` blocks on it. The exposure is new. Before this PR `destroy()` was not synchronized and did not close the client; it only cancelled the cache cleaner and cleared `INSTANCE`, all non-blocking. Serialising the double-init race is the right call, but it now puts an unbounded network shutdown under the same lock. This PR's macos CI run is consistent with that: `testDestroyClosesOwnedKvClient` spent 2.587 s before reaching its assertion, which fits waiting on this monitor. Requested change: keep only the `INSTANCE` transition under the monitor and move the blocking client close outside it. Capping the await inside `closeChannel()` would help too, but that is pre-existing code and better as a follow-up. ########## hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java: ########## @@ -180,50 +201,307 @@ public void onNext(WatchResponse value) { @Override public void onError(Throwable t) { - release(); - if (!closed.get()) { - clientId.set(0); - listenWrapper.accept(key, consumer); + if (isRetryableWatchError(t)) { + requestReconnect(subscription, this, shouldRotateWatchTransport(t)); + } else { + stopWatch(subscription, this, t); } } @Override public void onCompleted() { - + requestReconnect(subscription, this, false); } }; } public void listen(String key, Consumer<T> consumer) throws PDException { - long value = clientId.get(); - StreamObserver<WatchResponse> observer = getObserver(key, consumer, listenWrapper, value); - acquire(); + listen(key, consumer, throwable -> { }, false); + } + + public void listen(String key, Consumer<T> consumer, + Consumer<Throwable> errorConsumer) throws PDException { + listen(key, consumer, errorConsumer, false); + } + + public void listenPrefix(String prefix, Consumer<T> consumer) throws PDException { + listen(prefix, consumer, throwable -> { }, true); + } + + public void listenPrefix(String prefix, Consumer<T> consumer, + Consumer<Throwable> errorConsumer) throws PDException { + listen(prefix, consumer, errorConsumer, true); + } + + private void listen(String key, Consumer<T> consumer, + Consumer<Throwable> errorConsumer, + boolean prefix) throws PDException { + Objects.requireNonNull(key, "key"); + Objects.requireNonNull(consumer, "consumer"); + Objects.requireNonNull(errorConsumer, "errorConsumer"); + WatchSubscription subscription = + new WatchSubscription(key, consumer, errorConsumer, prefix); + subscriptions.add(subscription); try { - WatchRequest k = - WatchRequest.newBuilder().setClientId(clientId.get()).setKey(key).build(); - streamingCall(KvServiceGrpc.getWatchMethod(), k, observer, 1); - } catch (Exception e) { - release(); - throw new PDException(Pdpb.ErrorType.PD_UNREACHABLE_VALUE, e); + if (!startWatch(subscription)) { + throw new PDException(Pdpb.ErrorType.PD_UNREACHABLE_VALUE, + "KvClient is closed"); + } + } catch (PDException e) { + cleanupFailedStart(subscription); + subscriptions.remove(subscription); + throw e; + } catch (RuntimeException e) { + cleanupFailedStart(subscription); + subscriptions.remove(subscription); + throw e; } } - public void listenPrefix(String prefix, Consumer<T> consumer) throws PDException { - long value = clientId.get(); - StreamObserver<WatchResponse> observer = - getObserver(prefix, consumer, prefixListenWrapper, value); - acquire(); + private void cleanupFailedStart(WatchSubscription subscription) { + synchronized (subscription) { + subscription.observer.set(null); + cancelStartTimeout(subscription); + } + } + + private boolean startWatch(WatchSubscription subscription) throws PDException { + if (closed.get() || !subscriptions.contains(subscription)) { + return false; + } + + WatchRequest request = WatchRequest.newBuilder() + .setClientId(subscription.clientId.get()) + .setKey(subscription.key) + .build(); + StreamObserver<WatchResponse> observer = getObserver(subscription); try { - WatchRequest k = - WatchRequest.newBuilder().setClientId(clientId.get()).setKey(prefix).build(); - streamingCall(KvServiceGrpc.getWatchPrefixMethod(), k, observer, 1); + synchronized (this) { Review Comment: ⚠️ The client monitor is held across channel teardown, so gRPC callbacks for the other watches block on it. `streamingCall()` runs inside this `synchronized (this)` (lines 292 and 294). It reaches `getStub()`, and when `proxy.getStub()` is null that calls `AbstractClient.resetStub()`, then `closeConnections()`, then `closeChannel()`, whose loop is unbounded (`AbstractClient.java:292`): ```java while (channel != null && !channel.shutdownNow().awaitTermination(100, TimeUnit.MILLISECONDS)) { continue; } ``` The stub is null on exactly the common reconnect path, since `requestReconnect()` calls `invalidateAttemptStub()` (line 379) for `Leader_Changed` and `UNAVAILABLE`. Shutting that channel down makes gRPC deliver `onError` to the other live watches sharing it, and each of those callbacks runs `requestReconnect()` and then `invalidateAttemptStub()`, which is `synchronized` on the same client (line 384). They block on the monitor this thread is holding. The executor is single threaded (`pd-kv-watch-reconnect`, lines 88 and 89), and both `reconnect()` (line 462) and the `onStartTimeout()` watchdog (line 324) are scheduled on it, so one in-flight attempt stalls every other subscription's reconnect and its start timeout. I am not claiming a proven deadlock: whether the stall becomes permanent depends on whether gRPC channel termination waits on a blocked listener, which I could not exercise here. The contention itself follows from the source. Requested change: issue `streamingCall()` outside the client monitor, keeping only the observer and `attemptGeneration` bookkeeping inside it, so gRPC callback threads never contend with a thread that is inside `resetStub()`. -- 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]
