GIL и обработка сигналов

Небольшое добавление к статье о GIL.

Эта часть касается только поведения Питона на posix системах. Я проводил эксперимент на Linux, но на FreeBSD и MacOS результат должен быть тем же. На Windows свои тараканы, к последующему изложению не имеющие никакого отношения.

Описание проблемы

Итак, имеем простой код:

import threading

threads = []

running = True

def f():
    while running:
        pass

for i in range(1):
    th = threading.Thread(target=f)
    threads.append(th)
    th.start()

for th in threads:
    th.join()

Что произойдет, если после запуска пользователь нажмет <Ctrl+C>? Будет послан сигнал SIGINT, который мы не умеем обрабатывать.

Отлично, добавим нужное:

import threading
import signal

threads = []

running = True

def f():
    while running:
        pass

for i in range(1):
    th = threading.Thread(target=f)
    threads.append(th)
    th.start()

def sig_handler(sig_num, frame):
    print('SIGNAL')
    global running
    running = False

signal.signal(signal.SIGINT, sig_handler)

for th in threads:
    th.join()

Всё заработало?

Зависит от версии Питона. На 3.2 действительно всё отлично.

Python 2.7 не желает вызывать зарегистрированный обработчик сигнала. print не печатается, потоки не останавливаются.

В чём же дело?

Обработка сигналов в Питоне

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

Когда ОС вызывает этот обработчик, она предоставляет ему отдельный, очень маленький стек, на котором Питону просто не хватит места развернуться.

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

Есть еще одна особенность. Стандарт posix заявляет, что обработчик сигнала должен выполняться в главном потоке (том самом, который создается при запуске процесса). Разные операционные системы (и даже разные версии ядра linux) расширяют эти строгие рамки, но Питон в данном вопросе предельно консервативен — должен быть главный поток и всё тут!

