Към съдържанието
wannacode

Пул от нишки и задачи

Накратко: Пулът от нишки е няколко работника пред опашка от задачи, вместо нова нишка за всяка задача; в Python това е ThreadPoolExecutor, а за сметки ProcessPoolExecutor.

Проблем

Нишка за всяка снимка

Магазинът пуска разпродажба. Продавачите качват снимки на продуктите, а за всяка трябва умалено копие за списъка с продукти. Смаляването е работа, която смята: минава през всеки пиксел (виж работа, която чака, и работа, която смята).

Първият вариант пуска нишка за всяка качена снимка:

threads = []
for image in uploads:
    thread = threading.Thread(target=save_thumbnail, args=(image,))
    thread.start()
    threads.append(thread)

С 20 снимки при теб всичко минава: „Готови: 20 умалени снимки“. В първия час на разпродажбата идват 10 000 наведнъж.

  • Всяка нишка иска свой стек и място в операционната система. Хиляди нишки изяждат паметта, а около лимита на системата start() пада с RuntimeError: can't start new thread.
  • Дори да има памет, заради GIL само една нишка смята в даден момент. Останалите 9 999 чакат и се редуват.

Сървърът не забавя, а пада. И падналият сървър не обслужва и купувачите.

Идеята

Няколко работника и опашка

Пул от нишки (thread pool) е няколко работника, създадени веднъж, и опашка от задачи пред тях. Задачата е „смали тази снимка“. Свободният работник взима следващата задача от опашката, свършва я и се връща за нова.

  • Броят работници е постоянен. 10 000 снимки не правят 10 000 нишки, а 10 000 задачи в опашката. Задачата е малък обект, нишката е скъпа.
  • Работниците се ползват отново. Нишката не се създава и не умира за всяка снимка.
  • Опашката поема пика. Когато качванията са повече, отколкото работниците смогват, задачите чакат, вместо да сриват сървъра.

Това е производител и потребител: кодът, който приема качванията, слага задачи в опашката, а работниците са потребителите.

С пула вече се срещна в урока за нишките: ThreadPoolExecutor, submit и Future.result(). Тук ще видим какво става вътре, как да подредиш резултатите, какво става с грешките и колко работника трябват.

Виж го

Трима работника

Смаляването тук е пауза, 2 ms за всеки KB, а размерът идва от името на снимката. Така изходът е един и същ при всяко пускане:

# Смаляване на снимка: тук е пауза, а размерът идва от името.
def thumbnail(image):
    if image.startswith("повредена"):
        raise ValueError("повреден файл")
    kb = 20 + 3 * len(image)
    time.sleep(2 * kb / 1000)
    return kb

Осем снимки, пул от трима работника:

with ThreadPoolExecutor(max_workers=3) as pool:
    sizes = list(pool.map(thumbnail, images))
total = 0
for image, kb in zip(images, sizes):
    print(f"{image}: {kb} KB")
    total += kb
print(f"Общо: {total} KB")
Стъпка по стъпка, с думи
  1. 8 снимки чакат. Пулът е 3 нишки: ThreadPoolExecutor(max_workers=3).
  2. map прави всяка снимка задача в опашката. Трите нишки взимат първите три.
  3. Нишка 3 свършва първа: яке е третата снимка и резултатът отива на третото място. Работникът не спира: взима следващата задача.
  4. Опашката е празна. Нишка 1 свърши и няма какво да вземе, затова чака. Другите довършват последните снимки.
  5. map връща размерите по реда на снимките, не по реда на свършване. Общо 379 KB; краят на with затваря пула.
images = ["чанта.jpg", "обувки.jpg", "яке.jpg", "шапка.jpg", "часовник.jpg", "раница.jpg", "колан.jpg", "шал.jpg"]
with ThreadPoolExecutor(max_workers=3) as pool:
    sizes = list(pool.map(thumbnail, images))
total = 0
for image, kb in zip(images, sizes):
    print(f"{image}: {kb} KB")
    total += kb
