Как реализовать многопоточность на Python

Как работают потоки в Python

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

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

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

В Python, создание потоков осуществляется с использованием модуля `threading`. Этот модуль предоставляет множество функций и инструментов для управления потоками, позволяя вам создавать, запускать и управлять потоками выполнения в ваших приложениях. Работа с потоками в Python довольно интуитивная и удобная, что делает его привлекательным выбором для разработчиков.

Однако при работе с Python важно учитывать глобальную блокировку GIL. GIL (Global Interpreter Lock) — это механизм, который ограничивает выполнение Python кода только одним потоком в один момент времени. Это означает, что, даже если у вас есть несколько потоков в вашем приложении, они не могут полностью параллельно выполнять Python код. Это ограничение GIL может стать препятствием для достижения полной параллелизации в некоторых сценариях.

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

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

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

Параллелизм на уровне потоков TLP Python

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

1. threading — этот модуль предназначен для управления потоками выполнения.
2. queue — он используется для эффективной организации очередей задач.
3. multiprocessing — позволяет управлять процессами.

На данном этапе, мы сосредотачиваем внимание исключительно на первом пункте списка.

Как с Python выполнять несколько операций одновременно

Метод 1: подход с "функциональным" синтаксисом

Для инициации и управления потоками в Python, мы прибегаем к использованию модуля `threading`. Процесс начинается с импорта класса `Thread` следующим образом:

«`python

from threading import Thread

«`

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

«`python

variable = Thread(target=function_name, args=(arg1, arg2,))

«`

Первый аргумент, `target`, определяет "целевую" функцию, которая будет выполнена внутри потока. Далее следует список аргументов для этой функции. Если аргументы имеют фиксированное положение, как, например, в случае, когда один аргумент — делимое, а другой — делитель в уравнении, то они передаются как `args=(x, y)`. Если же вам требуются аргументы, заданные в виде пар "ключ-значение", вы можете использовать запись вида `kwargs={'prop': 120}`.

Для удобства отладки вы также можете присвоить каждому новому потоку имя. Это можно сделать, добавив параметр `name="Имя потока"` в параметры функции. По умолчанию, имя установлено в `None`. Кроме того, потоки можно группировать с помощью параметра `group`, который также по умолчанию равен `None`.

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

Вот как это выглядит на практике:

«`python

# coding: UTF-8

from threading import Thread

def write_lines_to_file(file_name, num_lines):

    with open(file_name, 'w') as file:

        for i in range(num_lines):

            if num_lines > 500:

                file.write('МногоБукв\n')

            else:

                file.write('МалоБукв\n')

thread1 = Thread(target=write_lines_to_file, args=('file1.txt', 200,))

thread2 = Thread(target=write_lines_to_file, args=('file2.txt', 1000,))

thread1.start()

thread2.start()

thread1.join()

thread2.join()

«`

Функция `start()` запускает созданный поток, как, вероятно, вы уже догадались. Метод `join()` служит для ожидания завершения выполнения потока, что особенно важно, когда имеются открытые файлы и другие ресурсы, которые нужно корректно освободить. Это можно представить как "Выключить свет, уходя." Завершать потоки внутри них самих и в явно заданный момент времени более надежно, чем ожидать случайные вмешательства снаружи. Метод `join()` также может принимать аргумент — количество секунд, на которое блокируется поток, прежде чем продолжает выполнение задачи.

Метод 2: подход с использованием классов

Для управления потоками с более сложным поведением обычно создают отдельные классы, которые наследуются от класса `Thread` в модуле `threading`. В таком случае, программа потока определяется в методе `run()`, созданного класса. Этот подход знаком многим из мира Java.

«`python

#coding: UTF-8

import threading

class MyThread(threading.Thread):

    def __init__(self, num):

        super().__init__(name="threddy" + num)

        self.num = num

    def run(self):

        print("Thread", self.num)

thread1 = MyThread("1")

thread2 = MyThread("2")

thread1.start()

thread2.start()

thread1.join()

thread2.join()

«`

