Производител и потребител (producer-consumer) в C#

Проблем

Черен петък

В 10:00 започва разпродажбата и сайтът приема по 50 поръчки в секунда. Складът опакова по 10. Има три лоши начина да се справиш с това.

Сайтът опакова сам. Клиентът натиска „Поръчай“ и чака, докато някой не опакова кашона. Сайтът е бърз колкото склада.

Сайтът слага поръчките в обикновен списък, а складът ги взима оттам. Два опаковчика могат да вземат една и съща поръчка, а друга да се загуби: това е race condition. Ако сложиш ключалка, опаковчик с празен списък трябва да проверява пак и пак, дали нещо е дошло.

Списъкът няма граница. Сайтът е доволен: никога не чака. Но всяка секунда списъкът расте с 40 поръчки. След час в него има 144 000, последната е чакала час, а паметта свършва.

Идеята

Опашка с ограничено място

Между двете страни се слага опашка. Производителят (producer) слага задачи в края ѝ, потребителите (consumers) ги взимат отначало. Това е шаблонът производител и потребител. Двете страни не се познават: сайтът не знае колко опаковчици има, а опаковчиците не знаят откъде идват поръчките.

Опашката е блокираща и с ограничено място:

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

В C# това е BlockingCollection<T> с boundedCapacity: 3. Add чака, ако е пълна. GetConsumingEnumerable() е цикъл, който взима, докато има, и чака, ако е празна. Тя е в System.Collections.Concurrent, като ConcurrentBag, в която събираме опакованите поръчки.

Виж го

Сайтът и опаковчиците

static void Front(List<Order> orders,
                  BlockingCollection<Order> queue)
{
    foreach (var order in orders)
    {
        queue.Add(order); // чака, ако е пълна
    }
    queue.CompleteAdding(); // край: опаковчиците спират
}

static void Pack(BlockingCollection<Order> queue,
                 ConcurrentBag<Order> packed)
{
    // чака, ако няма
    foreach (var order in queue.GetConsumingEnumerable())
    {
        packed.Add(order);
    }
}

Стъпка по стъпка, с думи

Сайтът (производител) слага поръчки в опашка с 3 места, двама опаковчици (потребители) ги взимат отпред. Тя е празна и те чакат в GetConsumingEnumerable.
Сайтът слага 1001 и 1002 в края на опашката.
Всеки опаковчик взима поръчката отпред и я опакова. Опаковането е по-бавно от приемането на поръчки.
Сайтът слага 1003, 1004 и 1005. И трите места са заети.
1006 не се събира: Add чака и сайтът спира да приема поръчки. Това е backpressure: бавната страна задържа бързата.
Опаковчик 1 приключва с 1001 и взима 1003. Освобождава се място и чакащият put пуска 1006.
И така нататък: опаковчиците взимат отпред, сайтът добавя отзад и в опашката никога няма повече от 3.
Поръчките свършиха. CompleteAdding казва на опашката, че няма да дойде нищо повече.
Празна и завършена: GetConsumingEnumerable свършва и всеки foreach спира. Без CompleteAdding опаковчиците щяха да чакат вечно.
8 от 8, всяка веднъж: 3 експресни за 90 €, 5 обикновени за 180 €. Кой коя опакова се сменя при всяко пускане, сумите не.

Поръчката:

record Order(int Id, int Euro, bool Express);

Сайтът и опаковчиците:

static void Front(List<Order> orders,
                  BlockingCollection<Order> queue)
{
    foreach (var order in orders)
    {
        queue.Add(order); // чака, ако е пълна
    }
    queue.CompleteAdding(); // край: опаковчиците спират
}

static void Pack(BlockingCollection<Order> queue,
                 ConcurrentBag<Order> packed)
{
    // чака, ако няма
    foreach (var order in queue.GetConsumingEnumerable())
    {
        packed.Add(order);
    }
}

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

var orders = new List<Order>
{
    new(1001, 25, false),
    new(1002, 40, true),
    new(1003, 15, false),
    new(1004, 60, false),
    new(1005, 30, true),
    new(1006, 45, false),
    new(1007, 20, true),
    new(1008, 35, false),
};
var queue = new BlockingCollection<Order>(boundedCapacity: 3);
var packed = new ConcurrentBag<Order>();
var threads = new List<Thread>
{
    new Thread(() => Front(orders, queue)),
    new Thread(() => Pack(queue, packed)),
    new Thread(() => Pack(queue, packed)),
};
foreach (var thread in threads) thread.Start();
foreach (var thread in threads) thread.Join();
Report(orders, packed);

Отчетът проверява неща, които не зависят от това кой опаковчик коя поръчка е взел:

static void Report(List<Order> orders, ConcurrentBag<Order> packed)
{
    int distinct = packed.Select(o => o.Id).Distinct().Count();
    Console.WriteLine($"Опаковани: {packed.Count} от {orders.Count}");
    Console.WriteLine($"Всяка поръчка веднъж: {(distinct == orders.Count ? "да" : "не")}");
    foreach (var express in new[] { true, false })
    {
        var group = packed.Where(o => o.Express == express).ToList();
        string kind = express ? "Експресни" : "Обикновени";
        Console.WriteLine($"{kind}: {group.Count} поръчки, {group.Sum(o => o.Euro)} €");
    }
}

