Node.js backpressure: write() false, drain va xotiraning to'lib ketishi

Qisqacha: Backpressure (orqaga bosim) — sekin yozuvchining tez manbaga "to'xta, ulgurmayapman" deyishi. Node'da bu signal — write() ning qaytargan qiymati: false bo'lsa, ichki bufer highWaterMark dan oshgan, "drain" hodisasini kutish kerak. Kanon o'lchadi: sekin oqimga 100 000 ta 1 KB lik bo'lak yozildi. Signalga qaramay yozilganda buferda 98 MB to'plandi, xotira 35 → 189 MB; pipeline bilan — buferda 16 KB dan oshmadi. false ni e'tiborsiz qoldirish hatto tezroq ham emas.

Bu darsda

• write() nega false qaytarishini va bu signalga qanday javob berishni bilasiz.
• "drain" hodisasini await once(w, "drain") bilan kutib, to'g'ri yozish siklini yozasiz.
• highWaterMark bufer chegarasi ekanini, objectMode da bo'laklar soni bilan o'lchanishini tushuntirasiz.
• Backpressure'siz yozishda xotira qanday o'sishini o'lchov bilan ko'rsatasiz.
• pipeline buni nega o'zi hal qilishini va valyuta da qayerda ishlashini bilasiz.

Oldin bilishingiz kerak: Stream nima: to'rt turi va rejimlari, EventEmitter chuqur, Buffer va binar ma'lumot, Event loop'ni bloklash.

1. Nega bu kerak?

Tasavvur qiling: diskdan fayl sekundiga 2 GB tezlikda o'qilyapti, tarmoq esa sekundiga 10 MB jo'natadi. Yoki CBU arxividan kurslar tez keladi, gzip esa ularni sekinroq siqadi. Manba qabul qiluvchidan tez — bu oqimlarda doimiy holat, istisno emas.

Agar manba to'xtamasa, ortiqcha ma'lumot qayerga ketadi? Hech qayerga yo'qolmaydi — Writable uni o'z ichki buferiga to'playveradi. 2 GB fayl — xotirada 2 GB navbat. Stream nima darsida oqimni xotirani tejash uchun tanlagan edik — backpressure'siz u xuddi readFile kabi butun ma'lumotni xotirada ushlaydi, faqat chalkashroq yo'l bilan.

O'xshatish: osh markazidagi oshpaz va ofitsiant. Oshpaz daqiqada 10 ta lagan tayyorlaydi, ofitsiant 3 tasini olib boradi. Agar oshpaz to'xtamasa, stol laganlarga to'lib ketadi va oxiri yiqiladi. To'g'ri tartib: stolda 5 ta lagan bo'lsa — ofitsiant "yetadi" deydi, oshpaz kutadi; stol bo'shasa — "davom et". "Yetadi" — write() ning false i, "davom et" — "drain".

