Backpressure

ปัญหาที่ backpressure แก้

ลองนึกภาพสายฉีดน้ำดับเพลิงที่ต่อเข้ากับหัวฉีดน้ำสวน ถ้าน้ำมาถึงเร็วกว่าที่หัวฉีดจะพ่นออกได้ สายจะแตกในที่สุด สิ่งเดียวกันเกิดขึ้นใน stream ของ Node: ถ้า Readable source ที่เร็วสร้างข้อมูลเร็วกว่าที่ Writable ที่ช้าจะบริโภคได้ chunk ที่ buffer ไว้จะสะสมใน memory จนกว่า process จะหมด RAM หรือ crash

Backpressure คือกลไกที่ consumer ที่ช้าบอกให้ producer ที่เร็วชะลอตัวลง มันคือ stream เวอร์ชันของ flow control ใน TCP

Backpressure ทำงานอย่างไรใน Node

method write() ของ Writable stream คือสัญญาณสำคัญ:

คืนค่า true — internal buffer อยู่ใต้ high-water mark (ดีฟอลต์บน Node 24: 65536 byte (64 KiB) สำหรับ stream แบบ byte-mode — ทั้ง Readable และ Writable — หรือ 16 object สำหรับ stream แบบ object-mode; fs.createReadStream ใช้ดีฟอลต์เดียวกันนี้ ไม่มีข้อยกเว้นแยกต่างหาก) producer เดินหน้าส่งต่อไปได้
คืนค่า false — buffer ถึงหรือเกิน high-water mark แล้ว producer ต้อง pause และรอ event drain ก่อนที่จะเขียนต่อ

high-water mark ไม่ใช่ hard limit; มันเป็นคำแนะนำแบบ soft Node ยัง buffer chunk ที่เกินมันอยู่ดี แต่การทำแบบนั้นมีต้นทุนเป็น memory การเคารพสัญญาณนี้ทำให้การใช้ memory คงที่

Backpressure แบบจัดการเอง

ก่อนที่จะมี pipe และ pipeline คุณต้องจัดการเรื่องนี้ด้วยตัวเอง:

function pumpWithBackpressure(readable, writable) {
  readable.on('data', function(chunk) {
    var ok = writable.write(chunk);
    if (!ok) {
      readable.pause();             // stop the source
      writable.once('drain', function() {
        readable.resume();          // resume when ready
      });
    }
  });

  readable.on('end', function() {
    writable.end();
  });
}

รูปแบบนี้ยาวและพลาดง่าย การลืมเรียก readable.pause() แค่จุดเดียวทำให้ buffer โตขึ้นโดยไม่มีขอบเขต

pipe และ pipeline จัดการมันให้อัตโนมัติอย่างไร

readable.pipe(writable) และ stream.pipeline() implement รูปแบบข้างบนไว้ภายในให้แล้ว มัน:

ส่งต่อแต่ละ chunk จาก readable ไปยัง writable
pause readable เมื่อ write() คืน false
resume readable เมื่อ event drain ยิง

คุณไม่ต้องจัดการ pause และ resume เองเลยเมื่อใช้ pipe หรือ pipeline นี่คือเหตุผลหลักที่ควรเลือกใช้มันแทน raw event listener

วัดต้นทุนจริงของการเพิกเฉยต่อ backpressure

การพูดถึง “memory ที่โตไม่มีขอบเขต” เป็นเรื่องนามธรรม demo นี้ทำให้มันเป็นรูปธรรม: source ที่มี 2,000 chunk เดียวกันถูก drain เข้า sink ที่ช้าตัวเดียวกันสองครั้ง — ครั้งหนึ่งผ่าน data listener ที่เพิกเฉยต่อ return value ของ write() และอีกครั้งผ่าน .pipe() ที่เคารพมัน ทั้งสองรอบติดตามจำนวน สูงสุด ของ chunk ที่ buffer ไว้ใน sink ในแต่ละครั้ง:

Node.js

