Коротко: в приведённом фрагменте основные проблемы — неверное замыкание переменной цикла (все потоки могут вызывать один и тот же таск), порождение слишком большого числа потоков (ресурсная «взрывчатка»), отсутствие ожидания завершения (join) и отсутствие контроля порядка/безопасного сбора результатов. GIL делает потоки бесполезными для CPU‑bound задач; для них лучше multiprocessing/ProcessPoolExecutor. Ниже — причины, последствия и исправления.
Проблемы и почему они приводят к потере данных/багам
- Замыкание цикла: в коде
tasks = [lambda i=i: i*i for i in range(1000)]
for t in tasks:
Thread(target=lambda: results.append(t())).start()
переменная цикла $t$ захватывается лексически, но поток может выполниться позже, когда $t$ уже изменена — в результате многие потоки могут вызвать один и тот же последний таск (данные теряются/дублируются). Правильный приём: захватить текущую задачу в аргументе лямбды, например `lambda tt=t: ...`, или передавать через args.
- Много потоков: создание $1000$ (и более) потоков ведёт к высокой нагрузке на планировщик, OOM/снижение производительности, возможным тайм‑аутам/потерям. Потоки — тяжёлые ресурсы; нельзя стартовать неограниченно.
- Нет join/синхронизации: main‑поток не ждёт завершения дочерних потоков — вы можете не получить все результаты или программа завершится раньше.
- Сбор результатов и порядок: простое `results.append` не гарантирует порядок; в CPython append является атомарным из‑за GIL, но логика порядка/индексации всё равно нарушается.
- GIL: если ваши таски CPU‑bound (например вычисление $i\ast i$ для многих больших чисел), потоки не дают параллелизма — всё равно будет последовательное исполнение под GIL. Для IO‑bound задач потоки полезны.
- multiprocessing нюансы: функции/лямбды должны быть picklable (обычные лямбды — не picklable), на Windows требуется `if __name__ == '__main__':`, и нужно корректно закрывать/ждать пул, иначе возможны дедлоки.
Исправления (минимальные правки)
1) Правильное захватывание текущей задачи + join:
from threading import Thread
results = []
threads = []
for t in tasks:
th = Thread(target=lambda tt=t: results.append(tt()))
th.start()
threads.append(th)
for th in threads:
th.join()
2) Использовать Lock (если хотите быть явными, хотя append в CPython атомарен):
from threading import Lock
lock = Lock()
def worker(tt):
r = tt()
with lock:
results.append(r)
3) Сохранять индекс, чтобы восстановить порядок:
results = [None] * len(tasks)
threads = []
for idx, t in enumerate(tasks):
th = Thread(target=lambda i=idx, tt=t: results.__setitem__(i, tt()))
th.start(); threads.append(th)
for th in threads: th.join()
Лучшие альтернативы (рекомендации с учётом GIL)
- Для IO‑bound задач: ThreadPoolExecutor (concurrent.futures)
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=50) as ex:
results = list(ex.map(lambda f: f(), tasks))
Здесь задайте разумный `max_workers` (не $1000$).
- Для CPU‑bound задач: ProcessPoolExecutor / multiprocessing.Pool
Важно: таски и функции должны быть picklable (не анонимные лямбды). Лучше передавать простые данные и использовать именованную функцию:
def square(i):
return i*i
with ProcessPoolExecutor() as ex:
results = list(ex.map(square, range(N)))
Учтите, что если у вас были `tasks = [lambda ...]`, замените их на данные (например числа) и функцию вычисления.
- Для большого количества мелких задач — батчирование и map/imap_unordered:
from multiprocessing import Pool
with Pool() as p:
for r in p.imap_unordered(square, iterable_of_inputs):
handle(r)
- Для асинхронного IO: asyncio + semafоры/граничение параллелизма, или `asyncio.get_event_loop().run_in_executor(...)` для запуска блокирующих функций в пуле.
Практические рекомендации
- Не порождайте тысячи потоков — используйте пул с ограничением числа рабочих.
- Не используйте лямбды для передачи в ProcessPool (они не picklable); используйте именованные функции.
- Если нужен порядок — передавайте индекс и сохраняйте по индексу; если порядок не важен — используйте imap_unordered.
- Всегда join/завершайте пулы/потоки, используйте контекстные менеджеры.
- Для CPU‑bound — ProcessPoolExecutor; для IO‑bound — ThreadPoolExecutor или asyncio.
Короткий пример исправления с ThreadPoolExecutor (IO‑bound):
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=50) as ex:
results = list(ex.map(lambda f: f(), tasks))
И для CPU‑bound (заменить лямбды на данные + функцию):
def square(i):
return i * i
with ProcessPoolExecutor() as ex:
results = list(ex.map(square, range(1000)))
(в тексте выше числовые литералы: $1000$, $i\ast i$ указаны как примеры)
Если нужно — могу привести конкретный готовый исправленный пример под вашу задачу (IO‑bound или CPU‑bound).