Python копирование файлов в несколько потоков

Обновлено: 06.05.2024

Способы реализации параллельных вычислений в программах на Python.

Что такое параллелизм?

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

Используя многопоточность threading , позволяя нескольким потокам работать по очереди.
Используя несколько ядер процессора multiprocessing . Делать сразу несколько вычислений, используя несколько ядер процессора. Это и называется параллелизмом.
Используя асинхронный ввод-вывод с модулем asyncio . Запуская какую то задачу, продолжать делать другие вычисления, вместо ожидания ответа от сетевого подключения или от операций чтения/записи.

Разница между потоками и процессами.

Поток threading - это независимая последовательность выполнения каких то вычислений. Поток thread делит выделенную память ядру процессора, а так же его процессорное время со всеми другими потоками, которые создаются программой в рамках одного ядра процессора. Программы на языке Python имеют, по умолчанию, один основной поток. Можно создать их больше и позволить Python переключаться между ними. Это переключение происходит очень быстро и кажется, что они работают параллельно.

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

Асинхронный ввод-вывод не является ни потоковым ( threading ), ни многопроцессорным ( multiprocessing ). По сути, это однопоточная, однопроцессная парадигма и не относится к параллельным вычислениям.

У Python есть одна особенность, которая усложняет параллельное выполнение кода. Она называется GIL, сокращенно от Global Interpreter Lock. GIL гарантирует, что в любой момент времени работает только один поток. Из этого следует, что с потоками невозможно использовать несколько ядер процессора.

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

Что такое условия гонки и потокобезопасность?

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

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

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

Алгоритм планирования доступа потоков к общим данным.

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

Например, есть общая переменная a = 2 . Теперь предположим, что есть два потока, thread_one и thread_two . Они выполняют следующие операции:

Если поток thread_one получит доступ к общей переменной a первым и thread_two вторым, то результат будет 12:

или наоборот, сначала запустится thread_two , а затем thread_one , то мы получим другой результат:

Таким образом очевидно, что порядок выполнения операций потоками имеет значение

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

Есть еще худший вариант развития событий, который может произойти без встроенной в Python блокировки потоков GIL . Например, если оба потока начинают читать глобальную переменную a одновременно. Оба потока увидят, что a = 2 , а дальше, в зависимости от того какой поток произведет вычисления последним, в конечном итоге и будет равна переменная a (4 или 6). Не то, что ожидалось!

Исследование разных подходов к параллельным вычислениям в Python.

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

Функция heavy() представляет собой вложенный цикл, который выполняет возведение в степень. Это функция связана со скоростью ядра процессора производить математические вычисления. Если понаблюдать за операционной системой во время выполнения функции, то можно увидеть загрузку ЦП близкую к 100%.

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

Однопоточный режим работы.

Каждая программа Python имеет по крайней мере один основной поток. Ниже представлен пример кода для запуска функции heavy() в одном основном потоке одного ядра процессора, который производит все операции последовательно и будет служить эталоном с точки зрения скорости выполнения:

Использование потоков threading .

В следующем примере будем использовать несколько потоков для выполнения функции heavy() . Так же произведем 80 циклов вычислений. Для этого разделим вычисления на 4 потока в каждом из которых запустим 20 циклов:

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

Если бы у Python не было GIL, то вычисления функции heavy() происходили быстрее, а общее время выполнения программы стремилось к времени выполнения однопоточной программы. Причина, по которой многопоточный режим в данном примере не будет работать быстрее однопоточного - это вычисления, связанные с процессором и заключаются в GIL!

Если бы функция heavy() имела много блокирующих операций, таких как сетевые вызовы или операции с файловой системой, то применение многопоточного режима работы было бы оправдано и дало огромное увеличение скорости!

Это утверждение можно проверить смоделировав операции ввода-вывода при помощи функции time.sleep() .

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

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

Использование многопроцессорной обработки multiprocessing .

Теперь попробуем настоящую параллельную обработку с использованием модуля multiprocessing . Модуль multiprocessing во многом повторяет API модуля threading , по этому изменения в коде будут незначительны.

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

