threading — Параллелизм на основе потоков

Локальные данные потока

Локальные данные потока — это данные, значения которых зависят от потока. Если вам нужны данные, локальные для потока, создайте объект local и используйте его атрибуты:

>>> mydata = local()
>>> mydata.number = 42
>>> mydata.number
42

Также можно обратиться к словарю объекта local:

>>> mydata.__dict__
{'number': 42}
>>> mydata.__dict__.setdefault('widgets', [])
[]
>>> mydata.widgets
[]

Если обратиться к данным из другого потока:

>>> log = []
>>> def f():
...     items = sorted(mydata.__dict__.items())
...     log.append(items)
...     mydata.number = 11
...     log.append(mydata.number)

>>> import threading
>>> thread = threading.Thread(target=f)
>>> thread.start()
>>> thread.join()
>>> log
[[], 11]

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

>>> mydata.number
42

Разумеется, значения, полученные из объекта local, включая его атрибут __dict__, относятся к тому потоку, который был текущим в момент чтения атрибута. Поэтому обычно не следует сохранять эти значения для использования в других потоках: они применимы только к тому потоку, из которого были получены.

Пользовательские объекты local можно создавать, наследуя класс local:

>>> class MyLocal(local):
...     number = 2
...     def __init__(self, /, **kw):
...         self.__dict__.update(kw)
...     def squared(self):
...         return self.number ** 2

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

Теперь создадим объект local:

>>> mydata = MyLocal(color='red')

получим число по умолчанию:

>>> mydata.number
2

исходный цвет:

>>> mydata.color
'red'
>>> del mydata.color

И метод, работающий с данными:

>>> mydata.squared()
4

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

>>> log = []
>>> thread = threading.Thread(target=f)
>>> thread.start()
>>> thread.join()
>>> log
[[('color', 'red')], 11]

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

>>> mydata.number
2
>>> mydata.color
Traceback (most recent call last):
...
AttributeError: 'MyLocal' object has no attribute 'color'

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

>>> class MyLocal(local):
...     __slots__ = 'number'

>>> mydata = MyLocal()
>>> mydata.number = 42
>>> mydata.color = 'red'

Таким образом, отдельный поток:

>>> thread = threading.Thread(target=f)
>>> thread.start()
>>> thread.join()

влияет на то, что мы видим:

>>> mydata.number
11

class threading.local

Класс, представляющий локальные данные потока.

Объекты потоков

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

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

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

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

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

Если метод run() вызывает исключение, для его обработки вызывается threading.excepthook(). По умолчанию threading.excepthook() молча игнорирует SystemExit.

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

Существует объект «главного потока», соответствующий исходному потоку управления в программе Python. Он не является потоком-демоном.

Также могут создаваться «фиктивные объекты потоков». Это объекты потоков, соответствующие «чужим потокам» — потокам управления, запущенным вне модуля threading, например непосредственно из кода C. Возможности фиктивных объектов потоков ограничены: они всегда считаются активными и являются потоками-демонами, их нельзя дождаться. Они никогда не удаляются, поскольку обнаружить завершение чужих потоков невозможно.

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

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

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

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

name — имя потока. По умолчанию формируется уникальное имя вида «Thread-N», где N — небольшое десятичное число, или «Thread-N (target)», где «target» — это target.__name__, если указан аргумент target.

args — список или кортеж аргументов для вызова целевого объекта. По умолчанию равен ().

kwargs — словарь именованных аргументов для вызова целевого объекта. По умолчанию равен {}.

Если значение не равно None, аргумент daemon явно задаёт, будет ли поток демоном. Если значение равно None (по умолчанию), свойство демона наследуется от текущего потока.

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

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

Изменено в версии 3.10: Если аргумент name не указан, используется имя target.

start()

Запускает действие потока.

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

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

run()

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

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

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

Пример:

>>> from threading import Thread
>>> t = Thread(target=print, args=[1])
>>> t.run()
1
>>> t = Thread(target=print, args=(1,))
>>> t.run()
1

join(timeout=None)

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

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

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

Один и тот же поток можно ожидать многократно.

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

name

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

getName()

setName()

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

Устарело с версии 3.10.

ident

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

native_id

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

Примечание

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

Доступность: Windows, FreeBSD, Linux, macOS, OpenBSD, NetBSD, AIX, DragonFlyBSD.

Добавлено в версии 3.8.

is_alive()

Возвращает признак активности потока.

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

daemon

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

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

isDaemon()

setDaemon()

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

Устарело с версии 3.10.

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

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

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

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

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

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

class threading.Lock

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

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

acquire(blocking=True, timeout=-1)

Захватить блокировку с ожиданием или без него.

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

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

Если аргумент с плавающей точкой timeout имеет положительное значение, ожидать не дольше указанного в timeout количества секунд и только пока блокировку нельзя захватить. Значение аргумента timeout, равное -1, означает неограниченное ожидание. Указывать timeout, если blocking имеет значение False, запрещено.

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

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

Изменено в версии 3.2: В POSIX захват блокировки теперь может быть прерван сигналами, если это поддерживается используемой реализацией потоков.

release()

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

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

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

Метод ничего не возвращает.

locked()

Возвращает True, если блокировка захвачена.

Объекты RLock

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

Потоки вызывают метод блокировки acquire(), чтобы заблокировать её, и метод release(), чтобы разблокировать её.

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

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

class threading.RLock

Этот класс реализует объекты рекурсивных блокировок. Рекурсивную блокировку должен освобождать поток, который её захватил. Захватив такую блокировку, поток может повторно захватывать её без ожидания; поток должен освобождать её по одному разу за каждый захват.

