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

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

Накратко: Пулът от нишки е няколко работника пред опашка от задачи, вместо нова нишка за всяка задача; в .NET това е общият басейн с Task.Run, Parallel.For и Task.WhenAll.

Проблем

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

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

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

var threads = new List<Thread>();
foreach (string image in uploads)
{
    var thread = new Thread(() => SaveThumbnail(image));
    thread.Start();
    threads.Add(thread);
}

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

  • Всяка нишка иска свой стек (обикновено до 1 MB) и място в операционната система. Хиляди нишки изяждат паметта на сървъра.
  • Дори да има памет, 10 000 нишки на 8 ядра само се редуват. Процесорът губи време да ги превключва, а всяка снимка става по-бавна.

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

Идеята

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

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

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

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

С пула вече се срещна в урока за нишките: Task.Run, Task<int> и .Result. Тук ще видим какво става вътре, как да подредиш резултатите, какво става с грешките и колко работника трябват.

Виж го

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

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

// Смаляване на снимка: тук е пауза, а размерът идва от името.
static int Thumbnail(string image)
{
    if (image.StartsWith("повредена"))
    {
        throw new ArgumentException("повреден файл");
    }
    int kb = 20 + 3 * image.Length;
    Thread.Sleep(2 * kb);
    return kb;
}

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

int[] sizes = new int[images.Length];
var options = new ParallelOptions
{
    MaxDegreeOfParallelism = 3,
};
Parallel.For(0, images.Length, options, i =>
{
    sizes[i] = Thumbnail(images[i]);
});
int total = 0;
for (int i = 0; i < images.Length; i++)
{
    Console.WriteLine($"{images[i]}: {sizes[i]} KB");
    total += sizes[i];
}
Console.WriteLine($"Общо: {total} KB");
Стъпка по стъпка, с думи
  1. 8 снимки чакат. Parallel.For ще ползва най-много 3 нишки от басейна на .NET.
  2. Parallel.For раздава индексите. Трите нишки взимат първите три снимки.
  3. Нишка 3 свършва първа: яке е третата снимка и резултатът отива на третото място. Работникът не спира: взима следващата задача.
  4. Опашката е празна. Нишка 1 свърши и няма какво да вземе, затова чака. Другите довършват последните снимки.
  5. Всеки резултат е в sizes[i], затова редът е този на снимките. Общо 379 KB.
