Make stdout and stderr async.

This required some minor changes to the Windows eventhandler, as we now spawn a new thread for writes (just like we do for reads).

BUG=https://code.google.com/p/dart/issues/detail?id=7218
R=sgjesse@google.com

Review URL: https://codereview.chromium.org//23609060

git-svn-id: https://dart.googlecode.com/svn/branches/bleeding_edge/dart@27811 260f80e4-7a28-3924-810f-c04153c831b5

runtime/bin/eventhandler.cc

@@ -74,6 +74,12 @@ void EventHandler::Stop() {
 }
 
 
+EventHandlerImplementation* EventHandler::delegate() {
+  if (event_handler == NULL) return NULL;
+  return &event_handler->delegate_;
+}
+
+
 /*
  * Send data to the EventHandler thread to register for a given instance
  * args[0] a ReceivePort args[1] with a notification event args[2].

runtime/bin/eventhandler.h

@@ -119,6 +119,8 @@ class EventHandler {
    */
   static void Stop();
 
+  static EventHandlerImplementation* delegate();
+
  private:
   friend class EventHandlerImplementation;
   EventHandlerImplementation delegate_;

runtime/bin/eventhandler_win.cc

@@ -107,6 +107,7 @@ Handle::Handle(HANDLE handle)
       pending_read_(NULL),
       pending_write_(NULL),
       last_error_(NOERROR),
+      thread_wrote_(0),
       flags_(0) {
   InitializeCriticalSection(&cs_);
 }
@@ -122,6 +123,7 @@ Handle::Handle(HANDLE handle, Dart_Port port)
       pending_read_(NULL),
       pending_write_(NULL),
       last_error_(NOERROR),
+      thread_wrote_(0),
       flags_(0) {
   InitializeCriticalSection(&cs_);
 }