Как высчитать оптимальное количество потоков в Питоне

Для эффективного управления потоками, важно иметь возможность наблюдать за их поведением. В модуле `threading` предоставлены специальные методы:

— `current_thread()` — позволяет определить, какой поток вызвал данную функцию.

— `active_count()` — позволяет подсчитать количество активных экземпляров класса `Thread` в данный момент.

— `enumerate()` — возвращает список всех работающих потоков.

Также существуют методы для управления потоками через объект класса:

— `is_alive()` — предоставляет информацию о том, работает ли поток в данный момент (возвращает `True` или `False`).

— `getName()` — позволяет получить имя потока.

— `setName(any_name)` — устанавливает имя потока.

Каждый поток, пока он активен, имеет уникальный идентификационный номер, который хранится в переменной `ident`. Например:

«`python

thread1.start()

print(thread1.ident)

«`

Вы также можете использовать таймер для задержки операций в вызываемых потоком функциях. В классе `Timer` из модуля `threading` всего два аргумента: время задержки в секундах и функция, которую нужно выполнить по истечении этого времени.

«`python

import threading

print("Waiting…")

def timer_test():

    print("The timer has done its job!")

tim = threading.Timer(5.0, timer_test)

tim.start()

«`

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

Что такое фоновые потоки в Питоне

Обычно Python-приложение не завершается, пока работает хотя бы один его поток. Но есть особые потоки, которые не препятствуют закрытию программы и завершаются вместе с ней. Эти потоки называются демонами (daemons). Вы можете проверить, является ли поток демоном с помощью метода `isDaemon()`. Если поток является демоном, метод вернет `True`.

Вы можете назначить поток демоном при создании, используя параметр "daemon=True" или аргумент в инициализаторе класса:

«`python

thread0 = Thread(target=target_func, kwargs={'x': 10}, daemon=True)

«`

Вы также можете превратить уже существующий поток в демона с помощью метода `setDaemon(daemonic)`.

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

Будущее параллелизма GIL и Python

Нельзя просто так взять и воспользоваться всеми преимуществами многопоточности в Python. Ваш путь будет усеян преградами, и, возможно, самой значительной из них будет огромная преграда, известная как Глобальный Шлюз (Global Interpreter Lock или GIL). Этот шлюз ограничивает многопоточность на уровне интерпретатора Python. Технически, это один для всех мьютекс, который создается по умолчанию. Это концепция, которой нет ни в C, ни в Java.

Цель этого шлюза — разрешать выполнение потоков поочередно, чтобы они не мешали друг другу, как уличные гонщики, и не создавали угрозу стабильности работы интерпретатора.

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

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

Чтобы избежать подобных проблем, GIL в определенные моменты, предсказуемым образом (по умолчанию каждые 5 миллисекунд для Python 3.2+), отправляет команду отключения активного потока с целью предостеречь его от мешанины в других потоках.

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

Приведенный пример в статье "Understanding Python GIL" от технического директора компании Gaglers Inc. и опытного разработчика, Четана Гиридхара, показывает:

«`python

from datetime import datetime

import threading

def factorial(number):

    fact = 1

    for n in range(1, number + 1):

        fact *= n

    return fact

number = 100000

thread = threading.Thread(target=factorial, args=(number,))

startTime = datetime.now()

thread.start()

thread.join()

endTime = datetime.now()

print("Время выполнения: ", endTime — startTime)

«`

Этот код вычисляет факториал числа 100 000 и показывает, сколько времени машина затратила на выполнение задачи. При тестировании на одном ядре и одном потоке вычисления заняли 3,4 секунды. Однако, когда Четан создал и запустил второй поток, вычисление факториала на двух ядрах заняло 6,2 секунды. По логике вещей, скорость вычислений не должна существенно измениться, но результат показывает, что выполнение задачи с двумя потоками заняло вдвое больше времени.

