bitflicker64 commented on code in PR #357:
URL:
https://github.com/apache/hugegraph-computer/pull/357#discussion_r3896555147
##########
computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/worker/WorkerService.java:
##########
@@ -176,7 +186,6 @@ public synchronized void close() {
this.computeManager.close();
} else {
LOG.warn("The computeManager is null");
- return;
}
Review Comment:
⚠️ Dropping the early `return` is the right fix, but the cleanup it unblocks
stops at the first manager that fails. `Managers.closeAll()`
(`computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/manager/Managers.java:69`)
calls `manager.close(config)` in a bare loop with no per-manager isolation,
and on exactly this partial-init path `DataServerManager.close(config)` calls
`address()` -> `TransportConnectionManager.getServer()`, which throws when the
transport server was never bound.
Evidence — harness compiled against this head (registers `DataServerManager`
holding a never-started `TransportConnectionManager`, plus one more manager
behind it, then calls `closeAll`):
```
closeAll threw: java.lang.IllegalArgumentException: The TransportServer has
not been initialized yet
later manager closed = false
```
In `initManagers()` the managers registered after `DataServerManager` are
`DataClientManager`, `SendSortManager`, `MessageSendManager`, `SnapshotManager`
and `WorkerInputManager`. So when `initAll()` fails before the data server
binds (`transport.server_port` already in use is the obvious trigger), none of
them are closed — and `DataClientManager.close()` is what calls
`sender.close()`, i.e. the `QueuedMessageSender` send-executor this PR hardens
is one of the threads left running. The surrounding `catch (Exception e)`
swallows the `IllegalArgumentException`, so nothing in the log says cleanup
stopped halfway.
Requested change: make `Managers.closeAll()` close every manager, wrapping
each `manager.close(config)` in its own try/catch and rethrowing the first
failure with the rest attached as suppressed (or, narrower, make
`DataServerManager.close()` a no-op when the server never bound). Please add a
regression that fails `initAll()` before `DataServerManager` binds and asserts
the managers registered after it were still closed.
##########
computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSender.java:
##########
@@ -63,15 +66,30 @@ public void init() {
}
public void close() {
+ this.closed = true;
this.sendExecutor.interrupt();
try {
this.sendExecutor.join();
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw new ComputerException("Interrupted when waiting for " +
"send-executor to stop", e);
}
}
+ @Override
+ public void checkFatal() {
+ Throwable error = this.fatalError.get();
+ if (error != null) {
+ throw new ComputerException("Send-executor encountered fatal
error",
+ error);
+ }
+ }
+
+ private void recordFatal(Throwable error) {
+ this.fatalError.compareAndSet(null, error);
Review Comment:
⚠️ `recordFatal()` stores the error but leaves every control future the
channels are already holding unresolved, so the new fail-fast path only takes
effect on the *next* caller. When the executor dies here, an in-flight or
queued START/FINISH is never completed, and
`MessageSendManager.sendControlMessageToWorkers()` is already parked in
`future.get(timeout, MILLISECONDS)` — the `this.sender.checkFatal()` calls
added to `startSend()`/`finishSend()` run only after that wait returns, so they
can neither shorten it nor surface the real cause.
Evidence — probe compiled against this head: one channel whose client always
returns `false` from `send()` (so the executor parks in
`waitAnyClientNotBusy()`), one queued data message plus a FINISH, then a
spurious interrupt with `closed == false`:
```
executor state before interrupt = WAITING
ComputerException: Interrupted when waiting any client not busy
at QueuedMessageSender.waitAnyClientNotBusy(QueuedMessageSender.java:256)
at QueuedMessageSender$Sender.run(QueuedMessageSender.java:206)
executor alive after fatal = false
FINISH future done = false
checkFatal() = Send-executor encountered fatal error
FINISH still incomplete after 3004 ms
```
With defaults the caller therefore blocks `transport.sync_request_timeout` =
10s for START and `10_000 * transport.max_pending_requests(8)` = 80s for FINISH
(`TransportConf.timeoutFinishSession()`), then reports `Timeout(80000ms) to
wait for controlling message(FINISH) to finished` rather than the fatal error
that was recorded 80 seconds earlier. `close()` has the same gap: it sets
`closed = true` and joins without resolving pending control futures.
Requested change: when the first fatal error is recorded (and on the
`close()` path), walk `channels` and hand the error to each one —
`channel.failDataSend(error)` already fails the registered control future and
poisons the channel against later ones — so `sendControlMessageToWorkers()`
returns immediately with the real cause. Please add a regression that
terminates the send-executor with a FINISH already queued and asserts the
future completes exceptionally with the recorded fatal error.
##########
computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/worker/WorkerService.java:
##########
@@ -101,7 +108,9 @@ public synchronized void init(Config config) {
this.workerInfo = new ContainerInfo();
LOG.info("{} Start to initialize worker", this);
- this.bsp4Worker = new Bsp4Worker(this.config, this.workerInfo);
+ if (this.bsp4Worker == null) {
Review Comment:
🧹 This null check and the package-private `WorkerService(Bsp4Worker)`
constructor at line 91 exist only so
`WorkerServiceTest#testInitFailsAfterRegistration` can inject a mock;
`initManagers()` was widened from `private` to package-private at line 320 for
the same test. Two consequences worth avoiding: production `init()` now carries
a test-only branch, and because `bsp4Worker` is never reset to `null` anywhere
in this class, an instance whose `init()` threw — which leaves `inited ==
false`, so the `E.checkArgument(!this.inited, ...)` guard at line 102 still
permits a retry — would re-`init()` against the already-closed `Bsp4Worker`
from the failed attempt instead of building a fresh one.
Requested change: keep `init()` constructing the `Bsp4Worker`
unconditionally through a package-private factory method (e.g. `Bsp4Worker
newBsp4Worker() { return new Bsp4Worker(this.config, this.workerInfo); }`) and
have the test override it — the test already subclasses `WorkerService` to
override `initManagers()`, so no extra seam is needed and the production path
keeps no test-only branch.
--
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]
