Cancelled TimerHandle keeps its callback, args and context alive until every earlier timer has fired (asyncio.timeout, asyncpg statement cache) #118

Summary

On rsloop, TimerHandle.cancel() only sets a flag. The callback, its arguments and its captured context stay in the timer heap. A cancelled entry is removed only when it reaches the top of the heap. So as long as any live timer has an earlier deadline, every cancelled timer behind it keeps its callback alive.

A long-running program almost always has such a timer, for example any background loop with asyncio.sleep(...). In practice a cancelled timer therefore holds its callback until its original deadline. asyncio and uvloop release the callback as soon as cancel() is called.

The common cases are:

• asyncio.timeout() / asyncio.wait_for(): every block that exits early cancels a timer whose callback is Timeout._on_timeout. That pins the Timeout, its finished Task and the task's result.
• asyncpg's statement cache (max_cached_statement_lifetime, 300 s by default): closing a connection cancels one timer per cached statement. That pins the _StatementCache and its PreparedStatementState objects for up to 300 s.

Under load this looks like a leak. In a Granian + SQLAlchemy asyncio + asyncpg service with one worker, RSS grew from 241 to 390 MiB in 300 s (about 117 MiB per 100k requests). uvloop and asyncio stayed flat in the same run.

This is separate from #115 and #116, which 0.1.63 fixes (PR #117 did not touch the timer code). On 0.1.63 the same service no longer leaks protocols, but it still grows through this path.

Minimal reproduction

No third-party code apart from the loops themselves:

"""Cancelled TimerHandles keep their callback (and args) alive while any live timer with an earlier deadline exists."""
import asyncio
import gc
import sys
import tracemalloc
import weakref

LOOP = sys.argv[1] if len(sys.argv) > 1 else "asyncio"
if LOOP == "rsloop":
    import rsloop
    factory = rsloop.new_event_loop
elif LOOP == "uvloop":
    import uvloop
    factory = uvloop.new_event_loop
else:
    factory = asyncio.new_event_loop

N = 20_000


def traced():
    gc.collect()
    return tracemalloc.get_traced_memory()[0]


class Payload:
    def __init__(self):
        self.blob = bytearray(4096)

    def fire(self):
        pass


def alive(refs):
    gc.collect()
    return sum(r() is not None for r in refs)


async def turns(seconds):
    loop = asyncio.get_running_loop()
    end = loop.time() + seconds
    while loop.time() < end:
        await asyncio.sleep(0.01)


async def ticker(period):
    while True:
        await asyncio.sleep(period)


def schedule_and_cancel(loop, delay):
    refs = []
    for _ in range(N):
        p = Payload()
        refs.append(weakref.ref(p))
        loop.call_later(delay, p.fire).cancel()
    return refs


async def case_no_other_timer():
    loop = asyncio.get_running_loop()
    refs = schedule_and_cancel(loop, 4.0)
    await turns(0.2)
    print(f"  A no other timer:           alive after 0.2 s: {alive(refs)}/{N}")


async def case_earlier_live_timer():
    loop = asyncio.get_running_loop()
    sentinel = loop.call_later(1.0, lambda: None)  # one live timer due before the cancelled ones
    tracemalloc.start()
    before = traced()
    refs = schedule_and_cancel(loop, 4.0)
    await turns(0.2)
    n = alive(refs)
    held = traced() - before - sys.getsizeof(refs) - N * sys.getsizeof(refs[0])  # minus the probe's own weakrefs
    tracemalloc.stop()
    print(f"  B live call_later(1) ahead: alive after 0.2 s: {n}/{N} (Python heap still held: {held // N} B per cancelled timer)")
    await turns(1.2)
    print(f"                              alive after 1.4 s (sentinel fired): {alive(refs)}/{N}")
    sentinel.cancel()


async def case_periodic_background():
    loop = asyncio.get_running_loop()
    tasks = [asyncio.create_task(ticker(0.2))]
    await asyncio.sleep(0.1)
    tasks.append(asyncio.create_task(ticker(0.2)))  # two staggered periodic sleeps, like background loops
    await asyncio.sleep(0.05)
    t0 = loop.time()
    refs = schedule_and_cancel(loop, 3.0)
    for mark in (1.0, 2.5, 3.5):
        await turns(mark - (loop.time() - t0))
        print(f"  C two periodic sleep(0.2) loops, cancelled call_later(3): alive at t={mark:.1f} s: {alive(refs)}/{N}")
    for t in tasks:
        t.cancel()


async def request(delay):
    # A typical request: work under asyncio.timeout() that finishes early; the task's result is a 4 KiB body.
    async with asyncio.timeout(delay):
        await asyncio.sleep(0)
    return bytearray(4096)


async def case_asyncio_timeout():
    loop = asyncio.get_running_loop()
    keepalive = loop.call_later(1.0, lambda: None)
    refs = []
    for i in range(0, N, 500):
        tasks = [asyncio.create_task(request(4.0)) for _ in range(500)]
        await asyncio.gather(*tasks)
        refs.extend(weakref.ref(t) for t in tasks)
        del tasks
    await turns(0.2)
    print(f"  D asyncio.timeout(4) exited early, live call_later(1) ahead: finished Tasks alive after 0.2 s: {alive(refs)}/{N}")
    await turns(1.2)
    print(f"                              alive after 1.4 s: {alive(refs)}/{N}")
    keepalive.cancel()