Глобальный шлюз — это наследие времен, когда программисты боролись за реализацию многозадачности, и у них не всегда получалось. Однако существуют вопросы: почему он все еще актуален сегодня, когда у нас есть процессоры с множеством ядер? Гвидо ван Россум, создатель Python, объясняет, что без GIL C-расширения для Python работать нормально не смогут. Кроме того, отказ от GIL может привести к ухудшению производительности однопоточных приложений, что не является желательным результатом, особенно в сравнении с Python 2, который до сих пор популярен.

Что такое GIL в Python

С технической точки зрения, GIL (глобальная блокировка интерпретатора) — это множество данных, которые управляют установкой этого флага, а также другие служебные переменные, такие как интервал переключения. Все эти вещи хранятся в структуре `_gil_runtime_state`:

«`c

struct _gil_runtime_state {

    /* микросекунды (Python API использует секунды) */

    unsigned long interval;

    /* Последний PyThreadState, удерживавший GIL. Это помогает нам узнать,

       был ли кто-то еще запланирован после того, как мы отпустили GIL. */

    _Py_atomic_address last_holder;

    /* Заблокирован ли GIL? (-1, если не инициализирован). Это

       атомарное, так как может быть прочитано без блокировки в ceval.c. */

    _Py_atomic_int locked;

    /* Количество переключений GIL с момента начала. */

    unsigned long switch_number;

    /* Эта условная переменная позволяет одному или нескольким потокам ждать

       освобождения GIL. Кроме того, мьютекс также защищает

       вышеуказанные переменные. */

    PyCOND_T cond;

    PyMUTEX_T mutex;

#ifdef FORCE_SWITCHING

    /* Эта условная переменная помогает потоку, освобождающему GIL, ждать,

       пока поток, ожидающий GIL, будет запланирован и получит GIL. */

    PyCOND_T switch_cond;

    PyMUTEX_T switch_mutex;

#endif

};

«`

Структура `_gil_runtime_state` является частью глобального состояния и хранится в структуре `_ceval_runtime_state`, которая в свою очередь является частью `_PyRuntimeState`, к которому имеют доступ все потоки Python:

«`c

struct _ceval_runtime_state {

    _Py_atomic_int signals_pending;

    struct _gil_runtime_state gil;

};

typedef struct pyruntimestate {

    // …

    struct _ceval_runtime_state ceval;

    struct _gilstate_runtime_state gilstate;

    // …

} _PyRuntimeState;

«`

Следует отметить, что `_gilstate_runtime_state` — это отдельная структура, отличная от `_gil_runtime_state`. Она содержит информацию о потоке, удерживающем GIL:

«`c

struct _gilstate_runtime_state {

    /* Флаг для отключения PyGILState_Check().

       Если установлено значение, отличное от нуля, PyGILState_Check() всегда вернет 1. */

    int check_enabled;

    /* Предполагая, что текущий поток удерживает GIL, это

       PyThreadState для текущего потока. */

    _Py_atomic_address tstate_current;

    /* Один PyInterpreterState, используемый этим процессом

       реализация GILState. */

    /* TODO: Учитывая interp_main, возможно, можно избавиться от этой ссылки. */

    PyInterpreterState *autoInterpreterState;

    Py_tss_t autoTSSkey;

};

«`

Наконец, есть структура `_ceval_state`, которая является частью `PyInterpreterState`. Она хранит флаги `eval_breaker` и `gil_drop_request`:

«`c

struct _ceval_state {

    int recursion_limit;

    int tracing_possible;

    /* Эта единая переменная консолидирует все запросы на выход из

       быстрого пути в цикле оценки. */

    _Py_atomic_int eval_breaker;

    /* Запрос на отпуск GIL */

    _Py_atomic_int gil_drop_request;

    struct _pending_calls pending;

};

«`

