16.2. threading – Параллелизм на основе потоков¶threading – Thread-based parallelism

16.2.1. Объекты потока¶Thread Objects

Этот класс представляет собой активность, выполняемую в отдельном потоке управления. Есть два способа определить активность: передать вызываемый объект в конструктор или переопределить метод run() в подклассе. Никакие другие методы (кроме конструктора) не следует переопределять в подклассе. Иными словами, только переопределяйте методы __init__() и run() этого класса.

После создания объекта потока его действие необходимо запустить вызовом метода start() этого потока. Это вызовет метод run() в отдельном потоке управления.

Когда действие потока запущено, поток считается «живым». Он перестаёт быть живым, когда его метод run() завершается – либо нормально, либо из-за необработанного исключения. Метод is_alive() проверяет, жив ли поток.

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

У потока есть имя. Имя можно передать конструктору, а также прочитать или изменить через атрибут name.

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

Существует объект «главный поток»; он соответствует начальному потоку управления в программе Python. Это не фоновый поток.

Существует возможность создания «фиктивных объектов потоков». Это объекты потоков, соответствующие «внешним потокам», то есть потокам управления, запущенным вне модуля threading, например, напрямую из C-кода. Фиктивные объекты потоков имеют ограниченную функциональность; они всегда считаются живыми и демоническими, и к ним нельзя применить join(). Они никогда не удаляются, так как невозможно обнаружить завершение внешнего потока.

class threading.Thread(group=None, target=None, name=None, args=(), kwargs={})¶

Этот конструктор всегда следует вызывать с именованными аргументами. Аргументы:

Параметр group должен быть None; зарезервирован для будущего расширения, когда будет реализован класс ThreadGroup.

Параметр target – это вызываемый объект, который будет вызван методом run(). По умолчанию None, то есть ничего не вызывается.

name – это имя потока. По умолчанию создаётся уникальное имя вида «Поток-N», где N – небольшое десятичное число.

args – кортеж аргументов для вызова целевой функции. По умолчанию равен ().

kwargs – это словарь именованных аргументов для вызова target. По умолчанию {}.

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

start()¶

Запускает выполнение потока.

Этот метод должен вызываться не более одного раза для каждого объекта потока. Он организует вызов метода run() объекта в отдельном потоке управления.

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

run()¶

Метод, представляющий действие потока.

Вы можете переопределить этот метод в подклассе. Стандартный метод run() вызывает вызываемый объект, переданный конструктору объекта как аргумент target (если он есть), с позиционными и именованными аргументами, взятыми из аргументов args и kwargs соответственно.

join(timeout=None)¶

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

Когда аргумент timeout присутствует и не равен None, он должен быть числом с плавающей точкой, задающим время ожидания операции в секундах (или долях секунды). Поскольку join() всегда возвращает None, необходимо вызвать is_alive() после join(), чтобы определить, произошло ли истечение времени – если поток всё ещё жив, вызов join() завершился по тайм-ауту.

Если аргумент timeout отсутствует или равен None, операция будет блокироваться до завершения потока.

К одному потоку можно применить join() много раз.

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

name¶

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

getName()¶

setName()¶

Устаревший API геттера/сеттера для name; вместо этого используйте его напрямую как свойство.

ident¶

«Идентификатор потока» данного потока или None, если поток не был запущен. Это ненулевое целое число. Смотрите функцию _thread.get_ident(). Идентификаторы потоков могут быть переиспользованы, когда поток завершается и создаётся новый поток. Идентификатор доступен даже после завершения потока.

is_alive()¶

Возвращает, жив ли поток.

Этот метод возвращает True начиная непосредственно перед запуском метода run() и до момента сразу после завершения метода run(). Функция модуля enumerate() возвращает список всех живых потоков.

daemon¶

Логическое значение, указывающее, является ли данный поток демоническим (True) или нет (False). Это значение должно быть установлено до вызова start(), иначе будет вызвано исключение RuntimeError. Его начальное значение наследуется от создающего потока; главный поток не является демоническим, и поэтому все потоки, созданные в главном потоке, по умолчанию имеют daemon = False.

Вся программа Python завершается, когда не остаётся ни одного живого потока, не являющегося демоном.

isDaemon()¶

setDaemon()¶

Устаревший API геттера/сеттера для daemon; вместо этого используйте его напрямую как свойство.

16.2.2. Объекты блокировки¶Lock Objects

Примитивная блокировка – это синхронизационный примитив, который не принадлежит конкретному потоку в заблокированном состоянии. В Python это в настоящее время самый низкоуровневый доступный примитив синхронизации, реализованный непосредственно модулем расширения _thread.

Примитивная блокировка находится в одном из двух состояний: «заблокировано» (locked) или «разблокировано» (unlocked). Она создаётся в разблокированном состоянии. Имеет два основных метода: acquire() и release(). Когда состояние разблокировано, acquire() переводит его в заблокированное и сразу возвращает управление. Когда состояние заблокировано, acquire() блокируется до тех пор, пока вызов release() из другого потока не переведёт его в разблокированное; после этого вызов acquire() снова устанавливает заблокированное состояние и возвращает управление. Метод release() следует вызывать только в заблокированном состоянии; он переводит состояние в разблокированное и немедленно возвращает управление. При попытке освободить разблокированную блокировку будет возбуждено исключение RuntimeError.

