Windows: make stdout/sterr pipes blocking

lib/net.js

@@ -26,6 +26,8 @@ var util = require('util');
 var assert = require('assert');
 var cares = process.binding('cares_wrap');
 var uv = process.binding('uv');
+var Pipe = process.binding('pipe_wrap').Pipe;
+
 
 var cluster;
 var errnoException = util._errnoException;
@@ -34,7 +36,6 @@ function noop() {}
 
 // constructor for lazy loading
 function createPipe() {
-  var Pipe = process.binding('pipe_wrap').Pipe;
   return new Pipe();
 }
 
@@ -147,6 +148,14 @@ function Socket(options) {
   } else if (!util.isUndefined(options.fd)) {
     this._handle = createHandle(options.fd);
     this._handle.open(options.fd);
+    if ((options.fd == 1 || options.fd == 2) &&
+        (this._handle instanceof Pipe) &&
+        process.platform === 'win32') {
+      // Make stdout and stderr blocking on Windows
+      var err = this._handle.setBlocking(true);
+      if (err)
+        throw errnoException(err, 'setBlocking');
+    }
     this.readable = options.readable !== false;
     this.writable = options.writable !== false;
   } else {

src/pipe_wrap.cc

@@ -91,6 +91,8 @@ void PipeWrap::Initialize(Handle<Object> target,
   NODE_SET_PROTOTYPE_METHOD(t, "unref", HandleWrap::Unref);
   NODE_SET_PROTOTYPE_METHOD(t, "ref", HandleWrap::Ref);
 
+  NODE_SET_PROTOTYPE_METHOD(t, "setBlocking", StreamWrap::SetBlocking);
+
   NODE_SET_PROTOTYPE_METHOD(t, "readStart", StreamWrap::ReadStart);
   NODE_SET_PROTOTYPE_METHOD(t, "readStop", StreamWrap::ReadStop);
   NODE_SET_PROTOTYPE_METHOD(t, "shutdown", StreamWrap::Shutdown);

src/stream_wrap.cc

@@ -87,7 +87,6 @@ void StreamWrap::UpdateWriteQueueSize() {
   object()->Set(env()->write_queue_size_string(), write_queue_size);
 }
 
-
 void StreamWrap::ReadStart(const FunctionCallbackInfo<Value>& args) {
   Environment* env = Environment::GetCurrent(args.GetIsolate());
   HandleScope scope(env->isolate());
@@ -496,6 +495,17 @@ void StreamWrap::WriteUcs2String(const FunctionCallbackInfo<Value>& args) {
   WriteStringImpl<UCS2>(args);
 }
 
+void StreamWrap::SetBlocking(const FunctionCallbackInfo<Value>& args) {
+  Environment* env = Environment::GetCurrent(args.GetIsolate());
+  HandleScope scope(env->isolate());
+
+  StreamWrap* wrap = Unwrap<StreamWrap>(args.This());
+
+  assert(args.Length() > 0);
+  int err = uv_stream_set_blocking(wrap->stream(), args[0]->IsTrue());
+
+  args.GetReturnValue().Set(err);
+}
 
 void StreamWrap::AfterWrite(uv_write_t* req, int status) {
   WriteWrap* req_wrap = CONTAINER_OF(req, WriteWrap, req_);

src/stream_wrap.h

@@ -126,6 +126,8 @@ class StreamWrap : public HandleWrap {
   static void WriteUtf8String(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void WriteUcs2String(const v8::FunctionCallbackInfo<v8::Value>& args);
 
+  static void SetBlocking(const v8::FunctionCallbackInfo<v8::Value>& args);
+
   inline StreamWrapCallbacks* callbacks() const {
     return callbacks_;
   }

test/simple/test-child-process-stdout-flush-exit.js

@@ -0,0 +1,69 @@
+// Copyright Joyent, Inc. and other Node contributors.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the
+// "Software"), to deal in the Software without restriction, including
+// without limitation the rights to use, copy, modify, merge, publish,
+// distribute, sublicense, and/or sell copies of the Software, and to permit
+// persons to whom the Software is furnished to do so, subject to the
+// following conditions:
+//
+// The above copyright notice and this permission notice shall be included
+// in all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
+// USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+
+
+
+var common = require('../common');
+var assert = require('assert');
+var path = require('path');
+
+// if child process output to console and exit
+if (process.argv[2] === 'child') {
+  console.log('hello');
+  for (var i = 0; i < 200; i++) {
+    console.log('filler');
+  }
+  console.log('goodbye');
+  process.exit(0);
+} else {
+  // parent process
+  var spawn = require('child_process').spawn;
+
+  // spawn self as child
+  var child = spawn(process.argv[0], [process.argv[1], 'child']);
+
+  var gotHello = false;
+  var gotBye = false;
+
+  child.stderr.setEncoding('utf8');
+  child.stderr.on('data', function (data) {
+    console.log('parent stderr: ' + data);
+    assert.ok(false);
+  });
+
+  // check if we receive both 'hello' at start and 'goodbye' at end
+  child.stdout.setEncoding('utf8');
+  child.stdout.on('data', function (data) {
+    if (data.slice(0, 6) == 'hello\n') {
+      gotHello = true;
+    } else if (data.slice(data.length - 8) == 'goodbye\n') {
+      gotBye = true;
+    } else {
+      gotBye = false;
+    }
+  });
+
+  child.on('close', function (data) {
+    assert(gotHello);
+    assert(gotBye);
+  });
+}

test/simple/test-stream2-stderr-sync.js

@@ -48,82 +48,45 @@ function parent() {
   });
 }
 
-function child0() {
-  // Just a very simple wrapper around TTY(2)
-  // Essentially the same as stderr, but without all the net stuff.
-  var Writable = require('stream').Writable;
-  var util = require('util');
-
-  // a lowlevel stderr writer
-  var TTY = process.binding('tty_wrap').TTY;
-  var handle = new TTY(2, false);
-
-  util.inherits(W, Writable);
-
-  function W(opts) {
-    Writable.call(this, opts);
-  }
-
-  W.prototype._write = function(chunk, encoding, cb) {
-    var req = { oncomplete: afterWrite };
-    var err = handle.writeUtf8String(req, chunk.toString() + '\n');
-    if (err) throw errnoException(err, 'write');
-    // here's the problem.
-    // it needs to tell the Writable machinery that it's ok to write
-    // more, but that the current buffer length is handle.writeQueueSize
-    if (req.writeQueueSize === 0)
-      req.cb = cb;
-    else
-      cb();
-  }
-  function afterWrite(status, handle, req) {
-    if (req.cb)
-      req.cb();
-  }
-
-  var w = new W
-  w.write('child 0');
-  w.write('foo');
-  w.write('bar');
-  w.write('baz');
-}
-
 // using console.error
-function child1() {
-  console.error('child 1');
+function child0() {
+  console.error('child 0');
   console.error('foo');
   console.error('bar');
   console.error('baz');
 }
 
 // using process.stderr
-function child2() {
-  process.stderr.write('child 2\n');
+function child1() {
+  process.stderr.write('child 1\n');
   process.stderr.write('foo\n');
   process.stderr.write('bar\n');
   process.stderr.write('baz\n');
 }
 
 // using a net socket
-function child3() {
+function child2() {
   var net = require('net');
-  var socket = new net.Socket({ fd: 2 });
-  socket.write('child 3\n');
+  var socket = new net.Socket({
+    fd: 2,
+    readable: false,
+    writable: true});
+  socket.write('child 2\n');
   socket.write('foo\n');
   socket.write('bar\n');
   socket.write('baz\n');
 }
 
 
-function child4() {
-  console.error('child 4\nfoo\nbar\nbaz');
+function child3() {
+  console.error('child 3\nfoo\nbar\nbaz');
 }
 
-function child5() {
-  process.stderr.write('child 5\nfoo\nbar\nbaz\n');
+function child4() {
+  process.stderr.write('child 4\nfoo\nbar\nbaz\n');
 }
 
-var children = [ child0, child1, child2, child3, child4, child5 ];
+var children = [ child0, child1, child2, child3, child4 ];
 
 if (!process.argv[2]) {
   parent();