Join embeder threads on Windows.

runtime/bin/eventhandler_win.cc

 // Copyright (c) 2013, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 #include "platform/globals.h"
 #if defined(TARGET_OS_WINDOWS)
 
 #include "bin/eventhandler.h"
 #include "bin/eventhandler_win.h"
 
@@ skipping 103 matching lines @@
 
 Handle::Handle(intptr_t handle)
     : DescriptorInfoBase(handle),
       handle_(reinterpret_cast<HANDLE>(handle)),
       completion_port_(INVALID_HANDLE_VALUE),
       event_handler_(NULL),
       data_ready_(NULL),
       pending_read_(NULL),
       pending_write_(NULL),
       last_error_(NOERROR),
-      flags_(0) {
-  InitializeCriticalSection(&cs_);
+      flags_(0),
+      read_thread_id_(Thread::kInvalidThreadId),
+      read_thread_starting_(false),
+      read_thread_finished_(false),
+      monitor_(new Monitor()) {
 }
 
 
 Handle::~Handle() {
-  DeleteCriticalSection(&cs_);
+  delete monitor_;
 }
 
 
-void Handle::Lock() {
-  EnterCriticalSection(&cs_);
-}
-
-
-void Handle::Unlock() {
-  LeaveCriticalSection(&cs_);
-}
-
-
 bool Handle::CreateCompletionPort(HANDLE completion_port) {
   completion_port_ = CreateIoCompletionPort(handle(),
                                             completion_port,
                                             reinterpret_cast<ULONG_PTR>(this),
                                             0);
   if (completion_port_ == NULL) {
     return false;
   }
   return true;
 }
 
 
 void Handle::Close() {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   if (!IsClosing()) {
     // Close the socket and set the closing state. This close method can be
     // called again if this socket has pending IO operations in flight.
     MarkClosing();
     // Perform handle type specific closing.
     DoClose();
   }
   ASSERT(IsHandleClosed());
 }
 
 
 void Handle::DoClose() {
   if (!IsHandleClosed()) {
     CloseHandle(handle_);
     handle_ = INVALID_HANDLE_VALUE;
   }
 }
 
 
 bool Handle::HasPendingRead() {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   return pending_read_ != NULL;
 }
 
 
 bool Handle::HasPendingWrite() {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   return pending_write_ != NULL;
 }
 
 
-void Handle::ReadComplete(OverlappedBuffer* buffer) {
-  ScopedLock lock(this);
-  // Currently only one outstanding read at the time.
-  ASSERT(pending_read_ == buffer);
-  ASSERT(data_ready_ == NULL);
-  if (!IsClosing() && !buffer->IsEmpty()) {
-    data_ready_ = pending_read_;
-  } else {
-    OverlappedBuffer::DisposeBuffer(buffer);
+void Handle::WaitForReadThreadStarted() {
+  MonitorLocker ml(monitor_);
+  while (read_thread_starting_) {
+    ml.Wait();
   }
-  pending_read_ = NULL;
 }
 
 
+void Handle::WaitForReadThreadFinished() {
+  // Join the Reader thread if there is one.
+  ThreadId to_join = Thread::kInvalidThreadId;
+  {
+    MonitorLocker ml(monitor_);
+    if (read_thread_id_ != Thread::kInvalidThreadId) {
+      while (!read_thread_finished_) {
+        ml.Wait();
+      }
+      read_thread_finished_ = false;
+      to_join = read_thread_id_;
+      read_thread_id_ = Thread::kInvalidThreadId;
+    }
+  }
+  if (to_join != Thread::kInvalidThreadId) {
+    Thread::Join(to_join);
+  }
+}
+
+
+void Handle::ReadComplete(OverlappedBuffer* buffer) {
+  WaitForReadThreadStarted();
+  {
+    MonitorLocker ml(monitor_);
+    // Currently only one outstanding read at the time.
+    ASSERT(pending_read_ == buffer);
+    ASSERT(data_ready_ == NULL);
+    if (!IsClosing() && !buffer->IsEmpty()) {
+      data_ready_ = pending_read_;
+    } else {
+      OverlappedBuffer::DisposeBuffer(buffer);
+    }
+    pending_read_ = NULL;
+  }
+  WaitForReadThreadFinished();
+}
+
+
 void Handle::RecvFromComplete(OverlappedBuffer* buffer) {
   ReadComplete(buffer);
 }
 
 
 void Handle::WriteComplete(OverlappedBuffer* buffer) {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   // Currently only one outstanding write at the time.
   ASSERT(pending_write_ == buffer);
   OverlappedBuffer::DisposeBuffer(buffer);
   pending_write_ = NULL;
 }
 
 
 static void ReadFileThread(uword args) {
   Handle* handle = reinterpret_cast<Handle*>(args);
   handle->ReadSyncCompleteAsync();
 }
 
 
+void Handle::NotifyReadThreadStarted() {
+  MonitorLocker ml(monitor_);
+  ASSERT(read_thread_starting_);
+  ASSERT(read_thread_id_ == Thread::kInvalidThreadId);
+  read_thread_id_ = Thread::GetCurrentThreadId();
+  read_thread_starting_ = false;
+  ml.Notify();
+}
+
+void Handle::NotifyReadThreadFinished() {
+  MonitorLocker ml(monitor_);
+  ASSERT(!read_thread_finished_);
+  ASSERT(read_thread_id_ != Thread::kInvalidThreadId);
+  read_thread_finished_ = true;
+  ml.Notify();
+}
+
+
 void Handle::ReadSyncCompleteAsync() {
+  NotifyReadThreadStarted();
   ASSERT(pending_read_ != NULL);
   ASSERT(pending_read_->GetBufferSize() >= kStdOverlappedBufferSize);
 
   DWORD buffer_size = pending_read_->GetBufferSize();
   if (GetFileType(handle_) == FILE_TYPE_CHAR) {
     buffer_size = kStdOverlappedBufferSize;
   }
+  char* buffer_start = pending_read_->GetBufferStart();
   DWORD bytes_read = 0;
   BOOL ok = ReadFile(handle_,
-                     pending_read_->GetBufferStart(),
+                     buffer_start,
                      buffer_size,
                      &bytes_read,
                      NULL);
   if (!ok) {
     bytes_read = 0;
   }
   OVERLAPPED* overlapped = pending_read_->GetCleanOverlapped();
   ok = PostQueuedCompletionStatus(event_handler_->completion_port(),
                                   bytes_read,
                                   reinterpret_cast<ULONG_PTR>(this),
                                   overlapped);
   if (!ok) {
     FATAL("PostQueuedCompletionStatus failed");
   }
+  NotifyReadThreadFinished();
 }
 
 
 bool Handle::IssueRead() {
-  ScopedLock lock(this);
   ASSERT(type_ != kListenSocket);
   ASSERT(pending_read_ == NULL);
   OverlappedBuffer* buffer = OverlappedBuffer::AllocateReadBuffer(kBufferSize);
   if (SupportsOverlappedIO()) {
     ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
 
     BOOL ok = ReadFile(handle_,
                        buffer->GetBufferStart(),
                        buffer->GetBufferSize(),
                        NULL,
                        buffer->GetCleanOverlapped());
     if (ok || GetLastError() == ERROR_IO_PENDING) {
       // Completing asynchronously.
       pending_read_ = buffer;
       return true;
     }
     OverlappedBuffer::DisposeBuffer(buffer);
     HandleIssueError();
     return false;
   } else {
     // Completing asynchronously through thread.
     pending_read_ = buffer;
+    read_thread_starting_ = true;
     int result = Thread::Start(ReadFileThread,
                                reinterpret_cast<uword>(this));
     if (result != 0) {
       FATAL1("Failed to start read file thread %d", result);
     }
     return true;
   }
 }
 
 
 bool Handle::IssueRecvFrom() {
   return false;
 }
 
 
 bool Handle::IssueWrite() {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   ASSERT(type_ != kListenSocket);
   ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
   ASSERT(pending_write_ != NULL);
   ASSERT(pending_write_->operation() == OverlappedBuffer::kWrite);
 
   OverlappedBuffer* buffer = pending_write_;
   BOOL ok = WriteFile(handle_,
                       buffer->GetBufferStart(),
                       buffer->GetBufferSize(),
                       NULL,
@@ skipping 36 matching lines @@
   if (error == ERROR_BROKEN_PIPE) {
     HandleClosed(this);
   } else {
     HandleError(this);
   }
   SetLastError(error);
 }
 
 
 void FileHandle::EnsureInitialized(EventHandlerImplementation* event_handler) {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   event_handler_ = event_handler;
   if (SupportsOverlappedIO() && completion_port_ == INVALID_HANDLE_VALUE) {
     CreateCompletionPort(event_handler_->completion_port());
   }
 }
 
 
 bool FileHandle::IsClosed() {
   return IsClosing() && !HasPendingRead() && !HasPendingWrite();
 }
 
 
 void DirectoryWatchHandle::EnsureInitialized(
     EventHandlerImplementation* event_handler) {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   event_handler_ = event_handler;
   if (completion_port_ == INVALID_HANDLE_VALUE) {
     CreateCompletionPort(event_handler_->completion_port());
   }
 }
 
 
 bool DirectoryWatchHandle::IsClosed() {
   return IsClosing() && pending_read_ == NULL;
 }
 
 
 bool DirectoryWatchHandle::IssueRead() {
-  ScopedLock lock(this);
   // It may have been started before, as we start the directory-handler when
   // we create it.
   if (pending_read_ != NULL || data_ready_ != NULL) return true;
   OverlappedBuffer* buffer = OverlappedBuffer::AllocateReadBuffer(kBufferSize);
   ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
   BOOL ok = ReadDirectoryChangesW(handle_,
                                   buffer->GetBufferStart(),
                                   buffer->GetBufferSize(),
                                   recursive_,
                                   events_,
                                   NULL,
                                   buffer->GetCleanOverlapped(),
                                   NULL);
   if (ok || GetLastError() == ERROR_IO_PENDING) {
     // Completing asynchronously.
     pending_read_ = buffer;
     return true;
   }
   OverlappedBuffer::DisposeBuffer(buffer);
   return false;
 }
 
 
 void DirectoryWatchHandle::Stop() {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   // Stop the outstanding read, so we can close the handle.
 
   if (pending_read_ != NULL) {
     CancelIoEx(handle(), pending_read_->GetCleanOverlapped());
     // Don't dispose of the buffer, as it will still complete (with length 0).
   }
 
   DoClose();
 }
 
@@ skipping 23 matching lines @@
                         NULL,
                         NULL);
   if (status == SOCKET_ERROR) {
     return false;
   }
   return true;
 }
 
 
 bool ListenSocket::IssueAccept() {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
 
   // For AcceptEx there needs to be buffer storage for address
   // information for two addresses (local and remote address). The
   // AcceptEx documentation says: "This value must be at least 16
   // bytes more than the maximum address length for the transport
   // protocol in use."
   static const int kAcceptExAddressAdditionalBytes = 16;
   static const int kAcceptExAddressStorageSize =
       sizeof(SOCKADDR_STORAGE) + kAcceptExAddressAdditionalBytes;
   OverlappedBuffer* buffer =
@@ skipping 19 matching lines @@
   }
 
   pending_accept_count_++;
 
   return true;
 }
 
 
 void ListenSocket::AcceptComplete(OverlappedBuffer* buffer,
                                   HANDLE completion_port) {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   if (!IsClosing()) {
     // Update the accepted socket to support the full range of API calls.
     SOCKET s = socket();
     int rc = setsockopt(buffer->client(),
                         SOL_SOCKET,
                         SO_UPDATE_ACCEPT_CONTEXT,
                         reinterpret_cast<char*>(&s), sizeof(s));
     if (rc == NO_ERROR) {
       // Insert the accepted socket into the list.
       ClientSocket* client_socket = new ClientSocket(buffer->client());
@@ skipping 40 matching lines @@
       client->Close();
       DeleteIfClosed(client);
     } else {
       break;
     }
   }
 }
 
 
 bool ListenSocket::CanAccept() {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   return accepted_head_ != NULL;
 }
 
 
 ClientSocket* ListenSocket::Accept() {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
 
   ClientSocket *result = NULL;
 
   if (accepted_head_ != NULL) {
     result = accepted_head_;
     accepted_head_ = accepted_head_->next();
     if (accepted_head_ == NULL) accepted_tail_ = NULL;
     result->set_next(NULL);
     accepted_count_--;
   }
 
   if (pending_accept_count_ < 5) {
     // We have less than 5 pending accepts, queue another.
     if (!IsClosing()) {
       if (!IssueAccept()) {
         HandleError(this);
       }
     }
   }
 
   return result;
 }
 
 
 void ListenSocket::EnsureInitialized(
     EventHandlerImplementation* event_handler) {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   if (AcceptEx_ == NULL) {
     ASSERT(completion_port_ == INVALID_HANDLE_VALUE);
     ASSERT(event_handler_ == NULL);
     event_handler_ = event_handler;
     CreateCompletionPort(event_handler_->completion_port());
     LoadAcceptEx();
   }
 }
 
 
 bool ListenSocket::IsClosed() {
   return IsClosing() && !HasPendingAccept();
 }
 
 
 intptr_t Handle::Available() {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   if (data_ready_ == NULL) return 0;
   ASSERT(!data_ready_->IsEmpty());
   return data_ready_->GetRemainingLength();
 }
 
 
 intptr_t Handle::Read(void* buffer, intptr_t num_bytes) {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   if (data_ready_ == NULL) return 0;
   num_bytes = data_ready_->Read(
       buffer, Utils::Minimum<intptr_t>(num_bytes, INT_MAX));
   if (data_ready_->IsEmpty()) {
     OverlappedBuffer::DisposeBuffer(data_ready_);
     data_ready_ = NULL;
     if (!IsClosing() && !IsClosedRead()) IssueRead();
   }
   return num_bytes;
 }
 
 
 intptr_t Handle::RecvFrom(
     void* buffer, intptr_t num_bytes, struct sockaddr* sa, socklen_t sa_len) {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   if (data_ready_ == NULL) return 0;
   num_bytes = data_ready_->Read(
       buffer, Utils::Minimum<intptr_t>(num_bytes, INT_MAX));
   if (data_ready_->from()->sa_family == AF_INET) {
     ASSERT(sa_len >= sizeof(struct sockaddr_in));
     memmove(sa, data_ready_->from(), sizeof(struct sockaddr_in));
   } else {
     ASSERT(data_ready_->from()->sa_family == AF_INET6);
     ASSERT(sa_len >= sizeof(struct sockaddr_in6));
     memmove(sa, data_ready_->from(), sizeof(struct sockaddr_in6));
   }
   // Always dispose of the buffer, as UDP messages must be read in their
   // entirety to match how recvfrom works in a socket.
   OverlappedBuffer::DisposeBuffer(data_ready_);
   data_ready_ = NULL;
   if (!IsClosing() && !IsClosedRead()) IssueRecvFrom();
   return num_bytes;
 }
 
 
 intptr_t Handle::Write(const void* buffer, intptr_t num_bytes) {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   if (pending_write_ != NULL) return 0;
   if (num_bytes > kBufferSize) num_bytes = kBufferSize;
   ASSERT(SupportsOverlappedIO());
   if (completion_port_ == INVALID_HANDLE_VALUE) return 0;
   int truncated_bytes = Utils::Minimum<intptr_t>(num_bytes, INT_MAX);
   pending_write_ = OverlappedBuffer::AllocateWriteBuffer(truncated_bytes);
   pending_write_->Write(buffer, truncated_bytes);
   if (!IssueWrite()) return -1;
   return truncated_bytes;
 }
 
 
 intptr_t Handle::SendTo(const void* buffer,
                         intptr_t num_bytes,
                         struct sockaddr* sa,
                         socklen_t sa_len) {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   if (pending_write_ != NULL) return 0;
   if (num_bytes > kBufferSize) num_bytes = kBufferSize;
   ASSERT(SupportsOverlappedIO());
   if (completion_port_ == INVALID_HANDLE_VALUE) return 0;
   pending_write_ = OverlappedBuffer::AllocateSendToBuffer(num_bytes);
   pending_write_->Write(buffer, num_bytes);
   if (!IssueSendTo(sa, sa_len)) return -1;
   return num_bytes;
 }
 
 
 static void WriteFileThread(uword args) {
   StdHandle* handle = reinterpret_cast<StdHandle*>(args);
   handle->RunWriteLoop();
 }
 
 
 void StdHandle::RunWriteLoop() {
-  write_monitor_->Enter();
+  MonitorLocker ml(monitor_);
   write_thread_running_ = true;
+  thread_id_ = Thread::GetCurrentThreadId();
   // Notify we have started.
-  write_monitor_->Notify();
+  ml.Notify();
 
   while (write_thread_running_) {
-    write_monitor_->Wait(Monitor::kNoTimeout);
+    ml.Wait(Monitor::kNoTimeout);
     if (pending_write_ != NULL) {
       // We woke up and had a pending write. Execute it.
       WriteSyncCompleteAsync();
     }
   }
 
   write_thread_exists_ = false;
-  write_monitor_->Notify();
-  write_monitor_->Exit();
+  ml.Notify();
 }
 
 
 void StdHandle::WriteSyncCompleteAsync() {
   ASSERT(pending_write_ != NULL);
 
   DWORD bytes_written = -1;
   BOOL ok = WriteFile(handle_,
                       pending_write_->GetBufferStart(),
                       pending_write_->GetBufferSize(),
                       &bytes_written,
                       NULL);
   if (!ok) {
     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");
   }
 }
 
 intptr_t StdHandle::Write(const void* buffer, intptr_t num_bytes) {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   if (pending_write_ != NULL) return 0;
   if (num_bytes > kBufferSize) num_bytes = kBufferSize;
   // In the case of stdout and stderr, OverlappedIO is not supported.
   // Here we'll instead use a thread, 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.
-  MonitorLocker locker(write_monitor_);
   if (thread_wrote_ > 0) {
     if (num_bytes > thread_wrote_) num_bytes = thread_wrote_;
     thread_wrote_ -= num_bytes;
     return num_bytes;
   }
   if (!write_thread_exists_) {
     write_thread_exists_ = true;
-    int result = Thread::Start(WriteFileThread,
-                                     reinterpret_cast<uword>(this));
+    int result = Thread::Start(
+        WriteFileThread, reinterpret_cast<uword>(this));
     if (result != 0) {
       FATAL1("Failed to start write file thread %d", result);
     }
     while (!write_thread_running_) {
       // Wait until we the thread is running.
-      locker.Wait(Monitor::kNoTimeout);
+      ml.Wait(Monitor::kNoTimeout);
     }
   }
   // Only queue up to INT_MAX bytes.
   int truncated_bytes = Utils::Minimum<intptr_t>(num_bytes, INT_MAX);
   // Create buffer and notify thread about the new handle.
   pending_write_ = OverlappedBuffer::AllocateWriteBuffer(truncated_bytes);
   pending_write_->Write(buffer, truncated_bytes);
-  locker.Notify();
+  ml.Notify();
   return 0;
 }
 
 
 void StdHandle::DoClose() {
-  MonitorLocker locker(write_monitor_);
+  MonitorLocker ml(monitor_);
   if (write_thread_exists_) {
     write_thread_running_ = false;
-    locker.Notify();
+    ml.Notify();
     while (write_thread_exists_) {
-      locker.Wait(Monitor::kNoTimeout);
+      ml.Wait(Monitor::kNoTimeout);
     }
+    Thread::Join(thread_id_);
   }
   Handle::DoClose();
 }
 
 
 bool ClientSocket::LoadDisconnectEx() {
   // Load the DisconnectEx function into memory using WSAIoctl.
   GUID guid_disconnect_ex = WSAID_DISCONNECTEX;
   DWORD bytes;
   int status = WSAIoctl(socket(),
@@ skipping 25 matching lines @@
 
 void ClientSocket::DoClose() {
   // Always do a shutdown before initiating a disconnect.
   shutdown(socket(), SD_BOTH);
   IssueDisconnect();
   handle_ = INVALID_HANDLE_VALUE;
 }
 
 
 bool ClientSocket::IssueRead() {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
   ASSERT(pending_read_ == NULL);
 
   // TODO(sgjesse): Use a MTU value here. Only the loopback adapter can
   // handle 64k datagrams.
   OverlappedBuffer* buffer = OverlappedBuffer::AllocateReadBuffer(65536);
 
   DWORD flags;
   flags = 0;
   int rc = WSARecv(socket(),
                    buffer->GetWASBUF(),
                    1,
                    NULL,
                    &flags,
                    buffer->GetCleanOverlapped(),
                    NULL);
   if (rc == NO_ERROR || WSAGetLastError() == WSA_IO_PENDING) {
     pending_read_ = buffer;
     return true;
   }
   OverlappedBuffer::DisposeBuffer(buffer);
   pending_read_ = NULL;
   HandleIssueError();
   return false;
 }
 
 
 bool ClientSocket::IssueWrite() {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
   ASSERT(pending_write_ != NULL);
   ASSERT(pending_write_->operation() == OverlappedBuffer::kWrite);
 
   int rc = WSASend(socket(),
                    pending_write_->GetWASBUF(),
                    1,
                    NULL,
                    0,
                    pending_write_->GetCleanOverlapped(),
@@ skipping 43 matching lines @@
   }
   if (!IsClosedWrite() && ((Mask() & (1 << kOutEvent)) != 0)) {
     Dart_Port port = NextNotifyDartPort(1 << kOutEvent);
     DartUtils::PostInt32(port, 1 << kOutEvent);
   }
 }
 
 
 void ClientSocket::EnsureInitialized(
     EventHandlerImplementation* event_handler) {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   if (completion_port_ == INVALID_HANDLE_VALUE) {
     ASSERT(event_handler_ == NULL);
     event_handler_ = event_handler;
     CreateCompletionPort(event_handler_->completion_port());
   }
 }
 
 
 bool ClientSocket::IsClosed() {
   return connected_ && closed_;
 }
 
 
 bool DatagramSocket::IssueSendTo(struct sockaddr* sa, socklen_t sa_len) {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
   ASSERT(pending_write_ != NULL);
   ASSERT(pending_write_->operation() == OverlappedBuffer::kSendTo);
 
   int rc = WSASendTo(socket(),
                      pending_write_->GetWASBUF(),
                      1,
                      NULL,
                      0,
                      sa,
                      sa_len,
                      pending_write_->GetCleanOverlapped(),
                      NULL);
   if (rc == NO_ERROR || WSAGetLastError() == WSA_IO_PENDING) {
     return true;
   }
   OverlappedBuffer::DisposeBuffer(pending_write_);
   pending_write_ = NULL;
   HandleIssueError();
   return false;
 }
 
 
 bool DatagramSocket::IssueRecvFrom() {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
   ASSERT(pending_read_ == NULL);
 
   OverlappedBuffer* buffer = OverlappedBuffer::AllocateRecvFromBuffer(1024);
 
   DWORD flags;
   flags = 0;
   int rc = WSARecvFrom(socket(),
                        buffer->GetWASBUF(),
                        1,
                        NULL,
                        &flags,
                        buffer->from(),
                        buffer->from_len_addr(),
                        buffer->GetCleanOverlapped(),
                        NULL);
   if (rc == NO_ERROR || WSAGetLastError() == WSA_IO_PENDING) {
     pending_read_ = buffer;
     return true;
   }
   OverlappedBuffer::DisposeBuffer(buffer);
   pending_read_ = NULL;
   HandleIssueError();
   return false;
 }
 
 
 void DatagramSocket::EnsureInitialized(
     EventHandlerImplementation* event_handler) {
-  ScopedLock lock(this);
+  MonitorLocker ml(monitor_);
   if (completion_port_ == INVALID_HANDLE_VALUE) {
     ASSERT(event_handler_ == NULL);
     event_handler_ = event_handler;
     CreateCompletionPort(event_handler_->completion_port());
   }
 }
 
 
 bool DatagramSocket::IsClosed() {
   return IsClosing() && !HasPendingRead() && !HasPendingWrite();
@@ skipping 19 matching lines @@
     shutdown_ = true;
   } else {
     Handle* handle = reinterpret_cast<Handle*>(msg->id);
     ASSERT(handle != NULL);
 
     if (handle->is_listen_socket()) {
       ListenSocket* listen_socket =
           reinterpret_cast<ListenSocket*>(handle);
       listen_socket->EnsureInitialized(this);
 
-      Handle::ScopedLock lock(listen_socket);
+      MonitorLocker ml(listen_socket->monitor_);
 
       if (IS_COMMAND(msg->data, kReturnTokenCommand)) {
         listen_socket->ReturnTokens(msg->dart_port, TOKEN_COUNT(msg->data));
       } else if (IS_COMMAND(msg->data, kSetEventMaskCommand)) {
         // `events` can only have kInEvent/kOutEvent flags set.
         intptr_t events = msg->data & EVENT_MASK;
         ASSERT(0 == (events & ~(1 << kInEvent | 1 << kOutEvent)));
         listen_socket->SetPortAndMask(msg->dart_port, events);
         TryDispatchingPendingAccepts(listen_socket);
       } else if (IS_COMMAND(msg->data, kCloseCommand)) {
         listen_socket->RemovePort(msg->dart_port);
 
         // We only close the socket file descriptor from the operating
         // system if there are no other dart socket objects which
         // are listening on the same (address, port) combination.
         ListeningSocketRegistry *registry =
             ListeningSocketRegistry::Instance();
         MutexLocker locker(registry->mutex());
         if (registry->CloseSafe(reinterpret_cast<intptr_t>(listen_socket))) {
           ASSERT(listen_socket->Mask() == 0);
           listen_socket->Close();
         }
 
         DartUtils::PostInt32(msg->dart_port, 1 << kDestroyedEvent);
       } else {
         UNREACHABLE();
       }
     } else {
       handle->EnsureInitialized(this);
-      Handle::ScopedLock lock(handle);
+      MonitorLocker ml(handle->monitor_);
 
       if (IS_COMMAND(msg->data, kReturnTokenCommand)) {
         handle->ReturnTokens(msg->dart_port, TOKEN_COUNT(msg->data));
       } else if (IS_COMMAND(msg->data, kSetEventMaskCommand)) {
         // `events` can only have kInEvent/kOutEvent flags set.
         intptr_t events = msg->data & EVENT_MASK;
         ASSERT(0 == (events & ~(1 << kInEvent | 1 << kOutEvent)));
 
         handle->SetPortAndMask(msg->dart_port, events);
 
@@ skipping 53 matching lines @@
     DeleteIfClosed(handle);
   }
 }
 
 
 void EventHandlerImplementation::HandleAccept(ListenSocket* listen_socket,
                                               OverlappedBuffer* buffer) {
   listen_socket->AcceptComplete(buffer, completion_port_);
 
   {
-    Handle::ScopedLock lock(listen_socket);
+    MonitorLocker ml(listen_socket->monitor_);
     TryDispatchingPendingAccepts(listen_socket);
   }
 
   DeleteIfClosed(listen_socket);
 }
 
 
 void EventHandlerImplementation::TryDispatchingPendingAccepts(
     ListenSocket *listen_socket) {
   if (!listen_socket->IsClosing() && listen_socket->CanAccept()) {
@@ skipping 143 matching lines @@
       HandleConnect(client_socket, bytes, buffer);
       break;
     }
     default:
       UNREACHABLE();
   }
 }
 
 
 EventHandlerImplementation::EventHandlerImplementation() {
+  startup_monitor_ = new Monitor();
+  handler_thread_id_ = Thread::kInvalidThreadId;
   completion_port_ =
       CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, NULL, 1);
   if (completion_port_ == NULL) {
     FATAL("Completion port creation failed");
   }
   shutdown_ = false;
 }
 
 
 EventHandlerImplementation::~EventHandlerImplementation() {
+  Thread::Join(handler_thread_id_);
+  delete startup_monitor_;
   CloseHandle(completion_port_);
 }
 
 
 int64_t EventHandlerImplementation::GetTimeout() {
   if (!timeout_queue_.HasTimeout()) {
     return kInfinityTimeout;
   }
   int64_t millis = timeout_queue_.CurrentTimeout() -
       TimerUtils::GetCurrentTimeMilliseconds();
@@ skipping 14 matching lines @@
     FATAL("PostQueuedCompletionStatus failed");
   }
 }
 
 
 void EventHandlerImplementation::EventHandlerEntry(uword args) {
   EventHandler* handler = reinterpret_cast<EventHandler*>(args);
   EventHandlerImplementation* handler_impl = &handler->delegate_;
   ASSERT(handler_impl != NULL);
 
+  {
+    MonitorLocker ml(handler_impl->startup_monitor_);
+    handler_impl->handler_thread_id_ = Thread::GetCurrentThreadId();
+    ml.Notify();
+  }
+
   while (!handler_impl->shutdown_) {
     DWORD bytes;
     ULONG_PTR key;
     OVERLAPPED* overlapped;
     int64_t millis = handler_impl->GetTimeout();
     ASSERT(millis == kInfinityTimeout || millis >= 0);
     if (millis > kMaxInt32) millis = kMaxInt32;
     ASSERT(sizeof(int32_t) == sizeof(DWORD));
     BOOL ok = GetQueuedCompletionStatus(handler_impl->completion_port(),
                                         &bytes,
@@ skipping 40 matching lines @@
 }
 
 
 void EventHandlerImplementation::Start(EventHandler* handler) {
   int result = Thread::Start(EventHandlerEntry,
                              reinterpret_cast<uword>(handler));
   if (result != 0) {
     FATAL1("Failed to start event handler thread %d", result);
   }
 
+  {
+    MonitorLocker ml(startup_monitor_);
+    while (handler_thread_id_ == Thread::kInvalidThreadId) {
+      ml.Wait();
+    }
+  }
+
   // Initialize Winsock32
   if (!Socket::Initialize()) {
     FATAL("Failed to initialized Windows sockets");
   }
 }
 
 
 void EventHandlerImplementation::Shutdown() {
   SendData(kShutdownId, 0, 0);
 }
 
 }  // namespace bin
 }  // namespace dart
 
 #endif  // defined(TARGET_OS_WINDOWS)