imbajin commented on code in PR #3308:
URL: https://github.com/apache/hugegraph/pull/3308#discussion_r4248129135
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/DistributedTaskScheduler.java:
##########
@@ -469,17 +550,200 @@ private void saveTaskWithoutResult(HugeTask<?> task) {
});
}
+ private boolean cancelLocal(PendingTask pending, Consumer<Boolean>
afterCancel) {
+ return this.cancelLocal(pending, null, afterCancel);
+ }
+
+ private boolean cancelLocal(PendingTask pending, Runnable beforeCancel,
+ Consumer<Boolean> afterCancel) {
+ return this.cancelLocal(pending, beforeCancel, afterCancel, false);
+ }
+
+ private boolean cancelLocal(PendingTask pending, Runnable beforeCancel,
+ Consumer<Boolean> afterCancel, boolean
terminalOnly) {
+ if (SCHEDULER_WORKER.get() == this.schedulerExecutor) {
+ return this.cancelOwned(pending, beforeCancel, afterCancel,
terminalOnly);
+ }
+ // Keep callbacks and final persistence off the request/close thread
and
+ // the DB worker, because callbacks may synchronously save to that
worker.
+ return this.call(() -> {
+ SCHEDULER_WORKER.set(this.schedulerExecutor);
+ try {
+ return this.cancelOwned(pending, beforeCancel, afterCancel,
terminalOnly);
+ } finally {
+ SCHEDULER_WORKER.remove();
+ // An interrupted request may stop waiting before this owner
finishes.
+ this.finishCancelledQueued(pending);
+ }
+ }, this.schedulerExecutor, true);
+ }
+
+ private boolean cancelOwned(PendingTask pending, Runnable beforeCancel,
+ Consumer<Boolean> afterCancel, boolean
terminalOnly) {
+ LockResult acquired = null;
+ boolean acquire;
+ synchronized (pending) {
+ if (this.runningTasks.get(pending.task.id()) != pending) {
+ return false;
+ }
+ acquire = !pending.task.ephemeralTask() && pending.taskLock ==
null;
+ pending.cancellationDepth++;
+ }
+ boolean entered = false;
+ boolean coordinated = false;
+ try {
+ if (acquire) {
+ // Don't block publication/release of this runner's token
during the lock RPC.
+ acquired = this.tryLockTask(pending.task.id().asString());
+ if (!acquired.lockSuccess()) {
+ return false;
+ }
+ synchronized (pending) {
+ pending.taskLock = acquired;
+ }
+ }
+ entered = true;
+ if (!pending.task.ephemeralTask()) {
+ HugeTask<?> stored;
+ try {
+ stored = this.taskWithoutResult(pending.task.id());
+ } catch (NotFoundException e) {
+ pending.discardQueued = true;
+ coordinated = true;
+ return false;
+ }
+ if (beforeCancel == null) {
+ if
(TaskStatus.COMPLETED_STATUSES.contains(stored.status())) {
+ pending.discardQueued = true;
+ coordinated = true;
+ return false;
+ }
+ if (terminalOnly) {
+ if (stored.status() == TaskStatus.DELETING) {
+ pending.discardQueued = true;
+ coordinated = true;
+ } else if (pending.task.isCancelled() &&
+ pending.task.status() ==
TaskStatus.CANCELLED &&
+ stored.status() == TaskStatus.CANCELLING) {
Review Comment:
Fixed the running-owner terminal-save recovery on the original source
branch. A close retry now retains and retries coordination for the owner that
already held the task lock, without replaying its completion callback. The
remote SUCCESS/DELETING protections remain in place.
A real running-task fault injection reproduces the stuck close before the
fix and verifies successful recovery after persistence is restored. The focused
regressions, combined task/graph suite, clean compilation and three independent
final reviews passed. New-head CI is pending; full PD/HStore/Raft topology
validation remains separate.
--
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]
