Skip to content

Commit

Permalink
http: fix stalled pipeline bug
Browse files Browse the repository at this point in the history 
This is a two-part fix:

- Fix pending data notification in `OutgoingMessage` to notify server
  about flushed data too
- Fix pause/resume behavior for the consumed socket. `resume` event is
  emitted on a next tick, and `socket._paused` can already be `true` at
  this time. Pause the socket again to avoid PAUSED error on parser.

Fix: #3332
PR-URL: #3342
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Trevor Norris <trev.norris@gmail.com>
  • Loading branch information
@indutny @jasnell
indutny authored and jasnell committed Oct 29, 2015
1 parent 4bac5d9 commit 0427cdf
Show file tree
Hide file tree
Showing 5 changed files with 120 additions and 47 deletions.
4 changes: 4 additions & 0 deletions lib/_http_common.js
View file
Expand Up @@ -140,6 +140,7 @@ var parsers = new FreeList('parsers', 1000, function() {

parser._headers = [];
parser._url = '';
parser._consumed = false;

// Only called in the slow case where slow means
// that the request headers were either fragmented
Expand Down Expand Up @@ -167,6 +168,9 @@ function freeParser(parser, req, socket) {
if (parser) {
parser._headers = [];
parser.onIncoming = null;
if (parser._consumed)
parser.unconsume();
parser._consumed = false;
if (parser.socket)
parser.socket.parser = null;
parser.socket = null;
Expand Down
66 changes: 31 additions & 35 deletions lib/_http_outgoing.js
View file
Expand Up @@ -131,7 +131,7 @@ OutgoingMessage.prototype._send = function(data, encoding, callback) {
this.outputEncodings.unshift('binary');
this.outputCallbacks.unshift(null);
this.outputSize += this._header.length;
if (this._onPendingData !== null)
if (typeof this._onPendingData === 'function')
this._onPendingData(this._header.length);
}
this._headerSent = true;
Expand All @@ -154,22 +154,7 @@ OutgoingMessage.prototype._writeRaw = function(data, encoding, callback) {
// There might be pending data in the this.output buffer.
var outputLength = this.output.length;
if (outputLength > 0) {
var output = this.output;
var outputEncodings = this.outputEncodings;
var outputCallbacks = this.outputCallbacks;
connection.cork();
for (var i = 0; i < outputLength; i++) {
connection.write(output[i], outputEncodings[i],
outputCallbacks[i]);
}
connection.uncork();

this.output = [];
this.outputEncodings = [];
this.outputCallbacks = [];
if (this._onPendingData !== null)
this._onPendingData(-this.outputSize);
this.outputSize = 0;
this._flushOutput(connection);
} else if (data.length === 0) {
if (typeof callback === 'function')
process.nextTick(callback);
Expand All @@ -194,7 +179,7 @@ OutgoingMessage.prototype._buffer = function(data, encoding, callback) {
this.outputEncodings.push(encoding);
this.outputCallbacks.push(callback);
this.outputSize += data.length;
if (this._onPendingData !== null)
if (typeof this._onPendingData === 'function')
this._onPendingData(data.length);
return false;
};
Expand Down Expand Up @@ -630,26 +615,11 @@ OutgoingMessage.prototype._finish = function() {
// to attempt to flush any pending messages out to the socket.
OutgoingMessage.prototype._flush = function() {
var socket = this.socket;
var outputLength, ret;
var ret;

if (socket && socket.writable) {
// There might be remaining data in this.output; write it out
outputLength = this.output.length;
if (outputLength > 0) {
var output = this.output;
var outputEncodings = this.outputEncodings;
var outputCallbacks = this.outputCallbacks;
socket.cork();
for (var i = 0; i < outputLength; i++) {
ret = socket.write(output[i], outputEncodings[i],
outputCallbacks[i]);
}
socket.uncork();

this.output = [];
this.outputEncodings = [];
this.outputCallbacks = [];
}
ret = this._flushOutput(socket);

if (this.finished) {
// This is a queue to the server or client to bring in the next this.
Expand All @@ -661,6 +631,32 @@ OutgoingMessage.prototype._flush = function() {
}
};

OutgoingMessage.prototype._flushOutput = function _flushOutput(socket) {
var ret;
var outputLength = this.output.length;
if (outputLength <= 0)
return ret;

var output = this.output;
var outputEncodings = this.outputEncodings;
var outputCallbacks = this.outputCallbacks;
socket.cork();
for (var i = 0; i < outputLength; i++) {
ret = socket.write(output[i], outputEncodings[i],
outputCallbacks[i]);
}
socket.uncork();

this.output = [];
this.outputEncodings = [];
this.outputCallbacks = [];
if (typeof this._onPendingData === 'function')
this._onPendingData(-this.outputSize);
this.outputSize = 0;

return ret;
};


OutgoingMessage.prototype.flushHeaders = function() {
if (!this._header) {
Expand Down
39 changes: 33 additions & 6 deletions lib/_http_server.js
View file
Expand Up @@ -343,8 +343,10 @@ function connectionListener(socket) {
socket.on = socketOnWrap;

var external = socket._handle._externalStream;
if (external)
if (external) {
parser._consumed = true;
parser.consume(external);
}
external = null;
parser[kOnExecute] = onParserExecute;

Expand Down Expand Up @@ -382,7 +384,7 @@ function connectionListener(socket) {
socket.removeListener('data', socketOnData);
socket.removeListener('end', socketOnEnd);
socket.removeListener('close', serverSocketCloseListener);
parser.unconsume(socket._handle._externalStream);
unconsume(parser, socket);
parser.finish();
freeParser(parser, req, null);
parser = null;
Expand Down Expand Up @@ -530,13 +532,38 @@ function connectionListener(socket) {
exports._connectionListener = connectionListener;

function onSocketResume() {
if (this._handle)
// It may seem that the socket is resumed, but this is an enemy's trick to
// deceive us! `resume` is emitted asynchronously, and may be called from
// `incoming.readStart()`. Stop the socket again here, just to preserve the
// state.
//
// We don't care about stream semantics for the consumed socket anyway.
if (this._paused) {
this.pause();
return;
}

if (this._handle && !this._handle.reading) {
this._handle.reading = true;
this._handle.readStart();
}
}

function onSocketPause() {
if (this._handle)
if (this._handle && this._handle.reading) {
this._handle.reading = false;
this._handle.readStop();
}
}

function unconsume(parser, socket) {
if (socket._handle) {
if (parser._consumed)
parser.unconsume(socket._handle._externalStream);
parser._consumed = false;
socket.removeListener('pause', onSocketPause);
socket.removeListener('resume', onSocketResume);
}
}

function socketOnWrap(ev, fn) {
Expand All @@ -546,8 +573,8 @@ function socketOnWrap(ev, fn) {
return res;
}

if (this._handle && (ev === 'data' || ev === 'readable'))
this.parser.unconsume(this._handle._externalStream);
if (ev === 'data' || ev === 'readable')
unconsume(this.parser, this);

return res;
}
17 changes: 11 additions & 6 deletions src/node_http_parser.cc
View file
Expand Up @@ -484,13 +484,18 @@ class Parser : public BaseObject {
if (parser->prev_alloc_cb_.is_empty())
return;

CHECK(args[0]->IsExternal());
Local<External> stream_obj = args[0].As<External>();
StreamBase* stream = static_cast<StreamBase*>(stream_obj->Value());
CHECK_NE(stream, nullptr);
// Restore stream's callbacks
if (args.Length() == 1 && args[0]->IsExternal()) {
Local<External> stream_obj = args[0].As<External>();
StreamBase* stream = static_cast<StreamBase*>(stream_obj->Value());
CHECK_NE(stream, nullptr);

stream->set_alloc_cb(parser->prev_alloc_cb_);
stream->set_read_cb(parser->prev_read_cb_);
}

stream->set_alloc_cb(parser->prev_alloc_cb_);
stream->set_read_cb(parser->prev_read_cb_);
parser->prev_alloc_cb_.clear();
parser->prev_read_cb_.clear();
}


Expand Down
41 changes: 41 additions & 0 deletions test/parallel/test-http-pipeline-regr-3332.js
View file
@@ -0,0 +1,41 @@
'use strict';
const common = require('../common');
const assert = require('assert');
const http = require('http');
const net = require('net');

const big = new Buffer(16 * 1024);
big.fill('A');

const COUNT = 1e4;

var received = 0;

var client;
const server = http.createServer(function(req, res) {
res.end(big, function() {
if (++received === COUNT) {
server.close();
client.end();
}
});
}).listen(common.PORT, function() {
var req = new Array(COUNT + 1).join('GET / HTTP/1.1\r\n\r\n');
client = net.connect(common.PORT, function() {
client.write(req);
});

// Just let the test terminate instead of hanging
client.on('close', function() {
if (received !== COUNT)
server.close();
});
client.resume();
});

process.on('exit', function() {
// The server should pause connection on pipeline flood, but it shoul still
// resume it and finish processing the requests, when its output queue will
// be empty again.
assert.equal(received, COUNT);
});

0 comments on commit 0427cdf

Please sign in to comment.