Обратите внимание: RLock на самом деле является фабричной функцией, которая возвращает экземпляр наиболее эффективной версии конкретного класса RLock, поддерживаемой платформой.

acquire(blocking=True, timeout=-1)

Захватить блокировку с ожиданием или без него.

См. также

Использование RLock в качестве менеджера контекста

По возможности рекомендуется вместо ручных вызовов acquire() и release().

При вызове со значением True аргумента blocking (по умолчанию):

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

При вызове со значением False аргумента blocking:

• Если блокировкой не владеет ни один поток, захватить её и немедленно вернуть управление.
• Если блокировкой владеет другой поток, немедленно вернуть управление.
• Если блокировкой владеет тот же поток, повторно захватить её и немедленно вернуть управление.

Во всех случаях, если потоку удалось захватить блокировку, возвращается True. Если потоку не удалось захватить блокировку (то есть ожидание отключено или истёк срок ожидания), возвращается False.

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

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

release()

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

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

Метод ничего не возвращает.

Объекты Condition

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

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

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

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

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

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

# Consume one item
with cv:
    while not an_item_is_available():
        cv.wait()
    get_an_available_item()

# Produce one item
with cv:
    make_an_item_available()
    cv.notify()

Цикл while, проверяющий выполнение условия приложения, необходим, потому что wait() может вернуться через произвольно долгое время, и условие, вызвавшее notify(), может к тому моменту перестать выполняться. Это неизбежная особенность многопоточного программирования. Метод wait_for() позволяет автоматизировать проверку условия и упрощает вычисление времени ожидания:

# Consume an item
with cv:
    cv.wait_for(an_item_is_available)
    get_an_available_item()

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

class threading.Condition(lock=None)

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

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

Изменено в версии 3.3: Фабричная функция заменена классом.

acquire(*args)

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

release()

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

wait(timeout=None)

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

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

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

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

Возвращаемое значение — True, если не истекло заданное значение timeout; в противном случае возвращается False.

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

wait_for(predicate, timeout=None)

Ожидать, пока условие не станет истинным. predicate должен быть вызываемым объектом, результат которого интерпретируется как логическое значение. Можно указать timeout — максимальное время ожидания.

Этот вспомогательный метод может многократно вызывать wait(), пока предикат не выполнится или не истечёт время ожидания. Возвращаемое значение — последнее возвращённое предикатом значение; если время ожидания истекло, оно будет False.

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

while not predicate():
    cv.wait()

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

Добавлено в версии 3.2.

notify(n=1)

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

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

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

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

notify_all()

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

Метод notifyAll — устаревший псевдоним этого метода.

Объекты семафоров

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

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

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

class threading.Semaphore(value=1)

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

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

Изменено в версии 3.3: Фабричная функция заменена классом.

acquire(blocking=True, timeout=None)

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

При вызове без аргументов:

• Если при входе внутренний счётчик больше нуля, уменьшить его на единицу и немедленно вернуть True.
• Если при входе внутренний счётчик равен нулю, приостановить выполнение до пробуждения вызовом release(). После пробуждения (и если счётчик больше 0) уменьшить счётчик на 1 и вернуть True. Каждый вызов release() пробуждает ровно один поток. Не следует полагаться на порядок пробуждения потоков.

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

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

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

release(n=1)

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

Изменено в версии 3.9: Добавлен параметр n, позволяющий за один раз пробуждать несколько ожидающих потоков.

class threading.BoundedSemaphore(value=1)

Класс, реализующий объекты семафоров с ограниченным счётчиком. Такой семафор проверяет, чтобы его текущее значение не превышало начального. Если это происходит, вызывается исключение ValueError. В большинстве случаев семафоры используют для контроля доступа к ресурсам ограниченной ёмкости. Если семафор освобождается слишком много раз, это указывает на ошибку. Если аргумент value не задан, по умолчанию он равен 1.

Изменено в версии 3.3: Фабричная функция заменена классом.

Semaphore: пример

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

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

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

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

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

Объекты событий

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

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

class threading.Event

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

Изменено в версии 3.3: фабричная функция заменена классом.

is_set()

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

Метод isSet является устаревшим псевдонимом этого метода.

set()

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

clear()

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

wait(timeout=None)

Блокирует выполнение, пока внутренний флаг имеет значение false и не истёк заданный тайм-аут. Возвращаемое значение указывает причину завершения этого блокирующего метода: True, если метод вернулся из-за того, что внутренний флаг установлен в значение true, или False, если задан тайм-аут, а внутренний флаг не стал равен true в течение указанного времени ожидания.

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

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

Объекты таймера

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

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

Например:

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

t = Timer(30.0, hello)
t.start()  # after 30 seconds, "hello, world" will be printed

class threading.Timer(interval, function, args=None, kwargs=None)

Создаёт таймер, который по истечении interval секунд вызовет function с аргументами args и именованными аргументами kwargs. Если args равно None (значение по умолчанию), будет использоваться пустой список. Если kwargs равно None (значение по умолчанию), будет использоваться пустой словарь.

Изменено в версии 3.3: фабричная функция заменена классом.

cancel()

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

Объекты барьера

Этот класс предоставляет простой примитив синхронизации для фиксированного числа потоков, которым необходимо дождаться друг друга. Каждый поток пытается пройти барьер, вызывая метод wait(), и блокируется, пока все потоки не вызовут 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:
    # Only one thread needs to print this
    print("passed the barrier")

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

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

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

reset()

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

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

abort()

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

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

parties

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

n_waiting

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

broken

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

exception threading.BrokenBarrierError

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