Потоки Python: threading для задач с ожиданием

Прямой ответ

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

import threading, queue

q = queue.Queue()
lock = threading.Lock()
results = []

def worker():
    while True:
        item = q.get()
        if item is None:
            break
        with lock:
            results.append(fetch(item))
        q.task_done()

threads = [threading.Thread(target=worker, daemon=True) for _ in range(4)]
for t in threads:
    t.start()

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

Блокировки, гонки и обработка исключений

Три аспекта определяют корректность работы с потоками.

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

Исключение в потоке. Исключение, возникшее в потоке, не распространяется в основной поток, поэтому ошибка остаётся незамеченной. Результат выполнения проверяется явно.

Завершение потока. Поток завершается по условию либо по сигналу, поэтому принудительная остановка не применяется. Фоновый поток завершается при выходе программы.

Практическое следствие: общий ресурс изменяется под блокировкой, а результат выполнения потока проверяется явно.

Отдельный аспект — объём блокировки. Блокировка удерживается на время изменения, поэтому длительный участок под блокировкой сокращает параллелизм. Участок ограничивается операциями, требующими исключительного доступа.

Три признака некорректной работы с потоками.

Изменение общего ресурса без блокировки. Результат зависит от порядка выполнения потоков, поэтому возникает гонка.

Отсутствие проверки результата потока. Исключение в потоке не распространяется, поэтому ошибка остаётся незамеченной.

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

О пулах потоков — в статье про concurrent.futures; о влиянии блокировки интерпретатора — в статье про GIL.

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

Таймауты и отмена ожидающих операций

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

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

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

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

Практическое следствие: каждая операция ожидания имеет таймаут, иначе поток удерживается неограниченно долго при недоступности ресурса. О таймаутах в асинхронном коде — в статье про отмену и таймауты asyncio.

Фоновые потоки и завершение программы

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

Завершение при выходе. Программа завершается, когда основной поток заканчивает работу, при этом фоновые потоки прерываются. Незавершённая работа в фоновом потоке теряется.

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

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

Практика: работа с потоками

1. Используйте очередь для передачи задач. Операции постановки и извлечения атомарны, поэтому дополнительный контроль не требуется.

2. Защищайте общий ресурс блокировкой. Изменение структуры под блокировкой исключает гонку за данные.

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

4. Проверяйте результат выполнения потока. Исключение в потоке не распространяется в основной поток, поэтому ошибка выявляется явно.

5. Завершайте потоки по условию. Принудительная остановка не применяется, поэтому поток завершается по сигналу.

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

7. Ограничивайте число потоков. Избыточное число увеличивает накладные расходы переключения без выигрыша.

Рекомендации:

1. Различайте конкурентность и параллелизм. Потоки обеспечивают конкурентность при ожидании, но не параллелизм вычислений.
2. Учитывайте стоимость переключения. Каждый поток требует ресурсов, поэтому избыточное число снижает производительность.
3. Проверяйте отсутствие гонок. Изменение общего ресурса без блокировки приводит к зависимости результата от порядка выполнения.
4. Фиксируйте способ завершения. Поток завершается по условию, поэтому способ сигнала определяется явно.
5. Пересматривайте число потоков при изменении нагрузки. Избыточное число потоков не даёт выигрыша при малом объёме задач.

Практический сценарий: данные терялись при параллельной записи

Контекст: результаты обработки терялись при выполнении в нескольких потоках.

1. Диагностика. Потоки добавляли элементы в общий список без блокировки. Операция добавления не атомарна, поэтому при одновременном выполнении часть элементов терялась.
2. Причина. Общий ресурс изменялся без исключительного доступа, поэтому результат зависел от порядка выполнения потоков. Потеря элементов обнаруживалась по расхождению числа результатов с числом задач.
3. Действия. (а) Заменили общий список на очередь, чтобы передача результатов выполнялась атомарно; (б) ограничили участок под блокировкой операцией изменения, исключив из него длительные вызовы; (в) добавили проверку числа результатов против числа задач; (г) включили учёт исключений в потоках, чтобы ошибка не оставалась незамеченной; (д) проверили отсутствие расхождений при повторных запусках.
4. Результат. Результаты не теряются, поскольку изменение общего ресурса выполняется под блокировкой либо через очередь.

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

FAQ

Нужна ли блокировка при работе с очередью? Нет, операции постановки и извлечения атомарны, поэтому дополнительный контроль не требуется. Блокировка нужна при изменении общей структуры напрямую.

Распространяется ли исключение из потока в основной? Нет, исключение остаётся в потоке, поэтому результат выполнения проверяется явно либо через пул с сбором результатов.

Как завершить поток? По условию либо по сигналу. Принудительная остановка не применяется, поэтому поток завершается по достижении условия.

Сколько потоков создавать? Определяется характером задач и числом ожидаемых операций. Избыточное число увеличивает накладные расходы переключения.

Источники: Python threading, queue, concurrent.futures — ссылки в шапке.