@@ -332,7 +334,7 @@ bool FileHandle::IsClosed() {
 
 
 void FileHandle::DoClose() {
-  if (GetStdHandle(STD_OUTPUT_HANDLE) == handle_) {
+  if (handle_ == GetStdHandle(STD_OUTPUT_HANDLE)) {
     int fd = _open("NUL", _O_WRONLY);
     ASSERT(fd >= 0);
     _dup2(fd, _fileno(stdout));
@@ -358,10 +360,6 @@ bool DirectoryWatchHandle::IsClosed() {
 }
 
 
-void DirectoryWatchHandle::DoClose() {
-  Handle::DoClose();
-}
-
 bool DirectoryWatchHandle::IssueRead() {
   ScopedLock lock(this);
   OverlappedBuffer* buffer = OverlappedBuffer::AllocateReadBuffer(kBufferSize);
@@ -562,30 +560,72 @@ int Handle::Read(void* buffer, int num_bytes) {
 }
 
 
+static unsigned int __stdcall WriteFileThread(void* args) {
+  Handle* handle = reinterpret_cast<Handle*>(args);
+  handle->WriteSyncCompleteAsync();
+  return 0;
+}
+
+
+void Handle::WriteSyncCompleteAsync() {
+  ASSERT(pending_write_ != NULL);
+
+  DWORD bytes_written = -1;
+  BOOL ok = WriteFile(handle_,
+                      pending_write_->GetBufferStart(),
+                      pending_write_->GetBufferSize(),
+                      &bytes_written,
+                      NULL);
+  if (!ok) {
+    if (GetLastError() != ERROR_BROKEN_PIPE) {
+      Log::PrintErr("WriteFile failed %d\n", GetLastError());
+    }
+    bytes_written = 0;
+  }
+  thread_wrote_ += bytes_written;
+  OVERLAPPED* overlapped = pending_write_->GetCleanOverlapped();
+  ok = PostQueuedCompletionStatus(event_handler_->completion_port(),
+                                  bytes_written,
+                                  reinterpret_cast<ULONG_PTR>(this),
+                                  overlapped);
+  if (!ok) {
+    FATAL("PostQueuedCompletionStatus failed");
+  }
+}
+
+
 int Handle::Write(const void* buffer, int num_bytes) {
   ScopedLock lock(this);
+  if (pending_write_ != NULL) return 0;
+  if (num_bytes > kBufferSize) num_bytes = kBufferSize;
   if (SupportsOverlappedIO()) {
-    if (pending_write_ != NULL) return 0;
     if (completion_port_ == INVALID_HANDLE_VALUE) return 0;
-    if (num_bytes > kBufferSize) num_bytes = kBufferSize;
     pending_write_ = OverlappedBuffer::AllocateWriteBuffer(num_bytes);
     pending_write_->Write(buffer, num_bytes);
     if (!IssueWrite()) return -1;
     return num_bytes;
   } else {
-    DWORD bytes_written = -1;
-    BOOL ok = WriteFile(handle_,
-                        buffer,
-                        num_bytes,
-                        &bytes_written,
-                        NULL);
-    if (!ok) {
-      if (GetLastError() != ERROR_BROKEN_PIPE) {
-        Log::PrintErr("WriteFile failed: %d\n", GetLastError());
-      }
-      event_handler_->HandleClosed(this);
+    // In the case of stdout and stderr, OverlappedIO is not supported.
+    // Here we'll instead spawn a new thread for each write, to make it async.
+    // This code is actually never exposed to the user, as stdout and stderr is
+    // not available as a RawSocket, but only wrapped in a Socket.
+    // Note that we return '0', unless a thread have already completed a write.
+    // TODO(ajohnsen): Don't spawn a new thread per write. Issue 13541.
+    if (thread_wrote_ > 0) {
+      if (num_bytes > thread_wrote_) num_bytes = thread_wrote_;
+      thread_wrote_ -= num_bytes;
+      return num_bytes;
     }
-    return bytes_written;
+    pending_write_ = OverlappedBuffer::AllocateWriteBuffer(num_bytes);
+    pending_write_->Write(buffer, num_bytes);
+    // Completing asynchronously through thread.
+    uint32_t tid;
+    uintptr_t thread_handle =
+        _beginthreadex(NULL, 32 * 1024, WriteFileThread, this, 0, &tid);
+    if (thread_handle == -1) {
+      FATAL("Failed to start write file thread");
+    }
+    return 0;
   }
 }
 

runtime/bin/eventhandler_win.h

@@ -211,6 +211,7 @@ class Handle {
   }
 
   void ReadSyncCompleteAsync();
+  void WriteSyncCompleteAsync();
 
   DWORD last_error() { return last_error_; }
   void set_last_error(DWORD last_error) { last_error_ = last_error; }
@@ -241,6 +242,7 @@ class Handle {
   OverlappedBuffer* pending_write_;  // Buffer for pending write
 
   DWORD last_error_;
+  DWORD thread_wrote_;
 
  private:
   int flags_;
@@ -272,7 +274,6 @@ class DirectoryWatchHandle : public Handle {
 
   virtual void EnsureInitialized(EventHandlerImplementation* event_handler);
   virtual bool IsClosed();
-  virtual void DoClose();
 
   virtual bool IssueRead();
 

runtime/bin/socket_android.cc

@@ -202,6 +202,7 @@ int Socket::GetType(intptr_t fd) {
 
 
 intptr_t Socket::GetStdioHandle(intptr_t num) {
+  Socket::SetNonBlocking(num);
   return num;
 }
 

runtime/bin/socket_linux.cc

@@ -202,6 +202,7 @@ int Socket::GetType(intptr_t fd) {
 
 
 intptr_t Socket::GetStdioHandle(intptr_t num) {
+  Socket::SetNonBlocking(num);
   return num;
 }
 

runtime/bin/socket_macos.cc

@@ -202,6 +202,7 @@ int Socket::GetType(intptr_t fd) {
 
 
 intptr_t Socket::GetStdioHandle(intptr_t num) {
+  Socket::SetNonBlocking(num);
   return num;
 }
 

runtime/bin/socket_win.cc

@@ -201,6 +201,7 @@ intptr_t Socket::GetStdioHandle(intptr_t num) {
   FileHandle* file_handle = new FileHandle(handle);
   if (file_handle == NULL) return -1;
   file_handle->MarkDoesNotSupportOverlappedIO();
+  file_handle->EnsureInitialized(EventHandler::delegate());
   return reinterpret_cast<intptr_t>(file_handle);
 }
 

sdk/lib/_internal/pub/bin/pub.dart

@@ -48,7 +48,6 @@ void main() {
       log.error('Could not find a command named "${options.rest[0]}".');
       log.error('Run "pub help" to see available commands.');
       flushThenExit(exit_codes.USAGE);
-      return;
     }
     return;
   }

tests/standalone/io/process_sync_script.dart

@@ -4,6 +4,7 @@
 //
 // Utility script to generate some output on stdout and stderr.
 
+import "dart:async";
 import "dart:math";
 import "dart:io";
 
@@ -20,5 +21,7 @@ main() {
     stdout.write(stdoutBlock);
     stderr.write(stderrBlock);
   }
-  exit(int.parse(options.arguments[3]));
+  Future.wait([stdout.close(), stderr.close()]).then((_) {
+    exit(int.parse(options.arguments[3]));
+  });
 }