Streams và backpressure — xử lý dữ liệu lớn không nổ RAM

Câu hỏi phỏng vấn thường gặp: "Export 1 triệu dòng ra CSV / cho user tải file 2GB — anh làm thế nào để server không hết RAM?". Câu trả lời là stream, nhưng interviewer Senior sẽ hỏi tiếp: backpressure là gì, .pipe() với pipeline() khác nhau chỗ nào, client mạng yếu thì sao? Bài này đo từng trường hợp bằng lab: RAM khi đọc file 300MB, RAM khi bỏ qua giá trị trả về của write(), file descriptor bị rò khi dùng .pipe().

Thuật ngữ dùng trong bài
stream (dòng dữ liệu xử lý từng mẩu thay vì cả khối) · chunk (một mẩu dữ liệu, mặc định 64KB) · Readable / Writable / Transform / Duplex (nguồn / đích / biến đổi / hai chiều) · highWaterMark (ngưỡng buffer nội bộ — vượt ngưỡng thì báo "dừng lại") · backpressure (áp lực ngược — bên nhận chậm báo bên gửi chậm lại, để dữ liệu không dồn vào RAM) · drain (sự kiện Writable phát khi buffer đã xả xuống, gửi tiếp được) · objectMode (stream truyền object JS thay vì byte) · file descriptor (fd) (số hiệu file/socket đang mở của process, có giới hạn).

Kiểm chứng: Context7: Node docs v22 (stream: writable.write, 'drain', highWaterMark, pipeline). Output thật trên Node 22.23.1 / Linux 6.8. Số thời gian bị nhiễu do máy dùng chung; số bộ nhớ ổn định giữa các lần chạy.

1. Bài toán gốc và bối cảnh ra đời

Mô hình tự nhiên nhất là "đọc hết vào RAM, xử lý, ghi ra" (readFile → xử lý → res.end). Nó có ba giới hạn:

1. RAM tỉ lệ với kích thước dữ liệu × số người dùng đồng thời. 10 người tải file 50MB = 500MB RAM (lab 3.4).
2. Độ trễ byte đầu tiên — client phải chờ đọc xong cả file mới nhận byte đầu.
3. Giới hạn cứng — Buffer có trần; readFile từ chối file > 2GiB (lab 5.3).

Node từ bản đầu đã có stream vì bản chất của nó là server mạng: socket vốn là dòng dữ liệu. API stream đã qua nhiều đời (streams1 kiểu "push", streams2 ở Node 0.10 thêm chế độ "pull" và highWaterMark, streams3, rồi pipeline ở Node 10, stream/promises ở Node 15). Lớp "lịch sử" này là lý do API stream khó dùng: nhiều cách làm cùng một việc, cách cũ (.pipe()) vẫn còn và có bẫy. (Mốc phiên bản theo changelog Node; không có lab.)

Tối ưu cho: dữ liệu lớn hoặc không biết trước kích thước, dữ liệu đến dần (mạng), xử lý tuần tự từng mẩu. Không tối ưu cho: dữ liệu nhỏ (vài KB — stream chỉ thêm độ phức tạp), việc cần truy cập ngẫu nhiên hoặc cần toàn bộ dữ liệu một lúc (sort cả file, parse JSON một khối).

2. Mô hình tư duy cốt lõi

Stream = băng chuyền có buffer giới hạn giữa hai bên chạy khác tốc độ; backpressure là tín hiệu "đầy rồi, dừng lại" chạy ngược từ đích về nguồn. Mỗi tầng giữ tối đa highWaterMark; khi buffer vượt ngưỡng, write() trả false, nguồn phải dừng tới khi có 'drain'. Mọi tính năng khác (pipe, pipeline, async iterator) chỉ là các cách tự động hoá việc tôn trọng tín hiệu này.

3. Internals và quyết định thiết kế

3.1 highWaterMark là ngưỡng mềm, không phải giới hạn cứng

Theo docs Node 22: mặc định 65536 byte (64KiB) cho stream byte và 16 object cho objectMode (đổi được bằng stream.setDefaultHighWaterMark). write() vẫn nhận dữ liệu khi đã vượt ngưỡng — nó chỉ trả false như một lời khuyên. Quyết định thiết kế này để API đơn giản (không throw, không chặn), đổi lại trách nhiệm tôn trọng backpressure nằm ở người gọi. Bỏ qua nó thì buffer lớn vô hạn — lab 3.3.

3.2 readFile vs stream: 300MB

// 40-readfile-vs-stream.js — tính sha256 file 300MB
if (mode === 'readFile') hash.update(await fs.promises.readFile('big.bin'));
else await pipeline(fs.createReadStream('big.bin'), hash);

