Deep Engineering

MEASUREMENT

bench/free-threading-races/shared_iterator.py

The script that produced the numbers in the article, and the record of the run. The file is read from the repository at build time — this is the code that was run, not a copy of it.

Cited in
/en/python/concurrency/free-threading-races

The run below is recorded in Russian. It is a lab record, kept in the language it was written in; the numbers, the tables and the code read the same either way.

Record of the run

Гонки без GIL: замеры для статьи

Проверяется утверждение: код, который в сборке с GIL годами работал правильно, в свободной сборке ломается — хотя гонка в нём была всегда. GIL не делал операции атомарными; он лишь не давал потоку уступить управление в середине короткой последовательности инструкций.

Один и тот же код запускается на трёх интерпретаторах:

сборка что это
3.14.10 обычная сборка, GIL включён
3.14.10t свободная сборка той же версии, GIL выключен
3.15.2t свободная сборка следующей версии: в ней исправлена гонка итератора range (cpython#129068)

Как запустить

uv python install 3.14.10 3.14.10t 3.15.2t
./run.sh        # записывает runs/<сборка>.txt

Нужен компилятор C (cc): gil_reenable.py собирает два крошечных расширения из ext/ под тот интерпретатор, которым запущен.

Скрипт Что меряет
bytecode.py инструкции между чтением и записью в двух формах приращения и есть ли среди них точка, где поток может уступить GIL
lost_update.py сколько приращений из 4 000 000 теряется: c.n += 1, d["n"] += 1, c.n = c.n + one(); два исправления и их цена
check_then_act.py ленивый кеш «проверил — записал»: два отдельных показателя — лишние вызовы make() (extra make() calls) и ключи, для которых потоки получили разные объекты (keys split); setdefault и блокировка. Раскол считается по самим объектам: каждый полученный объект удерживается до конца опыта, id сравниваются после join всех потоков
shared_iterator.py один итератор range или списка на четыре потока: повторы, пропуски, значения вне источника; очередь как исправление
gil_reenable.py импорт расширения многофазной инициализации без слота Py_mod_gil включает GIL для всего процесса: предупреждение и ускорение до и после (однофазный модуль заявляет поддержку вызовом PyUnstable_Module_SetGIL; здесь не меряется)
context_flags.py thread_inherit_context и context_aware_warnings: тот же код даёт другой результат

runs/3.14.10t-old-flags.txt — context_flags.py в свободной сборке с ключами -X thread_inherit_context=0 -X context_aware_warnings=0: так возвращается поведение сборки с GIL.

Машина замеров

2 vCPU (Intel Xeon; модель и частота — во второй строке каждой записи, текущие записи сняты на 2.10GHz), Linux, интерпретаторы из python-build-standalone (uv python install). Числа потерь от прогона к прогону разные — это и есть содержание замера, поэтому печатаются все пять прогонов, а не среднее. Времена — лучшее из пяти.

Script

84 lines
"""Один итератор на несколько потоков: «раздать работу» без очереди.

    it = iter(source)
    def worker():
        for x in it:
            out.append(x)

Четыре потока разбирают один итератор по миллиону элементов. Печатается:

  - выдано всего — сколько элементов получили потоки вместе;
  - повторов — сколько элементов выдано больше одного раза;
  - пропущено — сколько элементов источника не выдано ни разу;
  - вне источника — сколько выданных значений в источнике НЕТ вовсе;
    из них отдельно — за концом (x >= N) и до начала (x < 0).

Документация свободной сборки (howto/free-threading-python, раздел Iterators)
предупреждает о повторах и пропусках. Последнего столбца в ней нет; для
итератора range это гонка из cpython#129068, исправленная в 3.15.

Блок 3 — исправление: та же раздача через queue.Queue.
"""
import collections
import queue

from common import RUNS, THREADS, env_line, fmt_int, header, run_threads

N = 1_000_000


def trial(kind):
    if kind == "queue":
        q = queue.SimpleQueue()
        for x in range(N):
            q.put(x)
        for _ in range(THREADS):
            q.put(None)
    else:
        it = iter(range(N)) if kind == "range" else iter(list(range(N)))
    outs = [[] for _ in range(THREADS)]

    def worker(i):
        out = outs[i]
        if kind == "queue":
            while (x := q.get()) is not None:
                out.append(x)
        else:
            for x in it:
                out.append(x)

    run_threads(worker, args_for=lambda i: (i,))
    counts = collections.Counter(x for out in outs for x in out)
    given = sum(counts.values())
    dup = sum(c - 1 for c in counts.values() if c > 1)
    missing = sum(1 for x in range(N) if x not in counts)
    past_end = sum(1 for x in counts if x >= N)
    below = sum(1 for x in counts if x < 0)
    return given, dup, missing, past_end, below


def show(results):
    for given, dup, missing, past_end, below in results:
        print(
            f"handed out {fmt_int(given):>9}  duplicates {fmt_int(dup):>9}  "
            f"missing {fmt_int(missing):>7}  past the end {fmt_int(past_end):>7}  below zero {fmt_int(below):>3}"
        )


def main():
    print(env_line())
    print(f"threads {THREADS}, items {fmt_int(N)}, runs {RUNS}")

    header(1, "ONE RANGE ITERATOR, FOUR THREADS")
    show([trial("range") for _ in range(RUNS)])

    header(2, "ONE LIST ITERATOR, FOUR THREADS")
    show([trial("list") for _ in range(RUNS)])

    header(3, "FIX: A QUEUE INSTEAD OF A SHARED ITERATOR")
    show([trial("queue") for _ in range(RUNS)])


if __name__ == "__main__":
    main()