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

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

threading.TIMEOUT_MAX

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

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

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

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

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

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

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

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

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

class threading.local

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

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

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

Класс 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 если поток не запущен. Это целое число, отличное от нуля. См. функцию _thread.get_ident(). Идентификаторы потоков могут быть повторно использованы, когда поток завершается, а другой поток создается. Идентификатор доступен даже после завершения потока.

is_alive()

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

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

daemon

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

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

isDaemon()

setDaemon()

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

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 ложно.

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

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

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

release()

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

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

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

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

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.

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

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 за исключением случая истечения таймаута, в этом случае — False.

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

wait_for(predicate, timeout=None)

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

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

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

while not predicate():
    cv.wait()

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

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

notify(n=1)

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

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

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

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

notify_all()

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

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

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

Получить семафор.

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

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

При вызове со значением timeout, отличным от None, он будет блокироваться не более чем в течение timeout секунд. Если acquire не завершится успешно в этом интервале, возвращается false. В противном случае возвращается true.

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

release()

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

class threading.BoundedSemaphore(value=1)

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

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

6.1. Semaphore Пример

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

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

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

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

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

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

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

Объект события управляет внутренней меткой, которую можно установить в 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, или пока не произойдёт таймаут (если задан).

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

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

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

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()

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

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

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

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

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

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

reset()

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

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

abort()

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

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

parties

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

n_waiting

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

broken

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

exception threading.BrokenBarrierError

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

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.