Node 0.10 Streams2: Processing Huge Files Without Running Out of Memory

Node v0.10.0 came out on Monday, March 11, and the headline item in the release announcement is Streams2, a rework of the stream interface that every stream in core is now built on. It’s easy to read that as plumbing for people writing HTTP libraries, but it matters just as much for a boring everyday job: you have a log file that’s a few gigabytes, and you want the server errors out of it and into another file.

On the command line that’s what classic Unix tools are for. split chops a big file into pieces you can deal with one at a time, and awk walks it line by line and never cares how big it is. Say each line looks like 2013-03-15T00:00:01.000Z GET /login 503 1234, with the status in the fourth field. Then this does the job:

awk '$4 >= 500 { print $1, $3 }' access.log > errors.txt

The rest of this post builds the same thing in Node, and explains why the obvious first draft is a trap.

Why fs.readFile falls over

The first thing most of us type is fs.readFile, split the result on newlines, and loop. The fs docs for 0.10.0 are clear about what that does: it reads the entire contents of the file and passes them to your callback.

The source shows how. In the 0.10.0 lib/fs.js, readFile stats the file to get its size, allocates one Buffer that big, and keeps reading until it’s full. Nothing reaches your code until the whole file is sitting in memory. Then, to split it into lines, you call toString(), and now you’re holding the text a second time as a JavaScript string while the Buffer is still around.

There’s also a hard ceiling. A single Buffer in 0.10.0 can’t be larger than kMaxLength, which node_buffer.h sets to 0x3fffffff bytes, about 1 GB. Because that allocation happens inside readFile’s own internal callback, a file past the limit doesn’t give you a nice err to check. It throws. Well below that limit, loading a few hundred megabytes just to throw most of it away is a good way to find out how much RAM your small server really has.

fs.createReadStream has been in Node since 0.1.31 back in 2010, so streaming a file isn’t new. What’s new is that streams are now much harder to get wrong.

What Streams2 changed

Isaac Schlueter described the problem in a post in December, A New Streaming API for Node v0.10. In 0.8, 'data' events start firing right away whether you’re ready or not, and pause() is only advisory, so you still have to be ready for data after you’ve asked it to stop. Every stream author also had to solve buffering and pausing from scratch, and they tended to get it subtly wrong.

In 0.10 all the streams in core are built on a shared set of base classes you can load with require('stream'): Readable, Writable, Duplex and Transform. The release post says that makes behavior far more consistent and makes it easier to write your own streams. If you’re stuck on 0.8, the readable-stream package on npm gives you the same interface there.

The biggest shift is on the reading side. According to the 0.10.0 stream docs, a readable stream now holds data in an internal buffer until you ask for it. When there’s something to consume it emits 'readable', and you call read(), which returns a Buffer (or a string if you set an encoding), or null when the buffer is empty. A 'readable' event will fire again when more arrives. Here’s a pull style newline counter, basically wc -l:

var fs = require('fs');

var input = fs.createReadStream('access.log');
var newlines = 0;

input.on('readable', function () {
  var chunk;
  while (null !== (chunk = input.read())) {
    for (var i = 0; i < chunk.length; i++) {
      if (chunk[i] === 10) newlines++;
    }
  }
});

input.on('end', function () {
  console.log(newlines);
});

How much a stream buffers is set by highWaterMark, which the docs define as the most bytes to keep in the internal buffer before it stops reading from the underlying resource. The generic default is 16 KB. File read streams raise theirs to 64 KB in the 0.10.0 source. One small gotcha: the createReadStream section of the fs docs still lists a bufferSize option, but the ChangeLog for 0.9.12 says that option was removed, and the 0.10.0 fs.js source never reads it.

Old code keeps working. Adding a 'data' listener or calling pause() or resume() switches a stream into what the docs call “old mode”, and in that mode pause() now actually stops the 'data' events.

pipe() and backpressure

Pulling with read() is useful for parsers, but for moving data from one place to another you mostly want pipe(). The docs say readable.pipe(destination) writes incoming data to the destination and properly manages back pressure, so a slow destination isn’t overwhelmed by a fast source. It returns the destination, so pipes chain, and by default it calls end() on the destination when the source ends.

The mechanism is simple. On a writable stream, write() returns false once its buffer passes its own highWaterMark, and the stream emits 'drain' when it’s safe to write again. pipe() watches for that. When the destination says it’s full, the source stops handing it chunks until 'drain', the source’s buffer fills up, and a file read stream stops reading from disk. Memory stays roughly at the size of a few buffers no matter how large the input is.

