Skip to content

Cancelling Long-Running Work in a Process Pool

A report service renders PDFs in a ProcessPoolExecutor. A user closes the browser tab, the request is cancelled, the awaiting coroutine unwinds — and a worker process keeps rendering for another ninety seconds, holding a core the next request needs. Cancelling the asyncio side of an executor future does exactly what it promises: it stops waiting. It cannot reach into another process and interrupt a tight loop of Python bytecode, because there is no cooperative point to interrupt at and no signal that would leave the interpreter in a sane state. The result is a service whose CPU time is controlled by nobody: cancelled work keeps running, shutdown waits for it, and the pool's queue fills with jobs whose requesters are long gone. This guide measures that behaviour, then gives the three mechanisms that actually stop CPU work — a cooperative stop flag, a dedicated process per job that can be terminated, and a bounded shutdown — and shows when each is appropriate.

Prerequisites

What cancel() reaches and what it does not 3 lanes over time. What cancel() reaches and what it does not caller await future cancel() returns worker process still burning CPU pool shutdown waits for the worker time → Measured: the await returned in 0.00 s, shutdown then waited 1.81 s.

1. Measure what cancelling an executor future really does

Cancel a running job and time two things: how quickly the awaiting side returns, and how long the pool takes to shut down afterwards.

import asyncio
import time
from concurrent.futures import ProcessPoolExecutor


def burn(seconds: float) -> str:
    deadline = time.perf_counter() + seconds
    while time.perf_counter() < deadline:                 # pure CPU: no cooperative points
        pass
    return "done"


async def main() -> None:
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor(max_workers=1) as pool:
        future = loop.run_in_executor(pool, burn, 2.0)
        await asyncio.sleep(0.2)                          # the job is running in the worker
        started = time.perf_counter()
        future.cancel()
        try:
            await future
        except asyncio.CancelledError:
            print(f"await returned after {time.perf_counter() - started:.2f}s")
        shutdown_started = time.perf_counter()
    print(f"pool shutdown waited {time.perf_counter() - shutdown_started:.2f}s for the worker")


if __name__ == "__main__":
    asyncio.run(main())

The await returned immediately — in 0.00 seconds — and the pool's shutdown then blocked for 1.81 seconds, the remainder of the job. The work was never cancelled; only the waiting was. A future that is still queued is a different story: cancel() succeeds and the job never runs, which is why shutdown(cancel_futures=True) is worth calling even though it cannot touch running jobs.

Verify: the await returns in milliseconds while shutdown waits for the remaining job duration.

2. Add a cooperative stop flag

If the work has a loop — over pages, rows, tiles, frames — it can check a shared flag. A multiprocessing.Manager().Event() is visible in every worker and costs a cheap interprocess check per iteration.

import asyncio
import multiprocessing as mp
import time
from concurrent.futures import ProcessPoolExecutor


def render_pages(pages: int, stop) -> str:
    for page in range(pages):
        if stop.is_set():                                 # the cooperative point
            return f"cancelled after {page} pages"
        time.sleep(0.01)                                  # stands in for rendering one page
    return f"rendered {pages} pages"


async def main() -> None:
    loop = asyncio.get_running_loop()
    with mp.Manager() as manager:
        stop = manager.Event()
        with ProcessPoolExecutor(max_workers=1) as pool:
            started = time.perf_counter()
            future = loop.run_in_executor(pool, render_pages, 500, stop)
            await asyncio.sleep(0.2)
            stop.set()                                    # ask the worker to stop
            print(await future, f"in {time.perf_counter() - started:.2f}s")


if __name__ == "__main__":
    asyncio.run(main())

The job stopped about 0.2 seconds in and returned a partial result, instead of running for five seconds. Check the flag at a granularity that bounds your cancellation latency: once per page is fine, once per pixel is wasteful. The mechanism is the same cooperative cancellation asyncio uses, moved across the process boundary — and its limitation is the same: code that never reaches a check point, such as a single long call into a C library, cannot be stopped this way.

Verify: the result reports a partial render, and the elapsed time matches when the flag was set rather than the full job.

How long the CPU stays busy after the caller gives up 3 bars comparing future.cancel() only with the others. How long the CPU stays busy after the caller gives up future.cancel() only 1.81 s cooperative stop flag 0.21 s terminate the job process immediate Measured with the step 1 to 3 examples; the flag was checked once per rendered page. Cancellation latency is a design parameter, set by how often work checks in.

3. Give interruptible work its own process

When the work has no cooperative points — a single call into NumPy, a regex over a huge string, a third-party encoder — the only way to stop it is to end the process. Run such jobs in a dedicated child process you can terminate, instead of in a shared pool worker whose death would also affect queued jobs.

