Находки в опенсорсе: “Настало время ответов! Во-первых Код из примера упадет с ValueError: g…”

Настало время ответов!

Во-первых

Код из примера упадет с ValueError: generator already executing.
Почему так? Нельзя дважды запустить один и тот же генератор, даже в одном треде.

Самый простой пример:


def g():
    i = next(me)
    yield i
    
me = g()
next(me)  # ValueError

Исходник:


static PySendResult  // Objects/genobject.c
gen_send_ex(PyGenObject *gen, PyObject *arg, PyObject **presult)
{
    int8_t frame_state = FT_ATOMIC_LOAD_INT8_RELAXED(gen->gi_frame_state);
    // ...
        if (frame_state == FRAME_EXECUTING) {
            PyErr_SetString(PyExc_ValueError, "generator already executing");
            return PYGEN_ERROR;
        }
    // ...
}

Во-вторых

Переиспользовать генераторы в разных тредах - особенно плохая идея. О чем теперь явно написано в документации. Данный паттерн не поддерживается во Free-Threading нативно. Корневая причина, для самых любопытных: https://github.com/python/cpython/issues/120496

PR: https://github.com/python/cpython/pull/148894
Документация: https://docs.python.org/3.15/library/threading.html#iterator-synchronization

Был добавлен способ вызова __next__ под локом (как правильно догадались в комментариях):


class serialize_iterator:
    def __init__(self, iterable):
        self._iterator = iter(iterable)
        self._lock = Lock()
    def __iter__(self):
        return self
    def __next__(self):
        with self._lock:
            return next(self._iterator)

Для того, чтобы код из оригинального примера заработал, нужно заменить iterator = producer(limit) на iterator = threading.serialize_iterator(producer(limit)). Есть еще декоратор @synchronized_iterator для определения threadsafe генераторов сразу.

Теперь - должно быть понятно, почему такая фича была добавлена.

Было ли интересно поковырять? Было ли сложно? :)