async def main():
    print(f"{LOOP} python {sys.version.split()[0]}")
    await case_no_other_timer()
    await case_earlier_live_timer()
    await case_periodic_background()
    await case_asyncio_timeout()


asyncio.run(main(), loop_factory=factory)

Run as python repro.py asyncio, python repro.py uvloop and python repro.py rsloop.

Expected (asyncio and uvloop give the same output): a cancelled timer drops its callback immediately.

asyncio python 3.14.5
  A no other timer:           alive after 0.2 s: 0/20000
  B live call_later(1) ahead: alive after 0.2 s: 0/20000 (Python heap still held: 0 B per cancelled timer)
                              alive after 1.4 s (sentinel fired): 0/20000
  C two periodic sleep(0.2) loops, cancelled call_later(3): alive at t=1.0 s: 0/20000
  C two periodic sleep(0.2) loops, cancelled call_later(3): alive at t=2.5 s: 0/20000
  C two periodic sleep(0.2) loops, cancelled call_later(3): alive at t=3.5 s: 0/20000
  D asyncio.timeout(4) exited early, live call_later(1) ahead: finished Tasks alive after 0.2 s: 0/20000
                              alive after 1.4 s: 0/20000

uvloop 0.22.1 prints the same, except that case B shows 1 B per cancelled timer.

Actual (rsloop 0.1.63):

rsloop python 3.14.5
  A no other timer:           alive after 0.2 s: 0/20000
  B live call_later(1) ahead: alive after 0.2 s: 20000/20000 (Python heap still held: 4361 B per cancelled timer)
                              alive after 1.4 s (sentinel fired): 0/20000
  C two periodic sleep(0.2) loops, cancelled call_later(3): alive at t=1.0 s: 20000/20000
  C two periodic sleep(0.2) loops, cancelled call_later(3): alive at t=2.5 s: 20000/20000
  C two periodic sleep(0.2) loops, cancelled call_later(3): alive at t=3.5 s: 0/20000
  D asyncio.timeout(4) exited early, live call_later(1) ahead: finished Tasks alive after 0.2 s: 20000/20000
                              alive after 1.4 s: 0/20000

What each case shows:

• A: with no other timer the cancelled entries reach the heap top at once and are freed. A test that only cancels timers does not notice the problem.
• B: one live call_later(1) is enough to keep all 20,000 cancelled callbacks alive, together with their 4 KiB payloads: 4,361 B of Python heap per cancelled timer. They are freed only after that earlier timer fires.
• C: with two staggered periodic asyncio.sleep(0.2) loops, which any service with background tasks has, a live timer is always ahead. The cancelled callbacks then live until their own original deadline (3 s).
• D: the same happens through asyncio.timeout(). The exited Timeout pins its finished Task, so all 20,000 tasks and their results stay alive.

The output is identical:

• in Docker (python:3.14.5-slim, epoll fallback);
• on rsloop 0.1.61 and 0.1.62.

Real-library check: asyncpg statement cache

The setup: 500 × (asyncpg.connect(), three fetchval() calls each under asyncio.timeout(30), close()), against PostgreSQL 17, with two background asyncio.sleep(0.5) loops running. max_cached_statement_lifetime is set to 30 s instead of the default 300 s to keep the run short.

"""Closed asyncpg connections leave their statement caches alive until the caches' expiry timers' original deadlines."""
import asyncio
import gc
import os
import sys

import asyncpg

LOOP = sys.argv[1] if len(sys.argv) > 1 else "asyncio"
if LOOP == "rsloop":
    import rsloop
    factory = rsloop.new_event_loop
elif LOOP == "uvloop":
    import uvloop
    factory = uvloop.new_event_loop
else:
    factory = asyncio.new_event_loop

DSN = os.environ["PG_DSN"]
CONNS = 500
LIFETIME = 30.0  # asyncpg default max_cached_statement_lifetime is 300 s


def alive():
    gc.collect()
    counts = {"_StatementCache": 0, "PreparedStatementState": 0, "Protocol": 0}
    for o in gc.get_objects():
        name = type(o).__name__
        if name in counts:
            counts[name] += 1
    return " ".join(f"{k}={v}" for k, v in counts.items())


async def heartbeat():
    while True:
        await asyncio.sleep(0.5)


async def one_connection():
    conn = await asyncpg.connect(DSN, max_cached_statement_lifetime=LIFETIME)
    for i in range(3):
        async with asyncio.timeout(30):
            await conn.fetchval(f"SELECT $1::int + {i}", i)
    await conn.close()