Для этого Питон временно сбрасывает порог переключения GIL со 100 (или сколько там выставлено в sys.setcheckinterval до 1 в надежде быстро добраться до главного потока и асинхронно выполнить в нём свой питоновский код, зарегистрированный как обработчик сигнала.

Асинхронность в данном случае означает, что настоящий обработчик выполняется не сразу, а когда придет его очередь.

Так вот, Питон быстро молотит, переключая GIL в стремлении добраться до главного потока. Который, в свою очередь, упёрся в Thread.join.

А в питоне этот .join реализован через threading.Condition(threading.Lock()), а не через pthread_join. Так нужно — у питона поверх pthreads есть своя необходимая логика.

В результате главный поток блокирован Lock.acquire. Смотрим, что там происходит.

Открываем файл Python/thread_pthread.h из Python 2.7 и обращаем внимание на функцию PyThread_acquire_lock:

int PyThread_acquire_lock(PyThread_type_lock lock, int waitflag) {
    int success;
    sem_t *thelock = (sem_t *)lock;
    int status, error = 0;

    do {
        if (waitflag)
            status = fix_status(sem_wait(thelock));
        else
            status = fix_status(sem_trywait(thelock));
    } while (status == EINTR); /* Retry if interrupted by a signal */

    if (waitflag) {
        CHECK_STATUS("sem_wait");
    } else if (status != EAGAIN) {
        CHECK_STATUS("sem_trywait");
    }

    return (status == 0) ? 1 : 0;
}

Очень мило: если блокирующий вызов (sem_wait или sem_trywait) возвращает EINTR (произошло прерывание по сигналу) — игнорируем это и ждём дальше.

Беда в том, что весь этот код происходит при отпущенном GIL.

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

Еще раз.

Переключение на главный поток происходит, но не доходит до переключения GIL и выполнения питоновского кода.

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

Python 3.2

Теперь смотрим на ту же функцию в новом исполнении:

PyLockStatus
PyThread_acquire_lock_timed(PyThread_type_lock lock, PY_TIMEOUT_T microseconds,
                            int intr_flag)
{
    PyLockStatus success;
    sem_t *thelock = (sem_t *)lock;
    int status, error = 0;
    struct timespec ts;

    if (microseconds > 0)
        MICROSECONDS_TO_TIMESPEC(microseconds, ts);
    do {
        if (microseconds > 0)
            status = fix_status(sem_timedwait(thelock, &ts));
        else if (microseconds == 0)
            status = fix_status(sem_trywait(thelock));
        else
            status = fix_status(sem_wait(thelock));
        /* Retry if interrupted by a signal, unless the caller wants to be
           notified.  */
    } while (!intr_flag && status == EINTR);

    /* Don't check the status if we're stopping because of an interrupt.  */
    if (!(intr_flag && status == EINTR)) {
        if (microseconds > 0) {
            if (status != ETIMEDOUT)
                CHECK_STATUS("sem_timedwait");
        }
        else if (microseconds == 0) {
            if (status != EAGAIN)
                CHECK_STATUS("sem_trywait");
        }
        else {
            CHECK_STATUS("sem_wait");
        }
    }

    if (status == 0) {
        success = PY_LOCK_ACQUIRED;
    } else if (intr_flag && status == EINTR) {
        success = PY_LOCK_INTR;
    } else {
        success = PY_LOCK_FAILURE;
    }

    return success;
}

Добавилась поддержка таймаутов (и это само по себе здорово).

И, самое главное, появился новый параметр intr_flag. Если он установлен, то функция вернет управление наверх, с указанием, что произошел вызов сигнала. Дальше питон уже разберётся, что нужно сделать переключение GIL и дать возможность главному потоку запустить обработчики сигналов.

Теперь это стало стандартным поведением. Ядро Питона везде использует intr_flag, а старый способ остался для поддержки обратной совместимости со старым Python C API. Раздражающая вещь эта обратная совместимость, но без неё никуда...

Что делать, если приходится писать на Python 2.x?

Нужно давать возможность выполниться питоновскому коду в главном потоке до того, как посчитали всю работу законченной и приступили к освобождению ресурсов (.join).

Т.е. следует организовать другую схему сигнализации о завершении потоков, не упирающуюся в безнадежные блокировки (и сделать её перед .join, чтобы тот гарантировано «проскочил» проблемное место).

Решения могут быть разными.

В 2.7 нет ожидания по таймауту — но есть неблокирующие способы завладеть объектом threading.Lock. И если не получилось — поспать немного. Пример смотрите в реализации threading.Condition.wait для Python 2.7.

Или задействовать «непрофильно» select.select — например, создав на каждый рабочий поток локальный socket, в который этот самый поток запишет что-нибудь, а главный поток проснется и поймет, что вся работа закончена. У select есть таймаут, и это тоже может пригодится.

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

Заключение

Проблема в обработке сигналов есть, и теперь вы об этом знаете.

Как вы будете её решать — дело ваше. Предупреждён — значит вооружён.

Лучший способ — перейти на python 3.2. Если это невозможно по ряду причин — остаются другие, не столь элегантные но вполне рабочие решения.

Загадочный GIL

Наверное, каждый питонист слышал про существование Global Interpreter Lock (GIL). Обычно знание предмета исчерпывается фразой: "Это - та самая гадость, которая не позволяет запустить одновременно как минимум два потока, задействовав все имеющиеся ядра современного процессора".

Высказывание отчасти верное, но совершенно неконструктивное и не покрывающее всей многогранности рассматриваемого вопроса.

Позвольте мне пройтись по теме более подробно, рассмотрев вопросы GIL и реализации Питоном многопоточности с разных сторон.

Короткое определение

Прежде всего GIL — это блокировка, которая обязательно должна быть взята перед любым обращением к Питону (а это не только исполнение питоновского кода а еще и вызовы Python C API). Строго говоря, единственные вызовы, доступные после запуска интерпретатора при незахваченном GIL — это его захват. Нарушение правила ведет к мгновенному аварийному завершению (лучший вариант) или отложенному краху программы (куда более худший и труднее отлаживаемый сценарий).

Многопоточный код на питоне

Это — самый простой уровень. Имеем обычную программу, состоящую исключительно из питоновских модулей (.py файлы) и не содержащую Python C Extensions. Пусть в ней работают два потока: главный и запущенный нами.

import threading
import time

running = True

def f(delay):
    while running:
        time.sleep(delay)

th = threading.Thread(target=f, args=(0.5,))
th.start()

for i in range(10):
    time.sleep(0.7)

running = False

th.join()

Многопоточная программа на С не должна как-то отдельно регистрировать свои потоки — достаточно вызова API (pthread_create или CreateThread) для запуска потока. Интерпретатор питона для своей работы требует ряда структур. Давайте рассмотрим их подробнее.

Структуры интерпретатора, обеспечивающие многопоточную работу

PyInterpreterState содержит глобальное состояние интерпретатора: загруженные модули modules, указатель на первый (он же главный) поток tstate_head и кучу других важных для внутренней кухни вещей.

struct PyInterpreterState {
    PyInterpreterState *next;
    PyThreadState *tstate_head;

    PyObject *modules;
    PyObject *sysdict;
    PyObject *builtins;
    PyObject *modules_reloading;

    PyObject *codec_search_path;
    PyObject *codec_search_cache;
    PyObject *codec_error_registry;
};

PyThreadState позволяет узнать, какой кадр стека (frame) исполняется и какой номер у потока с точки зрения операционной системы. Остальные атрибуты сейчас не важны.

struct PyThreadState {
    PyThreadState *next;
    PyInterpreterState *interp;

    PyFrameObject *frame;
    int recursion_depth;

    Py_tracefunc c_profilefunc;
    Py_tracefunc c_tracefunc;
    PyObject *c_profileobj;
    PyObject *c_traceobj;

    PyObject *exc_type;
    PyObject *exc_value;
    PyObject *exc_traceback;

    PyObject *dict;  /* Stores per-thread state */

    long thread_id;
};

PyFrameObject — это объект кадра стека. Питоновский объект, в отличие от первых двух структур (на это указывает PyObject_VAR_HEAD). Имеет указатель на предыдущий кадр f_back, исполняемый код f_code и последнюю выполненную в этом коде инструкцию f_lasti, указатель на свой поток f_tstate и серию из глобального, локального и встроенного пространства имен (f_globals, f_locals и f_builtins соответственно).

struct PyFrameObject {
    PyObject_VAR_HEAD
    PyFrameObject *f_back;  /* previous frame, or NULL */
    PyCodeObject *f_code;   /* code segment */
    PyObject *f_builtins;   /* builtin symbol table (PyDictObject) */
    PyObject *f_globals;    /* global symbol table (PyDictObject) */
    PyObject *f_locals;     /* local symbol table (any mapping) */

    PyThreadState *f_tstate;
    int f_lasti;        /* Last instruction if called */
};

На самом деле членов в этих структурах поболее, и сами структуры отличаются от версии к версии (особенно заметны отличия между 2.x и 3.x) — но сейчас это не важно.

Важно понимать, что все три необходимых для исполнения структуры взаимно связаны между собой и PyThreadState_GET() возвращает указатель на текущий работающий поток:

GIL

Теперь пришло время показать, как именно работает GIL. Тут есть одна тонкость: в 3.2 его реализация довольно значительно изменилась. Для начала рассмотрим «старый» GIL, используемый в Python 2.x и 3.0/3.1.

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

def f(lst, val):
    return [i for i in lst if i != val]

Применим к ней дизассемблер:

import dis
dis.dis(f)
  2           0 BUILD_LIST               0
              3 LOAD_FAST                0 (lst)
              6 GET_ITER
        >>    7 FOR_ITER                24 (to 34)
             10 STORE_FAST               2 (i)
             13 LOAD_FAST                2 (i)
             16 LOAD_FAST                1 (val)
             19 COMPARE_OP               3 (!=)
             22 POP_JUMP_IF_FALSE        7
             25 LOAD_FAST                2 (i)
             28 LIST_APPEND              2
             31 JUMP_ABSOLUTE            7
        >>   34 RETURN_VALUE

Как видите, код превратился в последовательность этих самых инструкций, исполняемых интерпретатором. Скажу сразу: никакой внятной документации по байт-коду нет, инструкции добавляются и изменяются от версии к версии. Интересующиеся должны читать файл Python/ceval.c как первоисточник для понимания того, что какая инструкция делает.

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

Сам GIL устроен как обычная не-рекурсивная блокировка. Эта же структура лежит в основе threading.Lock. Реализуется через событие CreateEvent с бубенцами на Windows и семафор sem_t на Linux.

Давайте посмотрим на кусочек исходного кода функции PyEval_EvalFrameEx, которая представляет собой цикл с очень объемным switch/case внутри, исполняющим по одной инструкции за проход.

PyObject *
PyEval_EvalFrameEx(PyFrameObject *f, int throwflag)
{
    PyThreadState *tstate = PyThreadState_GET();
    /* ... */
    for (;;) {
        /* ... */
        if (--_Py_Ticker < 0) {
            /* ... */
            _Py_Ticker = _Py_CheckInterval;
            if (interpreter_lock) {
                /* Give another thread a chance */
                if (PyThreadState_Swap(NULL) != tstate)
                    Py_FatalError("ceval: tstate mix-up");
                PyThread_release_lock(interpreter_lock);

                /* Other threads may run now */

                PyThread_acquire_lock(interpreter_lock, 1);
                if (PyThreadState_Swap(tstate) != NULL)
                    Py_FatalError("ceval: orphan tstate");
                /* ... */
            }
        }
    /* instruction processing */
    }
}

Как видите, все просто: имея захваченный GIL (а поток уже им владеет перед вызовом PyEval_EvalFrameEx), каждый раз уменьшаем счетчик пока не дойдем до нуля. interpreter_lock — это наш GIL, указатель на объект блокировки. Если он есть (а есть всегда, за исключением специальных сборок питона с полностью отключенной многопоточностью), то происходит так называемое «переключение GIL».

PyThreadState_Swap сбрасывает указатель на текущий исполняемый поток (тот самый, который возвращается PyThreadState_GET) и освобождает GIL.

Затем следующей строкой пытается захватить этот GIL снова. Хитрость в том, что если работает несколько потоков одновременно, то операционная система сама будет определять, какой поток из ожидающих в PyThread_acquire_lock получит эту блокировку (остальные будут ждать следующего освобождения interpreter_lock). Современные операционные системы используют довольно замысловатые алгоритмы переключения потоков. Нам же нужно знать лишь то, что эти алгоритмы пытаются распределить время «справедливо», дав каждому поработать. Это означает, что только что освободивший GIL поток скорее всего обратно сразу же его не получит — а отдаст управление другому потоку и сам встанет в ожидание PyThread_acquire_lock.

Все работает, и схема получается надежная. Но она имеет ряд существенных недостатков:

• GIL переключается даже в однопоточной программе. Формально interpreter_lock создается не сразу при старте интерпретатора. Но импорт модуля threading или, к примеру, sqlite3 создаст GIL даже без создания второго потока. На практике правильней считать, что GIL есть всегда.
• Другими словами GIL переключается постоянно, независимо от того требует ли другой поток переключения или они все заблокированы ожиданием ввода-вывода или объектами синхронизации.
• Потоки «соревнуются» за захват GIL. Планировщик операционной системы — очень сложно устроенная штука. Поток, интенсивно использующий операции ввода-вывода, получает более высокий приоритет чем чисто вычислительный поток. Например, первый поток читает из файла и складывает прочитанное в очередь. Второй поток получает данные из очереди и обрабатывает их. Штука в том, что считывающий поток, обладая высоким приоритетом, может класть новые данные в очередь довольно долго, прежде чем обработчик получит возможность их обрабатывать. Да, первый поток регулярно освобождает GIL — но он тут же получает его назад (приоритет выше). Эта ситуация может быть исправлена выбором правильного способа взаимодействия между потоками, но решение зачастую неочевидно и, главное, проблема трудно локализуется.
• И, наконец, главная причина. Переключение происходит по количеству выполненных инструкций. Дело в том, что время выполнения инструкций может сильно отличаться (сравните простое сложение и создание списка на миллион элементов).

Для управления порогом переключения существуют функции:

• sys.getcheckinterval()
• sys.setcheckinterval(count)

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

Новый GIL

Он использует усовершенствованную схему, базирующуюся на времени. Кроме того, добавлен специальный механизм для предотвращения повторного захвата GIL.

Снова выдержка из PyEval_EvalFrameEx, на этот раз Python 3.2.

PyObject *
PyEval_EvalFrameEx(PyFrameObject *f, int throwflag)
{
    PyThreadState *tstate = PyThreadState_GET();
    /* ... */
    for (;;) {
        /* ... */
        if (_Py_atomic_load_relaxed(&eval_breaker)) {
            /* ... */
            if (_Py_atomic_load_relaxed(&gil_drop_request)) {
                /* Give another thread a chance */
                if (PyThreadState_Swap(NULL) != tstate)
                    Py_FatalError("ceval: tstate mix-up");
                drop_gil(tstate);

                /* Other threads may run now */

                take_gil(tstate);
                if (PyThreadState_Swap(tstate) != NULL)
                    Py_FatalError("ceval: orphan tstate");
            }
        }
    /* instruction processing */
    }
}

Как видите, внешне почти ничего не изменилось. Ушел счетчик _Py_Ticker, Появились две переменные: eval_breaker и gil_drop_request. Переключение произойдет, если обе установлены (ненулевые). Две переменные нужны потому, что один и тот же механизм используется для штатного переключения GIL и для обработки сигналов операционной системы.

eval_breaker указывает на необходимость переключения, а gil_drop_request используется для штатной ситуации переключения потоков.

_Py_atomic_load_relaxed — это просто макрос для атомарного чтения переменной.

Вся магия скрыта внутри функций drop_gil и take_gil, работающих в паре.

static void drop_gil(PyThreadState *tstate)
{
    if (!_Py_atomic_load_relaxed(&gil_locked))
        Py_FatalError("drop_gil: GIL is not locked");

    MUTEX_LOCK(gil_mutex);
    _Py_ANNOTATE_RWLOCK_RELEASED(&gil_locked, /*is_write=*/1);
    _Py_atomic_store_relaxed(&gil_locked, 0);
    COND_SIGNAL(gil_cond);
    MUTEX_UNLOCK(gil_mutex);

    if (_Py_atomic_load_relaxed(&gil_drop_request)) {
        MUTEX_LOCK(switch_mutex);
        /* Not switched yet => wait */
        RESET_GIL_DROP_REQUEST();
        COND_WAIT(switch_cond, switch_mutex);
        MUTEX_UNLOCK(switch_mutex);
    }
}

Наш герой теперь называется gil_locked — обычная целочисленная переменная. Используется блокировка gil_mutex в паре с условной переменной gil_cond для синхронизации доступа к GIL. gil_drop_request — запрос на переключение GIL, защищенный switch_mutex и switch_cond.

В «отпускающей» стороне нет ничего сложного: прикрываясь gil_mutex сбрасываем GIL (gil_locked) в нолик и сигналим об этом событии через gil_cond.

Затем смотрим, это была просьба о переключении от другого потока или наш поток сам попросил освободиться. Дело в том, что Питон отпускает GIL перед системными вызовами. Чтение из файла, к примеру, может занимать длительное время и совершенно не требует GIL — можно дать шанс другим потокам поработать.

Если GIL освобождается не по внешнему запросу — работа закончена. Иначе нужно дождаться, пока попросивший не захватит GIL. Таким образом форсируется переключение на другой поток.

static void take_gil(PyThreadState *tstate)
{
    MUTEX_LOCK(gil_mutex);

    while (_Py_atomic_load_relaxed(&gil_locked)) {
        int timed_out = 0;
        unsigned long saved_switchnum;

        saved_switchnum = gil_switch_number;
        COND_TIMED_WAIT(gil_cond, gil_mutex, INTERVAL, timed_out);
        /* If we timed out and no switch occurred in the meantime,
           it is time to ask the GIL-holding thread to drop it. */
        if (timed_out &&
            _Py_atomic_load_relaxed(&gil_locked) &&
            gil_switch_number == saved_switchnum) {
            SET_GIL_DROP_REQUEST();
        }
    }
    /* This mutex must be taken before modifying gil_last_holder. */
    MUTEX_LOCK(switch_mutex);
    /* We now hold the GIL */
    _Py_atomic_store_relaxed(&gil_locked, 1);
    ++gil_switch_number;
    _Py_ANNOTATE_RWLOCK_ACQUIRED(&gil_locked, /*is_write=*/1);

    COND_SIGNAL(switch_cond);
    MUTEX_UNLOCK(switch_mutex);

    if (_Py_atomic_load_relaxed(&gil_drop_request)) {
        RESET_GIL_DROP_REQUEST();
    }

    MUTEX_UNLOCK(gil_mutex);
}

Захват GIL зеркально отражает его освобождение. Сначала ждем, пока GIL не освободится. Если ждем долго (больше 5 мс по умолчанию) и при этом не произошло переключения (не важно, на нас или какой другой поток) — выставляем запрос на переключение.

Дождавшись наконец свободного GIL, захватываем его и сигналим отдавшему потоку что передача состоялась. Естественно, все обращения защищены блокировками.

Что получилось в итоге:

• поток, владеющий GIL, не отдает его пока об этом не попросят.
• если уж отдал по просьбе, то подождет окончания переключения и не будет сразу же пытаться захватить GIL назад.
• поток, у которого сразу не получилось захватить GIL, сначала выждет 5 мс и лишь потом пошлет запрос на переключение, принуждая текущего владельца освободить ценный ресурс. Таким образом переключение осуществляется не чаще чем раз в 5 мс, если только владелец не отдаст GIL добровольно перед выполнением системного вызова.

Управление временем переключения — через sys.getswitchinterval и sys.setswitchinterval. Обратите внимание: в python 3.2 остались sys.getcheckinterval и sys.setcheckinterval, но они ни на что не влияют.

GIL и системные вызовы

Почти любое обращение к ядру операционной системы — довольно затратная операция. Более того, этот вызов может блокировать поток на значительный промежуток времени. Скажем, открытие файла может потребовать нескольких обращений к диску, если только этот файл уже не находится в файловом кэше.

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

Поэтому перед вызовом такого долгоиграющего кода нужно отпустить GIL, а потом сразу же его захватить обратно:

Py_BEGIN_ALLOW_THREADS
errno = 0;
self->fd = open(name, flags, 0666);
Py_END_ALLOW_THREADS

Макросы Py_BEGIN_ALLOW_THREADS и Py_END_ALLOW_THREADS делают всю необходимую работу.

Обратите внимание, вкладывать друг в друга эти макросы запрещено! Один раз разрешив потоки, нельзя разрешить их вторично. Например, такой код ошибочный:

void f()
{
    Py_BEGIN_ALLOW_THREADS
    /* do something */
    Py_END_ALLOW_THREADS
}

void g()
{
    Py_BEGIN_ALLOW_THREADS
    f();
    Py_END_ALLOW_THREADS
}

Если очень хочется, то внутри функции можно писать Py_BLOCK_THREADS/Py_UNBLOCK_THREADS для временного получения GIL назад. Например, так:

PyObject* g()
{
    int ret;
    Py_BEGIN_ALLOW_THREADS
    ret = f();
    if (ret) {
        Py_BLOCK_THREADS
        PyErr_SetFromErrno(PyExc_IOerror);
        Py_UNBLOCK_THREADS
        return NULL;
    }
    Py_END_ALLOW_THREADS
}

Во вложенной функции испрользовать Py_BLOCK_THREADS не получится — эти макросы используют стандартные питоновские вызовы PyEval_SaveThread/PyEval_RestoreThread с сохранением структуры PyThreadState в локальной переменной _save.

Коротко говоря, следите за руками и одновременно изучайте исходники Питона. Они занятные — регулярно перечитываю перед сном.

GIL и потоки

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

Давайте посмотрим, как именно создается новый поток.

Вспомогательная структура:

struct bootstate {
    PyInterpreterState *interp;
    PyObject *func;
    PyObject *args;
    PyObject *keyw;
    PyThreadState *tstate;
};

Просто хранит функцию, которую нужно выполнить в новом потоке, и ее параметры. Состояние потока и интерпретатора тоже пригодится.

Код, запускающий поток:

static PyObject *
thread_PyThread_start_new_thread(PyObject *self, PyObject *fargs)
{
    PyObject *func, *args, *keyw = NULL;
    struct bootstate *boot;
    long ident;

    boot = PyMem_NEW(struct bootstate, 1);
    if (boot == NULL)
        return PyErr_NoMemory();
    boot->interp = PyThreadState_GET()->interp;
    boot->func = func; boot->args = args; boot->keyw = keyw;
    boot->tstate = _PyThreadState_Prealloc(boot->interp);
    if (boot->tstate == NULL) {
        PyMem_DEL(boot);
        return PyErr_NoMemory();
    }
    Py_INCREF(func); Py_INCREF(args); Py_XINCREF(keyw);
    PyEval_InitThreads(); /* Start the interpreter's thread-awareness */
    ident = PyThread_start_new_thread(t_bootstrap, (void*) boot);
    if (ident == -1) {
        PyErr_SetString(ThreadError, "can't start new thread");
        Py_DECREF(func); Py_DECREF(args); Py_XDECREF(keyw);
        PyThreadState_Clear(boot->tstate);
        PyMem_DEL(boot);
        return NULL;
    }
    return PyLong_FromLong(ident);
}

Ничего сложного: создаем bootstate и запускаем в новом потоке функцию t_bootstrap, которая должна закончить регистрацию PyThreadState. PyThread_start_new_thread — платформонезависимая обертка для создания потока ядра.

Сама запускаемая функция:

static void
t_bootstrap(void *boot_raw)
{
    struct bootstate *boot = (struct bootstate *) boot_raw;
    PyThreadState *tstate;
    PyObject *res;

    tstate = boot->tstate;
    tstate->thread_id = PyThread_get_thread_ident();
    _PyThreadState_Init(tstate);
    PyEval_AcquireThread(tstate);
    nb_threads++;
    res = PyEval_CallObjectWithKeywords(
        boot->func, boot->args, boot->keyw);
    if (res == NULL) {
        if (PyErr_ExceptionMatches(PyExc_SystemExit))
            PyErr_Clear();
        else {
            PyObject *file;
            PySys_WriteStderr(
                "Unhandled exception in thread started by ");
            file = PySys_GetObject("stderr");
            if (file != NULL && file != Py_None)
                PyFile_WriteObject(boot->func, file, 0);
            else
                PyObject_Print(boot->func, stderr, 0);
            PySys_WriteStderr("\n");
            PyErr_PrintEx(0);
        }
    }
    else
        Py_DECREF(res);
    Py_DECREF(boot->func); Py_DECREF(boot->args); Py_XDECREF(boot->keyw);
    PyMem_DEL(boot_raw);
    nb_threads--;
    PyThreadState_Clear(tstate);
    PyThreadState_DeleteCurrent();
    PyThread_exit_thread();
}

Код длинный, но простой. Из запущенного потока можно узнать его номер (идентификатор, используемый при обращениях к операционной системе). Осталось закончить инициализацию PyThreadState и выполнить запрашиваемую питоновскую функцию (PyEval_CallObjectWithKeywords). После окончания работы нужно почистить за собой. Если было исключение — записать его в stderr (исключения из потока не пробрасываются запустившему этот поток коду, остается только запись в поток ошибок).

Без приведенного пролога обращение к питону из нового потока приведет к краху интерпретатора. Причем не сразу же, а когда потребуется переключить GIL. Обращение к глобальным структурам может разрушить память.

Способ, используемый Питоном, рабочий. Но не самый подходящий для стороннего кода, желающего запускать питон в произвольном потоке, созданном без помощи Python API. Дело в том, что приведенные функции используют закрытую часть API — функции, начинающиеся с подчеркивания.

Код можно было бы переписать, полностью перенеся создание PyThreadState в t_bootstrap и сократив его до:

PyThreadState *tstate = PyThreadState_New(boot->interp);
PyEval_AcquireThread(tstate);

Вообще-то в наборе функций для работы с потоками и GIL наблюдается некоторый разброд и шатание. Например, PyEval_AcquireThread захватывает GIL. PyEval_RestoreThread делает практически то же самое плюс специальную проверку на случай завершения интерпретатора (которая в правильно написанной программе не нужна, порожденные потоки должны получить сигнал о завершении раньше, чем произойдет завершение питоновского кода в главном потоке). То же самое можно сказать про пару PyEval_ReleaseThread и PyEval_SaveThread и так далее.

В оправдание сложившегося положения вещей можно сказать, что эта часть API развивалась долго, постепенно переходя из закрытой в публичную и документированную. Существующие сторонние модули чаще всего писались по принципу "работает—и ладно", использование закрытого API авторов не волновало (тем более что открытое API появлялось с некоторым опозданием). На данный момент имеем множество библиотек, которые могут поломаться, вздумай разработчики Питона разом отрубить все устаревшие части API. Поэтому процесс удаления устаревшего кода занимает как минимум 2-3 версии питона (что составляет около 4-5 лет) со строгими предупреждениями и разъяснениями. И тем не менее всегда остаются недовольные авторы, чей код «внезапно» перестал компилироваться.

Временное получение GIL

Бывает так, что нужно выполнить питоновский вызов, не зная — зарегистрирован ли поток или еще нет. Для этого существует пара PyGILState_Ensure и PyGILState_Release:

PyGILState_STATE opaque = PyGILState_Ensure();
/* do stuff */
PyGILState_Release(opaque);

Между вызовами ensure/release можно делать всё, что угодно — GIL захвачен, PyThreadState настроен. Можно, например, использовать Py_BEGIN_ALLOW_THREADS/Py_END_ALLOW_THREADS и вообще вызывать любой питоновский код. Более того, можно делать вложенные вызовы PyGILState_Ensure — главное не забывать о необходимых PyGILState_Release.

Нужно только всегда помнить об одной маленькой детали. Дело в том, что при каждом вызове PyGILState_Ensure система смотрит, был ли зарегистрирован PyThreadState для исполняемого потока. Если был — то захват происходит быстро. Иначе нужно создать и зарегистрировать новый PyThreadState. Для существующей структуры достаточно просто увеличить счетчик использования. PyGILState_Release этот счетчик уменьшает и, досчитав до нуля, удаляет зарегистрированный PyThreadState. На удаление тоже нужно время. Потери относительно небольшие, если только код не исполняется очень много раз. Иными словами, вместо создания/удаления PyThreadState в цикле:

void f() {
    int i;
    PyGILState_STATE state;
    for(i=0; i<100000; ++i) {
        /* do C block */
        state = PyGILState_Ensure();
        /* call python API */
        PyGLIState_Release(state);
    }
}

лучше написать работающий практически так же, но более оптимальный по скорости код:

void f() {
    int i;
    PyGILState_STATE state, outer_state;
    outer_state = PyGILState_Ensure();
    Py_BEGIN_ALLOW_THREADS
    for(i=0; i<100000; ++i) {
        /* do C block */
        state = PyGILState_Ensure();
        /* call python API */
        PyGILState_Release(state);
    }
    Py_END_ALLOW_THREADS
    PyGILState_Release(outer_state);
}

Заключение

Я постарался как можно подробней описать, как работает GIL. Полностью покрыть эту тему невозможно даже в большой статье. Поэтому, если возникают вопросы — задавайте их или (что надежней и полезней) читайте ответ в исходном коде Питона, он простой и понятный.

Несколько слов «на общие темы».

• GIL не уберут никогда. Или, по крайней мере, в ближайший десяток лет. Сейчас никаких работ на эту тему не ведется. Если некий гений предъявит работающую реализацию без GIL, ничего не ломающую и работающую не медленней, чем существующая версия — предложению будет открыт зеленый свет. Пока же «убрать GIL» проходит по части благих, но невыполнимых пожеланий.

• В Java и C# никакого GIL нет. Потому что у них иначе устроен garbage collector. Если хотите, он более прогрессивный. Переделать GC Питона, не сломав обратной совместимости со всеми существующими библиотеками, использующими Python C API — невозможно. Сообщество и так уже который год лихорадит в связи с переходом на Python 3.x. Разработчики не желают выкатывать второе революционное изменение, не разобравшись с первым. Ждите Python 4.x (которого нет даже в планах) — до тех пор ничего не поменяется.

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

  • Во первых, если поток не делает вызовов Python C API — то GIL ему не нужен. Так можно держать много параллельно работающих потоков-числодробилок плюс несколько медленных питоновских потоков для управления всем хозяйством. Конечно, для этого нужно уметь писать Python C Extensions.
  • Второй способ еще лучше. Замените «поток» на «процесс». По настоящему высоконагруженная система в любом случае должна строится с учетом масштабируемости и высокой надежности. На эту тему можно говорить очень долго, но хорошая архитектура автоматически позволяет вам запускать несколько процессов на одной машине, которые общаются между собой через какую-либо систему сообщений. В качестве одного из приятных бонусов получается избавление от "проклятия GIL" — у каждого процесса он только один, но процессов много!

Надеюсь, прочитанное поможет вам понять, как GIL работает, какие у него есть неочевидные особенности. И, главное — запомнить, как он не работает никогда!

GIL и обработка сигналов

В дополнение к этой статье рекомендую прочесть про особенности обработки сигналов Питоном в posix средах.

Ссылки:

• Замечательная статья от признанного специалиста по GIL Дэвида Бизли (Dave Beazley)
• Описание проблем с PyGILState_Ensure в блоге Kristján Valur, ведущего специалиста по Питону в CCP Games (Eve Online)

Мультипоточность в Питоне. Часть 3 - высокоуровневые объекты

В частях 1 и 2 был описан низкоуровневый базис мультипоточных программ.

Эта основа крайне необходима для понимания. Но, тем не менее, работать с низким уровнем слишком утомительно и небезопасно.

Поэтому поговорим об объектах мультипоточности.

С точки зрения объектно ориентированного программирования в применении к многопоточности изменяемые объекты делятся на:

• Неизменяемые. Блокировать ничего не нужно, объект не может изменить свое состояние во время работы.

• Не имеющие представления по потоках. Доступ к ним должен обеспечиваться явными блокировками, выполнение которых - ответственность программиста. Иначе при одновременном обращении из разных потоков возможно повреждение объекта и прочие неприятности.

• Мониторы (monitors). Содержат внутри блокировку, которая берется при обращении к публичным атрибутам. Полностью потокобезопасные - одновременно с монитором работает только один поток.

• Активные объекты (active objects). Содержат внутри как минимум один поток. Вызовы публичных методов активного объекта возвращают отложенное значение (future), которое будет заполнено после того, как вложенный поток произведет нужные операции.

Мониторы

Назовем объектом-монитором такой объект, доступ к публичной части которого защищен блокировкой.

import threading

class A(object):
    def __init__(self, val):
        self._lock = threaing.RLock()
        self._val = val

    @property
    def val(self):
        with self._lock:
            return self._val

    @val.setter
    def val(self, val):
        with self._lock:
            self._val = val

    def do(self):
        with self._lock:
            pass # do stuff

Выглядит очень просто, верно? Ничем не отличается от примеров из предыдущей статьи.

Дело вот в чём: монитор - это концепт. Если часть публичного интерфейса защищена блокировками, а часть нет - концепт оказывается порванным.

Прелесть монитора в том, что он потокобезопасный по построению. Т.е. блокировка берется всегда - не смотря на то, нужна она или нет в конкретном случае. Например, мы уверены что метод .do из примера вызывается тогда, когда доступ к экземпляру класса имеет только один поток.

Возникает соблазн убрать эти "лишние" блокировки. С ним нужно бороться. Многопоточные программы - штука сложная. Отдельные потоки выполняются параллельно. Время их работы и график взаимодействия могут довольно сильно варьироваться от запуска к запуску. При развитии программы легко оказывается, что неблокирующая часть интерфейса начинает требовать синхронизации. Отсутствие блокировок в нужных местах ведет к непредсказуемому поведению, порче данных, обескураженным заказчикам и прочим неприятностям.

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

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

Явное выделение "чистых" мониторов способствует ясному, чистому и надежному дизайну.

Пулы потоков (thread pools)

Прежде чем перейти к обсуждению активных объектов давайте поговорим о пулах потоков.

Дело в следующем: создание потока (объекта ядра вашей операционной системы, между прочим) - довольно дорогая операция. Если код потока выполняется быстро, время создания потока может быть сопоставимо со временем исполнения этого кода.

Есть и еще одна проблема - операционные системы не очень любят когда потоков слишком много. У процесса заканчивается виртуальная память и количество доступных объектов ядра. Детали зависят от конкретной операционной системы, да на самом деле они и не нужны.

Важно запомнить: тысяча потоков - это практически всегда плохо, сотня - тоже скорее всего излишне.

Теория считает, что оптимальное количество активных (т.е. не спящих в ожидании того или иного события) потоков равно числу ядер процессора.

Для Питона это не совсем верно (всё тот же GIL), но примерный порядок чисел ясен.

Обе проблемы эффективно решаются созданием поточного пула.

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

Рабочий поток берет из очереди запрос и выполняет его в своем контексте. Получившийся результат возвращается обратно.

Пул потоков обычно создается при старте программы и живет до ее завершения. Чаще всего программе нужно очень немного пулов - несколько штук максимум, при том что чисто технически почти всегда можно обойтись лишь одним.

Я видел пару десятков различных реализаций этого механизма. К счастью, анархии пришел конец и в Питоне 3.2 появляется concurent.futures .

Короткий пример использования:

from concurent.futures import ThreadPoolExecutor, as_completed

def f(a):
    # do something
    return a

with ThreadPoolExecutor(max_workers=5) as pool:
    results = [pool.submit(f, i) for i in range(100)]

    for future in as_completed(results):
        print(future.result())

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

• Создается объект pool - пул потоков. Сколько этих потоков запущено - не важно. Потоки могут автоматически создаваться по мере необходимости. Создание и завершение потоков - часть внутренней реализации, которая может изменяться.

  max_workers - ограничение на максимально возможное для пула количество потоков.

  Пул работает в with блоке, чтобы вызвать .shutdown при выходе.

  Метод shutdown дождется завершения всех запущенных потоков.

• Самый важный метод - submit. Он ставит в очередь запрос на обработку вместе со всеми требуемыми параметрами и возвращает future - отложенное значение.

• У отложенного значения можно узнать, завершено ли асинхронное вычисление или еще нет, получить результат вычисления (или поймать исключение) и т.д.

• as_completed - очень удобная утилита. Принимает список отложенных вычислений и возвращает генератор, выдающий завершенные вычисления по мере готовности.

Конечно же, поддерживается принудительная отмена отложенного, но еще не начавшегося выполнятся вычисления, добавление обратного вызова на завершение future и таймауты.

Активные объекты

Активным объектом будем называть объект, выполняющим свои методы в отдельных потоках.

Имея пул потоков создать активный объект очень легко.

Давайте сделаем простенький пример, проверяющий - жив ли еще интернет?

#!/home/andrew/projects/py3k/python

import threading
from concurrent import futures
from collections import defaultdict, namedtuple
from urllib.request import urlopen, URLError

State = namedtuple('State', 'addr ok fail')

class Pinger:
    def __init__(self, pool):
        self._pool = pool
        self._lock = threading.RLock()
        self._results = defaultdict(lambda: {'ok': 0, 'fail': 0})
        self._pendings = set()

    def result(self, addr=None):
        def _make_state(addr, res):
            return State(addr=addr, ok=res['ok'], fail=res['fail'])
        with self._lock:
            if addr is not None:
                return _make_state(addr, self._results[addr])
            else:
                return {_make_state(addr, val)
                        for addr, val in self._results.items()}

    @property
    def pendings(self):
        with self._lock:
            return set(self._pendings)

    def ping(self, addr):
        with self._lock:
            future = self._pool.submit(self._ping, addr)
            self._pendings.add(future)
            future.add_done_callback(self._discard_pending)
            return future

    def _discard_pending(self, future):
        with self._lock:
            self._pendings.discard(future)

    def _ping(self, addr):
        try:
            ret = urlopen(addr)
            ret.read()
        except URLError:
            result = False
        else:
            result = True

        with self._lock:
            if result:
                self._results[addr]['ok'] += 1
            else:
                self._results[addr]['fail'] += 1

        return result

if __name__ == '__main__':
    from pprint import pprint

    with futures.ThreadPoolExecutor(max_workers=3) as pool:
        pinger = Pinger(pool)

        pinger.ping('http://google.com')
        pinger.ping('http://ya.ru')

        print("State for 'ya.ru'", pinger.result('http://ya.ru')) # 1

        future = pinger.ping('http://python.su/forum/index.php')
        print("Result for 'python.su'", future.result()) # 2

        pinger.ping('http://asvetlov.blogspot.com')

        futures.wait(pinger.pendings) # 3

        print("Total table")
        pprint(pinger.result()) # 4

Вышло не так уж и коротко - но оно того стоит.

Давайте разберем код по порядку, обращая внимание на существенные места.

• Имеем класс Pinger, который является активным объектом. Конструктор класса принимает параметром ThreadingPoolExecutor, а не создает его внутри.

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

  Лучше передавать один правильно сконфигурированный пул всем желающим, а они уж пусть сражаются за потоки в нем.

  Или создать несколько пулов для разных целей - но оставить себе контроль за их применением.

• Внутри класса есть блокировка self._lock (куда без нее), таблица результатов и множество ожидающих вызовов - тех future, которые были запланированы на выполнение но еще не завершились.

• Два публичных атрибута для опроса текущего состояния.

  • метод result возвращает всю таблицу или только ее строку для указанного адреса.

  • свойство pendings вернет список еще не выполненных запросов.

  Обратите внимание: эти действия берут блокировку _lock и возвращают копию изменяемых данных. Оба пункта очень важны.

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

  Если возвращать ссылку на внутреннюю структуру вместо копии - повреждение данных произойдет позже, при обработке результата вызова во внешнем коде.

  Вероятно, вы заметите проблемы далеко не сразу - и тем сложнее их будет выявить и поправить.

• Метод ping ставит обработчик (непубличный метод _ping, делающий всю работу) в очередь пула потоков.

  Блокировка нужна и здесь, так как изменяется _pendings.

  К полученному отложенному вызову добавляем callback, чтобы удалить наше вычисление из _pendings после того, как оно будет выполнено.

  Удаление (_discard_pending) тоже происходит с блокировкой.

• Рабочий код _ping пытается получить html страницу через urlopen.

  Блокировка не берется - доступ к интернету медленный, а мы как раз хотели получить возможность пинговать несколько сайтов одновременно. Столько, сколько окажется свободных потоков у _pool.

  Тем не менее для обновления внутренних структур класса блокировка нужна.

  Наконец, возвращаем результат, который попадет в future и заодно вызовет callback _discard_pending.

• _discard_pending будет вызван в контексте рабочего потока из пула, запомните этот факт. Не в потоке, вызвавшем 'ping' - а в рабочем, внутри пула.

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

andrew@ocean ~/p/a/threading> ./pinger.py 
State for 'ya.ru' State(addr='ya.ru', ok=0, fail=0)
Result for 'python.su' True
Total table
{State(addr='http://asvetlov.blogspot.com', ok=1, fail=0),
 State(addr='http://google.com', ok=1, fail=0),
 State(addr='http://python.su/forum/index.php', ok=1, fail=0),
 State(addr='http://ya.ru', ok=1, fail=0),

• На момент постановки 'ya.ru' в очередь пинг только выполняется, результата нет.

• Явный вызов future.result() для 'python.su' дожидается завершения, True демонстрирует вечноживучесть форума.

• futures.wait(pinger.pendings) ждет завершения всех отложенных задач. Если future на тот момент уже готова - ждать не потребуется.

Другие высокоуровневые мультипоточные объекты

Откровенно говоря, их нет.

Всё, что вы придумаете будет является комбинацией мониторов и активных объектов.

Иногда эти поделки удобны.

Временами, - в случае нарушения правил работы с потоками, - больно бьют по рукам.

Общее правило остается в силе - объект потокобезопасен только в двух случаях:

1. он константный

2. все публичные операции с объектом защищены тем или иным методом синхронизации и передаваемые/возвращаемые параметры тоже потокобезопасны.

Естественно, в реальной жизни всё куда сложнее. Сплошь и рядом приходится нарушать "правила безопасности жизнедеятельности", идя на компромисс в целях улучшения производительности или по каким-нибудь другим причинам.

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

Мультипоточность в питоне. Часть 2 - синхронизация

В части 1 мы рассмотрели базовые основы работы с потоками. Перейдем к примитивам синхронизации.

В мультипоточной программе доступ к объектам иногда нужно синхронизировать.

С этой точки зрения все объекты (переменные) разделяются на:

• Неизменяемые. Мои самые любимые. Если объект никто не меняет, то синхронизация доступа ему не нужна. К сожалению, таких не очень много.
• Локальные. Если объект не виден остальным потокам, то доступ к нему синхронизировать тоже не требуется.
• Разделяемые и изменяемые. Синхронизация необходима.

Синхронизация доступа к объектам осуществляется с помощью объектов синхронизации (простите за тавтологию).

Объектов синхронизации существует множество - и постоянно придумывают новые. Это не то чтобы плохо - но избыточно (по крайней мере на начальном этапе изучения).

Каждый объект синхронизации имеет свои плюсы и минусы (обычно это баланс между быстродействием и областью применения). Ошибка ведет к тому, что

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

Давным-давно Дейкстра работал не только над алгоритмами на графах, но и над проблемами многопоточного кода. Ему удалось в одной из своих работ доказать, что минимальный базис, необходимый для любой синхронизации, сводится к двум объектам:

• блокировки (они же lock, mutex).
• условные переменные (condition variable).

Давайте на них и остановимся.

Ремарка о скорости. Питон - не первый спринтер на деревне. Если для программы на С выигрыш от правильного использования специфичного объекта синхронизации может быть существенным, то для Питона в целом они все примерно одинаковы.

Выбирать нужно более безопасный способ, не обращая внимания на быстродействие.

Кстати, это не относится к реализации CPython как такового - в коде интерпретатора присутствуют довольно экстравагантные блокировки, дающие выигрыш в пару процентов. До тех пор, пока их использует небольшая очень квалифицированная команда разработчиков - это оправдано.

Вторая ремарка: рассматриваемые объекты блокировки очень примитивны.

Откровенно говоря, не существует высокоуровневой теории мультипоточных программ. Если сравнивать с развитием языков, то уровень мультипоточного программирования сейчас находится где-то там, где были процедурные языки программирования. Объекты синхронизации схожи с оператором goto - использовать приходится, но всегда следует внимательно смотреть, что вы делаете.

Более высокоуровневые концепты, такие как monitor и active object хороши, но не решают всех проблем.

Языки вроде erlang, прекрасно справляющиеся с распараллеливанием задач на оптимальное количество потоков, останутся в своей узкой нише - примерно как lisp и все семейство языков функционального программирования навсегда остануться инструментом для маргиналов.

Парадигма многопоточных языков общего употребления все еще не создана.

Блокировки

Это - основа всего.

Простейший пример использования:

import threading

class Point(object):
    def __init__(self):
        self._mutex = threading.RLock()
        self._x = 0
        self._y = 0

    def get(self):
        with self._mutex:
            return (self._x, self._y)

    def set(self, x, y):
        with self._mutex:
            self._x = x
            self._x = y

Итак, имеем тривиальный класс. Это точка в двухмерном пространстве. Пусть координаты _x и _y начинаются с подчеркивания - чтобы программист, пожелавший менять их непосредственно немного задумался.

Еще есть _mutex типа RLock.

Публичная часть - два метода get и set.

Работает все это так: - при вызове метода берем блокировку через with self._mutex: - весь код внутри with блока будет выполнятся только в одном потоке. Другими словами, если два разных потока вызовут .get то пока первый поток не выйдет из блока второй будет его ждать - и только потом продолжит выполнение.

Зачем это все нужно? Координаты нужно менять одновременно - ведь точка это цельный объект. Если позволить одному потоку поменять x, а другой в это же время поправит y логика алгоритма может поломаться.

Есть и другой вопрос: зачем методу get блокировка? В приведенном примере она действительно не нужна. Но я всё же настоятельно рекомендую использовать блокировки даже для методов, вроде бы не изменяющих содержимое.

Во первых, это просто хорошая привычка. Иногда блокировки можно опускать для повышения производительности - но только после тщательного изучения побочных эффектов.

Во вторых (и это важнее) - блокировки при чтении позволяют корректно работать с объектами, части которых могут изменятся независимо.

Для примера возьмет цветную точку.

class ColoredPoint(Point):
    def __init__(self):
        super(ColoredPoint, self).__init__()
        self._color = 'green'

    @property
    def color(self):
        with self._mutex:
            return self._color

    @color.setter
    def color(self, val):
        with self._mutex:
            self._color = val

    def do(self, observer):
        with self._mutex:
            if self._color == 'red':
                observer(self.get())

Без блокировки в методе .do возможна ситуация, при которой один поток поменяет цвет, в второй в это же время изменит координаты. И тогда observer будет вызван с неправильными значениями.

В модуле threading существует еще и Lock - никогда его не используйте. Дело в том, что RLock (recursive lock) допускает повторную блокировку.

В примере для цветной точки это хорошо видно - .do берет блокировку, а затем вызывает метод .get, который берет эту блокировку еще раз.

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

Говоря о блокировках, нужно упомянуть и две проблемы, которые возникают при их использовании:

• Race condition: неправильное поведение из за отсутствия блокировки.
• Dead lock: взаимная блокировка.

Первый случай я уже описал: при отсутствии блокировки один поток меняет часть объекта, временно приводя его в некорректное состояние - а второй поток именно в это время пытается работать с этим некорректным на данный момент объектом.

Пример для взаимных блокировок:

a = threading.RLock()
b = threading.RLock()

def f():
    with a:
        # do something
        with b:
            # do something also

def g():
    with b:
        # do something
        with a:
            # do something also

При одновременно вызове f из одного потока и g из другого оба потока навсегда повиснут: первый будет ждать захвата b а второй, захватив b остановится на ожидании блокировки a.

Обратите внимание: эта проблема может проявляться не сразу и нелегко диагностируется/воспроизводится.

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

Еще один пример:

def h():
    try:
        a.acquire()
        b.acquire()
    # do something
    finally:
        a.release()
        b.release()

При одновременном вызове h возможна взаимная блокировка.

Общее правило, немного помогающее при проектировании многопоточных систем: блокировки всегда должны браться в одном и том же порядке, а освобождаться в противоположном. Использование оператора with помогает решить эту проблему.

Условные переменные

С блокировками все более или менее ясно. Нужен еще один объект синхронизации.

Рассмотрим очередь на фиксированное количество элементов. Когда очередь пуста - поток, желающий получить новый элемент должен ждать. Аналогично с переполненной очередью.

Если решать все только на блокировках, придется проверять - а не пуста ли очередь? И если таки пуста - подождать еще немного. Сколько времени ждать - сложный вопрос.

Очевидно, нужен какой-то сигнал от заполняющего очередь потока - ожидающему.

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

Несмотря на то, что Питон непосредственно поддерживает сигналы (threading.Event) - я настоятельно не рекомендую их использовать.

Проблема в следующем: поток послал сигнал о поступлении элемента в очередь.

Кто его получит, если получение нового элемента ожидают несколько потоков?

Один из них забрал данные и очередь снова пуста. Как нужно дожидаться следующего поступления?

Условные переменные (condition variables) совмещают сигналы с блокировками.

Рассмотрим пример:

class Queue(object):
    def __init__(self, size=5):
        self._size = size
        self._queue = []
        self._mutex = threading.RLock()
        self._empty = threading.Condition(self._mutex)
        self._full = threading.Condition(self._mutex)

    def put(self, val):
        with self._full:
            while len(self._queue) >= self._size:
                self._full.wait()
            self._queue.append(val)
            self._empty.notify()

    def get(self):
        with self._empty:
            while len(self._queue) == 0:
                self._empty.wait()
            ret = self._queue.pop(0)
            self._full.notify()
            return ret

У нас два почти симметричных метода - положить в очередь и взять из нее.

Разберем .get. Сначала берем блокировку.

Обратите внимание - блокиратор один на обе условные переменные. Это важно Дело в том, что _full и _empty взаимозависимы. Хотя threading.Condition позволяет не указывать блокиратор, создавая новый RLock автоматически - не поступайте так. Можно поймать race condition, о котором я говорил раньше. Гораздо надежней и наглядней делать все явно. В нашем случае race condition был бы в явном виде - пусть и скрытый из за GIL.

• Входим в блокиратор.
• Проверяем наше условие - обычно это всегда цикл while.
• Если условие не выполнилось - вызываем .wait(). Этот метод освободит блокировку и будет ждать извещения .notify. После получения сигнала выполнение продолжится при снова взятой блокировке. Еще раз, простыми словами.
  1. Проверка условия всегда выполняется в блокировке. Никто другой в это же время условие поменять не сможет (если все сделано правильно, конечно).
  2. Если условие не выполнилось - ждем опять, отдав управление операционной системе.
  3. При повторном вхождении у нас снова есть блокировка.
• Условие выполнилось - очередь не пуста. Как минимум один элемент в ней есть. Может быть и больше - нас сейчас это не волнует. Блокировка все еще есть.
• Получаем этот элемент и извещаем очередь, что она стала не до конца заполненной - т.е. в нее можно еще что-то положить (self._full.notify()).
• Возвращаем полученное. Выходя из with освобождаем блокировку - другие потоки могут работать с очередью дальше.

Итак, условные переменные - это способ синхронизировать доступ к объектам при помощи блокировки и при этом возможность послать сигнал ожидающим потокам.

Альтернатива оператору with.

Вместо

with lock:
    # do something

можно использовать более традиционную форму:

try:
    lock.acquire()
    # do domething
finally:
    lock.release()

Эти два подхода абсолютно эквивалентны, но первый на три строчки короче. Выбросить try/finally блок нельзя - при возникновении исключения блокировка должна быть всё равно отпущена. К слову, неснятая блокировка при выходе из функции - одна из самых распространенных ошибок. Будьте внимательны - а еще лучше научитесь писать так, чтобы всё получалось просто, ясно и правильно.

Нужно отметить, что .acquire позволяет указывать параметры: blocking и, начиная с Python 3.2 - timeout. Подробности их использования неплохо описаны в документации.

Продолжение - в третьей части.

Мультипоточность в Питоне. Часть 1 - потоки

Продолжение - здесь

Писать многопоточные программы сложно. К сожалению, время от времени это приходится делать.

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

Для начала рассмотрим минимально необходимый базис:

• работа с потоками
• синхронизация разделяемых объектов (тех, которые одновременно используются в нескольких потоках)

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

Потоки.

Потоки нужно

• создавать
• завершать
• дожидаться завершения

Изначально у каждой программы есть один поток - он же "главный". Этот поток создается операционной системой при запуске процесса.

С точки зрения программиста почти не отличается от созданных вручную. Практически же существуют некоторые особенности.

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

Создание потока.

Работой с потоком занимается threading.Thread (документация).

Меня захотят поправить - есть еще thread. Добрый совет - никогда его не используйте. Потому что!.. Цитата из документации: "The threading module provides an easier to use and higher-level threading API built on top of thread module."

Поток может быть создан двумя способами: передачей исполняемой функции в конструктор и наследованием.

Первый вариант:

def f(a, b, c):
    # do something
    pass

th = threading.Thread(name='th1', target=f, args=(1, 2), kwargs={'c': 3})

• name - имя потока. Ни на что не влияет, но может быть полезно при отладке.
• target - точка входа. Любой callable object - функция, связязанный метод класса или что-то еще
• args - позиционные аргументы
• kwargs - именованные аргументы.

Поток с именем 'th1' будет создан, но не запущен. После запуска будет вызвана функция f с параметрами a=1, b=2, c=3. Все аргументы могут быть опущены.

Второй вариант:

class MyThread(threading.Thread):
    def __init__(self, a, b, c):
        threading.Thread.__init__(self)
        self.a = a
        self.b = b
        self.c = c

    def run(self):
        # do something, using self.a, self.b, self.c
        pass

th = MyThread(1, 2, 3)

Результат практически тот же самый, но в новом потоке будет запущен метод run.

После того, как объект создан - его нужно запустить. В обоих случаях это делается через вызов

th1.start()

Все очень просто и понятно, верно?

Завершение потока.

Любой поток рано или поздно нужно завершить. Делается это простым выходом из функции потока. Не существует ПРАВИЛЬНОГО способа завершить поток снаружи. Это - принципиальное ограничение. Т.е. если вы хотите завершить поток из другого - просигнализируйте ему о своей просьбе (выставив флаг-переменную, например). В конце статьи я объясню, почему это важно.

Присоединение потока.

Культурные люди должны убирать за собой. Поэтому после сигнала о завершении нужно дождаться, когда поток реально закончит свою работу. Делается это так:

# signal to th1 for exiting
th1.join()

.join приостановит выполнение потока, вызвавшего ее, и будет ждать когда поток th1 завершит свое выполнение. Зачастую поток, стартовавший th1 его же и ждет - но бывают и исключения.

Почему нельзя "прибить" поток снаружи?

Если подумать, то все очень просто.

С точки зрения операционной системы программа (запущенный процесс) и потоки - ортогональные вещи.

Процесс владеет ресурсами - памятью, открытыми файлами, сокетами и подключениями к базе данных, окнами пользовательского интерфейса Типов ресурсов очень много. Грубо говоря, каждую переменную можно рассматривать как ресурс.

Поток - это нить исполнения. Он просто исполняет код параллельно с другими потоками. При этом поток ресурсами не владеет.

Открытый файл нужно закрыть, выделенную память вернуть обратно. Поэтому только правильно написанный код самого потока может корректно завершить самого себя.

Возникает еще один вопрос: а почему нельзя послать потоку исключение, чтобы он корректно свернулся - освободив при этом все что нужно?

Забегая вперед, вспомним об объектах синхронизации. Поток может быть в состоянии блокировки, когда его исполнение приостановлено в ожидании освобождения мьютекса (критической секции), занятого другим потоком.

Объекты блокировки - это в общем случае объекты ядра операционной системы. Т.е. блокированный поток нельзя "разблокировать", не затронув другие потоки.

Т.е. принудительное завершение потока - это потенциально опасная операция. В некоторых случаях все получается, иногда происходит утечка ресурсов.

Несмотря на то, что в низкоуровневом API операционной системы почти всегда есть способ аварийно завершить поток, в Питоне он не реализован.

Давайте просто писать "правильно", к чему нас и подталкивают.

Зачем нужно ждать завершения потока?

Хорошо, поток создан и запущен. В свое время ему послана просьба завершить работу. Зачем ждать подтверждения?

Нет, это не паранойя.

Во первых, так заведено в Питоне (корни растут из posix threads) - процесс завершается только тогда, когда все его потоки завершены.

Т.е. окончание работы главного потока программы еще ни о чем не говорит - Питон будет ждать окончания работы остальных потоков. Это выглядит так, будто в конце программы добавили .join на каждый работающий поток (еще одна причина делать это явно - ведь явное лучше неявного, верно?)

Откровенно говоря, есть лазейка, позволяющая это правило обойти. Достаточно указать

th1.setDaemon(True)

и можно о нем забыть. Отличие потока-демона от "стандартного" только в том, что демоны "не считаются" при завершении процесса.

Вуаля, решение найдено и оно работает? Не совсем.

Дело в том, что программа завершается не мгновенно. После того, как закончился главный поток с точки зрения программиста, Питон начинает чистить свои ресурсы, разрушая интерпретатор.

Закрываются файлы, уничтожаются модули.

И такой безобидный код как

import logging

def f():
    a = logging.getLogger("a.b.c").debug("f is called")

Может "поломаться", потому что модуль logging уже не существует, вернее - он частично разрушен сборщиком мусора.

При аккуратном программировании можно обойти все проблемы, связанные с саморазрушением интерпретатора - но оно вам нужно?

Проще и безопасней писать "по правилам".

Заключение.

С потоками покончено.

• Потоки можно создавать и запускать.
• Можно просить их закончить свою работу, но нельзя приказывать.
• Завершения потока нужно дожидаться.

Но мультипоточное программирование только начинается.

Основная проблема здесь следующая: потоки должны взаимодействовать.

Или, другими словами, изменять состояние разделяемых между ними объектов. Несколько потоков могут пытаться изменить один объект одновременно, с непредсказуемым результатом.

Объекты могут быть сложными, и запись в объект "пользователь" фамилии одним потоком, а имени другим может привести к неожиданному результату. Существуют и более "разрушительные" примеры.

Для решения проблемы придуманы объекты синхронизации. О них - в следующей статье.