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

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

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

Модуль dummy_threading предоставляется для ситуаций, когда threading не может быть использован из-за отсутствия _thread.

Этот модуль определяет следующие функции:

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 для размера стека является рекомендуемым подходом в отсутствие более специфичной информации).

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

threading.TIMEOUT_MAX

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

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

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

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

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

17.1.1. Данные, локальные для потока

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

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

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

class threading.local

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

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

17.1.2. Объекты потока

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

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

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

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

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

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

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

Возможна ситуация создания «объектов фиктивных потоков». Это объекты потоков, соответствующие «внешним потокам», которые являются потоками управления, запущенными вне модуля потоков, например, непосредственно из кода 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; используйте его напрямую как свойство.

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

Примитивная блокировка — это синхронизирующий примитив, которым в момент блокировки не владеет ни один конкретный поток. В 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 если нет (например, если истекло время ожидания).

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

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

release()

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

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

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

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

17.1.4. Объекты 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.

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

17.1.5. Объекты 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.

17.1.6. Объекты Semaphore

Это одна из старейших примитивов синхронизации в истории компьютерной науки, изобретенная голландским компьютерным ученым Эдсгером В. Дейкстрой (он использовал имена 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 секунд. Если выполнение acquire не завершится успешно в течение этого интервала, возвращается значение false. В противном случае возвращается значение true.

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

release()

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

class threading.BoundedSemaphore(value=1)

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

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

17.1.6.1. Semaphore Пример

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

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

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

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

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

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

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

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

class threading.Event

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

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

is_set()

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

set()

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

clear()

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

wait(timeout=None)

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

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

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

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

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

Этот класс представляет собой действие, которое должно быть выполнено только после того, как пройдёт определённое количество времени — таймер. 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)

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

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

cancel()

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

17.1.9. Объекты барьеров

Этот класс предоставляет простую синхронизационную примитив для использования фиксированным числом потоков, которые нуждаются в ожидании друг друга. Каждый из потоков пытается пройти барьер, вызывая метод 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 количества потоков. действие, если предоставлено, — вызываемый объект, который должен быть вызван одним из потоков, когда они будут освобождены. 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 равно true, если барьер находится в состоянии разрыва.

exception threading.BrokenBarrierError

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

17.1.10. Использование блокировок, условий и семафоров в операторе 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.