imbajin commented on code in PR #357:
URL:
https://github.com/apache/hugegraph-computer/pull/357#discussion_r3652466487
##########
computer/computer-test/src/main/java/org/apache/hugegraph/computer/suite/integrate/SenderIntegrateTest.java:
##########
@@ -331,6 +442,103 @@ private WorkerService initWorker(String[] args) {
return service;
}
+ private static void waitForServices(List<CompletableFuture<Void>> futures)
{
+ CompletableFuture<Void> result = new CompletableFuture<>();
+ for (CompletableFuture<Void> future : futures) {
+ future.whenComplete((r, e) -> {
+ if (e != null) {
+ result.completeExceptionally(e);
+ }
+ });
+ }
+ CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
+ .whenComplete((r, e) -> {
+ if (e == null) {
+ result.complete(null);
+ } else {
+ result.completeExceptionally(e);
+ }
+ });
+ try {
+ result.get(SERVICE_WAIT_TIMEOUT, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ throw new ComputerException("Timed out to wait for master and " +
+ "worker services", e);
+ } catch (ExecutionException e) {
+ throw new ComputerException("Failed to wait for master and " +
+ "worker services", e.getCause());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new ComputerException("Interrupted when waiting for master "
+
+ "and worker services", e);
+ }
+ }
+
+ private static void closeServicesAndJoin(ServiceLifecycle lifecycle,
+ List<Thread> threads) {
+ lifecycle.closeAll();
Review Comment:
⚠️ `closeAll()` runs every closer sequentially before any service thread is
interrupted, but registration order is concurrent. If the master closer runs
before a still-active worker closer, `MasterService.close()` waits for
`workerCloseDone` that the queued worker closer has not had a chance to send,
so this fail-fast path can block for the full five-minute BSP timeout; one
thrown closer also skips the remaining cleanup. Please encode
worker-before-master shutdown (or close services concurrently), ensure one
close failure cannot prevent the rest, and cover a master-first registration
with one failed and one active worker.
##########
computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSender.java:
##########
@@ -227,75 +224,102 @@ 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;
-
+ public void sendStartMessage(CompletableFuture<Void> future)
+ throws TransportException {
+ try {
+ this.setControlFuture(future);
+ } catch (ComputerException e) {
+ // The control future has been completed exceptionally
+ return;
+ }
+ try {
+ this.client.startSessionAsync().whenComplete((r, e) -> {
Review Comment:
‼️ A synchronous unchecked exception from this call leaves
`controlFutureRef` pointing at `future`. `ClientSession.startAsync()` can throw
`IllegalArgumentException` from its state checks, and its send function can
propagate unchecked failures; neither reaches this `TransportException` catch,
so the returned future stays incomplete and `Sender.run()` also exits. Please
clear and complete the control future for synchronous runtime failures here and
in `finishSessionAsync()`, then add a client-stub regression that throws
synchronously and verifies both the returned future and the executor's intended
state.
##########
computer/computer-test/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSenderTest.java:
##########
@@ -64,4 +74,195 @@ public void testInitAndClose() {
Assert.assertTrue(ImmutableSet.of(Thread.State.TERMINATED)
.contains(sendExecutor.getState()));
}
+
+ @Test
+ public void
testControlFutureCanQueueNextControlBeforeCompletionDependentFinishes()
+ throws Exception {
+ QueuedMessageSender sender = new QueuedMessageSender(this.config);
+ ControlFutureClient client = new ControlFutureClient();
+ sender.addWorkerClient(1, client);
+ sender.addWorkerClient(2, new MockTransportClient());
+ sender.init();
+
+ CountDownLatch completionStarted = new CountDownLatch(1);
+ CountDownLatch allowCompletion = new CountDownLatch(1);
+ Thread completionThread = null;
+ try {
+ CompletableFuture<Void> startFuture = sender.send(1,
+
MessageType.START);
+ Assert.assertTrue(client.awaitStart());
+ startFuture.whenComplete((r, e) -> {
+ completionStarted.countDown();
+ try {
+ allowCompletion.await();
+ } catch (InterruptedException exception) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(exception);
+ }
+ });
+
+ completionThread = new Thread(client::completeStart);
+ completionThread.start();
+ Assert.assertTrue(completionStarted.await(1, TimeUnit.SECONDS));
+
+ CompletableFuture<Void> finishFuture = sender.send(1,
+
MessageType.FINISH);
+ allowCompletion.countDown();
Review Comment:
⚠️ This releases the blocked completion dependent before confirming that
FINISH was dispatched. An implementation that still clears the control
reference only after completion dependents return can therefore process FINISH
after this countdown and pass the test. Please await `client.awaitFinish()`
before releasing `allowCompletion` (while keeping the `finally` release) so the
test actually proves the next control message proceeds while the prior
dependent is still blocked.
##########
computer/computer-test/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSenderTest.java:
##########
@@ -64,4 +74,195 @@ public void testInitAndClose() {
Assert.assertTrue(ImmutableSet.of(Thread.State.TERMINATED)
.contains(sendExecutor.getState()));
}
+
+ @Test
+ public void
testControlFutureCanQueueNextControlBeforeCompletionDependentFinishes()
+ throws Exception {
+ QueuedMessageSender sender = new QueuedMessageSender(this.config);
+ ControlFutureClient client = new ControlFutureClient();
+ sender.addWorkerClient(1, client);
+ sender.addWorkerClient(2, new MockTransportClient());
+ sender.init();
+
+ CountDownLatch completionStarted = new CountDownLatch(1);
+ CountDownLatch allowCompletion = new CountDownLatch(1);
+ Thread completionThread = null;
+ try {
+ CompletableFuture<Void> startFuture = sender.send(1,
+
MessageType.START);
+ Assert.assertTrue(client.awaitStart());
+ startFuture.whenComplete((r, e) -> {
+ completionStarted.countDown();
+ try {
+ allowCompletion.await();
+ } catch (InterruptedException exception) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(exception);
+ }
+ });
+
+ completionThread = new Thread(client::completeStart);
+ completionThread.start();
+ Assert.assertTrue(completionStarted.await(1, TimeUnit.SECONDS));
+
+ CompletableFuture<Void> finishFuture = sender.send(1,
+
MessageType.FINISH);
+ allowCompletion.countDown();
+ completionThread.join(TimeUnit.SECONDS.toMillis(1));
+ Assert.assertFalse(completionThread.isAlive());
+ Assert.assertTrue(client.awaitFinish());
+ client.completeFinish();
+ finishFuture.get(1, TimeUnit.SECONDS);
+ } finally {
+ allowCompletion.countDown();
+ if (completionThread != null) {
+ completionThread.join(TimeUnit.SECONDS.toMillis(1));
+ }
+ sender.close();
+ }
+ }
+
+ @Test
+ public void testTransportExceptionCompletesInFlightControlFuture()
+ throws Exception {
+ QueuedMessageSender sender = new QueuedMessageSender(this.config);
+ ControlFutureClient client = new ControlFutureClient();
+ sender.addWorkerClient(1, client);
+ sender.addWorkerClient(2, new MockTransportClient());
+ sender.init();
+
+ try {
+ CompletableFuture<Void> startFuture = sender.send(1,
+
MessageType.START);
+ Assert.assertTrue(client.awaitStart());
+
+ TransportException cause =
+ new TransportException("connection failed");
+ sender.transportExceptionCaught(cause, client.connectionId());
+ assertFutureFailedWith(startFuture, cause);
+
+ client.completeStart();
Review Comment:
⚠️ This completes the stale START callback immediately after the transport
exception, without first installing a next-generation FINISH future. It
therefore does not protect the identity-sensitive CAS behavior in
`completeControlFuture()`; an unconditional clear could still pass while later
stranding FINISH. Please send and await FINISH first, then complete the stale
START callback, and finally complete/assert FINISH so the test locks in
cross-generation isolation.
--
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]