async def main():
    loop = asyncio.get_running_loop()
    beats = [asyncio.create_task(heartbeat())]
    await asyncio.sleep(0.25)
    beats.append(asyncio.create_task(heartbeat()))  # two staggered background loops, like any long-running service
    t0 = loop.time()
    for _ in range(CONNS // 10):
        await asyncio.gather(*(one_connection() for _ in range(10)))
    took = loop.time() - t0
    print(f"{LOOP} {CONNS} x connect/3 cached queries/close in {took:.1f} s, all connections closed; objects alive after gc.collect():")
    for mark in (took + 0.5, LIFETIME / 2, LIFETIME + took + 1.5):
        await asyncio.sleep(max(0.0, mark - (loop.time() - t0)))
        print(f"  t={loop.time() - t0:5.1f} s  {alive()}", flush=True)
    for b in beats:
        b.cancel()


asyncio.run(main(), loop_factory=factory)

Output (Docker python:3.14.5-slim, asyncpg 0.31.0, postgres:17-alpine):

asyncio 500 x connect/3 cached queries/close in 7.0 s, all connections closed; objects alive after gc.collect():
  t=  7.5 s  _StatementCache=0 PreparedStatementState=0 Protocol=0
  t= 15.0 s  _StatementCache=0 PreparedStatementState=0 Protocol=0
  t= 38.5 s  _StatementCache=0 PreparedStatementState=0 Protocol=0
uvloop 500 x connect/3 cached queries/close in 7.0 s, all connections closed; objects alive after gc.collect():
  t=  7.5 s  _StatementCache=0 PreparedStatementState=0 Protocol=0
  t= 15.0 s  _StatementCache=0 PreparedStatementState=0 Protocol=0
  t= 38.5 s  _StatementCache=0 PreparedStatementState=0 Protocol=0
rsloop 500 x connect/3 cached queries/close in 6.8 s, all connections closed; objects alive after gc.collect():
  t=  7.3 s  _StatementCache=500 PreparedStatementState=1500 Protocol=0
  t= 15.0 s  _StatementCache=500 PreparedStatementState=1500 Protocol=0
  t= 38.3 s  _StatementCache=0 PreparedStatementState=0 Protocol=0

On rsloop, every closed connection's statement cache and all 1,500 prepared statements stay alive until the 30 s expiry timers' original deadlines. Protocols are released, so the #116 fix holds. With asyncpg's default of 300 s, a service that opens and closes connections keeps every closed connection's cache for 5 minutes.

Service-level impact

The service is a Granian 2.7.5 + SQLAlchemy 2.0 asyncio + asyncpg 0.31.0 API with one worker, running on rsloop 0.1.63.

• Memory growth: RSS grew from 241 to 390 MiB in a 300 s load run, about 117 MiB per 100k requests. uvloop and asyncio stayed flat under the same load. The run was not extended past 300 s, so we did not check whether growth stops near the 300 s statement lifetime.
• Object dumps: we took three dumps from the live worker during load, each after gc.collect():
  • 9,262 asyncio.timeouts.Timeout objects, 9,252 of them already EXITED, and 3,457 live _StatementCache._on_entry_expired bound methods;
  • 9,136 Timeouts, of which 9,087 were EXITED and owned by request tasks that had already finished;
  • 9,343 exited Timeouts of finished tasks, none past its deadline. Deadlines ranged from 0 s to about 1,775 s ahead, median about 5 s.

Environment

• rsloop: 0.1.63, the latest release (cp314-cp314-manylinux_2_39_x86_64 wheel); master has no commits after v0.1.63. 0.1.61 and 0.1.62 behave the same.
• Python and controls: CPython 3.14.5, Linux x86_64. Controls: uvloop 0.22.1 and the stdlib loop.
• Reactors:
  • io_uring on the host (kernel 7.2, glibc 2.44);
  • the epoll fallback inside Docker (python:3.14.5-slim, Debian 13, glibc 2.41, default seccomp).

Likely cause (from reading the v0.1.63 source; not traced with a debugger)

• PyTimerHandle::cancel (src/engine/callbacks.rs:552) sets its flag and calls ReadyCallback::cancel (src/engine/callbacks.rs:259). That function only stores cancelled = true; its doc comment says cancellation "does not remove an already queued value". The callback, args and context fields stay in the Arc<ReadyCallback> held by the TimerEntry in the heap.
• LocalTimers::collect (src/engine/loop_core.rs:225) pops entries only while the heap top is due or cancelled (entry.when <= now || entry.callback.cancelled(), line 239). A cancelled entry behind a live, earlier entry is never looked at. Nothing counts cancelled entries or compacts the heap.

For comparison, asyncio does two things:

• Handle.cancel() clears _callback and _args right away (asyncio/events.py);
• BaseEventLoop counts cancelled timers and rebuilds the heap once more than half of it is cancelled (_MIN_CANCELLED_TIMER_HANDLES_FRACTION = 0.5 in asyncio/base_events.py).

uvloop's TimerHandle.cancel() also clears the callback and arguments and closes the libuv timer.

Possible fixes, either of which would remove the memory pinning:

• Release on cancel: on TimerHandle.cancel(), take the callback, args and context out of the ReadyCallback, for example by keeping them in an Option behind a cell that the GIL-holding loop thread can clear. The heap entry then holds only an empty shell.
• Compact the heap: count cancelled timers and rebuild the heap when they exceed a fraction of it, as asyncio does. This also bounds the heap itself when cancelled timers keep arriving behind a long-lived earlier one.