Transform streams

Between the file you read and the file you write, you need something that changes the data. That’s a Transform stream, which the docs describe as a duplex stream whose output is connected to its input, the way zlib and crypto streams are. You don’t implement _read() and _write() yourself. You implement _transform(chunk, encoding, callback), call this.push() zero or more times with output, and call callback when you’ve fully dealt with that chunk. If you need to emit something at the very end, you can also implement _flush(callback), which runs after all the input has been consumed and before the readable side ends. Subclasses need to call the Transform constructor so the buffering state gets set up.

Output doesn’t have to match input one to one. That’s exactly what a line filter needs, since a chunk might contain a hundred matching lines or none.

An awk style filter

Here’s the awk command from the top as a Transform. It takes the path to the log and the path to write to:

var fs = require('fs');
var util = require('util');
var Transform = require('stream').Transform;
var StringDecoder = require('string_decoder').StringDecoder;

function ServerErrors(options) {
  Transform.call(this, options);
  this._decoder = new StringDecoder('utf8');
  this._tail = '';
}
util.inherits(ServerErrors, Transform);

ServerErrors.prototype._transform = function (chunk, encoding, callback) {
  var lines = (this._tail + this._decoder.write(chunk)).split('\n');
  this._tail = lines.pop();
  this._pick(lines);
  callback();
};

ServerErrors.prototype._flush = function (callback) {
  var last = this._tail + this._decoder.end();
  if (last) this._pick([last]);
  callback();
};

ServerErrors.prototype._pick = function (lines) {
  var out = '';
  for (var i = 0; i < lines.length; i++) {
    var fields = lines[i].split(' ');
    if (Number(fields[3]) >= 500) out += fields[0] + ' ' + fields[2] + '\n';
  }
  if (out) this.push(out);
};

var input = fs.createReadStream(process.argv[2]);
var output = fs.createWriteStream(process.argv[3]);

input.on('error', function (err) { console.error(err.message); });
output.on('error', function (err) { console.error(err.message); });
output.on('finish', function () { console.log('done'); });

input.pipe(new ServerErrors()).pipe(output);

Save it as errors.js and run node errors.js access.log errors.txt. On a log in the format above, the output file matches what the awk one liner writes.

Two details make it correct rather than mostly correct. A read stream cuts chunks by byte count, not by line, so a chunk usually ends halfway through a line. After splitting, the last piece is kept in this._tail and glued onto the front of the next chunk, and _flush handles whatever is left at the end, including a final line with no trailing newline. The second detail is characters. A UTF-8 character can be split across two chunks too, and calling toString() on each chunk would mangle it. The StringDecoder module exists for this: write() returns the text it can decode and holds back an incomplete character until the next buffer, and end() returns whatever is left over.

_pick builds one string per chunk and pushes it once, instead of pushing every line separately, and skips the push when nothing matched. push() accepts strings as well as Buffers in 0.10. The 'finish' event is documented on writable streams as firing once end() has been called and everything has been written, which is why the “done” message hangs off the output stream.

Things to watch out for

Errors don’t travel down a pipe. In the 0.10.0 source, pipe() listens for errors on the destination so it can unpipe, but it never listens on the source, and nothing gets passed along the chain. The events docs say an 'error' with no listener prints a stack trace and exits the process. So give each stream that can fail its own 'error' handler, like the example does for the two file streams, and add one to your Transform too if its code can emit errors.

A stream you never read now waits forever. The compatibility section of the stream docs calls this out with a TCP server that listens for 'end' but never consumes the socket. Before 0.10 the incoming data was simply discarded, but now the socket stays paused and 'end' never comes. If you really don’t care about the data, call resume() to let it flow away.

The stream module is also still marked “Stability: 2 - Unstable” in the 0.10.0 docs, which the docs define as an API that is still settling and hasn’t had enough real world testing to be called stable. The December post already called the API feature complete, but it’s worth reading the ChangeLog before each upgrade.

The takeaway

If a file might ever be bigger than you’d happily keep in RAM, don’t readFile it. Open a read stream, do your work in a Transform, and pipe() into a write stream. Node 0.10 finally makes that the easy path, with backpressure handled for you and memory staying flat whether the log is 10 MB or 10 GB. It’s the same idea awk has always had, just in JavaScript.