Copilot commented on code in PR #357:
URL:
https://github.com/apache/hugegraph-computer/pull/357#discussion_r3684108823
##########
computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSender.java:
##########
@@ -227,75 +237,130 @@ 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;
+ private final AtomicReference<Throwable> dataFailureRef;
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<>();
+ this.dataFailureRef = 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(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.get();
+ if (future == null) {
+ this.failDataSend(cause);
+ } else {
+ this.completeControlFuture(future, cause);
+ }
+ }
+
+ public void failControlFuture(Throwable cause) {
+ CompletableFuture<Void> future =
this.controlFutureRef.getAndSet(null);
+ if (future != null) {
+ future.completeExceptionally(cause);
+ }
+ }
+
+ public void failDataSend(Throwable cause) {
+ this.dataFailureRef.compareAndSet(null, cause);
+ this.failControlFuture(this.dataFailureRef.get());
+ }
+
+ private boolean setControlFuture(CompletableFuture<Void> future) {
+ Throwable failure = this.dataFailureRef.get();
+ if (failure != null) {
+ future.completeExceptionally(failure);
+ return false;
+ }
+ if (this.controlFutureRef.compareAndSet(null, future)) {
+ failure = this.dataFailureRef.get();
+ if (failure == null) {
+ return true;
}
- });
+ this.completeControlFuture(future, failure);
+ return false;
+ }
+ ComputerException e = new ComputerException(
+ "The origin future must be null");
+ future.completeExceptionally(e);
+ return false;
}
- 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);
+ 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)) {
+ if (cause != null) {
+ this.failDataSend(cause);
}
- });
+ return;
+ }
+ if (cause == null) {
+ future.complete(null);
+ } else {
+ future.completeExceptionally(cause);
+ }
}
Review Comment:
`completeControlFuture()` unconditionally dereferences `future` after the
CAS. If `future` is ever null (e.g., a START/FINISH message created without
attaching a control future), `compareAndSet(null, null)` can succeed and then
`future.complete(...)` will throw a NullPointerException. If you keep
`controlFuture` nullable in `QueuedMessage`, add a guard here (or earlier) to
ensure null cannot reach this method.
##########
computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSender.java:
##########
@@ -227,75 +237,130 @@ 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;
+ private final AtomicReference<Throwable> dataFailureRef;
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<>();
+ this.dataFailureRef = 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);
}
}
Review Comment:
START/FINISH now rely on `QueuedMessage.controlFuture()`, but
`QueuedMessage` still has a public constructor that sets `controlFuture` to
null. If any code path queues a START/FINISH message without the new
(package-private) constructor, `sendStartMessage(null)` /
`sendFinishMessage(null)` can lead to inconsistent behavior and can ultimately
NPE when completing the future. Consider enforcing `controlFuture != null` for
START/FINISH (e.g., validate in `doSend()` and fail fast with a clear
exception), or make control messages be created only via dedicated factory
methods that always attach a non-null future.
--
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]