$ for m in readFile stream; do /usr/bin/time -f "  -> max RSS %MKB" node 40-readfile-vs-stream.js $m; done
readFile rss=345MB arrayBuffers=300MB heapUsed=5MB
  -> max RSS 353356KB
stream   rss=86MB arrayBuffers=16MB heapUsed=5MB
  -> max RSS 88076KB

Để ý heapUsed chỉ 5MB ở cả hai: Buffer nằm ngoài heap V8 (cột arrayBuffers). Đây là lý do --max-old-space-size không bảo vệ bạn khỏi OOM do Buffer — xem bài 6.

3.3 Bỏ qua giá trị trả về của write()

// 41-backpressure.js — producer nhanh, consumer ghi 1 chunk mỗi 1ms (vd client mạng yếu)
const slow = new Writable({ highWaterMark: 64 * 1024, write(chunk, enc, cb) { setTimeout(cb, 1); } });
function produce() {
  while (written < TOTAL) {                      // 3000 chunk x 64KB = ~188MB
    written++;
    const ok = slow.write(Buffer.alloc(64 * 1024, written % 256));
    if (mode === 'respect' && !ok) return slow.once('drain', produce);   // dừng, chờ drain
  }
  slow.end();
}

$ for m in ignore respect; do node 41-backpressure.js $m; done
ignore  peak rss=240MB | buffer nội bộ của writable cao nhất=187.5MB
respect peak rss=84MB | buffer nội bộ của writable cao nhất=0.1MB

3.4 Endpoint tải file: 10 người cùng tải 50MB

// 43-download.js
if (mode === 'readFile') res.end(await fs.promises.readFile('file50.bin'));
else pipeline(fs.createReadStream('file50.bin', { highWaterMark }), res, (err) => err && console.error('pipeline:', err.code));

$ for m in readFile pipeline pipeline1m; do node 43-download.js $m; done
readFile   10 lượt tải x 50MB: 3423ms | peak RSS server+client = 608MB
pipeline   10 lượt tải x 50MB: 3476ms | peak RSS server+client = 135MB
pipeline1m 10 lượt tải x 50MB: 2292ms | peak RSS server+client = 175MB     ← highWaterMark 1MB

RAM giảm ~4,5 lần (client chạy chung process nên RSS gồm cả phía client). Thời gian thì nhiễu (một lần khác readFile 10772ms / pipeline 14898ms), nhưng tăng highWaterMark lên 1MB giảm số lần gọi read() — đổi thêm 40MB RAM lấy throughput. Đây là núm chỉnh thật sự của stream: highWaterMark = RAM mỗi luồng × số luồng đồng thời.

3.5 Async generator + pipeline: backpressure miễn phí

// 44-generator.mjs — export 1 triệu dòng "từ DB" ra CSV, đích chậm
async function* rowsFromDb() {
  for (let page = 0; page < 1000; page++)
    for (let i = 0; i < 1000; i++) { produced++; yield { id: page * 1000 + i, email: `u${i}@x.vn` }; }
}
const toCsv = new Transform({ objectMode: true, transform(r, _, cb) { cb(null, `${r.id},${r.email}\n`); } });
await pipeline(Readable.from(rowsFromDb()), toCsv, slowSink);

$ node 44-generator.mjs
xong 1000000 dòng trong 5925ms | generator chạy trước consumer tối đa 4436 dòng | heapUsed 15MB

Generator chỉ bị "kéo" khi tầng sau còn chỗ, nên không bao giờ chạy trước quá xa. 4436 dòng = tổng buffer các tầng: 16 object (Readable) + 16 object (phía ghi Transform) + ~64KB byte ở đích (mỗi dòng CSV ~15 byte → ~4.300 dòng). Trong code thật, rowsFromDb đọc DB theo trang (keyset pagination — phân trang theo con trỏ, không dùng OFFSET) hoặc cursor — không bao giờ SELECT * hết vào mảng.

3.6 pipeline() vs .pipe()

.pipe() xử lý backpressure nhưng không lan lỗi và không dọn dẹp: đích lỗi thì nguồn vẫn mở. pipeline() destroy mọi stream trong chuỗi khi một cái lỗi hoặc kết thúc, và gọi callback / reject Promise đúng một lần.

// 42-pipe-leak.js — 50 lần: nguồn đọc file, đích lỗi ngay chunk đầu (vd client ngắt kết nối)
const bad = () => new Writable({ write(c, e, cb) { cb(new Error('đích hỏng (vd client ngắt kết nối)')); } });
if (process.argv[2] === 'pipe') src.pipe(dst);
else pipeline(src, dst, () => {});

$ for m in pipe pipeline; do node 42-pipe-leak.js $m; done
pipe     fd mở thêm: 50 | nguồn đã destroy: 0/50      ← rò 50 file descriptor
pipeline fd mở thêm: 0 | nguồn đã destroy: 50/50

