Прямой ответ
Модуль пулов задач предоставляет единый интерфейс для выполнения функций в потоках либо процессах, поэтому смена модели не требует изменения кода отправки задач.
from concurrent.futures import ThreadPoolExecutor, as_completed
def fetch(url):
return url, get(url)
with ThreadPoolExecutor(max_workers=8) as pool:
futures = {pool.submit(fetch, u): u for u in urls}
for fut in as_completed(futures):
try:
print(fut.result())
except Exception as exc:
print(futures[fut], 'ошибка', exc)
| Тип пула | Когда применяется |
|---|---|
| Пул потоков | Задачи с ожиданием ввода-вывода |
| Пул процессов | Вычислительные задачи |
Результат задачи получается через объект будущего, поэтому исключение возникает в момент получения результата, а не в момент отправки задачи. О работе с потоками напрямую — в статье про threading.
Обработка исключений и порядок сбора результатов
Два аспекта определяют корректность работы с пулом задач.
Исключение в задаче. Исключение сохраняется в объекте будущего и возникает при получении результата, поэтому ошибка не теряется, но требует явной обработки.
Порядок завершения. Задачи завершаются в порядке, определяемом временем выполнения, поэтому сбор результатов по мере готовности даёт результат раньше, чем ожидание всех задач.
Практическое следствие: результат каждой задачи получается с обработкой исключения, поскольку ошибка одной задачи не должна прерывать обработку остальных.
Отдельный аспект — завершение пула. Контекстный менеджер ожидает завершения всех отправленных задач, поэтому выход из блока не происходит до завершения работы. Принудительное завершение не выполняется.
Три признака некорректной работы с пулом.
Получение результата без обработки исключения. Ошибка задачи остаётся необработанной, поэтому программа завершается с исключением.
Ожидание всех задач вместо сбора по мере готовности. Результат доступен позже, поскольку обработка начинается после завершения самой длительной задачи.
Отправка задачи в закрытый пул. Пул не принимает задачи после завершения, поэтому отправка завершается ошибкой.
О многопроцессных вычислениях — в статье про multiprocessing; о выборе между моделями — в статье про выбор модели конкурентности.
Число работников выбирается по характеру задач: для ожидания — больше числа ядер, для вычислений — по числу ядер.
Ограничение числа одновременных задач и таймауты
Пул ограничивает число одновременно выполняемых задач, поэтому избыточная отправка не приводит к исчерпанию ресурсов.
Очередь задач. Задачи сверх числа работников помещаются в очередь, поэтому выполняются по мере освобождения работников. Объём очереди не ограничен, поэтому отправка большого числа задач занимает память.
Таймаут получения результата. Ожидание результата выполняется с таймаутом, поэтому задача, выполняющаяся неограниченно долго, не удерживает обработку остальных.
Отмена ожидающей задачи. Задача, не начатая к моменту отмены, исключается из очереди, поэтому не выполняется. Начатая задача отмене не подлежит.
Практическое следствие: таймаут получения результата ограничивает время ожидания, поэтому зависшая задача не блокирует обработку остальных.
Смена типа пула без изменения логики задач
Единый интерфейс пулов позволяет сменить модель выполнения без изменения кода отправки задач и сбора результатов.
Переход от потоков к процессам. Заменяется класс пула, при этом отправка задач и получение результатов выполняются тем же способом. Требуется проверка сериализуемости аргументов.
Переход от процессов к потокам. Заменяется класс пула, при этом ограничение на сериализацию снимается. Число работников увеличивается, поскольку поток в ожидании не занимает ядро.
Практическое следствие: выбор модели определяется типом пула, поэтому смена модели выполняется в одном месте без изменения логики задач.
Практика: работа с пулами задач
1. Выберите тип пула по характеру задачи. Задачи с ожиданием выполняются в пуле потоков, вычислительные — в пуле процессов.
2. Отправляйте задачи через пул. Единый интерфейс не зависит от типа пула, поэтому смена модели не требует изменения кода.
3. Собирайте результаты по мере готовности. Обработка начинается раньше, поскольку ожидание всех задач не требуется.
4. Обрабатывайте исключение при получении результата. Ошибка задачи возникает в момент получения, поэтому требуется явная обработка.
5. Используйте контекстный менеджер. Пул завершается корректно, поскольку ожидание задач выполняется автоматически.
6. Выбирайте число работников по характеру задач. Для ожидания число превышает число ядер, для вычислений — соответствует числу ядер.
7. Не отправляйте задачи в закрытый пул. Пул не принимает задачи после завершения, поэтому отправка завершается ошибкой.
Рекомендации:
- Различайте отправку задачи и получение результата. Отправка выполняется сразу, результат доступен после завершения задачи.
- Учитывайте стоимость смены типа пула. Пул процессов требует сериализации аргументов, поэтому не все объекты передаются.
- Проверяйте обработку исключений. Ошибка одной задачи не должна прерывать обработку остальных.
- Фиксируйте число работников в конфигурации. Изменение влияет на производительность, поэтому значение задаётся явно.
- Пересматривайте тип пула при изменении характера задачи. Переход от ожидания к вычислениям требует смены модели.
Практический сценарий: ошибка одной задачи прерывала обработку
Контекст: обработка набора адресов завершалась с исключением, при этом часть адресов не была обработана.
- Диагностика. Результаты собирались последовательно без обработки исключений. Ошибка в одной задаче возникала при получении результата и прерывала обработку остальных.
- Причина. Исключение задачи сохранялось в объекте будущего и возникало при получении результата. Отсутствие обработки приводило к завершению программы, поэтому оставшиеся задачи не обрабатывались.
- Действия. (а) Добавили обработку исключения при получении результата каждой задачи; (б) перешли на сбор результатов по мере готовности, чтобы обработка начиналась раньше; (в) включили учёт ошибок с указанием задачи, в которой возникло исключение; (г) добавили повторную отправку задач, завершившихся ошибкой; (д) проверили, что ошибка одной задачи не прерывает обработку остальных.
- Результат. Обработка продолжается при ошибке отдельной задачи, а ошибки учитываются для последующего разбора.
Вывод: исключение задачи возникает при получении результата, поэтому требуется явная обработка. Отсутствие обработки прерывает обработку остальных задач, хотя они могли завершиться успешно.
FAQ
Чем пул потоков отличается от пула процессов? Пул потоков выполняет задачи в потоках одного процесса, пул процессов — в отдельных процессах. Первый подходит для задач с ожиданием, второй — для вычислений.
Когда возникает исключение задачи? При получении результата из объекта будущего. Отправка задачи выполняется сразу, поэтому ошибка обнаруживается позже.
Нужно ли ожидать завершения всех задач? Нет, результаты собираются по мере готовности, поэтому обработка начинается раньше. Контекстный менеджер ожидает завершения при выходе из блока.
Как выбрать число работников? Для задач с ожиданием число превышает число ядер, для вычислительных задач — соответствует числу доступных ядер.
Источники: Python concurrent.futures, threading, multiprocessing — ссылки в шапке.