Cancellation
Soft-cancel an in-flight inference task with cancel_task_sync — overwrite its result with a CANCELLED status.
Overview
Cancellation lets a producer mark an enqueued task as no longer needed. cancel_task_sync(task_id)
is a soft cancel: it overwrites the task's stored result with a CANCELLED status and always
returns True. It does not interrupt a handler that is already running — the work still completes
on the consumer — but the producer-visible result reads CANCELLED. Because the result lives in the
queue-peek backend, cancellation is just another write to that result channel → see
Queues.
Soft cancel is best-effort. If the handler finishes and stores its outcome after the cancel
write lands, the later write wins, so the task can still end SUCCESS. There is no signal sent to
the consumer to stop the in-progress handler.
The cancel_task_sync method
cancel_task_sync writes a CANCELLED result record for the given task_id and returns True.
The cancel_task alias points at the same implementation.
def cancel_task_sync(self, task_id: str) -> bool:
...| Argument | Type | Description |
|---|---|---|
task_id | str | The id returned by enqueue_task_sync for the task to cancel. |
The call resolves the task's result channel ({result_channel_prefix}{task_id}) and stores a record
with status="CANCELLED", using the same purge-then-write path the result backend uses for every
state transition. A subsequent get_task_status_sync(task_id) reads CANCELLED — unless the handler
already stored a terminal result first.
Cancel a running task
Enqueue a long-running task, cancel it before the handler finishes, and confirm the result reads
CANCELLED. This mirrors the packaged examples/cancel_task/soft_cancel.py.
from __future__ import annotations
import os
import time
import uuid
from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter
BROKER = os.environ.get("KUBEMQ_ADDRESS", "localhost:50000")
def slow_handler(data: str) -> dict:
"""Slow handler to give time for cancellation."""
time.sleep(10)
return {"data": data, "completed": True}
def main():
channel = f"example-cancel-{uuid.uuid4().hex[:8]}"
config = KubeMQAdapterConfig(address=BROKER)
adapter = KubeMQTaskProcessorAdapter(config)
class _Cfg:
queue_name = channel
max_retries = 0
failed_task_queue_name = ""
unprocessable_task_queue_name = ""
adapter.initialize(consumer_concurrency=1, task_processor_config=_Cfg())
adapter.register_task_handler(slow_handler, name="slow_handler")
adapter.start_consumer()
try:
# Enqueue a long-running task
result = adapter.enqueue_task_sync("slow_handler", args=["work-item"])
task_id = result.id
# Cancel before it completes — overwrites the result with CANCELLED
cancelled = adapter.cancel_task(task_id)
print(f"cancel_task() returned: {cancelled}") # True
status = adapter.get_task_status_sync(task_id)
print(f"status={status.status}") # CANCELLED (unless the handler already finished)
finally:
adapter.stop_consumer()
if __name__ == "__main__":
main()cancel_task() returns True immediately. The 10-second handler keeps running on the consumer, but
the producer's next poll reads CANCELLED because the cancel write replaced the PENDING record.
Observe the status transition
To watch the full lifecycle, poll the task before and after the cancel. The
examples/cancel_task/cancel_and_check_status.py example drives a PENDING → CANCELLED transition
and confirms it by polling.
# 1. Enqueue a 15-second task
result = adapter.enqueue_task_sync("slow_handler", args=["work-item"])
task_id = result.id
# 2. Confirm the initial status
status = adapter.get_task_status_sync(task_id)
print(f"initial status={status.status}") # PENDING
# 3. Cancel
cancelled = adapter.cancel_task(task_id)
print(f"cancel_task() returned: {cancelled}") # True
# 4. Poll until CANCELLED (or a terminal status if the handler won the race)
for i in range(10):
status = adapter.get_task_status_sync(task_id)
if status.status == "CANCELLED":
print("Task confirmed as CANCELLED.")
break
if status.status in ("SUCCESS", "FAILURE"):
print(f"Task reached {status.status} before cancel took effect.")
break
time.sleep(0.5)The polling loop also guards against the race: if the handler stored SUCCESS or FAILURE before
the cancel write landed, the loop reports that terminal status instead of CANCELLED.
Cancellation only changes what the producer reads — it never reclaims the compute already spent on an in-flight handler. For tasks where stopping the work matters, keep handlers short and rely on the retry and DLQ controls to bound how long a single task can run.
Next steps
Was this page helpful?