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

Исходный код: Lib/threading.py

Этот модуль создаёт интерфейсы потоков высокого уровня на основе модуля низкого уровня _thread. Также см. модуль queue.

В этом модуле определены следующие функции:

threading.active_count()

Возвращает количество объектов Thread, которые в настоящее время активны. Возвращаемое значение равно длине списка, возвращаемого enumerate().

threading.current_thread()

Возвращает текущий объект Thread, соответствующий потоку управления вызывающего кода. Если поток управления вызывающего кода не был создан с помощью модуля threading, возвращается объект потока-заглушки с ограниченной функциональностью.

threading.get_ident()

Возвращает «идентификатор потока» текущего потока. Это целое число, отличное от нуля. Его значение не имеет прямого смысла; оно предназначено в качестве магической метки, например, для индексирования словаря данных, специфичных для потока. Идентификаторы потоков могут быть переиспользованы, когда поток завершает работу, а другой поток создаётся.

Введено в версии 3.3.

threading.enumerate()

Возвращает список всех объектов Thread, которые в настоящее время активны. Список включает демонические потоки, потоки-заглушки, созданные current_thread(), и основной поток. Из списка исключаются завершённые потоки и потоки, которые ещё не были запущены.

threading.main_thread()

Возвращает основной объект Thread. В нормальных условиях главный поток — это поток, с которого был запущен интерпретатор Python.

Введено в версии 3.4.

threading.settrace(func)

Устанавливает функцию отслеживания для всех потоков, запущенных из модуля threading. Функция func будет передана в sys.settrace() для каждого потока перед вызовом метода run().

threading.setprofile(func)

Устанавливает функцию профилирования для всех потоков, запущенных из модуля threading. Функция func будет передана в sys.setprofile() для каждого потока перед вызовом метода run().

threading.stack_size([size])

Возвращает размер стека потока, используемый при создании новых потоков. Необязательный аргумент size задаёт размер стека для последующих создаваемых потоков и должен быть равен 0 (используется платформа или конфигурированное значение по умолчанию) или положительным целым числом не менее 32 768 (32 КБ). Если size не указан, используется 0. Если изменение размера стека потока не поддерживается, генерируется исключение RuntimeError. Если указанный размер стека недействителен, генерируется исключение ValueError, и размер стека не изменяется. В настоящее время минимальный поддерживаемый размер стека 32 КБ гарантирует достаточное пространство для самого интерпретатора. Обратите внимание, что на некоторых платформах могут быть определённые ограничения на значения размера стека, такие как требование минимального размера стека > 32 КБ или требование выделения кратными размеру страницы системной памяти — для получения дополнительной информации следует обратиться к документации платформы (4 КБ — обычный размер страницы; использование кратных 4096 для размера стека является рекомендуемым подходом в отсутствие более конкретной информации).

Доступность: Windows, системы с POSIX-потоками.

В этом модуле также определена следующая константа:

threading.TIMEOUT_MAX

Максимальное значение, разрешённое для параметра timeout блокирующих функций (Lock.acquire(), RLock.acquire(), Condition.wait() и т. д.). Указание таймаута больше этого значения приведёт к возникновению исключения OverflowError.

Введено в версии 3.2.

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

Конструкция этого модуля в общих чертах основана на модели потоков Java. Однако, в отличие от Java, где блокировки и переменные условий являются базовым поведением каждого объекта, в Python они представляют собой отдельные объекты. Класс Thread Python поддерживает подмножество поведения класса Thread Java; в настоящее время нет приоритетов, нет групп потоков, и потоки нельзя уничтожить, остановить, приостановить, возобновить или прервать. Статические методы класса Thread Java, если они реализованы, отображаются в функции уровня модуля.

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

Данные, специфичные для потока

Данные, специфичные для потока — это данные, значения которых зависят от потока. Для управления данными, специфичными для потока, просто создайте экземпляр local (или подкласс) и храните атрибуты в нём:

mydata = threading.local()
mydata.x = 1

Значения экземпляра будут разными для отдельных потоков.

class threading.local

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

Для получения более подробной информации и расширенных примеров см. строку документации модуля _threading_local.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

start()

Запустить активность потока.

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

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

run()

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

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

join(timeout=None)

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

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

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

Поток может быть join() много раз.

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

name

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

getName()

setName()

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

ident

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

is_alive()

Возвращает, является ли поток активным.

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

daemon

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

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

isDaemon()

setDaemon()

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

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

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

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

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

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

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

class threading.Lock

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

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

acquire(blocking=True, timeout=-1)

Получение доступа к замку, блокирующим или неблокирующим способом.

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

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

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

Значение возврата — True , если замок успешно получен, False — если нет (например, если истекло время ожидания timeout).

Изменено в версии 3.2: Параметр timeout является новым.

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

release()

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

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

При вызове на разблокированном замке генерируется RuntimeError.

Значение возврата отсутствует.

locked()

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

Объекты RLock

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

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

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

class threading.RLock

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

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

acquire(blocking=True, timeout=-1)

Получение доступа к замку, блокирующим или неблокирующим способом.

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

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

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

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

Изменено в версии 3.2: Параметр timeout является новым.

release()

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

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

Значение возврата отсутствует.

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

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

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

Объекты Semaphore

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

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

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

class threading.Semaphore(value=1)

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

Необязательный аргумент задаёт начальное значение для внутреннего счётчика; по умолчанию он равен 1. Если заданное значение меньше 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()

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

class threading.BoundedSemaphore(value=1)

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

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

Semaphore Пример

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

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

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

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.

set()

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

clear()

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

wait(timeout=None)

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

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

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

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

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

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

Таймеры запускаются, как и потоки, вызовом метода 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)

Создать таймер, который выполнит функцию с аргументами 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)

Создайте объект барьера для n потоков. Если указано, действие — это вызываемый объект, который один из потоков должен вызвать при освобождении. timeout — это значение таймаута по умолчанию, если для метода wait() не указано другое.

wait(timeout=None)

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

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

i = barrier.wait()
if i == 0:
    # Only one thread needs to print this
    print("passed the barrier")

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

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

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

reset()

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

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

abort()

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

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

parties

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

n_waiting

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

broken

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

exception threading.BrokenBarrierError

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

Использование блокировок, условий и семафоров в инструкции with

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

with some_lock:
    # do something...

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

some_lock.acquire()
try:
    # do something...
finally:
    some_lock.release()

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