Влияние инкремента += в 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. Упражнение: придумайте код, который отберёт управление у всех остальных потоков минимум на одну секунду.

Пытался воспроизвести Race Condition на Python в разных вариантах.

Ответом на вопрос, почему не вышло, являются следующие факты:

1. Для возникновения "гонки", требуется несколько активных сторон;
2. Требуется, перекрытие по времени.

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

Необходимо более тщательно создавать условия для возникновения "гонки", например:

import threading

def Fib(n):
    return 0 if not n else 1 if 1 == n else Fib(n - 1) + Fib(n - 2)

class test_thread:
    def __init__(self, nthrs, nops, func=None, args=None):
        self.counter = 0
        self.nops = nops
        self.func = func
        self.args = args
        self.thrs = [threading.Thread(target=self.body) 
                     for i in range(nthrs)]
    def calc(self):
        if self.func:
            return self.func(*self.args)
        return 1
    def body(self):
        if self.func:
            for i in range(self.nops):
                self.counter += self.func(*self.args)
        else:
            for i in range(self.nops):
                self.counter += 1
    def test(self):
        start = self.counter
        for thr in self.thrs:
            thr.start()
        for thr in self.thrs:
            thr.join()
        finish = self.counter
        diff = (finish - start) - len(self.thrs)*self.nops*self.calc()
        status = "FAIL" if diff else "OK"
        print(f"{status}, func={self.func}, args={self.args} "
              f"diff={diff} start={start} finish={finish}")

x = test_thread(8, 100_000)
x.test()
x = test_thread(8, 100_000, Fib, [7])
x.test()

Легко видеть, для традиционного CPython (c GIL) с ошибки гонки с весьма большой вероятностью возникают в операторах вида counter += Fib(7):

$ python3.13 /tmp/test_thread.py
OK, func=None, args=None diff=0 start=0 finish=800000
FAIL, func=<function Fib at 0x10f23b560>, args=[7] diff=-8523476 start=0 finish=1876524

Впрочем, и на некоторых версиях с GIL, сравнительно легко увидеть, что и оператор counter += 1 неатомарен:

$ python3.8 /tmp/test_thread.py 
FAIL, func=None, args=None diff=-127286 start=0 finish=672714
FAIL, func=<function Fib at 0x10ad361f0>, args=[7] diff=-8939567 start=0 finish=1460433

Ну уж, тем более, для Free-Threading CPython (без GIL), они с большой вероятностью возникают, как для counter += Fib(7), так и для counter += 1:

$ python3.13t /tmp/test_thread.py 
FAIL, func=None, args=None diff=-656289 start=0 finish=143711
FAIL, func=<function Fib at 0x3d5d45808c0>, args=[7] diff=-9081085 start=0 finish=1318915

Подробности, наверное, стоит читать в PEP 583 – A Concurrency Memory Model for Python, вроде бы, реализация CPython 2.7 строго гарантировала непрерываемость counter += 1. На 3.х, даже таких гарантий нет, просто планировщик иногда затрудняет процесс.

P.S.

Выглядит так, будто операция инкремента += атомарная. То есть можно было бы предположить, что в данном примере из-за атомарности операции на уровне байт-кода

Так говорить не стоит, в байт-коде это даже несколько операций, а не одна специальная, так что, по-любому, существует момент времени, когда данные могут непредсказуемо измениться. Может, в отладчике, может при использовании разделяемой памяти multiprocessing.shared_memory, может и в иных случаях. Вопрос только в вероятности такого события.