Streams and backpressure: handling large files

Process multi-gigabyte files with constant memory thanks to streams, pipeline, and backpressure.

9 min · 3 questions

readFile loads an entire file into memory: on a 4 GB log, your process blows up. Streams process data in chunks with a constant memory footprint, bounded by highWaterMark (64 KiB by default for a file stream). Four types exist: Readable, Writable, Duplex, and Transform. The core problem they solve is called backpressure: what do you do when the producer is faster than the consumer?

import { createWriteStream } from "node:fs";

const out = createWriteStream("big.txt");
let i = 0;

function writeChunks() {
  let ok = true;
  while (i < 1e6 && ok) {
    ok = out.write(`line ${i}\n`);
    i += 1;
  }
  if (i < 1e6) out.once("drain", writeChunks);
  else out.end();
}

writeChunks();

When write() returns false, the internal buffer has exceeded highWaterMark: the data is not lost, but continuing to write accumulates everything in memory - exactly what streams were meant to avoid. The contract: stop writing and wait for the drain event. That's what pipe() does for you, and what pipeline() does even better: on top of backpressure, it propagates errors and cleanly destroys all streams, in both directions.

import { createReadStream, createWriteStream }
  from "node:fs";
import { Transform } from "node:stream";
import { pipeline } from "node:stream/promises";

const upper = new Transform({
  transform(chunk, encoding, callback) {
    callback(null, chunk.toString().toUpperCase());
  },
});

await pipeline(
  createReadStream("big.txt"),
  upper,
  createWriteStream("big-upper.txt"),
);
console.log("done");

Readable streams are async iterables: for await consumes chunks while respecting backpressure. Be careful, a chunk is an arbitrary slice of bytes: nothing guarantees it ends on a line boundary. For line-by-line processing, node:readline handles reassembling the pieces, including a \r\n split across two chunks thanks to crlfDelay: Infinity.

import { createReadStream } from "node:fs";
import { createInterface } from "node:readline";

const rl = createInterface({
  input: createReadStream("big.txt"),
  crlfDelay: Infinity,
});

let count = 0;
for await (const line of rl) {
  if (line.includes("42")) count += 1;
}
console.log(count, "matching lines");

Knowledge check

Make sure you remember the key points of this lesson.

1. What does it mean when writable.write(chunk) returns false?
   • The chunk was lost and must be rewritten
   • The internal buffer exceeds highWaterMark: stop writing until the drain event
   • The stream was destroyed by an error
2. What decisive advantage does pipeline() have over pipe()?
   • pipeline is faster because it's multithreaded
   • pipe handles backpressure, pipeline doesn't
   • pipeline propagates errors and destroys all streams involved
   • pipeline automatically converts encodings
3. In objectMode, what does highWaterMark represent?
   • A number of queued objects
   • A number of bytes, as in binary mode
   • The maximum size of a serialized object