bitflicker64 commented on code in PR #357:
URL:
https://github.com/apache/hugegraph-computer/pull/357#discussion_r4046533255
##########
computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSender.java:
##########
@@ -127,57 +173,75 @@ private class Sender implements Runnable {
public void run() {
LOG.info("The send-executor is running");
Thread thread = Thread.currentThread();
- while (!thread.isInterrupted()) {
- try {
- int emptyQueueCount = 0;
- int busyClientCount = 0;
- for (WorkerChannel channel : channels) {
- QueuedMessage message = channel.queue.peek();
- if (message == null) {
- ++emptyQueueCount;
- continue;
+ try {
+ while (!thread.isInterrupted()) {
+ try {
+ int emptyQueueCount = 0;
+ int busyClientCount = 0;
+ for (WorkerChannel channel : channels) {
+ QueuedMessage message = channel.queue.peek();
+ if (message == null) {
+ ++emptyQueueCount;
+ continue;
+ }
+ try {
+ if (channel.doSend(message)) {
+ // Only consume the message after it is
sent
+ channel.queue.take();
+ } else {
+ ++busyClientCount;
+ }
+ } catch (TransportException | RuntimeException e) {
+ channel.failDataSend(e);
+ // Discard the failed data message to keep
sending
+ channel.queue.take();
+ LOG.warn("Failed to send {} message to {}, " +
+ "discard it", message.type(),
channel, e);
+ }
}
- if (channel.doSend(message)) {
- // Only consume the message after it is sent
- channel.queue.take();
- } else {
- ++busyClientCount;
+ int channelCount = channels.length;
+ /*
+ * If all queues are empty, let send thread wait
+ * until any queue is available
+ */
+ if (emptyQueueCount >= channelCount) {
+ LOG.debug("The send executor was blocked " +
+ "to wait any queue not empty");
+ QueuedMessageSender.this.waitAnyQueueNotEmpty();
Review Comment:
🧹 The busy-wait path now treats an unexpected interrupt as fatal, but the
idle path here doesn't. `waitAnyQueueNotEmpty()` swallows the interrupt and
re-sets the flag, so the `while (!thread.isInterrupted())` check just ends the
loop: the executor returns normally, nothing goes through `recordFatal()`, and
`checkFatal()` keeps passing. The next START/FINISH is queued onto a dead
thread and the caller waits out the full control timeout (80s for FINISH with
defaults).
Probe against this head (one worker, idle `MockTransportClient`, interrupt
the executor while it is WAITING with `closed == false`, then send FINISH):
```
state before interrupt = WAITING
executor alive after interrupt = false
checkFatal() = no error recorded
FINISH still incomplete after 3005 ms
```
Could `waitAnyQueueNotEmpty()` handle the interrupt the same way
`waitAnyClientNotBusy()` does (return when `closed`, otherwise `recordFatal()`
and throw)? A test next to `testFatalExecutorCompletesQueuedFinish` that
interrupts an idle executor and asserts the next FINISH fails fast would cover
it.
##########
computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/network/DataServerManager.java:
##########
@@ -54,9 +54,8 @@ public void init(Config config) {
@Override
public void close(Config config) {
- InetSocketAddress address = this.address();
this.connectionManager.shutdownServer();
- LOG.info("DataServerManager closed with address '{}'", address);
+ LOG.info("DataServerManager closed");
Review Comment:
🧹 This makes `DataServerManager` safe to close after a partial init, but
`Managers.closeAll()` still stops at the first manager that throws, and the
last worker manager, `WorkerInputManager`, throws on the same path. If
`initAll()` fails before `WorkerInputManager.init()` runs (port already in use
in `DataServerManager.init()`, or a MinIO error in `SnapshotManager.init()`),
`LoadService.fetchers` is still all nulls and `LoadService.close()` NPEs on
`fetcher.close()`. `WorkerService.close()` then logs `Error while closing
managers` with that NPE, and `WorkerInputManager`'s `sendExecutor.shutdown()`
is skipped.
Probe against this head (real `WorkerInputManager` with mocked send/snapshot
managers, `close()` without `init()`):
```
close threw: java.lang.NullPointerException
at
org.apache.hugegraph.computer.core.worker.load.LoadService.close(LoadService.java:82)
at
org.apache.hugegraph.computer.core.input.WorkerInputManager.close(WorkerInputManager.java:97)
```
`testCloseManagersAfterPartialInitFailure` can't see this because `last` is
a mock. Could `LoadService.close()` skip null fetchers (or
`Managers.closeAll()` close each manager in its own try/catch and rethrow the
first failure with the rest suppressed)? Registering a real
`WorkerInputManager` after `DataServerManager` in that test would cover it.
--
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]
