Repository navigation
feat: interrupt the running action when a workflow is cancelled (opt-in) #31
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| package com.airbnb.skipper.internal | ||
|
|
||
| import java.util.concurrent.ConcurrentHashMap | ||
| import javax.inject.Singleton | ||
|
|
||
| /** | ||
| * The threads running actions, per workflow id, so cancelling a workflow can interrupt them in this | ||
| * process. Best effort: an action that never blocks, or swallows the interrupt, runs to completion. | ||
| * A workflow can run several actions at once (a signal handler runs beside the workflow method), so | ||
| * each thread gets its own [Registration]. [exit] and [interrupt] update a workflow's entry one at a | ||
| * time, so an interrupt never lands on a thread that has already left its action. | ||
| */ | ||
| @Singleton | ||
| class InFlightActions { | ||
|
Member
There was a problem hiding this comment. curious, did you consider doing the cancellation at the workflow level rather than the action level? is this to allow graceful handling of the action cancellation at the workflow layer? otherwise at the workflow level we can reuse the existing workflows in flight registry we currently use for lease renewal
Contributor
Author
There was a problem hiding this comment. I started from the lease-renewal registry and it can’t do this job: the future it holds is the A workflow-level thread registration would reach the action thread, since sync actions run on it, but the interrupt could then land between actions — in the checkpoint write, the status re-read, or the persist — turning store operations into failures. Bracketing only the action invocation closes the delivery window before any store call and lets the executor attribute the failure to the cancel rather than to the action. So it’s about keeping the interrupt out of engine and store code, not graceful handling at the workflow layer. |
||
| class Registration internal constructor(internal val workflowId: String) { | ||
| internal val thread: Thread = Thread.currentThread() | ||
| @Volatile internal var interruptRequested = false | ||
| } | ||
|
|
||
| private val entries = ConcurrentHashMap<String, MutableList<Registration>>() | ||
|
|
||
| /** Records the calling thread as running an action for [workflowId]. */ | ||
| fun enter(workflowId: String): Registration { | ||
| val registration = Registration(workflowId) | ||
| entries.compute(workflowId) { _, list -> (list ?: mutableListOf()).also { it.add(registration) } } | ||
| return registration | ||
| } | ||
|
|
||
| /** | ||
| * Forgets the registration and returns whether a cancel interrupted it. If so, it clears the | ||
| * thread's interrupt flag so the interrupt does not leak into what the thread runs next. | ||
| */ | ||
| fun exit(registration: Registration): Boolean { | ||
| entries.computeIfPresent(registration.workflowId) { _, list -> | ||
| list.remove(registration) | ||
| if (list.isEmpty()) null else list | ||
| } | ||
| if (registration.interruptRequested) Thread.interrupted() // reads and clears the flag | ||
| return registration.interruptRequested | ||
| } | ||
|
|
||
| /** The workflows with an action running here right now. */ | ||
| fun workflowIds(): Set<String> = entries.keys.toSet() | ||
|
|
||
| /** | ||
| * Interrupts each thread running an action for [workflowId] here, skipping any already interrupted, | ||
| * so an action is interrupted at most once. Returns whether this call interrupted any. | ||
| */ | ||
| fun interrupt(workflowId: String): Boolean { | ||
| var delivered = false | ||
| entries.computeIfPresent(workflowId) { _, list -> | ||
| for (registration in list) { | ||
| if (!registration.interruptRequested) { | ||
| registration.interruptRequested = true | ||
| registration.thread.interrupt() | ||
| delivered = true | ||
| } | ||
| } | ||
| list | ||
| } | ||
| return delivered | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| package com.airbnb.skipper.internal | ||
|
|
||
| import com.airbnb.skipper.FeatureGate | ||
| import com.airbnb.skipper.Metrics | ||
| import com.airbnb.skipper.WorkflowInstance | ||
| import com.airbnb.skipper.internal.storage.WorkflowStore | ||
| import java.time.Duration | ||
| import org.slf4j.LoggerFactory | ||
|
|
||
| /** | ||
| * Interrupts the actions running here whose workflow another process cancelled. Every [pollInterval] | ||
| * the scheduler manager calls [pollOnce], which reads the stored workflow of each in-flight action and | ||
| * interrupts on CANCELLED. It reads nothing unless both in-flight cancellation flags are on. | ||
| */ | ||
| class InFlightCancelWatcher( | ||
| private val inFlightActions: InFlightActions, | ||
| private val workflowStore: WorkflowStore, | ||
| private val featureGate: FeatureGate, | ||
| private val metrics: Metrics, | ||
| val pollInterval: Duration, | ||
| ) { | ||
| init { | ||
| require(pollInterval.toMillis() > 0) { "inflightCancellationPollInterval must be at least 1 ms" } | ||
| } | ||
|
|
||
| fun pollOnce() { | ||
| if (!featureGate.isEnabled(FeatureGate.Keys.INFLIGHT_CANCELLATION_INTERRUPT) || | ||
| !featureGate.isEnabled(FeatureGate.Keys.INFLIGHT_CANCELLATION_CHECKPOINTS) | ||
| ) { | ||
| return | ||
| } | ||
| for (workflowId in inFlightActions.workflowIds()) { | ||
| try { | ||
| val stored = workflowStore.getWorkflow(workflowId) | ||
| if (stored.isDefined && stored.get().status == WorkflowInstance.Status.CANCELLED && | ||
| inFlightActions.interrupt(workflowId) | ||
| ) { | ||
| metrics.counter(METRICS_COMPONENT, "watcherInterrupts").inc() | ||
| log.info("workflowId={} was cancelled elsewhere; interrupted its running action", workflowId) | ||
| } | ||
| } catch ( | ||
| @Suppress("TooGenericExceptionCaught") e: Exception | ||
| ) { | ||
| // One unreadable workflow must not stop the others from being checked. | ||
| log.warn("could not check whether workflowId={} was cancelled", workflowId, e) | ||
| metrics.counter(METRICS_COMPONENT, "watcherErrors").inc() | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private companion object { | ||
| private val log = LoggerFactory.getLogger(InFlightCancelWatcher::class.java) | ||
| private const val METRICS_COMPONENT = "inflightCancellation" | ||
| } | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -59,6 +59,7 @@ open class SkipperEngine | |
| private val eventPublisher: EventPublisher, | ||
| private val metrics: Metrics, | ||
| private val callbackHandlerInjector: SkipperInjector, | ||
| private val inFlightActions: InFlightActions, | ||
| ) { | ||
| /** | ||
| * Creates and starts the execution of a new workflow instance. | ||
|
|
@@ -654,6 +655,17 @@ open class SkipperEngine | |
| if (task.isDefined) { | ||
| persistentScheduler.remove(task.get()) | ||
| } | ||
| // Needs the checkpoints gate too: it ends the interrupted execution as CANCELLED without a | ||
| // status write, and stops any action that starts after this interrupt. | ||
| if (featureGate.isEnabled(FeatureGate.Keys.INFLIGHT_CANCELLATION_INTERRUPT) && | ||
| featureGate.isEnabled(FeatureGate.Keys.INFLIGHT_CANCELLATION_CHECKPOINTS) | ||
| ) { | ||
| val interrupted = inFlightActions.interrupt(workflowId) | ||
|
Member
There was a problem hiding this comment. I think this will not work as expected. this will only work in the case where the process that serves this cancellation request is the same process that happens to host the workflow executions. In practice where we'd have multiple process, this will rarely happen, so we'd need either a broadcast mechanism or a way to route the cancellation request to the process that is actively executing the workflow
Contributor
Author
There was a problem hiding this comment. Agreed — as built this only fires when the cancel lands on the pod running the workflow, so in a multi-pod service it rarely will. The cheapest cross-process delivery is to have the executing process poll the persisted CANCELLED status for workflows with an action currently in flight and interrupt locally when it sees it: no broadcast or pod routing, the store row stays the only cross-pod truth, and latency is the poll interval. The cost is read amplification — one status read per in-flight action per interval, only when the flag is on. Does that trade sound right to you, or would you rather route or broadcast?
Contributor
Author
There was a problem hiding this comment. Implemented in 1d688a1. The scheduler manager runs one more loop: every |
||
| metrics | ||
| .counter(mapOf("interrupted" to interrupted.toString()), METRICS_COMPONENT, "cancelInterrupts") | ||
| .inc() | ||
| log.info("cancelled workflowId={}: running action interrupted={}", workflowId, interrupted) | ||
| } | ||
| try { | ||
| if (instance.callbackHandler != null) { | ||
| val handler = callbackHandlerInjector.getInstance(instance.callbackHandler) | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I wonder if we need to evaluate wether the action is async and exit the inFlightAction only after the async execution completed rather than synchronously here, otherwise we'll exit immeidately and the action might still be executing async?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Agreed it’s a gap. Keeping the thread registered until the future completes would interrupt the wrong thread, though — by then it’s back in the workflow method. The right handle for an async action is the future: register it and
cancel(true)it. Follow-up PR.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
sounds good. should we leave a TODO here for the record?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Added the TODO at the future-returning branch in 1d688a1.