print(f"Общо: {total} KB")
Изход
чанта.jpg: 47 KB
обувки.jpg: 50 KB
яке.jpg: 41 KB
шапка.jpg: 47 KB
часовник.jpg: 56 KB
раница.jpg: 50 KB
колан.jpg: 47 KB
шал.jpg: 41 KB
Общо: 379 KB
  • ThreadPoolExecutor(max_workers=3) прави три нишки и опашка пред тях.
  • map прави задача за всяка снимка и връща резултатите по реда на снимките, каквото и да е свършило първо. Ако не ги чакаш всичките, взимаш ги един по един от map: той чака точно следващия.
  • with затваря пула накрая: не приема нови задачи, изчаква започнатите и спира нишките.

Нишките тук помагат, защото смаляването е пауза. Истинско смаляване смята, а заради GIL нишките не смятат едновременно. За него има процеси, малко по-надолу.

По-точно, за любопитните

Без max_workers пулът има min(32, os.cpu_count() + 4) нишки. Опашката му е без край: ако качванията винаги идват по-бързо, тя расте, докато паметта свърши.

Как е в другите езици: в Java пулът е ExecutorService (там нишките смятат истински паралелно), в C# басейнът е един за цялата програма, а Node няма готов пул от Worker-и.

Пробвай

Намери грешката

Колега е пренаписал смаляването с пул, а разпродажбата не стана по-бърза. Шест снимки отнемат колкото без пул:

start = time.monotonic()
sizes = []
with ThreadPoolExecutor(max_workers=3) as pool:
    for image in images:
        size = pool.submit(thumbnail, image)
        sizes.append(size.result())
print(f"Отне над 0,5 s: {time.monotonic() - start > 0.5}")
Изход
Отне над 0,5 s: True

Избери реда с грешката.

Идеята

Резултатите по реда на входа

Задачите свършват в ред, който не знаеш: зависи от снимките, от натоварването, от това кой работник е бил свободен. Затова никога не печатай „по реда на свършване“, ако искаш един и същ изход.

Два начина дават еднакъв резултат при всяко пускане:

  • По реда на входа. Пазиш обещанията в списък по реда на задачите и ги чакаш в този ред. Така работи кодът отгоре.
  • Сбор или брой. Ако ти трябва само общото, редът няма значение: 47 + 50 е колкото 50 + 47.

Понякога искаш резултатите по реда на свършване: за лента „готови 3 от 8“, или за да покажеш първите готови снимки. as_completed ги дава така. Тогава събираш само неща, които не зависят от реда:

count = 0
total = 0
with ThreadPoolExecutor(max_workers=3) as pool:
    futures = [pool.submit(thumbnail, image) for image in images]
    for future in as_completed(futures):
        total += future.result()
        count += 1
print(f"Готови: {count} от {len(images)}, общо: {total} KB")
Изход
Готови: 8 от 8, общо: 379 KB
as_completed дава всяко Future, щом е готово. result() на него не чака.

Идеята

Грешка в задачата

Една от снимките е повредена и thumbnail хвърля изключение. То не излиза в нишката, която е пуснала задачата: тя отдавна е продължила. Пулът го хваща и го пази в обещанието за резултата. Излиза, когато чакаш този резултат:

images = ["чанта.jpg", "обувки.jpg", "повредена.jpg", "яке.jpg"]
with ThreadPoolExecutor(max_workers=3) as pool:
    futures = [pool.submit(thumbnail, image) for image in images]
for image, future in zip(images, futures):
    try:
        print(f"{image}: {future.result()} KB")
    except ValueError as error:
        print(f"{image}: грешка: {error}")
Изход
чанта.jpg: 47 KB
обувки.jpg: 50 KB
повредена.jpg: грешка: повреден файл
яке.jpg: 41 KB
result() хвърля наново същото изключение (ValueError). Другите снимки не страдат. С pool.map грешката излиза, когато обхождането стигне до повредената снимка, и резултатите след нея се губят. Затова, ако очакваш грешки, ползвай submit и result() за всяка.

Пробвай

Къде отиде грешката

Втората снимка е повредена. Какво ще отпечата програмата?

images = ["чанта.jpg", "повредена.jpg", "яке.jpg"]
with ThreadPoolExecutor(max_workers=3) as pool:
    for image in images:
        pool.submit(thumbnail, image)
print(f"Готово: {len(images)} снимки")
Отговор

Идеята

Колко работника

