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, context=None)

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

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

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

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

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

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

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

context — значение Context, используемое при запуске потока. Значение по умолчанию — None, что означает, что поведение определяется флагом sys.flags.thread_inherit_context. Если флаг имеет значение true, потоки запускаются с копией контекста вызывающего метода start(). Если значение флага false, они запускаются с пустым контекстом. Чтобы явно запустить поток с пустым контекстом, передайте новый экземпляр Context(). Чтобы явно запустить поток с копией текущего контекста, передайте значение, возвращаемое copy_context(). В сборках со свободной многопоточностью по умолчанию флаг имеет значение true, в остальных случаях — false.

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

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

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

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

start()

Запускает деятельность потока.

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

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

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

Изменено в версии 3.14: Задаётся имя потока в операционной системе.

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() для потока до его запуска; такие попытки вызывают то же исключение.

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

Изменено в версии 3.14: Может быть вызвано исключение PythonFinalizationError.

name

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

На некоторых платформах имя потока задаётся на уровне операционной системы при запуске потока, поэтому оно отображается в диспетчерах задач. Это имя может быть усечено в соответствии с системным ограничением (например, до 15 байт в Linux или 63 байт в macOS).

Изменения name отражаются на уровне ОС, только если переименовывается текущий выполняющийся поток. (При изменении атрибута name другого потока обновляется только объект Thread в Python.)

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, если это поддерживается базовой реализацией потоков.

Изменено в версии 3.14: Теперь получение блокировки может прерываться сигналами в Windows.

release()

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

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

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

Возвращаемого значения нет.

locked()

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

Объекты RLock

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

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

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

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

class threading.RLock

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

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

acquire(blocking=True, timeout=-1)

Получает блокировку с ожиданием или без него.

См. также

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

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

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

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

Если аргумент blocking имеет значение False:

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

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

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

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

release()

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

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

Возвращаемого значения нет.

locked()

Возвращает логическое значение, указывающее, заблокирован ли этот объект в данный момент.

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

Объекты условий

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

Переменная условия поддерживает протокол управления контекстом: использование оператора 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()

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

locked()

Вернуть логическое значение, указывающее, заблокирован ли сейчас этот объект.

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

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 является устаревшим псевдонимом этого метода.

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

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

Семафор управляет внутренним счётчиком, который уменьшается при каждом вызове 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 секунд. Если за это время захватить семафор не удалось, вернуть 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()

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

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

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

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

class threading.Event

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

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

is_set()

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

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

set()

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

clear()

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

wait(timeout=None)

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

Если аргумент 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)

Создать таймер, который выполнит function с аргументами args и именованными аргументами kwargs через interval секунд. Если 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 сбрасывается или ломается.