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

Код на:JavaC#PythonJavaScript

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

Проблем

Черен петък

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

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

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

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

Идеята

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

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

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

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

В JavaScript готова такава опашка няма. В една нишка с async функции тя се пише за по-малко от 40 реда с обещания: put и take връщат обещание, което се изпълнява, когато има място или поръчка. Ключалка не трябва, защото кодът между два await не може да бъде прекъснат.

BoundedQueue

class BoundedQueue {
  #items = [];
  #putters = []; // чакащите да сложат
  #takers = []; // чакащите да вземат

  constructor(capacity) {
    this.capacity = capacity;
  }

  get size() {
    return this.#items.length;
  }

  async put(item) {
    while (this.#items.length >= this.capacity) {
      await new Promise((resolve) => this.#putters.push(resolve));
    }
    this.#items.push(item);
    this.#takers.shift()?.();
  }

  tryPut(item) {
    if (this.#items.length >= this.capacity) return false;
    this.#items.push(item);
    this.#takers.shift()?.();
    return true;
  }

  async take() {
    while (this.#items.length === 0) {
      await new Promise((resolve) => this.#takers.push(resolve));
    }
    const item = this.#items.shift();
    this.#putters.shift()?.();
    return item;
  }
}

#putters и #takers са опашки от resolve функции: кой чака място и кой чака поръчка. take събужда един чакащ производител, put един чакащ потребител. Събуденият проверява пак (while), защото някой друг може да е бил по-бърз.

Виж го

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

const DONE = null;

async function front(orders, queue, packers) {
  for (const order of orders) {
    await queue.put(order); // чака, ако е пълна
  }
  for (let i = 0; i < packers; i++) {
    await queue.put(DONE); // хапче за всеки опаковчик
  }
}

async function pack(queue, packed) {
  while (true) {
    const order = await queue.take(); // чака, ако няма
    if (order === DONE) return; // хапчето: спри
    packed.push(order);
  }
}

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

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

Поръчката:

// поръчката е { id, euro, express }

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

const DONE = null;

async function front(orders, queue, packers) {
  for (const order of orders) {
    await queue.put(order); // чака, ако е пълна
  }
  for (let i = 0; i < packers; i++) {
    await queue.put(DONE); // хапче за всеки опаковчик
  }
}

async function pack(queue, packed) {
  while (true) {
    const order = await queue.take(); // чака, ако няма
    if (order === DONE) return; // хапчето: спри
    packed.push(order);
  }
}

front и двата pack са три async функции в една нишка. await е мястото, където всяка може да чака и да пусне другите. Това е разделяне на работата, не паралелна работа: за опаковане, което смята много, трябват Worker-и (виж по-долу).

const orders = [
  { id: 1001, euro: 25, express: false },
  { id: 1002, euro: 40, express: true },
  { id: 1003, euro: 15, express: false },
  { id: 1004, euro: 60, express: false },
  { id: 1005, euro: 30, express: true },
  { id: 1006, euro: 45, express: false },
  { id: 1007, euro: 20, express: true },
  { id: 1008, euro: 35, express: false },
];
const queue = new BoundedQueue(3);
const packed = [];
await Promise.all([
  front(orders, queue, 2),
  pack(queue, packed),
  pack(queue, packed),
]);
report(orders, packed);

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

function report(orders, packed) {
  const distinct = new Set(packed.map((order) => order.id)).size;
  console.log(`Опаковани: ${packed.length} от ${orders.length}`);
  console.log(`Всяка поръчка веднъж: ${distinct === orders.length ? "да" : "не"}`);
  for (const express of [true, false]) {
    const group = packed.filter((order) => order.express === express);
    const euro = group.reduce((sum, order) => sum + order.euro, 0);
    const kind = express ? "Експресни" : "Обикновени";
    console.log(`${kind}: ${group.length} поръчки, ${euro} €`);
  }
}

Изход

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

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

Идеята

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

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

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

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

Тук хапчето е DONE = null.

Идеята

Backpressure

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

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

tryPut(order) в нашата опашка връща false, ако няма място.

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

Пробвай

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

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

const queue = new BoundedQueue(2);
console.log(queue.tryPut("1001"));
console.log(queue.tryPut("1002"));
console.log(queue.tryPut("1003"));
console.log(await queue.take());
console.log(queue.tryPut("1003"));
console.log(queue.size);

Отговор

true
true
false
1001
true
2

true
true
false
1002
true
2

true
true
true
1001
true
3

Провери

Пробвай

Попълни

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

for (let i = 0; i < ; i++) {
  await queue.put(DONE);
}

Провери

Пробвай

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

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

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

1async function front(orders, queue, packers) {2 for (const order of orders) {3 queue.tryPut(order);4 }5 for (let i = 0; i < packers; i++) {6 await queue.put(DONE);7 }8}

Провери

Виж го

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

Когато опаковането смята много, трябват истински нишки: Worker-и. Между тях опашка с общи обекти няма. Главната нишка праща всяка поръчка на Worker със postMessage, а той я получава в on("message"):

import { Worker, isMainThread, parentPort } from "node:worker_threads";
import { once } from "node:events";

if (isMainThread) {
  const orders = [
    { id: 1001, euro: 25, express: false },
    { id: 1002, euro: 40, express: true },
    { id: 1003, euro: 15, express: false },
    { id: 1004, euro: 60, express: false },
    { id: 1005, euro: 30, express: true },
    { id: 1006, euro: 45, express: false },
    { id: 1007, euro: 20, express: true },
    { id: 1008, euro: 35, express: false },
  ];
  const packers = [0, 1].map(() => new Worker(new URL(import.meta.url)));
  const packed = [];
  const reports = packers.map((packer) => once(packer, "message"));
  orders.forEach((order, i) => packers[i % 2].postMessage(order)); // не чака никога
  packers.forEach((packer) => packer.postMessage(null));
  for (const [done] of await Promise.all(reports)) packed.push(...done);
  report(orders, packed);
} else {
  const done = [];
  parentPort.on("message", (order) => {
    if (order === null) {
      parentPort.postMessage(done);
      parentPort.close();
    } else {
      done.push(order);
    }
  });
}

Изход

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

Това също е производител и потребител, но с две разлики. Първо, граница няма: postMessage никога не чака, съобщенията се трупат в паметта на Worker-а и backpressure трябва да си направиш сам (Worker-ът отговаря „готов“ и чак тогава получава следващата). Второ, поръчките не се взимат от обща опашка, а главната нишка решава коя на кой Worker да отиде (тук поред). Поръчката се копира при пращането, затова race condition върху нея няма.

Упражнение

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

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

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

Изход

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

Още задачи

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

Тест

Провери се

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

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

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

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

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

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

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

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

Провери

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

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

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