Блокировки также поддерживают протокол контекстного менеджера.

Когда несколько потоков заблокированы в вызове acquire() в ожидании разблокировки состояния, только один поток продолжает работу после того, как вызов release() переведёт состояние в разблокированное; какой именно из ожидающих потоков продолжит работу – не определено и может различаться в разных реализациях.

Все методы выполняются атомарно.

Lock.acquire(blocking=True, timeout=-1)¶

Захватывает блокировку, блокирующую или неблокирующую.

При вызове с аргументом blocking, установленным в True (по умолчанию), блокирует выполнение до тех пор, пока блокировка не будет разблокирована, после чего устанавливает её в заблокированное состояние и возвращает True.

При вызове с аргументом blocking, установленным в False, блокировка не выполняется. Если вызов с blocking, установленным в True, привёл бы к блокировке, немедленно возвращается False; в противном случае блокировка устанавливается в заблокированное состояние и возвращается True.

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

Возвращаемое значение – True, если блокировка успешно захвачена, или False в противном случае (например, если истёк тайм-аут).

Изменено в версии 3.2: Добавлен параметр timeout.

Изменено в версии 3.2: Захват блокировок теперь может прерываться сигналами на POSIX.

Lock.release()¶

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

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

При вызове на разблокированной блокировке вызывается исключение ThreadError.

Возвращаемое значение отсутствует.

16.2.3. Объекты RLock¶RLock Objects

Повторно входимая блокировка – это примитив синхронизации, который может быть захвачен одним и тем же потоком несколько раз. Внутри она использует понятия «поток-владелец» и «уровень рекурсии» в дополнение к состоянию «заблокировано/разблокировано», используемому простыми блокировками. В заблокированном состоянии блокировкой владеет какой-то поток; в разблокированном состоянии ею не владеет ни один поток.

Для блокировки поток вызывает свой метод acquire(); этот метод возвращает управление, когда поток получает во владение блокировку. Для разблокировки поток вызывает свой метод release(). Пары вызовов acquire()/release() могут быть вложенными; только последний вызов release() (самый внешний release()) переводит блокировку в разблокированное состояние и позволяет другому потоку, заблокированному в acquire(), продолжить работу.

Рекурсивные блокировки также поддерживают протокол контекстного менеджера.

RLock.acquire(blocking=True, timeout=-1)¶

Захватывает блокировку, блокирующую или неблокирующую.

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

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

При вызове с аргументом blocking, установленным в false, не блокируется. Если вызов без аргументов привёл бы к блокировке, немедленно возвращает false; в противном случае делает то же самое, что и при вызове без аргументов, и возвращает true.

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

Изменено в версии 3.2: Добавлен параметр timeout.

RLock.release()¶

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

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

Возвращаемое значение отсутствует.

16.2.4. Объекты Condition¶Condition Objects

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

Переменная условия подчиняется протоколу контекстного менеджера: использование оператора with захватывает связанную блокировку на время вложенного блока. Методы acquire() и release() также вызывают соответствующие методы связанной блокировки.

Другие методы должны вызываться только при удержании связанной блокировки. Метод wait() освобождает блокировку и затем блокируется, пока другой поток не пробудит его вызовом notify() или notify_all(). После пробуждения wait() повторно захватывает блокировку и возвращает управление. Также можно указать тайм-аут.

Метод notify() пробуждает один из потоков, ожидающих на данной условной переменной (если такие есть). Метод notify_all() пробуждает все потоки, ожидающие на условной переменной.

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

# Потребить один элемент
with cv:
    while not an_item_is_available():
        cv.wait()
    get_an_available_item()

# Произвести один элемент
with cv:
    make_an_item_available()
    cv.notify()

# Потребить элемент
with cv:
    cv.wait_for(an_item_is_available)
    get_an_available_item()

16.2.5. Объекты семафоров¶Semaphore Objects

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

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

Семафоры также поддерживают протокол менеджера контекста.

class threading.Semaphore(value=1)¶

Необязательный аргумент задаёт начальное value для внутреннего счётчика; по умолчанию оно равно 1. Если указанное value меньше 0, возникает исключение ValueError.

acquire(blocking=True, timeout=None)¶

Захватывает семафор.

При вызове без аргументов: если внутренний счётчик при входе больше нуля, уменьшить его на единицу и сразу вернуться. Если он равен нулю, заблокироваться, ожидая, пока другой поток вызовет release(), чтобы сделать его больше нуля. Это делается с надлежащей взаимной блокировкой, так что если несколько вызовов acquire() заблокированы, release() пробудит ровно один из них. Реализация может выбрать любой случайным образом, поэтому на порядок пробуждения заблокированных потоков полагаться не следует. Возвращает true (или блокируется на неопределённый срок).

При вызове с blocking, установленным в false, не блокироваться. Если вызов без аргумента заблокировался бы, немедленно вернуть false; в противном случае сделать то же самое, что и при вызове без аргументов, и вернуть true.

