http: provide backpressure for pipeline flood

If a client sends a lot more pipelined requests than we can handle, then
we need to provide backpressure so that the client knows to back off.
Do this by pausing both the stream and the parser itself when the
responses are not being read by the downstream client.

Backport of 085dd30

1 files changed, 37 insertions, 5 deletions

diff --git a/lib/http.js b/lib/http.js
index ec68f54668..32e2ef3072 100644
--- a/lib/http.js
+++ b/lib/http.js
@@ -37,7 +37,8 @@ if (process.env.NODE_DEBUG && /http/.test(process.env.NODE_DEBUG)) {
 }
 
 function readStart(socket) {
-  if (!socket || !socket._handle || !socket._handle.readStart) return;
+  if (!socket || !socket._handle || !socket._handle.readStart || socket._paused)
+    return;
   socket._handle.readStart();
 }
 
@@ -172,10 +173,8 @@ function parserOnMessageComplete() {
     stream.push(null);
   }
 
-  if (parser.socket.readable) {
-    // force to read the next incoming message
-    readStart(parser.socket);
-  }
+  // force to read the next incoming message
+  readStart(parser.socket);
 }
 
 
@@ -1963,6 +1962,7 @@ function connectionListener(socket) {
   });
 
   socket.ondata = function(d, start, end) {
+    assert(!socket._paused);
     var ret = parser.execute(d, start, end - start);
     if (ret instanceof Error) {
       debug('parse error');
@@ -1989,6 +1989,12 @@ function connectionListener(socket) {
         socket.destroy();
       }
     }
+
+    if (socket._paused) {
+      // onIncoming paused the socket, we should pause the parser as well
+      debug('pause parser');
+      socket.parser.pause();
+    }
   };
 
   socket.onend = function() {
@@ -2017,9 +2023,35 @@ function connectionListener(socket) {
   // The following callback is issued after the headers have been read on a
   // new message. In this callback we setup the response object and pass it
   // to the user.
+
+  socket._paused = false;
+  function socketOnDrain() {
+    // If we previously paused, then start reading again.
+    if (socket._paused) {
+      socket._paused = false;
+      socket.parser.resume();
+      readStart(socket);
+    }
+  }
+  socket.on('drain', socketOnDrain);
+
   parser.onIncoming = function(req, shouldKeepAlive) {
     incoming.push(req);
 
+    // If the writable end isn't consuming, then stop reading
+    // so that we don't become overwhelmed by a flood of
+    // pipelined requests that may never be resolved.
+    if (!socket._paused) {
+      var needPause = socket._writableState.needDrain;
+      if (needPause) {
+        socket._paused = true;
+        // We also need to pause the parser, but don't do that until after
+        // the call to execute, because we may still be processing the last
+        // chunk.
+        readStop(socket);
+      }
+    }
+
     var res = new ServerResponse(req);
 
     res.shouldKeepAlive = shouldKeepAlive;