API Python/C предоставляет функции `PyEval_RestoreThread()` и `PyEval_SaveThread()` для захвата и освобождения GIL. Эти функции также отвечают за настройку `gilstate->tstate_current`. Фактически, весь этот процесс выполняют функции `take_gil()` и `drop_gil()`. Они вызываются потоком, удерживающим GIL, когда он приостанавливает выполнение байт-кода.

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

Как остановить threading

Шлюз, ограничивающий параллельное выполнение кода в Python, может быть временно выключен. Для этого необходимо "отвлечь" интерпретатор Python, вызывая функции из внешних библиотек или взаимодействуя с операционной системой. Например, GIL может быть выключен на время операций ввода-вывода, таких как сохранение или открытие файлов. Вспомните пример с записью строк в файлы? Как только вызываемая функция завершит свою работу и вернет управление коду Python или интерфейсу Python C API, GIL восстанавливается.

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

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

Numba для математических расчетов

Numba — это библиотека, которая динамически компилирует Python-код "на лету" в машинный код для выполнения на процессоре и GPU. Этот вид компиляции называется JIT ("Just in Time"). Он помогает оптимизировать производительность программ, ускоряя выполнение циклов и компиляцию функций при первом запуске.

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

Например, для математических вычислений библиотеку можно удобно использовать в сочетании с NumPy. Допустим, вам нужно сложить одномерные массивы поэлементно:

«`python

import numba

import numpy as np

@numba.jit

def arr_sum(x, y):

    result_arr = np.empty_like(x)

    for i in range(len(x)):

        result_arr[i] = x[i] + y[i]

    return result_arr

«`

Метод `np.empty_like()` принимает массив и возвращает (но не инициализирует) другой массив, который соответствует по форме и типу исходному. Добавив аннотацию `@jit` в начале кода, вы можете значительно ускорить выполнение операции — более чем в 100 раз! Если вас заинтересовал этот подход, рекомендуется изучить, какие результаты можно достичь, сравнив скорость выполнения математических расчетов с использованием разных библиотек для Python.

PyCUDA и Numba для графики

Numba также может пригодиться в графических вычислениях. Она позволяет взаимодействовать с программной моделью CUDA для визуализации научных данных, выполнения алгоритмов и получения информации о GPU. И снова, в этом контексте, мы сталкиваемся с многопоточностью.

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

«`python

from numba import cuda

@cuda.jit

def call_for_kernel(io_arr):

    # Идентификатор потока в одномерном блоке

    thread_x = cuda.threadIdx.x

    # Идентификатор блока в одномерной сетке

    thread_y = cuda.blockIdx.x

    # Число потоков в блоке (ширина блока)

    block_width = cuda.blockDim.x

    # Находим положение в массиве

    t_position = thread_x + thread_y * block_width

    if t_position < io_arr.size:  # Убеждаемся, что не выходим за границы массива

        io_arr[t_position] *= 2  # Выполняем вычисления

«`

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

Когда многопоточность в Python оправдана

Теперь давайте разберемся, когда стоит запустить функцию в отдельном потоке в Python и когда она принесет больше пользы, чем вреда:

1. Для длительных и независимых операций ввода-вывода. Например, если вам нужно обрабатывать большое количество отдельных запросов с большой задержкой на ожидание. В этом случае лучше распараллелить задачу.
2. Когда вы хотите экономить время через параллельное выполнение вычислений. Если операции занимают более 1 миллисекунды, и вы хотите сэкономить время путем их параллельного выполнения. Если операции выполняются за 1 мс, многопоточность может оказаться менее эффективной из-за высоких накладных расходов.
3. Когда количество потоков не превышает количество ядер. Если у вас больше потоков, чем ядер, то параллельное выполнение всех потоков не будет возможным, и в этом случае вы можете больше потерять, чем выиграть.

Когда лучше работать с одним потоком

Существуют ситуации, когда лучше ограничиться одним потоком:

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