imbajin commented on code in PR #3308:
URL: https://github.com/apache/hugegraph/pull/3308#discussion_r4247598006
##########
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:
[P1] Retry cancellation persistence for an already-running owner
This only repairs the queued-reservation case, whose record has already been
changed to CANCELLING. For a persistent task holding pending.taskLock, cancel()
skips requestCancellation and cancelOwned() skips the acquired != null
transition, so a transient failure of done/cancelled/final save can leave the
record RUNNING while the in-memory task is CANCELLED and coordinationPending
stays true. Once the worker exits, close() retries through terminalOnly, but
this condition never persists the terminal state or clears the ticket. Because
close() already cancelled cron, later close retries remain blocked even after
the store recovers.
Please cover a running LockHoldingJob failure/recovery case and allow the
owned RUNNING case to finish safely without replaying callbacks or overwriting
a remote terminal/DELETING state. The new checkCancellationPersistence test
gates execution before the owner acquires its lock, so it covers the queued
case only.
Evidence boundary: independently traced the source and reproduced this with
extracted cancel/cancelOwned/terminal-cleanup methods and real JDK FutureTask
in a bounded cloud model. After three recovered retries, the running case
retained stored RUNNING/ticket=1; the queued control reached
CANCELLED/ticket=0, with callbacks remaining exactly once. Storage/PD
locks/dispatch were stubbed; this is not a live PD/HStore validation.
--
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]
