diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js index afe3b3baf065c2..c5c4f2eaac94f6 100644 --- a/lib/_stream_readable.js +++ b/lib/_stream_readable.js @@ -274,7 +274,15 @@ function howMuchToRead(n, state) { } // you can override either this method, or the async _read(n) below. -Readable.prototype.read = function(n) { +Readable.prototype.read = function(arg) { + var n; + var buf; + if (arg instanceof Buffer) { + buf = arg; + } else { + n = arg; + } + debug('read', n); var state = this._readableState; var nOrig = n; @@ -352,7 +360,7 @@ Readable.prototype.read = function(n) { if (state.length === 0) state.needReadable = true; // call internal read method - this._read(state.highWaterMark); + this._read(state.highWaterMark, buf); state.sync = false; } diff --git a/lib/net.js b/lib/net.js index 509112f58f2077..54251fad3ccc59 100644 --- a/lib/net.js +++ b/lib/net.js @@ -401,12 +401,14 @@ Object.defineProperty(Socket.prototype, 'bufferSize', { // Just call handle.readStart until we have enough in the buffer -Socket.prototype._read = function(n) { +Socket.prototype._read = function(n, buf) { debug('_read'); + if (this._handle && !this._handle[0]) + this._handle[0] = buf; if (this.connecting || !this._handle) { debug('_read wait for connection'); - this.once('connect', () => this._read(n)); + this.once('connect', () => this._read(n, buf)); } else if (!this._handle.reading) { // not already reading, start the flow debug('Socket._read readStart'); @@ -526,6 +528,9 @@ function onread(nread, buffer) { var self = handle.owner; assert(handle === self._handle, 'handle != self._handle'); + var prealloc = buffer === handle[0]; + handle[0] = undefined; + self._unrefTimer(); debug('onread', nread); @@ -539,7 +544,11 @@ function onread(nread, buffer) { // called again. // Optimization: emit the original buffer with end points - var ret = self.push(buffer); + var ret; + if (prealloc) + ret = self.push(buffer.slice(0, nread)); + else + ret = self.push(buffer); if (handle.reading && !ret) { handle.reading = false; diff --git a/src/stream_wrap.cc b/src/stream_wrap.cc index b942e48078e29a..6d3739e3431832 100644 --- a/src/stream_wrap.cc +++ b/src/stream_wrap.cc @@ -149,13 +149,26 @@ void StreamWrap::OnAlloc(uv_handle_t* handle, void StreamWrap::OnAllocImpl(size_t size, uv_buf_t* buf, void* ctx) { - buf->base = static_cast(malloc(size)); - buf->len = size; + StreamWrap* wrap = static_cast(ctx); + Environment* env = wrap->env(); + HandleScope handle_scope(env->isolate()); + Context::Scope context_scope(env->context()); - if (buf->base == nullptr && size > 0) { - FatalError( - "node::StreamWrap::DoAlloc(size_t, uv_buf_t*, void*)", - "Out Of Memory"); + Local prealloc = wrap->object()->Get(0); + if (Buffer::HasInstance(prealloc)) { + buf->base = Buffer::Data(prealloc); + buf->len = Buffer::Length(prealloc); + wrap->allocated_ = false; + } else { + buf->base = static_cast(malloc(size)); + buf->len = size; + wrap->allocated_ = true; + + if (buf->base == nullptr && size > 0) { + FatalError( + "node::StreamWrap::DoAlloc(size_t, uv_buf_t*, void*)", + "Out Of Memory"); + } } } @@ -192,19 +205,25 @@ void StreamWrap::OnReadImpl(ssize_t nread, Local pending_obj; if (nread < 0) { - if (buf->base != nullptr) + if (buf->base != nullptr && wrap->allocated_) free(buf->base); + wrap->allocated_ = false; wrap->EmitData(nread, Local(), pending_obj); return; } if (nread == 0) { - if (buf->base != nullptr) + if (buf->base != nullptr && wrap->allocated_) free(buf->base); + wrap->allocated_ = false; return; } - char* base = static_cast(realloc(buf->base, nread)); + char* base; + if (wrap->allocated_) + base = static_cast(realloc(buf->base, nread)); + else + base = buf->base; CHECK_LE(static_cast(nread), buf->len); if (pending == UV_TCP) { @@ -217,8 +236,14 @@ void StreamWrap::OnReadImpl(ssize_t nread, CHECK_EQ(pending, UV_UNKNOWN_HANDLE); } - Local obj = Buffer::New(env, base, nread).ToLocalChecked(); + Local obj; + if (!wrap->allocated_) { + obj = wrap->object()->Get(0).As(); + } else { + obj = Buffer::New(env, base, nread).ToLocalChecked(); + } wrap->EmitData(nread, obj, pending_obj); + wrap->allocated_ = false; } diff --git a/src/stream_wrap.h b/src/stream_wrap.h index b0d9986db50c5d..9ad43a5a06afff 100644 --- a/src/stream_wrap.h +++ b/src/stream_wrap.h @@ -98,6 +98,7 @@ class StreamWrap : public HandleWrap, public StreamBase { void* ctx); uv_stream_t* const stream_; + bool allocated_; };