imbajin commented on code in PR #357:
URL:
https://github.com/apache/hugegraph-computer/pull/357#discussion_r3661395670
##########
computer/computer-test/src/main/java/org/apache/hugegraph/computer/suite/integrate/SenderIntegrateTest.java:
##########
@@ -68,10 +76,212 @@ public static void clear() {
// pass
}
+ @Test
+ public void testWaitForServicesFailsFast() {
+ CompletableFuture<Void> failedWorker = new CompletableFuture<>();
+ CompletableFuture<Void> waitingMaster = new CompletableFuture<>();
+ IllegalStateException cause = new IllegalStateException("worker
failed");
+ failedWorker.completeExceptionally(cause);
+
+ try {
+ waitForServices(Arrays.asList(failedWorker, waitingMaster));
+ Assert.fail("Expected worker failure to stop service wait");
+ } catch (ComputerException e) {
+ Assert.assertSame(cause, e.getCause());
+ }
+ }
+
+ @Test
+ public void testCleanupFailureDoesNotHideWaitFailure() {
+ ServiceLifecycle lifecycle = new ServiceLifecycle();
+ CompletableFuture<Void> failedWorker = new CompletableFuture<>();
+ IllegalStateException waitFailure =
+ new IllegalStateException("worker failed");
+ IllegalStateException cleanupFailure =
+ new IllegalStateException("worker close failed");
+ failedWorker.completeExceptionally(waitFailure);
+ lifecycle.registerWorker(() -> {
+ throw cleanupFailure;
+ });
+
+ try {
+ waitForServicesAndClose(lifecycle, Arrays.asList(failedWorker),
+ new ArrayList<>(), null);
+ Assert.fail("Expected worker failure to be preserved");
+ } catch (ComputerException e) {
+ Assert.assertSame(waitFailure, e.getCause());
+ Assert.assertEquals(1, e.getSuppressed().length);
+ Assert.assertSame(cleanupFailure, e.getSuppressed()[0].getCause());
+ }
+ }
+
+ @Test
+ public void testCiTimeoutsAllowHeavyInputStep() {
+ Assert.assertEquals(TimeUnit.MINUTES.toMillis(5L), BSP_WAIT_TIMEOUT);
+ Assert.assertEquals(BSP_WAIT_TIMEOUT + TimeUnit.SECONDS.toMillis(10L),
+ SERVICE_WAIT_TIMEOUT);
+ }
+
+ @Test
+ public void testServiceLifecycleClosesLateRegisteredService() {
+ ServiceLifecycle lifecycle = new ServiceLifecycle();
+ AtomicBoolean closed = new AtomicBoolean();
+
+ lifecycle.closeAll();
+
+ Assert.assertFalse(lifecycle.registerWorker(() -> closed.set(true)));
+ Assert.assertTrue(closed.get());
+ }
+
+ @Test
+ public void testServiceLifecycleClosesWorkersBeforeMasterAfterFailure() {
+ ServiceLifecycle lifecycle = new ServiceLifecycle();
+ AtomicBoolean activeWorkerClosed = new AtomicBoolean();
+ AtomicBoolean masterClosed = new AtomicBoolean();
+ RuntimeException workerFailure =
+ new IllegalStateException("worker close failed");
+
+ lifecycle.registerMaster(() -> {
+ Assert.assertTrue(activeWorkerClosed.get());
+ masterClosed.set(true);
+ });
+ lifecycle.registerWorker(() -> {
+ throw workerFailure;
+ });
+ lifecycle.registerWorker(() -> activeWorkerClosed.set(true));
+
+ Throwable failure = lifecycle.closeAll();
+
+ Assert.assertSame(workerFailure, failure);
+ Assert.assertTrue(activeWorkerClosed.get());
+ Assert.assertTrue(masterClosed.get());
+ }
+
+ @Test
+ public void testInitializeServiceClosesPartiallyInitializedService() {
+ AtomicBoolean closed = new AtomicBoolean();
+ RuntimeException cause = new IllegalStateException("init failed");
+
+ try {
+ initializeService(new Object(), service -> {
+ throw cause;
+ }, service -> closed.set(true));
+ Assert.fail("Expected initialization to fail");
+ } catch (RuntimeException e) {
+ Assert.assertSame(cause, e);
+ }
+
+ Assert.assertTrue(closed.get());
+ }
+
+ @Test
+ public void testCloseServicesAndJoinStopsSpawnedThreads() throws Exception
{
+ ServiceLifecycle lifecycle = new ServiceLifecycle();
+ AtomicBoolean closed = new AtomicBoolean();
+ CountDownLatch started = new CountDownLatch(1);
+ Thread thread = new Thread(() -> {
+ started.countDown();
+ try {
+ Thread.sleep(Long.MAX_VALUE);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ lifecycle.registerWorker(() -> closed.set(true));
+ thread.start();
+ Assert.assertTrue(started.await(1, TimeUnit.SECONDS));
Review Comment:
⚠️ These cleanup regressions can leak the very non-daemon threads they
create when a startup assertion fails. This thread is started before the
one-second await, but cleanup is not in a `finally`; the worker/master test
below has the same gap before lines 271-272. Under scheduler delay, the failed
assertion leaves a thread sleeping indefinitely and can hang or contaminate the
remaining suite. Please put unconditional interrupt plus bounded join cleanup
in `finally` blocks around both tests.
##########
computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSender.java:
##########
@@ -227,75 +232,101 @@ private static class WorkerChannel {
private final MessageQueue queue;
// Each target worker has a TransportClient
private final TransportClient client;
- private final AtomicReference<CompletableFuture<Void>> futureRef;
+ private final AtomicReference<CompletableFuture<Void>>
controlFutureRef;
public WorkerChannel(int workerId, MessageQueue queue,
TransportClient client) {
this.workerId = workerId;
this.queue = queue;
this.client = client;
- this.futureRef = new AtomicReference<>();
- }
-
- public CompletableFuture<Void> newFuture() {
- CompletableFuture<Void> future = new CompletableFuture<>();
- if (!this.futureRef.compareAndSet(null, future)) {
- throw new ComputerException("The origin future must be null");
- }
- return future;
- }
-
- public void resetFuture(CompletableFuture<Void> future) {
- if (!this.futureRef.compareAndSet(future, null)) {
- throw new ComputerException("Failed to reset futureRef, " +
- "expect future object is %s, " +
- "but some thread modified it",
- future);
- }
+ this.controlFutureRef = new AtomicReference<>();
}
public boolean doSend(QueuedMessage message)
throws TransportException, InterruptedException {
switch (message.type()) {
case START:
- this.sendStartMessage();
+ this.sendStartMessage(message.controlFuture());
return true;
case FINISH:
- this.sendFinishMessage();
+ this.sendFinishMessage(message.controlFuture());
return true;
default:
return this.sendDataMessage(message);
}
}
- public void sendStartMessage() throws TransportException {
- this.client.startSessionAsync().whenComplete((r, e) -> {
- CompletableFuture<Void> future = this.futureRef.get();
- assert future != null;
-
- if (e != null) {
- LOG.info("Failed to start session connected to {}", this);
- future.completeExceptionally(e);
- } else {
- LOG.info("Start session connected to {}", this);
- future.complete(null);
- }
- });
+ public void sendStartMessage(CompletableFuture<Void> future) {
+ if (!this.controlFutureInFlight(future)) {
+ return;
+ }
+ try {
+ this.client.startSessionAsync().whenComplete((r, e) -> {
+ if (e != null) {
+ LOG.info("Failed to start session connected to {}",
this);
+ } else {
+ LOG.info("Start session connected to {}", this);
+ }
+ this.completeControlFuture(future, e);
+ });
+ } catch (TransportException e) {
+ this.completeControlFuture(future, e);
+ } catch (RuntimeException e) {
+ this.completeControlFuture(future, e);
+ }
}
- public void sendFinishMessage() throws TransportException {
- this.client.finishSessionAsync().whenComplete((r, e) -> {
- CompletableFuture<Void> future = this.futureRef.get();
- assert future != null;
-
- if (e != null) {
- LOG.info("Failed to finish session connected to {}", this);
- future.completeExceptionally(e);
- } else {
- LOG.info("Finish session connected to {}", this);
- future.complete(null);
- }
- });
+ public void sendFinishMessage(CompletableFuture<Void> future) {
+ if (!this.controlFutureInFlight(future)) {
+ return;
+ }
+ try {
+ this.client.finishSessionAsync().whenComplete((r, e) -> {
+ if (e != null) {
+ LOG.info("Failed to finish session connected to {}",
this);
+ } else {
+ LOG.info("Finish session connected to {}", this);
+ }
+ this.completeControlFuture(future, e);
+ });
+ } catch (TransportException e) {
+ this.completeControlFuture(future, e);
+ } catch (RuntimeException e) {
+ this.completeControlFuture(future, e);
+ }
+ }
+
+ public void transportExceptionCaught(TransportException cause) {
+ CompletableFuture<Void> future =
this.controlFutureRef.getAndSet(null);
+ if (future != null) {
+ future.completeExceptionally(cause);
+ }
+ }
+
+ private boolean setControlFuture(CompletableFuture<Void> future) {
+ if (this.controlFutureRef.compareAndSet(null, future)) {
+ return true;
+ }
+ ComputerException e = new ComputerException(
+ "The origin future must be null");
+ future.completeExceptionally(e);
+ return false;
+ }
+
+ private boolean controlFutureInFlight(CompletableFuture<Void> future) {
+ return this.controlFutureRef.get() == future;
+ }
+
+ private void completeControlFuture(CompletableFuture<Void> future,
+ Throwable cause) {
+ if (!this.controlFutureRef.compareAndSet(future, null)) {
+ return;
+ }
+ if (cause == null) {
+ future.complete(null);
+ } else {
+ future.completeExceptionally(cause);
+ }
}
public boolean sendDataMessage(QueuedMessage message)
Review Comment:
‼️ A synchronous failure on this data path still terminates the sole send
executor without resolving control work. `TransportClient.send()` declares
`TransportException` and the underlying session/send function can also throw
unchecked state failures; either escapes the loop, while a FINISH future
already reserved behind that data message remains queued and incomplete. Please
transition the sender to an explicit failed state that exceptionally completes
every queued/in-flight control future (or otherwise keep the executor alive
while propagating the failure), and add regressions for both synchronous
`TransportException` and `RuntimeException` with a queued FINISH.
--
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]
