stream: Raise readable high water mark in powers of 2

isaacs · isaacs · commit 9208c8905823 · 2013-03-06T11:44:30.000-08:00
This prevents excessively raising the buffer level in tiny increments in
pathological cases.
diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
@@ -161,6 +161,19 @@ Readable.prototype.setEncoding = function(enc) {
   this._readableState.decoder = new StringDecoder(enc);
 };
 
+// Don't raise the hwm > 128MB
+var MAX_HWM = 0x800000;
+function roundUpToNextPowerOf2(n) {
+  if (n >= MAX_HWM) {
+    n = MAX_HWM;
+  } else {
+    // Get the next highest power of 2
+    n--;
+    for (var p = 1; p < 32; p <<= 1) n |= n >> p;
+    n++;
+  }
+  return n;
+}
 
 function howMuchToRead(n, state) {
   if (state.length === 0 && state.ended)
@@ -181,9 +194,11 @@ function howMuchToRead(n, state) {
     return 0;
 
   // If we're asking for more than the target buffer level,
-  // then raise the water mark.
+  // then raise the water mark.  Bump up to the next highest
+  // power of 2, to prevent increasing it excessively in tiny
+  // amounts.
   if (n > state.highWaterMark)
-    state.highWaterMark = n;
+    state.highWaterMark = roundUpToNextPowerOf2(n);
 
   // don't have that much.  return null, unless we've ended.
   if (n > state.length) {
diff --git a/test/simple/test-stream-readable-flow-recursion.js b/test/simple/test-stream-readable-flow-recursion.js
@@ -34,9 +34,15 @@ process.throwDeprecation = true;
 
 var stream = new Readable({ highWaterMark: 2 });
 var reads = 0;
+var total = 5000;
 stream._read = function(size) {
   reads++;
-  stream.push(new Buffer(size));
+  size = Math.min(size, total);
+  total -= size;
+  if (size === 0)
+    stream.push(null);
+  else
+    stream.push(new Buffer(size));
 };
 
 var depth = 0;
@@ -61,8 +67,9 @@ flow(stream, 5000, function() {
 process.on('exit', function(code) {
   assert.equal(reads, 2);
   // we pushed up the high water mark
-  assert.equal(stream._readableState.highWaterMark, 5000);
-  assert.equal(stream._readableState.length, 5000);
+  assert.equal(stream._readableState.highWaterMark, 8192);
+  // length is 0 right now, because we pulled it all out.
+  assert.equal(stream._readableState.length, 0);
   assert(!code);
   assert.equal(depth, 0);
   console.log('ok');