import asyncio
import multiprocessing as mp
import time


def _worker_main(conn) -> None:
    while (msg := conn.recv()) is not None:
        seconds = msg
        deadline = time.perf_counter() + seconds
        while time.perf_counter() < deadline:             # uninterruptible CPU work
            pass
        conn.send("done")


class TerminableJob:
    """One child process per job, so cancellation is a terminate()."""

    def __init__(self) -> None:
        ctx = mp.get_context("spawn")
        self.parent, child = ctx.Pipe()
        self.process = ctx.Process(target=_worker_main, args=(child,), daemon=True)
        self.process.start()

    async def run(self, seconds: float, timeout: float) -> str:
        loop = asyncio.get_running_loop()
        self.parent.send(seconds)
        try:
            async with asyncio.timeout(timeout):
                return await loop.run_in_executor(None, self.parent.recv)
        except TimeoutError:
            await self.stop()
            return "terminated"

    async def stop(self, grace: float = 2.0) -> None:
        if self.process.is_alive():
            self.process.terminate()                      # SIGTERM to the child
            await asyncio.get_running_loop().run_in_executor(None, self.process.join, grace)
            if self.process.is_alive():
                self.process.kill()                       # SIGKILL if it ignored SIGTERM
                await asyncio.get_running_loop().run_in_executor(None, self.process.join, grace)


async def main() -> None:
    job = TerminableJob()
    started = time.perf_counter()
    print(await job.run(seconds=5.0, timeout=0.2), f"in {time.perf_counter() - started:.2f}s")
    print("exit code:", job.process.exitcode)             # -15: SIGTERM


if __name__ == "__main__":
    asyncio.run(main())

Termination is immediate — the process exits with -15 — and no other job is affected, because the process was serving only this one. The costs are process start-up per job (tens of milliseconds with spawn) and losing any in-process caches the worker had warmed, so reserve this for jobs whose duration dwarfs start-up. Note that recv() blocks, so it is awaited through the default thread executor; the same bridging rules as in awaiting concurrent.futures Futures in asyncio apply.

Verify: the call returns terminated about 0.2 seconds in, and the child's exit code is -15 rather than 0.

4. Stop accepting work the caller no longer wants

Cancellation is cheapest before a job starts. Track which submissions are still wanted, skip queued jobs whose caller has gone, and bound the queue so a burst of abandoned work cannot fill it.

import asyncio
import time
from concurrent.futures import ProcessPoolExecutor


def quick(n: int) -> int:
    return n * n


class CancellableOffload:
    def __init__(self, workers: int = 2, max_queued: int = 8) -> None:
        self.pool = ProcessPoolExecutor(max_workers=workers)
        self.slots = asyncio.Semaphore(max_queued)
        self.skipped = 0

    async def run(self, fn, *args):
        async with self.slots:                             # bound work waiting for a worker
            job = self.pool.submit(fn, *args)              # the concurrent.futures side
            try:
                return await asyncio.wrap_future(job)
            except asyncio.CancelledError:
                if job.cancel():                           # True only if it never started
                    self.skipped += 1
                raise

    def shutdown(self) -> None:
        self.pool.shutdown(wait=False, cancel_futures=True)   # drop everything still queued


async def main() -> None:
    offload = CancellableOffload(workers=1, max_queued=100)
    tasks = [asyncio.create_task(offload.run(quick, i)) for i in range(50)]
    await asyncio.sleep(0)
    for task in tasks[10:]:                                # most callers give up immediately
        task.cancel()
    results = await asyncio.gather(*tasks, return_exceptions=True)
    print("completed:", sum(1 for r in results if isinstance(r, int)),
          "| cancelled before running:", offload.skipped)
    offload.shutdown()


if __name__ == "__main__":
    asyncio.run(main())

Jobs still waiting in the executor's queue were genuinely cancelled and never ran; only the handful already dispatched to the worker completed. cancel_futures=True at shutdown does the same for everything left. This is the cheapest of the three mechanisms and should be in place regardless of which of the others you add.

Verify: the number of completed jobs is small and skipped accounts for most of the cancelled ones.

5. Bound shutdown so abandoned work cannot block a deploy

ProcessPoolExecutor.__exit__ waits for running jobs. With minute-long jobs, that turns a rolling deploy into a stall and eventually a SIGKILL from the orchestrator. Make shutdown explicit and bounded: ask cooperatively, then terminate.

import asyncio
import multiprocessing as mp
import time
from concurrent.futures import ProcessPoolExecutor


