16.2. threading – Высокоуровневый интерфейс для работы с потоками ¶threading – Higher-level threading interface

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])¶

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

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

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

Поток можно join() много раз.

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

name¶

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

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

getName()¶

setName()¶

API до версии 2.6 для name.

ident¶

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

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

is_alive()¶

isAlive()¶

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

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

Изменено в версии 2.6: Добавлено написание is_alive().

daemon¶

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

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

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

isDaemon()¶

setDaemon()¶

API, предшествовавший версии 2.6, для daemon.

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

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

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

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

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

Lock.acquire([blocking])¶

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

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

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

Lock.release()¶

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

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

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

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

locked()¶

Return true if the lock is acquired.

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

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

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

RLock.acquire([blocking=1])¶

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

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

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

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

RLock.release()¶

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

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

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

16.2.4. Объекты условий¶Condition Objects

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

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

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

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

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

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

# Потребить один элемент
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() и notifyAll(), подумайте, может ли одно изменение состояния быть интересным только одному или нескольким ожидающим потокам. Например, в типичной ситуации «производитель-потребитель» добавление одного элемента в буфер требует пробуждения только одного потока-потребителя.

class threading.Condition([lock])¶

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

acquire(*args)¶

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

release()¶

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

wait([timeout])¶

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

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

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

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

notify(n=1)¶

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

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

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

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

notify_all()¶

notifyAll()¶

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

Изменено в версии 2.6: Добавлено написание notify_all().

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

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

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

class threading.Semaphore([value])¶

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

acquire([blocking])¶

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

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

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

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

release()¶

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

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

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

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

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

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

class threading.Event¶

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

is_set()¶

isSet()¶

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

Изменено в версии 2.6: Добавлено написание is_set().

set()¶

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

clear()¶

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

wait([timeout])¶

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

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

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

Изменено в версии 2.7: Ранее метод всегда возвращал 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. 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"

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

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

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

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