Зависи от това какво правят:

  • Работа, която смята (смаляване, PDF, компресия): колкото са ядрата. Девети работник на 8 ядра няма къде да върви, само се редува с другите.
  • Работа, която чака (имейл, плащане, база): повече. Докато един работник чака мрежата, ядрото е свободно за друг. Ако задачата чака W ms и смята C ms, на ядро се събират около 1 + W / C работника.
def pool_size(cores, wait_ms, compute_ms):
    return cores * (1 + wait_ms // compute_ms)




print(f"Снимки (само смятане): {pool_size(8, 0, 50)}")
print(f"Имейли (90 ms чакане, 10 ms смятане): {pool_size(8, 90, 10)}")
Изход
Снимки (само смятане): 8
Имейли (90 ms чакане, 10 ms смятане): 80

Броят ядра дава os.cpu_count(). Формулата е начало, не закон: измери. Ако имейлите ползват един сървър с ограничение от 20 връзки, 80 работника само ще чакат на опашка пред него.

В Python има уловка: заради GIL нишките не смятат едновременно. За работа, която смята, пулът е от процеси: ProcessPoolExecutor, всеки процес със свой Python и свой GIL. Задачите и резултатите минават между процесите като копия:

from concurrent.futures import ProcessPoolExecutor




def checksum(invoice):
    return sum((invoice * 31 + i) % 256 for i in range(200_000))




if __name__ == "__main__":
    invoices = [1001, 1002, 1003, 1004]
    with ProcessPoolExecutor(max_workers=4) as pool:
        sums = list(pool.map(checksum, invoices))
    for invoice, total in zip(invoices, sums):
        print(f"Фактура {invoice}: {total}")
Изход
Фактура 1001: 25497376
Фактура 1002: 25499360
Фактура 1003: 25501344
Фактура 1004: 25503328

if __name__ == "__main__": пази главния код да не се изпълни наново във всеки процес: на macOS и Windows процесът тръгва, като зарежда файла отначало.

Пробвай

Попълни

Колко работника за задача, която чака waitMs и смята computeMs?

def pool_size(cores, wait_ms, compute_ms):
    return cores * (1 + )
По-точно, за любопитните

За работа, която чака, в Python има и asyncio: една нишка чака хиляди заявки (урокът async). Пулът от нишки остава за библиотеки, които чакат, без да могат await.

Упражнение

Фактурите на разпродажбата

Реши задачата в редактора си.

След разпродажбата всяка поръчка иска фактура. invoice(amount) прави PDF-а (тук е пауза от 50 ms) и връща сумата с 20% ДДС. Поръчка със сума 0 или по-малко е грешна и функцията хвърля изключение с текст „невалидна сума“.

Поръчките са 1001 до 1008 със суми 50, 125, 80, 0, 240, 35, 90 и 15 €. Направи фактурите в пул от 4 работника. Отпечатай ги по реда на поръчките, а грешната с текста на грешката. Накрая: сборът на добрите фактури и колко са с грешка.

Изход
Поръчка 1001: 60 €
Поръчка 1002: 150 €
Поръчка 1003: 96 €
Поръчка 1004: грешка: невалидна сума
Поръчка 1005: 288 €
Поръчка 1006: 42 €
Поръчка 1007: 108 €
Поръчка 1008: 18 €
Общо: 762 €, с грешка: 1

Още задачи

  • Ограничена опашка. Направи пула така, че в опашката да чакат най-много 10 задачи. Какво става с единадесетата? (threading.BoundedSemaphore около submit)
  • Напредък. Отпечатвай „готови 3 от 8“ след всяка свършена фактура, а накрая пак подредения списък.
  • Таймаут. Фактура, която се прави над 1 секунда, да се брои за грешна, без да спира останалите.

Тест

Провери се

1. Защо пул от няколко нишки, а не нишка за всяка качена снимка?
2. Пул от 3 работника смалява 8 снимки. Третата свършва първа. Къде е размерът ѝ, когато ги взимаш по реда на списъка с get, result, Promise.all или sizes[i]?
3. Задача в пула хвърля изключение. Кога го виждаш?
4. Сървърът има 8 ядра. Всеки имейл чака мрежата 90 ms и смята 10 ms. Около колко работника трябват, за да са заети ядрата?
5. Колко работника за смаляване на снимки, ако то само смята, а сървърът има 8 ядра?

Създаден от Иво Гергов като част от кампанията на Evo Tech за достъпно обучение по програмиране ·