class BoundedPool:
    def __init__(self, workers: int, manager: mp.managers.SyncManager) -> None:
        self.stop_flag = manager.Event()
        self.pool = ProcessPoolExecutor(max_workers=workers)

    async def shutdown(self, grace: float = 5.0) -> str:
        self.stop_flag.set()                                # cooperative: jobs may exit early
        loop = asyncio.get_running_loop()
        self.pool.shutdown(wait=False, cancel_futures=True)  # drop queued work immediately
        try:
            async with asyncio.timeout(grace):
                await loop.run_in_executor(None, self.pool.shutdown, True)
            return "drained"
        except TimeoutError:
            for process in list(self.pool._processes.values()):   # internal: last resort
                process.terminate()
            return "terminated"


def slow_job(seconds: float, stop) -> str:
    deadline = time.perf_counter() + seconds
    while time.perf_counter() < deadline:
        if stop.is_set():
            return "stopped early"
        time.sleep(0.01)
    return "finished"


async def main() -> None:
    with mp.Manager() as manager:
        pool = BoundedPool(workers=2, manager=manager)
        loop = asyncio.get_running_loop()
        job = loop.run_in_executor(pool.pool, slow_job, 30.0, pool.stop_flag)
        await asyncio.sleep(0.2)
        started = time.perf_counter()
        print(await pool.shutdown(grace=5.0), f"in {time.perf_counter() - started:.2f}s")
        print("job result:", await job)


if __name__ == "__main__":
    asyncio.run(main())

The cooperative flag let the 30-second job exit within milliseconds, so shutdown reported drained well inside the grace period. For jobs without cooperative points, the timeout branch terminates the workers; reaching into pool._processes is an internal detail, so prefer the per-job processes from step 3 when termination must be part of the design. Wire this into the service's graceful shutdown sequence with a grace period shorter than the orchestrator's kill timeout.

Verify: shutdown returns drained quickly with the flag in place, and terminated within the grace period when the job ignores it.

Choosing a cancellation mechanism A grid of 4 rows by 3 columns. Choosing a cancellation mechanism mechanism stops running work? cost fits cancel queued futures no none always cooperative stop flag yes, at check points a check per iteration loops you control per-job process + terminate yes, immediately process start per job opaque C calls bounded shutdown terminates at the end partial results lost deploys The first row is free and belongs in every design; the others are chosen per workload.

Verification

Process-pool cancellation is handled when:

  • Expectations are explicit: the team knows that cancelling a future does not stop running CPU work.
  • Long jobs are interruptible: they check a stop flag at a known granularity, or run in a process that can be terminated.
  • Queued work is dropped promptly: cancelled callers' jobs never start, and shutdown passes cancel_futures=True.
  • Shutdown is bounded: draining has a grace period shorter than the orchestrator's kill timeout, with termination as the fallback.
  • Cancellation latency is measured: the time between a caller giving up and the CPU being released is a known number.

Pitfalls & edge cases

  • Missing if __name__ == "__main__":. With spawn or forkserver, child processes re-import the main module; without the guard, they start more pools recursively and fail.
  • Passing an unmanaged multiprocessing.Event. A plain mp.Event() cannot be pickled into a pool job; use one from a Manager, or pass it through the pool initialiser as a global.
  • Terminating shared pool workers. Killing a worker in a shared pool loses every job it was running and can break the pool; recreate the executor or use per-job processes.
  • Ignoring partial results. Cooperative cancellation usually leaves partial output; decide whether to discard it, cache it for resumption, or return it to the caller.
  • Forgetting that threads are worse. A ThreadPoolExecutor job cannot be cancelled either, and it cannot be terminated; CPU work that must be interruptible belongs in a process.

Frequently Asked Questions

Does cancelling a ProcessPoolExecutor future stop the running job?

No. If the job is still queued, cancel prevents it from starting; if it is already running in a worker, cancellation only abandons the wait. The worker keeps executing, and the executor's shutdown waits for it. In our measurement the await returned immediately while shutdown blocked for the rest of the job.

How do I stop CPU-bound work in another process?

Either make it cooperative — pass a multiprocessing Manager Event and check it inside the work loop, returning early when it is set — or run the job in its own process and terminate that process. Cooperative checks are cheap but need a loop; termination works for any code but loses the process.

Why does my asyncio service hang at shutdown with a process pool?

ProcessPoolExecutor's shutdown waits for running jobs by default, so a long CPU job blocks the exit. Call shutdown with wait=False and cancel_futures=True to drop queued work, wait for the remainder with a bounded timeout, and terminate workers if that timeout expires.

Can I use a signal to interrupt a process pool worker?

It is unsafe in general: Python handles signals between bytecodes, so a signal cannot interrupt a long C call, and killing a worker of a shared pool destroys jobs belonging to other callers. Prefer cooperative flags, or a dedicated process per job that you can terminate cleanly.