import { Readable, Writable } from 'node:stream';

function makeSource() {
  var i = 0;
  return new Readable({
    objectMode: true,
    read() {
      if (i < 2000) { this.push('chunk' + i); i++; }
      else this.push(null);
    }
  });
}

function makeSlowSink() {
  return new Writable({
    objectMode: true,
    highWaterMark: 1,
    write(chunk, enc, cb) { setImmediate(cb); } // slow relative to the fast source above
  });
}

// --- Wrong: 'data' handler ignores write()'s return value, never pauses ---
await new Promise(function(resolve) {
  var source = makeSource();
  var sink = makeSlowSink();
  var maxBuffered = 0;
  source.on('data', function(chunk) {
    sink.write(chunk); // return value ignored -- source is never paused
    if (sink.writableLength > maxBuffered) maxBuffered = sink.writableLength;
  });
  source.on('end', function() {
    console.log('ignoring backpressure -- peak buffered chunks:', maxBuffered);
    resolve();
  });
});

// --- Right: pipe() respects write()'s return value automatically ---
await new Promise(function(resolve) {
  var source = makeSource();
  var sink = makeSlowSink();
  var maxBuffered = 0;
  var timer = setInterval(function() {
    if (sink.writableLength > maxBuffered) maxBuffered = sink.writableLength;
  }, 0);
  source.pipe(sink);
  sink.on('finish', function() {
    clearInterval(timer);
    console.log('using pipe() -- peak buffered chunks:', maxBuffered);
    resolve();
  });
});

Output (verified on Node 24)

ignoring backpressure -- peak buffered chunks: 2000
using pipe() -- peak buffered chunks: 1

การเพิกเฉยต่อ source ทั้งหมด (ไม่เคยเรียก .pause()) ทำให้ทุก chunk ในจำนวน 2,000 ตัวเข้าคิวใน sink พร้อมกันได้ — source ไม่เคยได้รับสัญญาณให้ชะลอตัวเลย .pipe() รักษา buffer ของ sink ไว้ที่ high-water mark ของมันคือ 1 ตลอดเวลา เพราะมัน pause source ทันทีที่ write() คืน false

มองเห็น backpressure ด้วย high-water mark ที่กำหนดเอง

demo ด้านล่างสร้าง Readable และ Writable ที่มี high-water mark เล็กมาก (1 byte) เพื่อให้ backpressure trigger ทุก chunk ดูว่า readable ถูก pause และ resume อัตโนมัติอย่างไร

Node.js

const { Readable, Writable, pipeline } = require('node:stream');

const source = Readable.from(['a', 'b', 'c', 'd', 'e'], {
  objectMode: true,
  highWaterMark: 1
});

const sink = new Writable({
  objectMode: true,
  highWaterMark: 1,
  write(chunk, encoding, callback) {
    console.log('sink writing: ' + chunk);
    // simulate a slow sink
    setTimeout(callback, 10);
  }
});

pipeline(source, sink, function(err) {
  if (err) {
    console.error('error:', err.message);
  } else {
    console.log('done — pipeline handled backpressure automatically');
  }
});

Output (verified on Node 24)

sink writing: a
sink writing: b
sink writing: c
sink writing: d
sink writing: e
done — pipeline handled backpressure automatically

Mental Model

flowchart TB
  Fast["fast Readable source"] -->|"data listener, no pause()"| Ignored["chunks queue up in the sink without limit"]
  Fast -->|".pipe() / pipeline()"| Managed["source is paused the instant write() returns false"]
  Managed -->|"sink emits 'drain'"| Resume["source is resumed"]
  Resume --> Managed

การเพิกเฉยต่อสัญญาณของ write() ทำให้ source ที่เร็วท่วม sink ที่ช้า; pipe()/pipeline() pause source จนกว่า drain จะยิง

Trade-offs