sequenceDiagram
  participant M as Manba (tez)
  participant W as Writable (sekin)
  M->>W: write(bo'lak) → true
  M->>W: write(bo'lak) → false
  Note over M: to'xtaydi, kutadi
  W-->>W: buferni bo'shatadi
  W-->>M: "drain"
  M->>W: write(bo'lak) → true

Diagrammada false dan keyin manba hech narsa yozmaydi. Bufer bo'shagach, Writable "drain" hodisasi bilan "yana yozsa bo'ladi" deydi.

2. write() ning javobi

2.1 Kichik misol

Bufer chegarasi — highWaterMark. Obyekt rejimida u bo'laklar soni: highWaterMark: 2 — navbatda 2 ta bo'lak bo'lsa, "to'ldi". Har yozuv 10 ms davom etadigan sekin Writable'ga to'rtta taomni birdaniga yozamiz:

import { Writable } from "node:stream";

const waiter = new Writable({
  objectMode: true,
  highWaterMark: 2,
  write(dish, encoding, callback) {
    setTimeout(callback, 10); // har lagan 10 ms
  },
});

for (const dish of ["osh", "manti", "somsa", "choy"]) {
  const ok = waiter.write(dish);
  console.log(`${dish}: ${ok} (navbatda ${waiter.writableLength})`);
}
waiter.on("drain", () =>
  console.log("drain:", waiter.writableLength),
);

Konsolda:

osh: true (navbatda 1)
manti: false (navbatda 2)
somsa: false (navbatda 3)
choy: false (navbatda 4)
drain: 0

Qatorma-qator:

• osh — navbatda 1 ta, chegaradan (2) kam → true.
• manti — navbatda 2 ta, chegaraga yetdi → false. Bu "endi to'xta" signali.
• somsa, choy — biz signalga qaramadik va yozaverdik. Ular qabul qilindi — write ma'lumotni rad etmaydi! Navbat 4 ga o'sdi, chegaradan ikki baravar.
• Hamma yozuv tugab, navbat 0 ga tushganda — "drain".

Eng muhim xulosa: false — taqiq emas, iltimos. Node buferga yozishni to'xtatmaydi. Iltimosni bajarish — sizning kodingiz vazifasi. writableLength — hozir navbatda qancha (bayt rejimida — bayt, obyekt rejimida — bo'lak).

2.2 To'g'ri sikl: false → drain ni kutish

import { Writable } from "node:stream";
import { once } from "node:events";

const waiter = new Writable({
  objectMode: true,
  highWaterMark: 2,
  write(dish, encoding, callback) {
    setTimeout(callback, 10);
  },
});

let maxQueue = 0;
for (let i = 1; i <= 20; i++) {
  const ok = waiter.write(`lagan ${i}`);
  maxQueue = Math.max(maxQueue, waiter.writableLength);
  if (!ok) await once(waiter, "drain"); // ✅ bo'shaguncha kutamiz
}
waiter.end();
await once(waiter, "finish");
console.log("eng ko'p navbat:", maxQueue);

Konsolda:

eng ko'p navbat: 2

20 ta lagan, lekin navbat hech qachon 2 dan oshmadi. await once(waiter, "drain") (EventEmitter chuqur) — false kelganda sikl to'xtaydi va navbat bo'shaganda davom etadi. Qoida bitta qatorda: if (!w.write(x)) await once(w, "drain").

Tekshirib ko'ring: write() false qaytardi, lekin siz baribir yana 1000 ta bo'lak yozdingiz. Bu bo'laklar yo'qoladimi?

3. Kanon o'lchovi: 98 MB va 16 KB

3.1 Tajriba

Endi haqiqiy hajmda. Kanon tajribasi (tajriba/t31-backpressure.mjs): sekin Writable (highWaterMark 16 KB, har yozuv bitta event loop aylanishi) va unga 100 000 ta 1 KB lik yangi bufer — jami ~98 MB. Ikki usul:

import { Writable, Readable } from "node:stream";
import { pipeline } from "node:stream/promises";

const slow = () => new Writable({
  highWaterMark: 16 * 1024,
  write(_chunk, _enc, cb) { setImmediate(cb); },
});
const piece = () => Buffer.alloc(1024, 97); // har safar yangi bufer

// 1) write() natijasiga qaramay
let w = slow(), falseCount = 0, maxBuffer = 0;
for (let i = 0; i < 1e5; i++) {
  if (!w.write(piece())) falseCount++;
  maxBuffer = Math.max(maxBuffer, w.writableLength);
}

// 2) pipeline — backpressure'ni o'zi boshqaradi
const source = Readable.from((function* () {
  for (let i = 0; i < 1e5; i++) yield piece();
})());
await pipeline(source, slow());

Kanon xom natijasi (natijalar/tajriba/t31-backpressure.txt, Node 24.21.0, Windows 11):

1) write() natijasiga qaramay: 99985 marta false, buferda eng ko'p 98 MB, rss 35 → 189 MB
2) pipeline (backpressure hurmat qilinadi): buferda eng ko'p 16 KB, rss 230 → 147 MB
Readable.from objectMode: true | sukut highWaterMark (bayt oqim): 16384

Birinchi usulda write() 100 000 dan 99 985 marta false dedi — faqat birinchi 15 ta yozuv 16 KB ga sig'di. Dastur 99 985 marta e'tiborsiz qoldirdi va butun 98 MB buferga tushdi. Jarayon xotirasi (rss) 35 MB dan 189 MB ga o'sdi. Ikkinchi usulda bufer hech qachon 16 KB dan oshmadi — highWaterMark ning o'zi. (2-qatordagi rss 230 dan boshlanadi — u birinchi usuldan qolgan xotira; uni ajratib o'lchash uchun pastdagi alohida o'lchovga qarang.)

Kanon izohi muhim: birinchi variantda tajriba bitta buferni qayta-qayta yozgan va rss faqat 48 MB bo'lgan — chunki buferda 100 000 ta havola bitta obyektga edi. Halol o'lchov uchun har bo'lak yangi yaratildi. Haqiqiy dasturda (fayl, tarmoq) har bo'lak doim yangi.

3.2 Uch usul alohida jarayonlarda

