Copilot commented on code in PR #357:
URL:
https://github.com/apache/hugegraph-computer/pull/357#discussion_r3650287187
##########
computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSender.java:
##########
@@ -227,75 +224,92 @@ 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 {
+ this.setControlFuture(future);
+ try {
+ this.client.startSessionAsync().whenComplete((r, e) -> {
Review Comment:
`setControlFuture()` throws `ComputerException` when a previous control
future is still set. In the current flow this exception would propagate out of
`sendStartMessage()` and is not caught by the sender loop (which only catches
`InterruptedException`/`TransportException`), so the send-executor thread can
die unexpectedly. Since `setControlFuture()` already completes the passed
future exceptionally, it’s safer to catch `ComputerException` here and return
to keep the sender thread alive.
This issue also appears on line 269 of the same file.
--
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]
