imbajin commented on code in PR #357:
URL:
https://github.com/apache/hugegraph-computer/pull/357#discussion_r3653428272
##########
computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSender.java:
##########
@@ -227,75 +224,106 @@ 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) -> {
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);
}
- });
+ this.completeControlFuture(future, e);
+ });
+ } catch (TransportException e) {
+ this.completeControlFuture(future, e);
+ throw 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;
-
+ public void sendFinishMessage(CompletableFuture<Void> future)
+ throws TransportException {
+ try {
+ this.setControlFuture(future);
+ } catch (ComputerException e) {
+ // The control future has been completed exceptionally
+ return;
+ }
+ try {
+ this.client.finishSessionAsync().whenComplete((r, e) -> {
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);
}
- });
+ this.completeControlFuture(future, e);
+ });
+ } catch (TransportException e) {
+ this.completeControlFuture(future, e);
+ throw e;
+ } catch (RuntimeException e) {
+ this.completeControlFuture(future, e);
+ }
+ }
+
+ public void transportExceptionCaught(TransportException cause) {
+ CompletableFuture<Void> future =
this.controlFutureRef.getAndSet(null);
Review Comment:
‼️ `send()` now only places the control future in the queue (lines 88-93),
while `controlFutureRef` stays null until the sender thread dispatches that
entry at lines 254/279. A connection callback in that window therefore makes
this `getAndSet(null)` discard the original failure; FINISH may later succeed,
or the caller may wait until timeout without the real cause, even though
preceding data was lost. Please reserve the future before enqueue while
retaining clear-before-complete ordering, or explicitly fail queued as well as
in-flight controls, and add a regression that injects the exception after
`send()` returns but before the session method is called.
##########
computer/computer-test/src/main/java/org/apache/hugegraph/computer/suite/integrate/SenderIntegrateTest.java:
##########
@@ -81,18 +167,27 @@ public void testOneWorker() {
.withResultClass(DoubleValue.class)
.withMessageClass(DoubleValue.class)
.withMaxSuperStep(3)
- .withComputationClass(COMPUTATION)
- .withWorkerCount(1)
- .withBufferThreshold(50)
+ .withComputationClass(COMPUTATION)
+ .withWorkerCount(1)
+ .withTestBspTimeouts()
+ .withBufferThreshold(50)
.withBufferCapacity(60)
.withRpcServerHost("127.0.0.1")
.withRpcServerPort(8611)
.withRpcServerPort(0)
.build();
- try (MasterService service = initMaster(args)) {
- masterServiceRef.set(service);
- service.execute();
- masterFuture.complete(null);
+ try {
+ MasterService service = initMaster(args);
Review Comment:
⚠️ `initMaster()`/`initWorker()` allocates and initializes the service
before it is registered with `ServiceLifecycle`. If `init()` throws after
creating BSP or network resources, neither lifecycle registration nor the inner
`finally` runs, so the new fail-fast cleanup still cannot close that partially
initialized service and CI can hang or leak resources. Please register a safely
closeable service before initialization, or close it inside the helper on
initialization failure, and add an init-failure cleanup test.
##########
computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSender.java:
##########
@@ -227,75 +224,106 @@ 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) -> {
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);
}
- });
+ this.completeControlFuture(future, e);
+ });
+ } catch (TransportException e) {
+ this.completeControlFuture(future, e);
+ throw e;
Review Comment:
‼️ After completing only this control future, rethrowing the declared
`TransportException` reaches `Sender.run()`, which converts it to
`ComputerException` and permanently terminates the sole send executor. A
control already queued for another worker can then remain unresolved until its
timeout; the new synchronous-failure test covers only `RuntimeException`, not
this checked path. Please keep the executor alive after completing the affected
future, or transition it to an explicit failed state that completes every
queued/in-flight control future, and add a two-client regression for a
synchronous `TransportException` here and in the FINISH path.
--
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]