ทางเลือก	ข้อดี	ต้นทุน
เพิกเฉยต่อ return value ของ `write()`	code ง่ายที่สุดเท่าที่จะทำได้	memory โตไม่มีขอบเขตทันทีที่ source เร็วกว่า sink — ท้ายที่สุด OOM crash ภายใต้ load จริง
pause/resume เองบน `drain`	ควบคุมได้เต็มที่ว่าจะชะลอตัวเมื่อไหร่	ยาวและทุก stream เพิ่มเติมใน chain เพิ่มโอกาสพลาดกรณีใดกรณีหนึ่ง
`.pipe()` / `pipeline()`	จัดการ backpressure ได้ถูกต้องโดยไม่ต้องทำ bookkeeping เอง	มองเห็นได้น้อยลงว่า pause เกิดขึ้นตอนไหนแน่ ๆ — คุณต้องเชื่อ implementation

Common Misconceptions

“data listener ปลอดภัยตราบใดที่ฉันไม่ console.log มากเกินไป” — ความจริง: อันตรายไม่ใช่การ log แต่คือการเขียน chunk ไปยังปลายทางที่ ช้ากว่า (socket, ไฟล์, API อื่น) โดยไม่ตรวจว่ามันตามทันหรือเปล่า — นั่นคือสิ่งที่สะสมใน memory จริง ๆ
“Backpressure เป็นแนวคิดเฉพาะของ Node” — ความจริง: มันคือปัญหาทั่วไปของ producer ที่เร็วท่วม consumer ที่ช้า — แนวคิดเดียวกับ TCP flow control และ stream ของ Node แก้มันด้วยวิธีเดียวกัน (สัญญาณให้ pause และสัญญาณให้ resume)
“high-water mark เล็ก ๆ แปลว่าการใช้ memory โดยรวมเล็กด้วย” — ความจริง: มันจำกัดว่า stream ตัวเดียว buffer ได้เท่าไหร่ แต่ถ้าคุณเพิกเฉยต่อสัญญาณ write() ของมันทั้งหมด (ตามที่ demo ข้างบนแสดง) high-water mark ก็ไม่มีความหมาย — ผู้เรียกยังคงเขียนต่อไปอยู่ดี

อ่านโค้ดที่ AI เขียน

data listener เขียน (หรือส่งต่อ) ไปยังสิ่งที่อาจช้า — socket, stream อื่น, การเรียก external API — โดยไม่เคยตรวจสัญญาณ backpressure ของปลายทางนั้นเองเลยหรือเปล่า?
code implement pump ที่ทำเองระหว่าง stream สองตัว (แทนที่จะใช้ .pipe()/pipeline()) โดยไม่มี drain listener อยู่เลยหรือเปล่า?
มี comment ที่อ้างว่า “จัดการ backpressure” อยู่ข้าง ๆ code ที่ไม่เคยเรียก .pause(), ตรวจ return value ของ write() หรือมอบหมายให้ .pipe()/pipeline() เลยหรือเปล่า?

🐛 หา bug ในโค้ดนี้

const { Readable } = require('node:stream');
const net = require('node:net');

function forwardToSocket(readable, socket) {
readable.on('data', function(chunk) {
  socket.write(chunk); // return value never checked, readable never paused
});
}

เฉลย

forwardToSocket ไม่เคยตรวจ return value ของ socket.write(chunk) และไม่เคยเรียก readable.pause() ถ้า network socket ช้ากว่าความเร็วข้อมูลของ readable (สถานการณ์ทั่วไปบน network จริง เทียบกับ source ในตัว process ที่เร็ว) Node จะยังคง buffer chunk ที่ยังไม่ได้ส่งไว้ใน write buffer ภายในของ socket โดยไม่มีขอบเขต และ memory จะโตขึ้นจนกว่า process จะหมดหรือการเชื่อมต่อถูกตัด วิธีแก้คือรูปแบบ pause/drain ที่ทำเองจากก่อนหน้านี้ในบทนี้ หรือแค่ readable.pipe(socket) ซึ่งใช้รูปแบบนั้นให้อัตโนมัติเป๊ะ