4. Đảm bảo và giới hạn

Độ phức tạp: RAM O(highWaterMark × số tầng) thay vì O(kích thước dữ liệu); thời gian vẫn O(n) nhưng thêm chi phí mỗi chunk (callback, microtask) — chunk nhỏ quá thì chi phí này chiếm ưu thế.

5. Failure modes

5.1 Client chậm làm RAM server tăng dần

Code ghi res.write() trong vòng lặp không kiểm tra giá trị trả về (lab 3.3). Với client mạng tốt thì không sao — test ở local luôn qua. Với client 3G, dữ liệu dồn trong buffer của socket phía Node. Nhiều client chậm cùng lúc → RSS tăng → OOMKilled. Bẫy nổ trễ và chỉ ở production. Hệ quả: process chết, mất mọi request đang chạy.

5.2 Rò file descriptor khi client ngắt giữa chừng

Dùng .pipe() để trả file; client đóng tab giữa chừng → đích lỗi/đóng → nguồn không bị đóng (lab 3.6: rò 50/50). Sau vài nghìn lần: EMFILE: too many open files, server không mở được file lẫn socket mới. (Thông báo EMFILE là thông báo chuẩn của libuv; không chạy tới mức cạn fd trong lab.)

5.3 readFile file lớn

$ truncate -s 3G sparse.bin
$ node -e "require('fs').readFile('sparse.bin', (e) => console.log(e.code, '-', e.message))"
ERR_FS_FILE_TOO_LARGE - File size (3221225472) is greater than 2 GiB

Trước khi chạm giới hạn này thì thường đã OOM (RAM của container nhỏ hơn 2GB).

5.4 Lỗi bị nuốt, request treo mãi

Với .pipe(), nếu nguồn lỗi (vd đọc S3 bị ngắt) mà không ai nghe 'error' trên nguồn: hoặc crash (sự kiện 'error' không có listener là throw), hoặc — nếu có listener nhưng quên res.destroy() — response không bao giờ kết thúc, client treo tới timeout. pipeline destroy res giúp bạn.

5.5 Transform làm việc CPU nặng mỗi chunk

Transform chạy trên main thread. Nén, mã hoá, parse phức tạp trong transform() = chặn event loop theo từng chunk (bài 4). zlib stream thì không — nó đẩy việc nén vào thread pool.

6. Vận hành trong production

• Metric: RSS và process.memoryUsage().arrayBuffers (Buffer ngoài heap), số fd đang mở (/proc/<pid>/fd hoặc process_open_fds của prom-client), số response đang stream, thời gian mỗi lượt tải.
• Tham số: highWaterMark (RAM × throughput), timeout của socket (server.requestTimeout, socket.setTimeout) để cắt client quá chậm thay vì giữ buffer mãi.
• Capacity planning (ví dụ minh hoạ): 500 lượt tải đồng thời × (64KB đọc + ~64KB socket) ≈ 64MB — chấp nhận được. Cùng tải đó mà dùng readFile file 50MB = 25GB.
• File lớn cho người dùng tải: tốt nhất là không đi qua Node — trả presigned URL của S3/MinIO hoặc để nginx X-Accel-Redirect. Node chỉ kiểm tra quyền.
• Debug lúc 3h: RSS tăng mà heapUsed đứng yên → nghi Buffer/stream (arrayBuffers tăng). fd tăng dần → nghi .pipe() không dọn.

7. So sánh và ma trận quyết định

Dùng stream khi dữ liệu lớn/không biết trước kích thước hoặc đi qua mạng. Tránh khi dữ liệu nhỏ và cần toàn bộ một lúc — stream chỉ thêm code và thêm chỗ để lỗi.

8. Góc nhìn tech lead: tác động tổ chức

• Quy ước: cấm .pipe() trong code mới (lint/review), dùng pipeline từ node:stream/promises.
• Kiến trúc: file lớn đi thẳng object storage (presigned URL) — giảm tải Node, giảm chi phí băng thông qua app server. Export lớn chạy nền qua queue rồi gửi link, không giữ HTTP request mở 10 phút.
• Onboarding: stream là phần API khó nhất của Node; dev mới nên có một ví dụ mẫu trong codebase (export CSV bằng generator + pipeline) để chép theo.
• Exit strategy: Web Streams là chuẩn chung với browser/Deno/Bun/edge runtime; code mới hướng về Web Streams giúp chuyển runtime dễ hơn, nhưng hệ sinh thái Node streams vẫn lớn hơn — chuyển dần tại ranh giới, không viết lại.

9. Anti-patterns và bài học thực tế

