Concurrent partial.__setstate__() can free args during flattening

[Nievesjyl]

Description

functools.partial.__setstate__() independently replaces fn, args, kw, dict, phcount, and vectorcall without a critical section. partial_new(), partial_call(), and partial_vectorcall() read these fields and borrow tuple items without taking owned references to one coherent state. A setter can therefore release the old args tuple while another thread concatenates or indexes it, and readers can observe mixed state from different replacements.

Observed Behavior

One thread repeatedly installed fresh valid states while seven threads called the partial and flattened it into another partial. The free-threaded ASan build crashed after 0.6 seconds in tuple_concat() from partial_new() at Modules/_functoolsmodule.c:287. The same executable completed 340,357 call/construction cycles in eight seconds with the GIL enabled.

Affected Version

CPython 3.14.7 at commit 823f0323ee6ec1402088b73bce1a38473cac36dc, tested with a free-threaded ASan build.

Reproduction

ASAN_OPTIONS=abort_on_error=1:detect_leaks=0 PYTHON_GIL=0 python3.14 poc/reproduce.py 10

Setting PYTHON_GIL=1 provides the GIL-enabled control.

PoC Source Code

poc/reproduce.py

import functools
import sys
import threading
import time


class Payload:
    __slots__ = ("data",)

    def __init__(self):
        self.data = bytearray(8192)


def target_a(*args, **kwargs):
    return len(args)


def target_b(*args, **kwargs):
    return len(kwargs)


def fresh_state(turn):
    func = target_a if turn & 1 else target_b
    args = tuple(Payload() for _ in range(96))
    return (func, args, {}, None)


duration = float(sys.argv[1]) if len(sys.argv) > 1 else 10.0
shared = functools.partial(target_a, *fresh_state(1)[1])
stop = threading.Event()
counts = [0] * 7


def mutate():
    turn = 0
    while not stop.is_set():
        shared.__setstate__(fresh_state(turn))
        turn += 1


def exercise(slot):
    while not stop.is_set():
        try:
            shared()
            functools.partial(shared, Payload())
            counts[slot] += 1
        except (TypeError, ValueError):
            pass


threads = [threading.Thread(target=mutate)]
threads.extend(threading.Thread(target=exercise, args=(i,)) for i in range(7))
for thread in threads:
    thread.start()
time.sleep(duration)
stop.set()
for thread in threads:
    thread.join()
print("completed", sum(counts))

[ZeroIntensity]

Refer to #157124 (comment)