17.1. threading – параллелизм на основе потоков

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

---

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

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

Примечание

Хотя они не перечислены ниже, имена camelCase, использовавшиеся для некоторых методов и функций в этом модуле в серии Python 2.x, по-прежнему поддерживаются этим модулем.

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

threading.active_count()¶

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

threading.current_thread()¶

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

threading.get_ident()¶

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

Новое в версии 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 при реализации сопоставляются функциям уровня модуля.

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

17.1.1. Потоковые локальные данные

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

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

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

classthreading.local¶

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

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

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

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

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

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

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

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

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

Примечание

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

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

Существует возможность создания «фиктивных объектов потоков» (dummy thread objects). Это объекты потоков, соответствующие «внешним потокам» (alien threads), то есть потокам управления, запущенным вне модуля 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 (по умолчанию), свойство фоновости наследуется от текущего потока.

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

Особенность реализации CPython: В CPython из-за глобальной блокировки интерпретатора (GIL) только один поток может одновременно выполнять код Python (хотя некоторые библиотеки, ориентированные на производительность, могут обойти это ограничение). Если вы хотите, чтобы ваше приложение лучше использовало вычислительные ресурсы многоядерных машин, рекомендуется использовать multiprocessing или concurrent.futures.ProcessPoolExecutor. Однако модель потоков по-прежнему подходит, если вы хотите одновременно выполнять несколько задач, связанных с вводом-выводом.

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

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

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

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

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

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

classthreading.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.

Возвращаемое значение отсутствует.

17.1.4. Объекты RLock

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

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

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

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

# Потребить один элемент
with cv:
    while not an_item_is_available():
        cv.wait()
    get_an_available_item()

# Произвести один элемент
with cv:
    make_an_item_available()
    cv.notify()

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

# Потребить элемент
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.

classthreading.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)¶

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

Если аргумент timeout указан и не равен 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()  # через 30 секунд будет напечатано "hello, world"

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

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

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

cancel()¶

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

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

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

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

Барьер может быть повторно использован любое количество раз для того же числа потоков.

В качестве примера вот простой способ синхронизации потока клиента и сервера:

b = Barrier(2, timeout=5)

def server():
    start_server()
    b.wait()
    while True:
        connection = accept_connection()
        process_server_connection(connection)

def client():
    b.wait()
    while True:
        connection = make_connection()
        process_client_connection(connection)

class threading.Barrier(parties, action=None, timeout=None)¶

Создает объект барьера для parties потоков. action, если задан, представляет собой вызываемый объект, который будет вызван одним из потоков при их освобождении. timeout – это значение тайм-аута по умолчанию, если не указано для метода wait().

wait(timeout=None)¶

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

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

i = barrier.wait()
if i == 0:
    # Только одному потоку нужно напечатать это
    print("passed the barrier")

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

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

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

reset()¶

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

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

abort()¶

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

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

parties¶

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

n_waiting¶

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

broken¶

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

exceptionthreading.BrokenBarrierError¶

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

17.1.10. Using locks, conditions, and semaphores in the with statement

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

with some_lock:
    # сделать что-то...

эквивалентно следующему:

some_lock.acquire()
try:
    # сделать что-то...
finally:
    some_lock.release()

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