threading – интерфейс многопоточности высокого уровня¶threading – Higher-level threading interface

Объекты потоков¶Thread Objects

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Thread.start()¶

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

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

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

Thread.run()¶

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

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

Thread.join([timeout])¶

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

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

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

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

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

Thread.name¶

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

Thread.getName()¶

Thread.setName()¶

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

Thread.ident¶

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

Thread.is_alive()¶

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

Грубо говоря, поток активен с момента возврата метода start() до завершения его метода run(). Функция модуля enumerate() возвращает список всех активных потоков.

Thread.daemon¶

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

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

Thread.isDaemon()¶

Thread.setDaemon()¶

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

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

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

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

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

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

Lock.acquire([blocking=1])¶

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

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

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

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

Lock.release()¶

Освободить блокировку.

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

Не вызывайте этот метод, если блокировка уже разблокирована.

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

Объекты RLock¶RLock Objects

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

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

RLock.acquire([blocking=1])¶

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

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

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

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

RLock.release()¶

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

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

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

Объекты Condition¶Condition Objects

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

Переменная условия имеет методы acquire() и release(), которые вызывают соответствующие методы связанной блокировки. Также у неё есть метод wait(), а также методы notify() и notify_all(). Эти три метода можно вызывать только после того, как вызывающий поток захватил блокировку; в противном случае возбуждается RuntimeError.

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

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

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

Совет: типичный стиль программирования с переменными условия использует блокировку для синхронизации доступа к некоторому общему состоянию; потоки, заинтересованные в определённом изменении состояния, вызывают wait() повторно, пока не увидят желаемое состояние, а потоки, изменяющие состояние, вызывают notify() или notify_all(), когда изменяют состояние таким образом, что оно может стать желаемым для одного из ожидающих. Например, следующий код демонстрирует обобщённую ситуацию «производитель-потребитель» с неограниченной ёмкостью буфера:

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

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

Чтобы выбрать между notify() и notify_all(), подумайте, может ли одно изменение состояния быть интересным только одному или нескольким ожидающим потокам. Например, в типичной ситуации «производитель-потребитель» добавление одного элемента в буфер требует пробуждения только одного потока-потребителя.

class threading.Condition([lock])¶

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

Condition.acquire(*args)¶

Захватывает базовую блокировку. Этот метод вызывает соответствующий метод базовой блокировки; возвращаемое значение совпадает с возвращаемым значением этого метода.

Condition.release()¶

Освобождает базовую блокировку. Этот метод вызывает соответствующий метод базовой блокировки; возвращаемого значения нет.

Condition.wait([timeout])¶

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

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

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

Если базовая блокировка является RLock, она не освобождается с помощью его метода release(), поскольку это может на самом деле не разблокировать блокировку, если она была захвачена рекурсивно несколько раз. Вместо этого используется внутренний интерфейс класса RLock, который действительно разблокирует её, даже если она была рекурсивно захвачена несколько раз. Затем другой внутренний интерфейс используется для восстановления уровня рекурсии при повторном захвате блокировки.

Condition.notify()¶

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

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

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

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

Condition.notify_all()¶

Пробуждает все потоки, ожидающие на этом условии. Этот метод работает как notify(), но пробуждает все ожидающие потоки, а не один. Если вызывающий поток не захватил блокировку при вызове этого метода, возбуждается RuntimeError.

Объекты Semaphore¶Semaphore Objects

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

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

class threading.Semaphore([value])¶

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

Semaphore.acquire([blocking])¶

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

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

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

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

Semaphore.release()¶

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

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

pool_sema.acquire()
conn = connectdb()
... use connection ...
conn.close()
pool_sema.release()

Объекты Event¶Event Objects

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

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

class threading.Event¶

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

Event.is_set()¶

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

Event.set()¶

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

Event.clear()¶

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

Event.wait([timeout])¶

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

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

Объекты Timer¶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 секунд.

Timer.cancel()¶

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

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

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

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

import threading

some_rlock = threading.RLock()

with some_rlock:
    print("some_rlock is locked while this executes")

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

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

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