Biz uch usulni alohida jarayonda, eng yuqori xotirani har 5 ms da kuzatib o'lchadik (Node 24.21.0, Windows 11, i5-12500H, ikki marta):

ignore: 321 ms, eng ko'p rss 230 MB
drain: 272 ms, eng ko'p rss 44 MB
pipeline: 320 ms, eng ko'p rss 47 MB
ignore: 308 ms, eng ko'p rss 231 MB
drain: 286 ms, eng ko'p rss 45 MB
pipeline: 302 ms, eng ko'p rss 47 MB

Grafikda farq besh baravar. Vaqtga qarang: backpressure'ni e'tiborsiz qoldirish tezroq emas — 308–321 ms, drain bilan 272–286 ms. Writable baribir o'z tezligida ishlaydi; manba uni "quvib o'tsa" ham ish tezlashmaydi, faqat xotira to'ladi. Ustiga, ulkan bufer bilan ishlash axlat yig'uvchiga (GC) qo'shimcha yuk beradi.

Tekshirib ko'ring: Server bir vaqtda 50 ta foydalanuvchiga 100 MB lik faylni sekin internet orqali beryapti va backpressure'ni e'tiborsiz qoldiradi. Eng yomon holatda xotira qancha kerak?

4. highWaterMark va objectMode aniq

highWaterMark haqida uchta noto'g'ri tasavvur bor:

1. "Bu maksimal hajm" — yo'q. Bu chegara, undan oshganda write() false qaytaradi. Kanon o'lchovida bufer 98 MB ga yetdi, chegara esa 16 KB edi. U tavsiya, devor emas.
2. "Kattaroq — har doim tezroq" — shart emas. Katta bufer kamroq drain aylanishi demak, lekin har oqim ko'proq xotira oladi. Fayl o'qish 64 KB, boshqa bayt oqimlari Windows'da 16 KB (Linux/macOS'da 64 KB) — ko'p hollarda yaxshi muvozanat.
3. "Bayt bilan o'lchanadi" — faqat bayt rejimida. objectMode da — bo'laklar soni (sukut 16). Bitta bo'lak 10 MB lik obyekt bo'lsa, 16 ta bo'lak — 160 MB. Obyekt rejimida chegarani obyekt hajmiga qarab tanlang.

Readable tomonda ham xuddi shu mexanizm bor: o'z Readable'ingizni yozganda this.push(chunk) ham false qaytaradi — "o'quvchi ulgurmayapti, endi push qilma". Uni O'z Readable/Transform stream'ingizni yozish darsida ko'ramiz.

5. pipeline hammasini o'zi qiladi

Har joyda if (!write) await drain yozish — xato qilish oson. pipe() va pipeline() bu siklni o'zi bajaradi:

• Manbadan bo'lak oladi, qabul qiluvchiga yozadi.
• write() false qaytarsa — manbani pause() qiladi (Stream nima dagi paused rejim).
• "drain" kelganda — resume().

import { Readable, Writable } from "node:stream";
import { pipeline } from "node:stream/promises";

const waiter = new Writable({
  objectMode: true,
  highWaterMark: 2,
  write(dish, encoding, callback) {
    setTimeout(callback, 5);
  },
});

// kuzatish uchun: pipeline chaqiradigan write() ni o'rab olamiz
let maxQueue = 0, falses = 0;
const write = waiter.write.bind(waiter);
waiter.write = (chunk) => {
  const ok = write(chunk);
  if (!ok) falses++;
  maxQueue = Math.max(maxQueue, waiter.writableLength);
  return ok;
};

const orders = Array.from({ length: 30 }, (_, i) => `lagan ${i + 1}`);
await pipeline(Readable.from(orders), waiter);
console.log("eng ko'p navbat:", maxQueue, "| false:", falses);

Konsolda:

eng ko'p navbat: 2 | false: 15

pipeline ham write() dan 15 marta false oldi — lekin har safar to'xtadi. Shuning uchun 30 ta lagan navbatni 2 dan oshirmay yozildi. Bizning kodimizda esa sikl ham, drain ham yo'q — pipeline chaqiruvining o'zi. pipe va pipeline farqi — xatolar: pipe xatoni keyingi oqimga uzatmaydi va ochiq qoldiradi, pipeline esa hammasini yopadi va xatoni bitta joyga beradi. Kanon buni haqiqiy xato sifatida topgan — pipeline, finished va stream xatolari darsining markaziy misoli. Hozircha qoida: zanjir — pipeline bilan.

for await ham backpressure'ni hurmat qiladi: sikl tanasi tugamaguncha keyingi bo'lak so'ralmaydi. Tanada sekin ish (masalan bazaga yozish) bo'lsa, manba o'zi kutadi.