При вызове с timeout, отличным от None, он будет блокироваться не более timeout секунд. Если acquire не завершится успешно за это время, вернуть false. В противном случае вернуть true.

Изменено в версии 3.2: Добавлен параметр timeout.

release()¶

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

maxconnections = 5
...
pool_sema = BoundedSemaphore(value=maxconnections)

with pool_sema:
    conn = connectdb()
    try:
        ... use connection ...
    finally:
        conn.close()

16.2.6. Объекты событий¶Event Objects

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

Объект-событие управляет внутренним флагом, который может быть установлен в true методом set() и сброшен в false методом clear(). Метод wait() блокируется, пока флаг не станет true.

class threading.Event¶

Внутренний флаг изначально равен false.

is_set()¶

Возвращает true тогда и только тогда, когда внутренний флаг равен true.

set()¶

Устанавливает внутренний флаг в true. Все потоки, ожидающие его установки в true, пробуждаются. Потоки, которые вызывают wait() после того, как флаг уже true, не будут блокироваться вообще.

clear()¶

Сбрасывает внутренний флаг в false. После этого потоки, вызывающие wait(), будут блокироваться до тех пор, пока не будет вызван set(), чтобы снова установить флаг в true.

wait(timeout=None)¶

Блокируется, пока внутренний флаг не станет true. Если при входе флаг уже true, вернуться немедленно. В противном случае блокироваться до тех пор, пока другой поток не вызовет set(), чтобы установить флаг в true, или пока не истечет необязательный тайм-аут.

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

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

Изменено в версии 3.1: Ранее метод всегда возвращал None.

16.2.7. Объекты таймеров¶Timer Objects

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

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

Например:

def hello():
    print("hello, world")

t = Timer(30.0, hello)
t.start() # через 30 секунд будет выведено "hello, world"

class threading.Timer(interval, function, args=[], kwargs={})¶

Создаёт таймер, который выполнит function с аргументами args и именованными аргументами kwargs по прошествии interval секунд.

cancel()¶

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

16.2.8. Объекты барьеров¶Barrier Objects

Новое в версии 3.2.

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

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

В качестве примера приведён простой способ синхронизации потоков клиента и сервера:

b = Barrier(2, timeout=5)

def server():
    start_server()
    b.wait()
    while True:
        connection = accept_connection()
        process_server_connection(connection)

def client():
    b.wait()
    while True:
        connection = make_connection()
        process_client_connection(connection)

class threading.Barrier(parties, action=None, timeout=None)¶

Создает объект барьера для parties потоков. Параметр action, если указан, должен быть вызываемым объектом, который будет вызван одним из потоков при их освобождении. timeout – это значение таймаута по умолчанию, если оно не задано для метода wait().

wait(timeout=None)¶

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

Возвращаемое значение – целое число в диапазоне от 0 до parties – 1, различное для каждого потока. Это можно использовать для выбора потока, который выполнит специальные вспомогательные действия, например:

i = barrier.wait()
if i == 0:
    # Выводить это должен только один поток
    print("passed the barrier")

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

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

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

reset()¶

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

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

abort()¶

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

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

parties¶

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

n_waiting¶

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

broken¶

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

exception threading.BrokenBarrierError¶

Это исключение, являющееся подклассом RuntimeError, возбуждается, когда объект Barrier сбрасывается или повреждается.

16.2.9. Использование блокировок, условий и семафоров в операторе with¶Using locks, conditions, and semaphores in the with statement

Все объекты, предоставляемые этим модулем, которые имеют методы acquire() и release(), могут использоваться как контекстные менеджеры в операторе with. Метод acquire() будет вызван при входе в блок, а release() – при выходе из блока. Следовательно, следующий фрагмент:

with some_lock:
    # выполнить какие-то действия...

эквивалентно:

some_lock.acquire()
try:
    # выполнить какие-то действия...
finally:
    some_lock.release()

В настоящее время объекты Lock, RLock, Condition, Semaphore и BoundedSemaphore могут использоваться как контекстные менеджеры в операторе with.

16.2.10. Импортирование в многопоточном коде¶Importing in threaded code

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

• Во-первых, за исключением главного модуля, импорт не должен вызывать побочный эффект в виде создания нового потока с последующим ожиданием этого потока любым способом. Несоблюдение этого ограничения может привести к взаимоблокировке, если созданный поток прямо или косвенно попытается импортировать модуль.
• Во-вторых, все попытки импорта должны быть завершены до того, как интерпретатор начнёт завершение работы. Проще всего этого добиться, выполняя импорт только из недемоновых потоков, созданных через модуль threading. Потоки-демоны и потоки, созданные напрямую с помощью модуля thread, потребуют какой-либо иной синхронизации, чтобы гарантировать, что они не будут пытаться выполнять импорт после начала завершения работы интерпретатора. Несоблюдение этого ограничения приведёт к периодическим исключениям и сбоям при завершении работы интерпретатора (поскольку поздние попытки импорта пытаются получить доступ к механизмам, которые уже находятся в недопустимом состоянии).