Код выполнился почти в 5 раз быстрее. Это прекрасно демонстрирует линейное увеличение скорости вычислений от количества ядер процессора.

Использование многопроцессорной обработки с пулом.

Можно сделать предыдущую версию программы немного более элегантной, используя multiprocessing.Pool() . Объект пула, управляет пулом рабочих процессов, в который могут быть отправлены задания. Используя метод Pool.starmap() , можно произвести инициализацию функции sequential () для каждого процесса.

В целях эксперимента в функции запуска пула процессов pooled(core) предусмотрено ручное указание количества ядер процессора. Если не указывать значение core , то по умолчанию будет использоваться количество ядер процессора вашей системы, что является разумным выбором:

Из результатов работы видно, что время работы незначительно увеличилось.

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

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

Как работает многопоточность

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

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

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

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

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

Можно ли считать threading многопоточным?

В Python используется GIL (Global Interpreter Lock), который однопоточный. Все потоки, которые создаются с помощью threading будут работать внутри потока GIL. В связи с этим они будут обрабатываться только одним ядром. Ни о какой работе одновременно на нескольких физических ядрах процессора не может быть и речи.

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

В чём преимущества тогда модуля Threading по сравнению с Multiprocessing? Рассмотрим их:

Подключение библиотеки threading

Threading – это стандартный модуль, который поставляется вместе с интерпретатором. Программисту не нужно устанавливать его, достаточно просто подключить модуль с помощью команды:

Работать с потоками можно, создавая экземпляры класса Thread. Чтобы создать отдельный, поток нужно создать экземпляр класса и применить к нему метод start() . Вот пример:

Здесь мы функцию mydef запустили в отдельном потоке. В качестве аргументов функции передали числа 1 и 2.

threading.Thread()

Эта конструкция позволяет создать новый поток, создав экземпляр класса Thread. Вот как выглядят её аргументы:

Она принимает аргументы:

Рассмотрим их подробнее:

Демоны

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

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

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

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

Методы для работы с потоками

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

start()

Он используется для запуска созданного потока. После использования threading.Thread() создаётся новый поток, однако он неактивен. Для того чтобы он начал работу, используется метод start().

Здесь пока мы не вызвали метод start , функция myfunc не будет запущена.

Этот метод блокирует выполнение потока, который его вызвал, до тех пор пока не завершится поток, метод которого был вызван. То есть если в потоке thread1 был вызван метод потока thread2: thread2.join(), то поток thread1 будет приостановлен до тех пор, пока выполнение thread2 не завершится.

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

У метода join() есть аргумент timeout . По умолчанию он имеет значение None, но программист может передать в него число с плавающей точкой.

Если аргумент имеет значение по умолчанию, то выполнение потока приостанавливается, пока выполняется поток метода.

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

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

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

Здесь мы делаем поток демоническим, чтобы программа не дожидалась окончания выполнения функции. Подключаем модуль time, для того, чтобы сделать задержку в функции на 2.5 секунд. После старта потока, мы приостанавливаем основной поток на 0.125 секунд. Потом выполняем проверку is_alive(). Если выведет True, значит поток не закончил выполнение за 0.125 секунды.

В этом методе описываются операции, выполняемые потоком. Он используется, когда явно создается экземпляр класса. Пример:

Остановка потока

В бесконечном цикле делать постоянно её проверку и если она True, то завершать его.
Не использовать функции, которые могут блокировать выполнение на длительное время. Всегда использовать timeout.

Вот пример такой программы:

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

Состояние гонки

Состояние гонки или race condition – это ошибка, возникающая при неправильном проектировании многопоточной программы. Она возникает тогда, когда несколько потоков обращаются к одним и тем же данным. Например, переменная хранит число, которое пытаются одновременно изменить потоки thread1 и thread2, что приводит к непредсказуемым результатам или ошибке.

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

deadlock

При использовании Lock возникает серьезная проблема, которая приводит к полной остановки работы программы. Если вызвать метод acquire(), а объект Lock уже заблокирован, то вызвавший acquire() поток будет ждать, пока заблокировавший объект поток не вызовет release().

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