6. Backpressure qayerda buziladi

Odatiy xatolar ro'yxati — ularning hammasi "oqimni oddiy massivdek ishlatish"dan kelib chiqadi:

• "data" tinglovchisi ichida asinxron ish. readable.on("data", async (chunk) => { await db.insert(chunk); }) — "data" await ni kutmaydi, keyingi bo'laklar ketma-ket keladi va minglab insert bir vaqtda boshlanadi. Yechim: for await yoki pipeline + Writable.
• Siklda write() va natijaga qaramaslik — «Kanon o'lchovi» bo'limidagi 98 MB.
• Promise.all(bo'laklar.map(...)) — hamma bo'lakni birdaniga ishga tushirish. Cheklangan parallellik kerak.
• process.stdout.write bilan ulkan chiqish. stdout ham Writable; quvur sekin bo'lsa (| less), u ham buferlaydi.

7. CLI qadami: holat 35 da backpressure (hali commit qilmang)

Stream nima darsida aytilgandek, valyuta ning stream qismi holat 35 da quriladi va commit — Katta faylni stream bilan qayta ishlash darsida. Bu darsda hali commit qilmang. Lekin backpressure qayerda ishlashini hozirdan ko'rib qo'ying — kanon src/tarix.ts dagi oqimniYoz (kod qatorlari aynan, ikkitasi uzun; faqat izoh telefon uchun ikki qatorga bo'lindi):

/** pipeline: manba → (gzip) → fayl (vaqtinchalik + rename)
 * yoki stdout. Backpressure'ni pipeline boshqaradi */
export async function oqimniYoz(bolaklar: AsyncIterable<string>, s: { gzip: boolean; fayl?: string | undefined; signal?: AbortSignal | undefined }): Promise<void> {
  const manba = Readable.from(bolaklar);
  // … fayl berilmagan bo'lsa:
  const chiqish: Writable = process.stdout;
  await (s.gzip ? pipeline(manba, createGzip(), chiqish, { end: false, signal: s.signal }) : pipeline(manba, chiqish, { end: false, signal: s.signal }));

Kanon izohining o'zi — "Backpressure'ni pipeline boshqaradi". Zanjir: async generator (kunma-kun kurslar, CSV qatorlari) → Readable.from → createGzip (Transform) → process.stdout yoki fayl. Gzip yoki terminal sekin bo'lsa, pipeline generatorni to'xtatadi — keyingi kun so'ralmaydi ham. Bu tarmoq uchun ham foydali: CBU'ga ortiqcha so'rov ketmaydi.

{ end: false } — process.stdout ni zanjir oxirida yopmaslik: stdout butun jarayonga tegishli, undan keyin xato xabari ham chiqishi mumkin.

Kanon jonli o'lchovi (natijalar/35-jonli.txt): valyuta tarix … | head -2 — head 2 qatordan keyin quvurni yopadi. Bu backpressure'ning eng keskin ko'rinishi: qabul qiluvchi butunlay yo'qoldi. Natija: EPIPE xatosi chiqmadi, exit 0. Bu xatolarni qanday qayta ishlash — pipeline darsida.

Bu darsning amaliy qadami — o'lchovni o'z mashinangizda takrorlash (4-mashq).

8. Ko'p uchraydigan xatolar

9. Mashqlar

Mashqlar kurs/mashqlar/24/31-backpressure/ da.

1-mashq (oson): bo'shliqlarni to'ldiring

• Bufer to'lganini bildiruvchi write() qiymati —
• Bufer bo'shaganda chiqadigan hodisa — ""
• Windows'da bayt oqimining sukut chegarasi — bayt
• objectMode da chegara nimani sanaydi —
• Backpressure'ni o'zi boshqaradigan funksiya —

2-mashq (o'rta): writeAll

writeAll(writable, items) funksiyasini yozing: massivdagi hamma elementni backpressure'ni hurmat qilib yozsin, oxirida end() qilib, "finish" ni kutsin va nechta marta drain kutilganini qaytarsin. Sinang: obyekt rejimidagi Writable, highWaterMark: 3, har yozuv 2 ms, 10 ta element.

import { Writable } from "node:stream";
import { once } from "node:events";

async function writeAll(writable, items) {
  let waits = 0;
  for (const item of items) {
    if (!writable.write(item)) {
      waits++;
      await once(writable, "drain");
    }
  }
  writable.end();
  await once(writable, "finish");
  return waits;
}

const written = [];
const sink = new Writable({
  objectMode: true,
  highWaterMark: 3,
  write(item, encoding, callback) {
    written.push(item);
    setTimeout(callback, 2);
  },
});

const items = Array.from({ length: 10 }, (_, i) => i);
const waits = await writeAll(sink, items);
console.log("yozildi:", written.length, "| drain kutildi:", waits);

yozildi: 10 | drain kutildi: 3

3-mashq (qiyin): "data" ichidagi await tuzog'i

Manba — 6 ta buyurtma (Readable.from). Har buyurtmani "saqlash" 20 ms davom etadi va bir vaqtda nechta saqlash ketayotganini sanang (active, eng ko'pi — peak). Ikki usulni solishtiring: (a) on("data", async …), (b) for await. Har usulda peak ni chiqaring.

import { Readable } from "node:stream";
import { once } from "node:events";

function makeSaver() {
  const state = { active: 0, peak: 0 };
  state.save = async () => {
    state.active++;
    state.peak = Math.max(state.peak, state.active);
    await new Promise((r) => setTimeout(r, 20));
    state.active--;
  };
  return state;
}

const dishes = ["osh", "manti", "somsa", "lag'mon", "choy", "non"];
const orders = () => Readable.from(dishes);

const a = makeSaver();
const streamA = orders();
streamA.on("data", async () => { await a.save(); }); // ❌
await once(streamA, "end");
await new Promise((r) => setTimeout(r, 50));
console.log("on('data', async):", a.peak);

const b = makeSaver();
for await (const _ of orders()) await b.save(); // ✅
console.log("for await:", b.peak);

on('data', async): 6
for await: 1

4-mashq: CLI qadami — o'lchovni takrorlang

«Kanon o'lchovi» bo'limidagi tajribani o'z kompyuteringizda ishga tushiring (loyihaga qo'shmasdan, alohida fayl). rss raqamlarini yozib, kanon natijasi (98 MB / 16 KB) bilan solishtiring. Keyin highWaterMark ni 64 KB qilib, buferdagi eng katta hajm qanday o'zgarishini ko'ring. Hali commit qilmang.

10. Real ishda

• Fayl yuklab berish — pipeline(createReadStream(fayl), res) — har HTTP serverdagi standart. Qo'lda res.write sikli — xotira sizishi manbai.
• ETL va import — million qatorli CSV'ni bazaga yozishda for await yoki Writable + pipeline; on("data", async) — eng ko'p uchraydigan xato.
• Navbatlar va brokerlar (Kafka, RabbitMQ) — "prefetch", "consumer lag" — xuddi shu g'oya: iste'molchi ulgurmasa, ishlab chiqaruvchi sekinlashadi.
• Brauzer — Web Streams'da ham backpressure bor (desiredSize, ready) — Async iteratorlar va Web Streams darsida.
• Intervyuda: "Backpressure nima?", "write() false qaytarsa nima qilasiz?", "pipe va pipeline farqi?".

Xulosa

• Backpressure — sekin qabul qiluvchining "to'xta" signali; Node'da write() → false, keyin "drain".
• false — iltimos: ma'lumot baribir buferga tushadi; e'tiborsiz qoldirsangiz, bufer cheksiz o'sadi (kanon: 98 MB).
• To'g'ri sikl: if (!w.write(x)) await once(w, "drain"); yoki pipeline / for await — ular buni o'zi qiladi (kanon: 16 KB).
• E'tiborsiz yozish tezroq emas (biz o'lchadik: 308–321 ms va 272–286 ms), faqat xotira 5 baravar ko'p.
• highWaterMark — chegara, devor emas; objectMode da bo'laklar soni.

Keyingi dars: pipeline, finished va stream xatolari — pipe nega xatoni yo'qotadi, pipeline va finished qanday tozalaydi, AbortSignal bilan bekor qilish.

Manbalar

• Node.js hujjati — Stream: writable.write() qaytaradigan qiymat, "drain", highWaterMark, writableLength, "Buffering": https://nodejs.org/docs/latest-v24.x/api/stream.html#buffering
• Node.js qo'llanmasi — Backpressuring in Streams: https://nodejs.org/en/learn/modules/backpressuring-in-streams
• Kanon: tajriba/t31-backpressure.mjs, natijalar/tajriba/t31-backpressure.txt (Node 24.21.0, Windows 11, 2026-10-23), valyuta holat 35 (src/tarix.ts — oqimniYoz), natijalar/35-jonli.txt. Uch usulni alohida jarayonda o'lchash va kichik misollar — biz ishga tushirdik (Node 24.21.0, Windows 11 Pro, i5-12500H).