string[] images = { "чанта.jpg", "обувки.jpg", "яке.jpg", "шапка.jpg", "часовник.jpg", "раница.jpg", "колан.jpg", "шал.jpg" };
int[] sizes = new int[images.Length];
var options = new ParallelOptions
{
    MaxDegreeOfParallelism = 3,
};
Parallel.For(0, images.Length, options, i =>
{
    sizes[i] = Thumbnail(images[i]);
});
int total = 0;
for (int i = 0; i < images.Length; i++)
{
    Console.WriteLine($"{images[i]}: {sizes[i]} KB");
    total += sizes[i];
}
Console.WriteLine($"Общо: {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
  • .NET има един басейн от нишки (ThreadPool) за цялата програма и сам решава колко нишки да държи. Не го създаваш и не го затваряш.
  • Parallel.For пуска тялото за всеки индекс в басейна и чака всички. MaxDegreeOfParallelism = 3 казва „най-много три наведнъж“: това е пулът от трима.
  • Всяка задача пише в своето място sizes[i], затова не трябва lock и резултатите са по реда на снимките.
По-точно, за любопитните

Колко нишки държи басейнът, .NET решава сам: започва с толкова, колкото са ядрата, и добавя, ако задачите стоят дълго, без да смятат. ThreadPool.GetMaxThreads и SetMinThreads го настройват, но рядко има нужда.

Parallel.ForEach(images, options, image => …) е същото за колекция, а Parallel.ForEachAsync е за задачи с await. Нишката, която вика Parallel.For, също работи и чака накрая, затова в уеб приложение не я викай в нишката на заявката.

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

Пробвай

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

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

var watch = Stopwatch.StartNew();
var sizes = new List<int>();
foreach (string image in images)
{
    Task<int> size = Task.Run(() => Thumbnail(image));
    sizes.Add(await size);
}
Console.WriteLine($"Отне над 0,5 s: {watch.ElapsedMilliseconds > 500}");
Изход
Отне над 0,5 s: True

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

Идеята

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

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

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

  • По реда на входа. Пазиш обещанията в списък по реда на задачите и ги чакаш в този ред. Така работи кодът отгоре. В C# Task.WhenAll дава масив по реда на задачите, а не по реда на свършване:
  • Сбор или брой. Ако ти трябва само общото, редът няма значение: 47 + 50 е колкото 50 + 47.
Task<int>[] tasks = images.Select(image => Task.Run(() => Thumbnail(image))).ToArray();
int[] sizes = await Task.WhenAll(tasks);
for (int i = 0; i < images.Length; i++)
{
    Console.WriteLine($"{images[i]}: {sizes[i]} KB");
}
Изход
чанта.jpg: 47 KB
обувки.jpg: 50 KB
яке.jpg: 41 KB
шапка.jpg: 47 KB

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

var running = images.Select(image => Task.Run(() => Thumbnail(image))).ToList();
int count = 0;
int total = 0;
while (running.Count > 0)
{
    Task<int> done = await Task.WhenAny(running);
    running.Remove(done);
    total += await done;
    count++;
}
Console.WriteLine($"Готови: {count} от {images.Length}, общо: {total} KB");
Изход
Готови: 8 от 8, общо: 379 KB
Task.WhenAny дава първата готова задача от списъка. Махаш я и чакаш следващата. При много задачи това е O(n²), затова .NET 9 добави Task.WhenEach, който ги дава една по една.

Идеята

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

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

string[] images = { "чанта.jpg", "обувки.jpg", "повредена.jpg", "яке.jpg" };
var tasks = images.Select(image => Task.Run(() => Thumbnail(image))).ToList();
for (int i = 0; i < images.Length; i++)
{
    try
    {
        int kb = await tasks[i];
        Console.WriteLine($"{images[i]}: {kb} KB");
    }
    catch (ArgumentException e)
    {
        Console.WriteLine($"{images[i]}: грешка: {e.Message}");
    }
}
Изход
чанта.jpg: 47 KB
обувки.jpg: 50 KB
повредена.jpg: грешка: повреден файл
яке.jpg: 41 KB
await хвърля самото изключение (ArgumentException). .Result би го хвърлил увито в AggregateException. Другите снимки не страдат.

Пробвай

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

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

string[] images = { "чанта.jpg", "повредена-1.jpg", "повредена-2.jpg" };
Task<int[]> all = Task.WhenAll(images.Select(image => Task.Run(() => Thumbnail(image))));
try
{
    await all;
}
catch (ArgumentException e)
{
    Console.WriteLine($"Хванато: {e.Message}");
}
Console.WriteLine($"Грешки в задачата: {all.Exception!.InnerExceptions.Count}");
Отговор

Идеята

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

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

  • Работа, която смята (смаляване, PDF, компресия): колкото са ядрата. Девети работник на 8 ядра няма къде да върви, само се редува с другите.
  • Работа, която чака (имейл, плащане, база): повече. Докато един работник чака мрежата, ядрото е свободно за друг. Ако задачата чака W ms и смята C ms, на ядро се събират около 1 + W / C работника.
static int PoolSize(int cores, int waitMs, int computeMs)
{
    return cores * (1 + waitMs / computeMs);
}


Console.WriteLine($"Снимки (само смятане): {PoolSize(8, 0, 50)}");
Console.WriteLine($"Имейли (90 ms чакане, 10 ms смятане): {PoolSize(8, 90, 10)}");
Изход
Снимки (само смятане): 8
Имейли (90 ms чакане, 10 ms смятане): 80

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

Пробвай

Попълни

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

static int PoolSize(int cores, int waitMs, int computeMs)
{
    return cores * (1 + );
}
По-точно, за любопитните

В C# за работа, която чака, не се държат нишки изобщо: await освобождава нишката, докато заявката чака мрежата (урокът async и await). Затова уеб сървърът обслужва хиляди заявки с малък басейн. Ресурсите зад тях (връзките към базата) се ограничават, например със SemaphoreSlim.

Упражнение

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

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

След разпродажбата всяка поръчка иска фактура. 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 задачи. Какво става с единадесетата? (Channel.CreateBounded или SemaphoreSlim)
  • Напредък. Отпечатвай „готови 3 от 8“ след всяка свършена фактура, а накрая пак подредения списък.
  • Таймаут. Фактура, която се прави над 1 секунда, да се брои за грешна, без да спира останалите.

Тест

Провери се

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

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