1. res.write() trong vòng lặp không kiểm tra giá trị trả về (lab 3.3).
2. .pipe() không xử lý lỗi → rò fd (lab 3.6), request treo.
3. Gom chunk vào mảng rồi Buffer.concat — mất hết lợi ích stream, chỉ là readFile viết dài hơn.
4. SELECT * rồi mới stream ra CSV — RAM đã nổ ở bước query.
5. Đặt highWaterMark rất lớn (100MB) "cho nhanh" — RAM mỗi luồng tăng tương ứng.
6. Trộn on('data') với pipe/async iterator trên cùng stream — hai bên tranh dữ liệu, chế độ flowing/paused lẫn lộn.
7. Tưởng 'finish' = dữ liệu đã an toàn trên đĩa — chỉ là đã giao cho kernel; cần fsync nếu yêu cầu bền vững.

10. Câu hỏi phụ interviewer hay hỏi tiếp

Backpressure là gì, Node làm thế nào?

Tín hiệu từ đích về nguồn để giảm tốc. Node: write() trả false khi buffer vượt highWaterMark, phát 'drain' khi xả xong; pipe/pipeline tự pause/resume nguồn theo tín hiệu này.

highWaterMark mặc định?

64KiB cho byte stream, 16 object cho objectMode (Node 22). Là ngưỡng mềm: vượt vẫn nhận, chỉ trả false.

pipe vs pipeline?

Cả hai có backpressure. pipeline lan lỗi, destroy mọi stream, có bản Promise. pipe thì không — lab rò 50/50 fd.

Buffer nằm ở đâu trong bộ nhớ?

Ngoài heap V8 (arrayBuffers/external). Lab: đọc 300MB, heapUsed vẫn 5MB. Heap limit không chặn được OOM do Buffer.

Export 1 triệu dòng thế nào?

Đọc DB theo trang/cursor trong async generator → Transform sang CSV → pipeline vào response hoặc S3 multipart; chạy nền nếu lâu. Lab: 1 triệu dòng, heap 15MB.

Stream có chạy song song không?

Không. Các tầng chạy xen kẽ trên main thread (trừ fs/zlib dùng thread pool). Stream giải bài toán RAM và độ trễ byte đầu, không phải bài toán CPU.

Phụ lục: lab đã chạy

Môi trường: Node v22.23.1, Linux 6.8, 4 vCPU dùng chung. Tạo dữ liệu: head -c 300M /dev/urandom > big.bin, head -c 50M /dev/urandom > file50.bin.

// 42-pipe-leak.js (đầy đủ)
const fs = require('node:fs'); const { Writable, pipeline } = require('node:stream');
const fds = () => fs.readdirSync('/proc/self/fd').length;
const bad = () => new Writable({ write(c, e, cb) { cb(new Error('đích hỏng (vd client ngắt kết nối)')); } });
const base = fds(); const srcs = [];
for (let i = 0; i < 50; i++) {
  const src = fs.createReadStream('big.bin'); srcs.push(src);
  const dst = bad(); dst.on('error', () => {});
  if (process.argv[2] === 'pipe') src.pipe(dst);
  else pipeline(src, dst, () => {});
}
setTimeout(() => {
  console.log(`${process.argv[2].padEnd(8)} fd mở thêm: ${fds() - base} | nguồn đã destroy: ${srcs.filter((s) => s.destroyed).length}/50`);
}, 500);

// 43-download.js (đầy đủ)
const http = require('node:http'); const fs = require('node:fs'); const { pipeline } = require('node:stream');
const mode = process.argv[2]; let peak = 0;
setInterval(() => { peak = Math.max(peak, process.memoryUsage().rss); }, 20).unref();
const srv = http.createServer(async (req, res) => {
  if (mode === 'readFile') res.end(await fs.promises.readFile('file50.bin'));
  else pipeline(fs.createReadStream('file50.bin', { highWaterMark: mode === 'pipeline1m' ? 2 ** 20 : 64 * 1024 }), res,
    (err) => err && console.error('pipeline:', err.code));
}).listen(3102, '127.0.0.1', async () => {
  const t0 = performance.now();
  await Promise.all(Array.from({ length: 10 }, async () => {
    const r = await fetch('http://127.0.0.1:3102/'); let n = 0;
    for await (const chunk of r.body) n += chunk.length;
    if (n !== 50 * 2 ** 20) throw new Error('thiếu dữ liệu');
  }));
  console.log(`${mode} 10 lượt tải x 50MB: ${(performance.now() - t0).toFixed(0)}ms | peak RSS = ${(peak / 2 ** 20).toFixed(0)}MB`);
  srv.close(); process.exit(0);
});

Chưa verify: cạn fd tới EMFILE; client mạng yếu thật (lab dùng Writable giả lập chậm); Web Streams; hành vi fsync.