Skip to content

with lock: can skip __exit__ when a signal handler raises right after __enter__ #148874

Description

@colesbury

Bug report

A with X: statement can return without calling X.__exit__ when a Python signal handler raises between __enter__ returning and the body starting.

This leaks whatever resource __enter__ acquired (e.g. a threading.Lock stays locked forever).

Cause

In 3.14, the following changed how with X: compiles:

Before 3.14:

LOAD X
BEFORE_WITH        # single op: calls enter, sets up exit on stack
POP_TOP            # <- exception table starts here
... body ...

BEFORE_WITH has no periodic/eval-breaker check, so there was no window for a signal handler to fire between __enter__ returning and the with-statement's exception handler being established.

From 3.14 onward:

LOAD X
COPY 1
LOAD_SPECIAL exit
SWAP 2 / SWAP 3
LOAD_SPECIAL enter
CALL 0             # ends with _CHECK_PERIODIC_AT_END
POP_TOP            # <- exception table starts here
... body ...

CALL includes _CHECK_PERIODIC_AT_END, which runs the Python signal handler. If that handler raises, JUMP_TO_LABEL(error) fires with frame->instr_ptr still pointing at CALL (set before the micro-ops run).

Reproducer (via Claude Code)

test_with_signal_leak.py
import ctypes, ctypes.util, signal, sys, threading

# signal.pthread_kill / os.kill both call PyErr_CheckSignals after the syscall,
# which perturbs timing enough to suppress the race. Use libc.pthread_kill
# directly.
_pthread_kill = ctypes.CDLL(ctypes.util.find_library("c")).pthread_kill
_pthread_kill.argtypes = [ctypes.c_ulong, ctypes.c_int]
_pthread_kill.restype = ctypes.c_int

_MAIN_TID = threading.get_ident()

def _handler(signum, frame):
    raise RuntimeError("signal")

def _send():
    _pthread_kill(_MAIN_TID, signal.SIGUSR1)

def run_trial(lock, iterations=200):
    t = threading.Thread(target=_send)
    t.start()
    try:
        for _ in range(iterations):
            with lock:
                pass
    except BaseException:
        pass
    try:
        t.join()
    except BaseException:
        pass
    if lock.locked():
        lock.release()
        return True
    return False

signal.signal(signal.SIGUSR1, _handler)
lock = threading.Lock()
leaked = 0
for _ in range(2000):
    try:
        if run_trial(lock):
            leaked += 1
    except BaseException:
        if lock.locked():
            lock.release()
print(f"leaked={leaked}/2000")

cc @markshannon

Linked PRs

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    3.14bugs and security fixes3.15pre-release feature fixes, bugs and security fixesinterpreter-core(Objects, Python, Grammar, and Parser dirs)type-bugAn unexpected behavior, bug, or error

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions