Влияние инкремента += в Python на возникновение Race Condition

Код count = count + 1 выполняется атомарно. Код count += 1 тоже. А вот этот кусочек уже нет:

    def term():
        return 1

    count += term()

Сам язык не даёт никаких гарантий атомарности кода. Но в вашей версии CPython так получилось, что какие-то варианты выполняются атомарно. Это зависит от интерпретатора и его версии. Например, в CPython 3.9 атомарности не было. А в 3.10 и 3.11 она появилась. У меня стоит CPython 3.11, на нём и покажу.

NB Не полагайтесь на атомарность. Она не гарантирована и может быть нарушена в любой момент. Найдут в интерпретаторе баг, исправят его, и атомарность пропадёт. Не надо рисковать.

Питон компилирует исходный код в байт код, который затем интерпретируется. count = count + 1 компилируется в такую последовательность инструкций (это дизассемблер, сам код, естественно, двоичный, удобный для интерпретации):

  8           2 LOAD_GLOBAL              0 (count)
             14 LOAD_CONST               1 (1)
             16 BINARY_OP                0 (+)
             20 STORE_GLOBAL             0 (count)

Для обработки данных используется стек. Последовательность действий:
положить на стек значение count,
сверху положить единицу,
вызвать инструкцию сложения,
результат сохранить в count.

Интерпретатор _PyEval_EvalFrameDefault в цикле читает инструкции и выполняет их.

Обработка инструкции LOAD_CONST:

        TARGET(LOAD_CONST) {
            PREDICTED(LOAD_CONST);
            PyObject *value = GETITEM(consts, oparg);
            Py_INCREF(value);
            PUSH(value);
            DISPATCH();
        }

Здесь много макросов, но смысл не теряется: из какого-то хранилища констант достаём значение и помещаем его на стек. Обратите внимание на вызов DISPATCH();. Этот макрос скрывает goto на метку, которая обработает следующую инструкцию.

Если в процессе запущены, скажем, два потока, Питон исполняет немного кода из одного, немного из другого. Потоки никогда не исполняются параллельно, за это отвечает GIL. Прервать исполнение потока в Питоне нельзя. Вместо этого интерпретатор регулярно проверяет, не ожидают ли другие потоки доступа к процессору. Если ожидают, интерпретатор останавливает интерпретацию в текущем потоке, освобождает GIL, ставит текущий поток в очередь на исполнение.

Слишком частые переключения воруют время, Питон переключает потоки не чаще чем раз 5 миллисекунд. Это можно регулировать вызывая sys.setswitchinterval.

То есть, когда поток получил управление, он регулярно проверяет флаг eval_breaker. Через пять миллисекунд этот флаг будет поднят, интерпретатор заметит это и отдаст управление. Управление отдаётся всегда между инструкциями.

Как часто интерпретатор проверяет eval_breaker? Не часто. Проверку флага делает вызов CHECK_EVAL_BREAKER();. Посмотрите ещё раз на код обработки LOAD_CONST. Там нет этого вызова. Интерпретатор перейдёт к следующей инструкции безусловно. После LOAD_CONST текущий поток не может быть прерван.

Если вы проверите ещё LOAD_GLOBAL, BINARY_OP и STORE_GLOBAL, то во всех них проверки нет. Этот кусок кода выполняется как одна атомарная инструкция.

Почему тогда ломается код с вызовом функции?

    def term():
        return 1

    count += term()

Потому что в нём есть вызов функции:

 12           8 LOAD_GLOBAL              0 (count)
             20 PUSH_NULL
             22 LOAD_FAST                0 (term)
             24 PRECALL                  0
             28 CALL                     0
             38 BINARY_OP                0 (+)
             42 STORE_GLOBAL             0 (count)

CALL выполняет проверку флага:

        TARGET(CALL) {
            ...
            CHECK_EVAL_BREAKER();
            DISPATCH();
        }

Ваш первый код ломался не потому что вы вызывали time.sleep(...), а потому что вы вообще вызывали какую-то функцию между чтением значения глобальной переменной и записью его обратно.

Рецепт приготовления гонки данных:

• поток должен работать дольше пяти миллисекунд;
• между чтением и записью должна быть инструкция, после которой интерпретатор готов отдать управление (вызов функции – простейший вариант).

import threading

m = 1_000_000
count = 0


def counter():
    global count

    def one():
        return 1

    for _ in range(m):
        count += one()


threads = [threading.Thread(target=counter) for _ in range(2)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(count, len(threads) * m)

$ python --version
Python 3.11.2

$ python race.py
1634669 2000000

P.S. Упражнение: придумайте код, который отберёт управление у всех остальных потоков минимум на одну секунду.