Опаковани: 8 от 8
Всяка поръчка веднъж: да
Експресни: 3 поръчки, 90 €
Обикновени: 5 поръчки, 180 €

Изходът е един и същ при всяко пускане. Кой опаковчик колко поръчки е взел, се сменя, затова програмата не го печата.

Идеята

Как спират опаковчиците

Опаковчикът чака в GetConsumingEnumerable(), докато не дойде поръчка. Ако сайтът просто свърши, опаковчиците ще чакат вечно и програмата няма да свърши.

Затова сайтът вика накрая CompleteAdding(): „няма да има още“. След това Add хвърля изключение, а GetConsumingEnumerable() свършва, щом опашката е празна. Всеки foreach по нея спира сам, колкото и опаковчици да има. Поръчките, които са вече вътре, се опаковат преди това.

Другият начин, който ще срещнеш в чужд код, е отровното хапче (poison pill): специална стойност, която не е поръчка, по една за всеки потребител. В C# не трябва.

Идеята

Backpressure

Пълната опашка кара сайта да чака. Това не е грешка, а целта: бавната страна дърпа назад бързата. На английски това е backpressure. Сайтът не може да изпревари склада с повече от размера на опашката, паметта не расте, а последната поръчка чака най-много толкова, колкото трябва да се опаковат 3 пред нея.

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

TryAdd(order) връща false, ако няма място. TryAdd(order, 1000) чака най-много секунда.

С него сайтът може да покаже „Голям интерес, опитайте след минута“ вместо да виси. Важното е, че решението е твое: поръчката не изчезва тихо.

Пробвай

Предвиди изхода

Опашка с 2 места и без опаковчици. Какво ще отпечата?

var queue = new BlockingCollection<string>(boundedCapacity: 2);
Console.WriteLine(queue.TryAdd("1001"));
Console.WriteLine(queue.TryAdd("1002"));
Console.WriteLine(queue.TryAdd("1003"));
Console.WriteLine(queue.Take());
Console.WriteLine(queue.TryAdd("1003"));
Console.WriteLine(queue.Count);

Отговор

True
True
True
1001
True
3

True
True
False
1001
True
2

True
True
False
1002
True
2

Пробвай

Попълни

Какво вика сайтът накрая, за да спрат опаковчиците?

queue.(); // край: опаковчиците спират

CompleteAdding(): след него GetConsumingEnumerable() свършва, щом опашката се изпразни.

Пробвай

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

Колега не иска сайтът да чака и сменя един ред. Програмата вече не виси никога, а отчетът понякога казва „Опаковани: 5 от 8“. На кой ред е грешката?

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

Виж го

Нишки, Worker-и и съобщения

В нов C# код ще срещнеш и Channel<T> (System.Threading.Channels). Channel.CreateBounded<Order>(3) е същата опашка с 3 места, но за async код: await writer.WriteAsync(order) чака без да блокира нишка, writer.Complete() е CompleteAdding, а await foreach (var o in reader.ReadAllAsync()) е потребителят. BlockingCollection блокира нишки, Channel ги пуска, докато чака.

Упражнение

Два входа, три опаковчика

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

Поръчките идват от два входа: сайта (2001–2006) и мобилното приложение (3001–3006). Всеки вход е отделен производител, а опаковчиците са трима. Опашката е с 4 места. Хапчетата (CompleteAdding) идват чак след като и двата входа свършат. Отпечатай:

Опаковани: 12 от 12
Всяка поръчка веднъж: да
Общо: 325 €

Още задачи

Отказ. Направи опашката с 1 място и сайта с TryAdd(order, 10). Брой отказаните поръчки и отпечатай „приети + отказани = 12“.
По вид. Експресните поръчки трябва да се опаковат първи. две опашки и опаковчик, който гледа първо експресната помага ли? Какво става с хапчетата?
Верига. Добави втора опашка след опаковчиците: те слагат опакованите кашони в нея, а куриер ги взима. Колко хапчета трябват сега и кой ги слага?

Тест

Провери се

1. Опашката между сайта и опаковчиците е пълна. Какво прави put?Чака, докато опаковчик не вземе поръчка и не освободи мястоХвърля поръчкатаУвеличава опашката с още едно място

2. Защо опашката изобщо има граница? Нали без нея сайтът никога не чака?Защото опашките без граница не са безопасни между нишкиБез граница бавният склад не спира бързия сайт: опашката расте, докато не свърши паметта, а последната поръчка чака все по-дългоНяма причина: границата само забавя програмата

3. Два сайта слагат поръчки, трима опаковчици ги взимат. Колко хапчета трябват и кога се слагат?Три, по едно за опаковчик, след като и двата сайта са свършили (след join)Едно: първият опаковчик, който го види, казва на другитеДве, по едно от всеки сайт, когато свърши

4. Как разбираш, че програмата с опашката е вярна, щом редът на опаковане е различен при всяко пускане?Проверяваш неща, които не зависят от реда: всяка поръчка е опакована точно веднъж, броят и сумите по видСлагаш sleep, за да е редът винаги същиятПускаш я, докато два пъти подред не излезе един и същ ред

Производител и потребител

Бисквитки