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

Код на:JavaC#PythonJavaScript

Накратко: Производител и потребител е шаблон, в който едни нишки слагат задачи в опашка с ограничено място, а други ги взимат; пълната опашка кара производителя да чака (backpressure).

Проблем

Черен петък

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

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

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

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

Идеята

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

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

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

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

В Python това е queue.Queue(maxsize=3). put чака, ако е пълна, а get чака, ако е празна. (Модулът queue е за нишки. Има и asyncio.Queue със същите методи за async код.)

Виж го

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

DONE = None


def front(orders, queue, packers):
    for order in orders:
        queue.put(order)  # чака, ако е пълна
    for _ in range(packers):
        queue.put(DONE)  # хапче за всеки опаковчик


def pack(queue, packed):
    while True:
        order = queue.get()  # чака, ако няма
        if order is DONE:
            return  # хапчето: спри
        packed.append(order)

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

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

Поръчката:

@dataclass(frozen=True)
class Order:
    id: int
    euro: int
    express: bool

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

DONE = None


def front(orders, queue, packers):
    for order in orders:
        queue.put(order)  # чака, ако е пълна
    for _ in range(packers):
        queue.put(DONE)  # хапче за всеки опаковчик


def pack(queue, packed):
    while True:
        order = queue.get()  # чака, ако няма
        if order is DONE:
            return  # хапчето: спри
        packed.append(order)

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

orders = [
    Order(1001, 25, False),
    Order(1002, 40, True),
    Order(1003, 15, False),
    Order(1004, 60, False),
    Order(1005, 30, True),
    Order(1006, 45, False),
    Order(1007, 20, True),
    Order(1008, 35, False),
]
queue = Queue(maxsize=3)
packed = []  # list.append е безопасен от много нишки
threads = [threading.Thread(target=front, args=(orders, queue, 2)),
           threading.Thread(target=pack, args=(queue, packed)),
           threading.Thread(target=pack, args=(queue, packed))]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()
report(orders, packed)

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

def report(orders, packed):
    distinct = len({order.id for order in packed})
    print(f"Опаковани: {len(packed)} от {len(orders)}")
    print("Всяка поръчка веднъж:", "да" if distinct == len(orders) else "не")
    for express in (True, False):
        group = [order for order in packed if order.express == express]
        kind = "Експресни" if express else "Обикновени"
        print(f"{kind}: {len(group)} поръчки, {sum(o.euro for o in group)} €")

Изход

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

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

Идеята

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

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

Затова сайтът слага накрая отровно хапче (poison pill): специална стойност, която не е поръчка. Опаковчик, който вземе хапчето, спира. Хапчетата са по едно за всеки опаковчик: всеки взима само едно и не го връща.

Хапчетата влизат в опашката след поръчките, затова всяка поръчка се опакова преди тях.

Тук хапчето е DONE = None, а is DONE го сравнява по идентичност.

Идеята

Backpressure

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

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

put_nowait(order) хвърля queue.Full, ако няма място. put(order, timeout=1) чака най-много секунда и тогава хвърля Full.

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

Пробвай

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

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

def offer(queue, item):
    try:
        queue.put_nowait(item)
        return True
    except Full:
        return False


queue = Queue(maxsize=2)
print(offer(queue, "1001"))
print(offer(queue, "1002"))
print(offer(queue, "1003"))
print(queue.get())
print(offer(queue, "1003"))
print(queue.qsize())

Отговор

True
True
True
1001
True
3

True
True
False
1001
True
2

True
True
False
1002
True
2

Провери

Пробвай

Попълни

Колко хапчета слага сайтът?

for _ in range():
    queue.put(DONE)

Провери

Пробвай

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

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

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

1def front(orders, queue, packers):2 for order in orders:3 if not queue.full():4 queue.put(order)5 for _ in range(packers):6 queue.put(DONE)

Провери

Виж го

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

Queue() без maxsize е без граница: put никога не чака и проблемът от началото на урока се връща. В async код опашката е asyncio.Queue(maxsize=3) с await queue.put(...) и await queue.get(): същото, но в една нишка.

Упражнение

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

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

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

Изход

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

Още задачи

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

Тест

Провери се

1. Опашката между сайта и опаковчиците е пълна. Какво прави put?

Чака, докато опаковчик не вземе поръчка и не освободи мястоХвърля поръчкатаУвеличава опашката с още едно място

2. Защо опашката изобщо има граница? Нали без нея сайтът никога не чака?

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

3. Два сайта слагат поръчки, трима опаковчици ги взимат. Колко хапчета трябват и кога се слагат?

Три, по едно за опаковчик, след като и двата сайта са свършили (след join)Едно: първият опаковчик, който го види, казва на другитеДве, по едно от всеки сайт, когато свърши

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

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

Провери

Започни урока отначало

Свързани уроци

• Нужно отпреди: Race condition · Заключване · Атомарни операции и неизменими обекти · Deadlock
• Продължава в: Пул от нишки и задачи · Събития и опашки
• Задачи по темата: Нишки и race conditions