From d11df78618799dd8900a9e89b6090e3d11cdeb2d Mon Sep 17 00:00:00 2001 From: "ajohnsen@google.com" Date: Tue, 24 Sep 2013 11:47:53 +0000 Subject: [PATCH] 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 | 6 ++ runtime/bin/eventhandler.h | 2 + runtime/bin/eventhandler_win.cc | 78 +++++++++++++++----- runtime/bin/eventhandler_win.h | 3 +- runtime/bin/socket_android.cc | 1 + runtime/bin/socket_linux.cc | 1 + runtime/bin/socket_macos.cc | 1 + runtime/bin/socket_win.cc | 1 + sdk/lib/_internal/pub/bin/pub.dart | 1 - tests/standalone/io/process_sync_script.dart | 5 +- 10 files changed, 77 insertions(+), 22 deletions(-) diff --git a/runtime/bin/eventhandler.cc b/runtime/bin/eventhandler.cc index c25a2fac4e1..8de036eb8bf 100644 --- a/runtime/bin/eventhandler.cc +++ b/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]. diff --git a/runtime/bin/eventhandler.h b/runtime/bin/eventhandler.h index a9b0bcd0896..64f93063372 100644 --- a/runtime/bin/eventhandler.h +++ b/runtime/bin/eventhandler.h @@ -119,6 +119,8 @@ class EventHandler { */ static void Stop(); + static EventHandlerImplementation* delegate(); + private: friend class EventHandlerImplementation; EventHandlerImplementation delegate_; diff --git a/runtime/bin/eventhandler_win.cc b/runtime/bin/eventhandler_win.cc index 1584c8b71e4..59b22077d4d 100644 --- a/runtime/bin/eventhandler_win.cc +++ b/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(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(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; } } diff --git a/runtime/bin/eventhandler_win.h b/runtime/bin/eventhandler_win.h index 3adaf5795a1..945277b78dc 100644 --- a/runtime/bin/eventhandler_win.h +++ b/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(); diff --git a/runtime/bin/socket_android.cc b/runtime/bin/socket_android.cc index 31495dd3195..18b2bd88133 100644 --- a/runtime/bin/socket_android.cc +++ b/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; } diff --git a/runtime/bin/socket_linux.cc b/runtime/bin/socket_linux.cc index 72a6330867f..24e688d41a5 100644 --- a/runtime/bin/socket_linux.cc +++ b/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; } diff --git a/runtime/bin/socket_macos.cc b/runtime/bin/socket_macos.cc index 6670d48a3c0..2b6138fa8d0 100644 --- a/runtime/bin/socket_macos.cc +++ b/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; } diff --git a/runtime/bin/socket_win.cc b/runtime/bin/socket_win.cc index 696f8cc6744..87a0b30e2f5 100644 --- a/runtime/bin/socket_win.cc +++ b/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(file_handle); } diff --git a/sdk/lib/_internal/pub/bin/pub.dart b/sdk/lib/_internal/pub/bin/pub.dart index d972e63fe3e..ecd58bdee25 100644 --- a/sdk/lib/_internal/pub/bin/pub.dart +++ b/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; } diff --git a/tests/standalone/io/process_sync_script.dart b/tests/standalone/io/process_sync_script.dart index d8aab9c7b28..53691242514 100644 --- a/tests/standalone/io/process_sync_script.dart +++ b/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])); + }); }