Как сопровождающий пул соединений для PyMongo, официального драйвера MongoDB для Python, я получил гораздо более глубокие знания о потоках Python, чем я когда-либо хотел.

Одна из проблем, с которыми я сталкиваюсь: если пул соединений назначает сокет потоку, а поток умирает, как мы можем вернуть сокет для общего пула? Я думал, что прибил его в прошлом году, используя слабый обратный вызов для локального потока, но в этом методе есть ошибка. Джастин Патрин из Idle Games обнаружил это во время тестирования вклада PyMongo, который он вносит. Я собираюсь описать ошибку, ее влияние, причину и исправление. В заключение я расскажу о поддержке архаичных версий Python.

Букашка

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

import threading
import weakref

nthreads = 10000
ncallbacks = 0
ncallbacks_lock = threading.Lock()
local = threading.local()
refs = set()

class Vigil(object):
    pass

def run():
    def on_thread_died(ref):
        global ncallbacks
        ncallbacks_lock.acquire()
        ncallbacks += 1
        ncallbacks_lock.release()

    vigil = Vigil()
    local.vigil = vigil
    refs.add(weakref.ref(vigil, on_thread_died))

threads = [threading.Thread(target=run)
           for _ in range(nthreads)]
for t in threads: t.start()
for t in threads: t.join()
getattr(local, 'c', None)  # Trigger cleanup in <= 2.7.0
assert ncallbacks == nthreads, \
    'only %d callbacks run' % ncallbacks

Это метод, который я представил в «Знании, когда поток Python умер» . Каждый поток создает объект «бодрствования» и помещает его в локальный поток. Поскольку только нить локальна относится к бдению, бдение должно быть уничтожено, когда нить умирает. Я делаю слабый ответ на бдение и регистрирую слабый обратный вызов . Если все идет хорошо, обратный вызов запускается, когда поток умирает. Причудой Python 2.7.0 или ниже является то, что обратный вызов запускается, когда следующий поток обращается к локальному потоку. Эта странность является следствием Python Issue 1868 , исправленного Antoine Pitrou в конце 2010 года и выпущенного в Python 2.7.1.

Также обратите внимание, что я синхронизируюсь ncallbacks += 1с мьютексом, поскольку +=не является атомарным в Python . Этот невинно выглядящий мьютекс таит в себе мрачные намерения, как мы скоро обнаружим.

В Python 2.7.1 и новее приведенный выше код работает должным образом: ncallbacksон равен 1000 сразу после объединения всех потоков. В Python 2.7.0 ncallbacksдолжно быть 999 после объединения потоков, а затем 1000 после того, как основной поток getattrвыполнит последний запуск очистки.

Ошибка: в Python 2.7.0 и более ранних, ncallbacksиногда несколько обратных вызовов стесняются тысячи. Несколько нитей были похоронены в безымянных могилах ….

Его влияние

Я обнаружил, что приложение, работающее на Python 2.7.0 или старше, если оно создает и уничтожает очень большое количество потоков непрерывно в течение длительного времени, и если каждый поток вызывает end_requestпо крайней мере один и start_requestболее раз end_request, время от времени оставляет сокет привязанным к мертвая нить. Эти сокеты в конечном итоге превысят ulimit процесса или MongoDB.

Этот шаблон приложения будет настолько странным и необычным, насколько это звучит, поэтому никто не жаловался на ошибку.

Исправление

После того, как я написал тестовый код выше, я потратил несколько часов на то, чтобы с ним справиться — Черт, я думал, что это сработало! Я пробовал разные методы, чтобы заставить Python 2.7.0 надежно выполнить обратный вызов тысячу раз. В конце дня прозвучал божественный голос: «Синхронизируйте назначение с локальным потоком». Поэтому я добавил блокировку:

local_lock = threading.Lock()
# ...
    vigil = Vigil()
    local_lock.acquire()
    local.vigil = vigil
    local_lock.release()
    refs.add(weakref.ref(vigil, on_thread_died))

Это сработало! Теперь я был злее. Как назначение в локальный поток не может быть потокобезопасным?

Причина

Давайте снова рассмотрим пример кода выше. Байткод для назначения vigilна local.vigilNST:

28 LOAD_FAST        1 (vigil)
31 LOAD_GLOBAL      3 (local)
34 STORE_ATTR       4 (vigil)

STORE_ATTRзвонки PyObject_SetAttr, которые звонки local_setattro, определены в Modules / threadmodule.c:

static int
local_setattro(localobject *self, PyObject *name, PyObject *v)
{
    PyObject *ldict;

    ldict = _ldict(self);
    if (ldict == NULL)
        return -1;

    return PyObject_GenericSetAttr((PyObject *)self, name, v);
}

На выделенную строку он звонит _ldict. Как _ldictя уже давно знаю, эта функция — жалкая часть poo в Python 2.7.0 и старше. Вот какашка, отредактированная немного:

static PyObject *
_ldict(localobject *self)
{
    PyObject *tdict, *ldict;

    tdict = PyThreadState_GetDict();
    ldict = PyDict_GetItem(tdict, self->key);
    if (ldict == NULL) {
        ldict = PyDict_New(); /* we own ldict */

        PyDict_SetItem(tdict, self->key, ldict);
        Py_DECREF(ldict); /* now ldict is borrowed */
        if (i < 0)
            return NULL;

        Py_CLEAR(self->dict);
        Py_INCREF(ldict);
        self->dict = ldict; /* still borrowed */
    }

    /* The call to tp_init above may have caused
       another thread to run.
       Install our ldict again. */
    if (self->dict != ldict) {
        Py_CLEAR(self->dict);
        Py_INCREF(ldict);
        self->dict = ldict;
    }

    return ldict;
}

Мы не видели никакого использования Py_BEGIN_ALLOW_THREADSмакроса, поэтому один поток имел GIL все время. Блокировка вокруг задания не должна иметь никакого эффекта, верно?

Что ж, взгляните на выделенное Py_CLEAR(self->dict)утверждение — преступник есть. Этот оператор получает ldictпоследний поток, который получил доступ к этому локальному потоку, меняет его на NULL и дешифрует. Если это последняя ссылка на ldict(потому что последний поток умер), тогда decf’ing уничтожает его, и выполняется обратный вызов слабой ссылки vigil. Делается обратный вызов ncallbacks_lock.acquire, который освобождает GIL перед попыткой получить мьютекс.

Вот такой сценарий, который я предотвратил, заблокировав присваивание локальному потоку:

1. Поток A запускается, присваивается локальному потоку, умирает.
2. Поток А ldictтеперь self->dictявляется локальным и имеет счет 1.
3. Поток B запускается, начинает присваиваться локальному потоку, входит в _ldictфункцию.
4. _ldictустанавливает self->dictв NULL и decfs Thread A ldict, который работает on_thread_died, который вызывает ncallbacks_lock.acquireи освобождает GIL.
5. Теперь начинается поток C, начинается присвоение локальному потоку, ввод _ldict.
6. self->dictПоток C находит значение NULL, увеличивает свой локальный ldictи назначает его self->dict. Это выходит _ldict.
7. Py_CLEAR(self->dict)Поток B возобновляется с , увеличивает свой собственный ldictи назначает его self->dict.