Самоблокировку можно предотвратить, если удалить лишний вызов acquire(), но это не всегда возможно. Самоблокировка может происходить из-за следующий вещей:

Возникновение ошибок, когда Lock остаётся заблокированным.
Неправильное проектирование программы, когда одна функция вызывается другой функцией, у которой отсутствует блокировка.

В случае возникновения ошибок достаточно воспользоваться конструкцией try-finally или оператором with.

Вот пример с with:

Конструкция try-finally позволяет удалять блокировку даже в случае возникновения ошибок, что позволяет избежать deadblock. Пример:

Конструкция try-finally гарантирует, что код в finally будет исполнен всегда, независимо от ошибок и результатов блока try.

Однако это не работает в случае самоблокировки из-за неправильного проектирования программы. Для этого был создан объект RLock.

RLock

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

Данный код будет работать, но его проблема заключается в том, что при вызове функции both_parts , в ней вызываются функции part1 и part2 . Между вызовами этих функций может получить доступ к данным какой-нибудь другой поток и их поменять. А что делать, если нужно избежать изменения другим потоком?

Чтобы решить проблему, нужно заблокировать lock1 и в both_parts , перепишем её:

Идея проста: внешняя both_parts блокирует поток на время выполнения функций part1 и part1 . Каждая из функций также блокирует поток для суммирования своей части объекта. Однако объект Lock не позволит этого сделать, этот код приведет к полному зависанию программы, потому что для Lock нет разницы, где в потоке был вызван acquire().

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

Использовать RLock нужно для управления вложенным доступом к разделяемым объектам. Чтобы решить возникшую проблему с Lock в коде выше, достаточно заменить строчку « lock1 = threading.Lock() » на « lock1 = threading.RLock() ».

Также следует помнить, что, хотя и можно вызывать acquire() несколько раз, метод release() нужно вызвать столько же раз. При каждом вызове acquire() уровень рекурсии увеличивается на единицу, соответственно при каждом вызове release() он уменьшается на единицу.

Передача данных с помощью очередей (Queue)

Библиотеке queue содержит все необходимые инструменты для передачи данных между потоками и реализует нужные механизмы блокировки.

Класс Queue реализует очередь FIFO, который работает так: первый элемент, который пошел в очередь, первым и выйдет из неё. Эту очередь можно сравнить с вертикальной полой трубой, в которую сверху бросают элементы.

Queue имеет параметр maxsize, принимающий только целочисленные значения. Он указывает максимальное количество элементов, которое можно поместить в очередь. Когда максимум достигается, добавление в очередь элементов блокируется, пока в ней не освободится место. Если maxsize принимает значение <= 0, то очередь является бесконечной.

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

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

qsize()

Возвращает примерный размер очереди. Важно понимать две вещи:

Если qsize() больше нуля, следующий метод get() всё равно может быть заблокирован.
Если qsize() меньше maxsize, следующий метод put() может быть заблокирован.

Это может возникнуть из-за того что к очереди могут обратиться другие потоки и получить/записать данные сразу после того как вы получили её размер.

empty()

Метод проверяет, содержится ли что-то в очереди. Если очередь пуста, возвращается True, если очередь содержит элементы, возвращается False.

Как и с методом qsize(), возврат True или False не гарантирует, что следующий метод put() или get() не будут заблокированы.

Проверяет, заполнена ли очередь. Если очередь заполнена, возвращает True, иначе возвращает False.

Как и в предыдущих методах, возврат True или False не даёт гарантий, что put() и get() не будут заблокированы.

Метод помещает новый объект в очередь, имеет обязательный аргумент item и два необязательных аргумента: block = True и timeout = None.

В зависимости от указанных аргументов, ожидание места в очереди будет вести себя по-разному:

Если аргумент block имеет значение True, а timeout — None, объект, который нужно загрузить в очередь, будет бесконечно ждать свободного места.
При timeout больше нуля, ожидание свободного места будет длиться не дольше указанного числа секунд, если за это время свободного места так и не появилось, возбудится исключение.
Если block имеет значение False, аргумент timeout игнорируется, и элемент можно поместить в очередь, только если есть свободное место, иначе сразу же возбуждается исключение.

Вот пример создания очереди на Python и добавления в неё элемента:

put_nowait()

Эквивалентно вызову put(item, False). То есть помещает элемент в очередь, только если есть место, иначе вызывает исключение.

Удаляет и возвращает элемент из очереди. Имеет два необязательных аргумента: block = True и timeout = None.

В зависимости от значений аргументов ожидание объекта ведёт себя по разному:

Если аргументы имеют значению по умолчанию, метод ожидает объект из очереди до тех пор, пока тот не станет доступен.
При timeout — положительное число, то объект из очереди ожидается определенное время, по истечении которого вызовется исключение.
Если block имеет значение False, элемент возвращается, только если он доступен, иначе вызывается исключение (аргумент timeout игнорируется).

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

get_nowait()

Эквивалентно вызову get(False).

task_done()

Этот метод работает в связке с методом join().

Метод показывает, что ранее поставленная задача была выполнена. После получения каждого элемента из очереди, допустим с помощью get(), нужно вызывать task_done(), чтобы уменьшить счётчик задач. Ниже описан метод join с примером

Если task_done() вызывается больше раз, чем количество элементов, помещенное в очередь, то возбуждается исключение ValueError.

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

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

Этот пример для того чтобы показать работу join и task_done. Здесь всё происходит в одном потоке. Обычно пишет в очередь один поток данные, потом ждёт когда их обработают с помощью join. Другой же поток при получении каждого нового значения вызывает task_done.

Пример программы

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

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

Полезные инструменты модуля threading

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

Semaphore

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

Здесь привёл два варианта получения доступа к данным:

C помощью acquire и release.
Используя with.

Timer

Этот класс позволяет контролировать время запуска какого-либо действия. Timer является подклассом Thread.

Вот его аргументы:

Таймеры запускаются также, как и потоки, с помощью метода start(). Их можно остановить, используя метод cancel().

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

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

Здесь функция myfunc выполнится через 4 секунды после вызова метода start().

Barrier

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

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

Так выглядят его аргументы:

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

Здесь выставляю barrier на 2 вызова wait. То есть, для того, чтобы выполнился код после wait, wait должен быть вызван в 2 потоках. В данном случае функция myfunc сразу запускается в потоке, но она сразу не выведет 'отработал barrier' в консоль, а дождётся когда в основном потоке будет вызван wait тоже.

Event

Event представляет собой простой механизм реализации связи между потоками: один поток даёт сигнал о событии, другие ожидают этого сигнала.

Объект события управляет внутренним флагом, который может быть установлен в True или False с помощью методов set() и clear(). Также есть методы is_set(), которым можно проверить состояние внутреннего флага. С помощью метода wait(timeout=None) можно ждать пока не выставлен флаг в True. Так же при необходимости можно задать время ожидания.

Вот пример:

Возможность управления потоками в Python – это мощный инструмент в разработке больших программ. Для работы с ними используется модуль Threading и библиотека queue в связке с ним.

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

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

Небольшое предупреждение для тех, кто впервые слышит о параллельных вычислениях. Что такое поток и чем он отличается от процесса, мы выяснили в статье «Внутри процесса: многопоточность и пинг-понг mutex'ом». Тогда мы приводили примеры на Java, но теоретические основы многопоточности верны и для Python. Совпадают, в том числе, механизмы синхронизации потоков: семафоры, взаимные исключения (mutex), условия, события. Поэтому сегодня сделаем акцент на особенностях Python, его механизмах и инструментах, связанных с многопоточностью.

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

threading — для управления потоками.
queue — для организации очередей.
multiprocessing — для управления процессами.

Пока нас интересует только первый пункт списка.

Как создавать потоки в Python

Метод 1 — «функциональный»

Для работы с потоками из модуля threading импортируем класс Thread. В начале кода пишем:

После этого нам будет доступна функция Thread() — с ней легко создавать потоки. Синтаксис такой:

Первый параметр target — это «целевая» функция, которая определяет поведение потока и создаётся заранее. Следом идёт список аргументов. Если судьбу аргументов (например, кто будет делимым, а кто делителем в уравнении) определяет их позиция, их записывают как args=(x,y). Если же вам нужны аргументы в виде пар «ключ-значение», используйте запись вида kwargs=.

Ради удобства отладки можно также дать новому потоку имя. Для этого среди параметров функции прописывают name=«Имя потока». По умолчанию name хранит значение null. А ещё потоки можно группировать с помощью параметра group, который по умолчанию — None.

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

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

Метод 2 — «классовый»

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

Стандартные методы работы с потоками

Чтобы управлять потоками, нужно следить, как они себя ведут. И для этого в threading есть специальные методы:

current_thread() — смотрим, какой поток вызвал функцию;

active_count() — считаем работающие в данный момент экземпляры класса Thread;

enumerate() — получаем список работающих потоков.

Ещё можно управлять потоком через методы класса:

is_alive() — спрашиваем поток: «Жив ещё, курилка?» — получаем true или false;

getName() — узнаём имя потока;

setName(any_name) — даём потоку имя;

У каждого потока, пока он работает, есть уникальный идентификационный номер, который хранится в переменной ident.

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

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

Потусторонние потоки

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

Назначить поток демоном можно при создании — через параметр “daemon=True” или аргумент в инициализаторе класса.

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

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

Приключение начинается. У древнего шлюза

Питон слывёт дружелюбным и простым в общении, но есть у него причуды. Нельзя просто взять и воспользоваться всеми преимуществами многопоточности в Python! Дорогу вам преградит огромный шлюз… Даже так — глобальный шлюз (Global Interpreter Lock, он же GIL), который ограничивает многопоточность на уровне интерпретатора. Технически, это один на всех mutex, созданный по умолчанию. Такого нет ни в C, ни в Java.

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

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

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

Чтобы такого не было, GIL в предсказуемый момент (по умолчанию раз в 5 миллисекунд для Python 3.2+) командует отработавшему потоку: «СПАААТЬ!» — тот отключается и не мешает проезжать следующему желающему. Даже если желающего нет, блокировщик всё равно подождёт, прежде чем вернуться к предыдущему активному потоку.

Благодаря шлюзу однопоточные приложения работают быстро, а потоки не конфликтуют. Но, к сожалению, многопоточные программы при таком подходе выполняются медленнее — слишком много времени уходит на регулировку «дорожного движения». А значит обработка графики, расчет математических моделей и поиск по большим массивам данных c GIL идут неприемлемо долго.

В статье «Understanding Python GIL»технический директор компании Gaglers Inc. и разработчик со стажем Chetan Giridhar приводит такой пример:

Код вычисляет факториал числа 100 000 и показывает, сколько времени ушло у машины на эту задачу. При тестировании на одном ядре и с одним потоком вычисления заняли 3,4 секунды. Тогда Четан создал и запустил второй поток. Расчет факториала на двух ядрах длился 6,2 секунды. А ведь по логике скорость вычислений не должна была существенно измениться! Повторите этот эксперимент на своей машине и посмотрите, насколько медленнее будет решена задача, если вы добавите thread2. Я получила замедление ровно вдвое.

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

«Нормальные герои всегда идут в обход»

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

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

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

Numba для математики

Numba — динамически, «на лету» компилирует Python-код, превращая его в машинный код для исполнения на CPU и GPU. Такая технология компиляции называется JIT — “Just in time”. Она помогает оптимизировать производительность программ за счет ускорения работы циклов и компиляции функций при первом запуске.

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

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

Метод nupmy.empty_like() принимает массив и возвращает (но не инициализирует!) другой — соответствующий исходному по форме и типу. Чтобы ускорить выполнение кода, импортируем класс jit из модуля numba и добавляем в начало кода аннотацию @jit:

Это скромное дополнение способно ускорить выполнение операции более чем в 100 раз! Если интересно, посмотрите замеры скорости математических расчётов при использовании разных библиотек для Python.

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

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

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

Главный плюс этого кода даже не в скорости исполнения, а в прозрачности и простоте. Снова сошлюсь на Хабр, где есть сравнение скорости GPU-расчетов при использовании Numba, PyCUDA и эталонного С CUDA. Небольшой спойлер: PyCUDA позволяет достичь скорости вычислений, сопоставимой с Cи, а Numba подходит для небольших задач.

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

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

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

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

При взаимозависимых вычислениях. Считать что-то в одном потоке и передавать для дальнейшей обработки второму — плохая идея. Возникает лишняя зависимость, которая приводит к снижению производительности, а в случае ошибки — к ступору и краху программы.
При работе через GIL. Это мы уже выяснили выше.
Когда важна хорошая переносимость на разных устройствах. Правильно подобрать число потоков для машины пользователя — задача не из легких. Если вы пишете под известное вам «железо», всё можно решить тестированием. Если же нет — понадобится дополнительно создавать гибкую систему подстройки под аппаратную часть, что потребует времени и умения.

Анонс — взаимные блокировки в Python

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

Python включает в себя ряд разных параллельных конструкций, таких как как threading, queues и multiprocessing. Модуль threading использовался как главный способ достижения параллельности. Несколько лет назад, модуль multiprocessing был добавлен в пакет стандартных библиотек Python. В этой статье мы сфокусируемся на том, как использовать очереди и потоки (queues и threads).

Использование потоков

Мы начнем с простого примера, который демонстрирует работу потоков. Мы наследуем класс Thread в класс MyThread и укажем, чтобы его имя выводилось как stdout. Попробуем!

В этом коде мы импортировали модули random и time, также мы импортировали класс Thread из модуля threading Python. Далее, мы наследуем класс Thread, и переопределили его метод __init__ для принятия аргумента, под названием name. Для начала потока, вам нужно вызывать метод start().

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

Написание потокового загрузчика

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

msg = "%s закончил загрузку %s!" % ( self . name , self . url )

Это, в общем, полностью переписанный первый скрипт. Здесь мы импортировали наши модули os, urllib2, и threading python. Мы используем urllib2 для непосредственной загрузки в класс потока. Модуль os мы используем для извлечения имени файла, который мы загружаем, так что мы можем использовать его для создания файла с таким же названием на нашем компьютере. В классе DownloadThread мы настраиваем __init__ для принятия url и наименований для потока. В методе run, мы открываем url, извлекаем название файла, после чего используем это название для того, чтобы создать файл на диске.

Есть вопросы по Python?

На нашем форуме вы можете задать любой вопрос и получить ответ от всего нашего сообщества!

Telegram Чат & Канал

Вступите в наш дружный чат по Python и начните общение с единомышленниками! Станьте частью большого сообщества!

Паблик VK

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

После этого мы используем цикл для загрузки файла по килобайту за раз, и сохранять его на диск. После того, как сохранение файла завершится, мы выводим название потока и тот url, который загрузился. В Python 3 этот код немного отличаться. Нам нужно импортировать urllib вместо urllib2 и использовать urllib.request.urlopen вместо urllib2.urlopen. Вот код, в котором вы можете увидеть разницу:

msg = "%s закончил загрузку %s!" % ( self . name , self . url )

Использование Queues

Очередь(Queues Python) может быть использована для стековых реализаций «пришел первым – ушел первым» (first-in-first-out (FIFO)) или же «пришел последним – ушел последним» (last-in-last-out (LILO)) , если вы используете их правильно.

В данном разделе, мы смешаем потоки и создадим простой скрипт файлового загрузчика, чтобы продемонстрировать, как работает Queues Python со случаями, которые мы хотим паралеллизировать. Чтобы помочь объяснить, как работает Queues, мы перепишем загрузочный скрипт из предыдущей секции для использования Queues. Приступим!

Подведем итоги

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

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

Читайте также: