Reland: [dart:io] Adds waitForEventSync

The only fix needed for relanding is adding _ensureScheduleImmediate
to the list of vm entrypoints in //runtime/vm/compiler/aot/precompiler.cc

Original commit message:

Adds a top-level call waitForEventSync to dart:io that blocks the
thread an Isolate is running on until messages are available.
Before the thread blocks, the microtask queue is drained.
Before waitForEventSync returns, all messages are handled.

Lifting this up from a comment:

This is apropos of the request that nweiz@ sent to the mailing list a
couple weeks back. I'm not sure we should land this. We certainly
shouldn't land it without some annotations that will make the analyzer
complain a lot in most configurations, but I don't know what those
annotations are.

fixes #31102

Change-Id: Id96de46cc5f10e1847045cfafb7cfed6a38bce16
Reviewed-on: https://dart-review.googlesource.com/28920
Reviewed-by: Siva Annamalai <asiva@google.com>
Commit-Queue: Zach Anderson <zra@google.com>

runtime/bin/common.cc

@@ -0,0 +1,27 @@
+// Copyright (c) 2017, 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 "bin/builtin.h"
+#include "bin/dartutils.h"
+
+#include "include/dart_api.h"
+
+namespace dart {
+namespace bin {
+
+void FUNCTION_NAME(IOUtils_WaitForEventSync)(Dart_NativeArguments args) {
+  int64_t timeout_millis;
+  Dart_Handle result = Dart_GetNativeIntegerArgument(args, 0, &timeout_millis);
+  if (Dart_IsError(result)) {
+    Dart_PropagateError(result);
+  }
+  result = Dart_WaitForEventSync(timeout_millis);
+  if (Dart_IsError(result)) {
+    Dart_PropagateError(result);
+  }
+  Dart_SetReturnValue(args, result);
+}
+
+}  // namespace bin
+}  // namespace dart

runtime/bin/common_patch.dart

@@ -59,3 +59,13 @@ _setupHooks() {
   VMLibraryHooks.eventHandlerSendData = _EventHandler._sendData;
   VMLibraryHooks.timerMillisecondClock = _EventHandler._timerMillisecondClock;
 }
+
+// The implementation of waitForEventSync for embedders in which the VM
+// manages the event loop and microtask queue.
+// IOUtils_WaitForEventSync throws an exception if there is an error.
+void _standaloneWaitForEventSync(int timeoutMillis)
+    native "IOUtils_WaitForEventSync";
+
+void Function(int timeoutMillis) _getWaitForEventSyncClosure() {
+  return _standaloneWaitForEventSync;
+}

runtime/bin/dartutils.cc

@@ -572,6 +572,12 @@ Dart_Handle DartUtils::SetupIOLibrary(const char* namespc_path,
   Dart_Handle set_script_name =
       Dart_SetField(platform_type, script_name, dart_script);
   RETURN_IF_ERROR(set_script_name);
+
+  Dart_Handle wait_for_event =
+      Dart_Invoke(io_lib, NewString("_getWaitForEventSyncClosure"), 0, NULL);
+  RETURN_IF_ERROR(wait_for_event);
+  RETURN_IF_ERROR(Dart_SetField(io_lib, NewString("_waitForEventSyncImpl"),
+                                wait_for_event));
   return Dart_Null();
 }
 

runtime/bin/io_impl_sources.gni

@@ -5,6 +5,7 @@
 # This file contains some C++ sources for the dart:io library.  The other
 # implementation files are in builtin_impl_sources.gypi.
 io_impl_sources = [
+  "common.cc",
   "eventhandler.cc",
   "eventhandler.h",
   "eventhandler_android.cc",

runtime/bin/io_natives.cc

@@ -81,6 +81,7 @@ namespace bin {
   V(Filter_Processed, 3)                                                       \
   V(InternetAddress_Parse, 1)                                                  \
   V(IOService_NewServicePort, 0)                                               \
+  V(IOUtils_WaitForEventSync, 1)                                               \
   V(Namespace_Create, 2)                                                       \
   V(Namespace_GetDefault, 0)                                                   \
   V(Namespace_GetPointer, 1)                                                   \

runtime/bin/main.cc

@@ -676,6 +676,8 @@ static Dart_QualifiedFunctionName standalone_entry_points[] = {
     {"dart:io", "::", "_makeDatagram"},
     {"dart:io", "::", "_makeUint8ListView"},
     {"dart:io", "::", "_setupHooks"},
+    {"dart:io", "::", "_getWaitForEventSyncClosure"},
+    {"dart:io", "::", "_waitForEventSyncImpl"},
     {"dart:io", "_EmbedderConfig", "_mayExit"},
     {"dart:io", "_ExternalBuffer", "get:end"},
     {"dart:io", "_ExternalBuffer", "get:start"},

runtime/include/dart_api.h

@@ -1215,6 +1215,16 @@ DART_EXPORT Dart_Handle Dart_HandleMessage();
  */
 DART_EXPORT Dart_Handle Dart_HandleMessages();
 
+/**
+ * Drains the microtask queue, then blocks the calling thread until the current
+ * isolate recieves a message, then handles all messages.
+ *
+ * \param timeout_millis When non-zero, the call returns after the indicated
+          number of milliseconds even if no message was received.
+ * \return A valid handle if no error occurs, otherwise an error handle.
+ */
+DART_EXPORT Dart_Handle Dart_WaitForEventSync(int64_t timeout_millis);
+
 /**
  * Handles any pending messages for the vm service for the current
  * isolate.

runtime/lib/isolate_patch.dart

@@ -100,7 +100,8 @@ _ImmediateCallback _pendingImmediateCallback;
 /// The closure that should be used as scheduleImmediateClosure, when the VM
 /// is responsible for the event loop.
 void _isolateScheduleImmediate(void callback()) {
-  assert(_pendingImmediateCallback == null);
+  assert((_pendingImmediateCallback == null) ||
+      (_pendingImmediateCallback == callback));
   _pendingImmediateCallback = callback;
 }
 

runtime/lib/schedule_microtask_patch.dart

@@ -24,3 +24,7 @@ class _ScheduleImmediate {
 void _setScheduleImmediateClosure(_ScheduleImmediateClosure closure) {
   _ScheduleImmediate._closure = closure;
 }
+
+void _ensureScheduleImmediate() {
+  _AsyncRun._scheduleImmediate(_startMicrotaskLoop);
+}

runtime/vm/compiler/aot/precompiler.cc

@@ -626,6 +626,7 @@ void Precompiler::PrecompileConstructors() {
 
 static Dart_QualifiedFunctionName vm_entry_points[] = {
     // Functions
+    {"dart:async", "::", "_ensureScheduleImmediate"},
     {"dart:core", "::", "_completeDeferredLoads"},
     {"dart:core", "::", "identityHashCode"},
     {"dart:core", "AbstractClassInstantiationError",

runtime/vm/dart_api_impl.cc

@@ -1672,6 +1672,36 @@ DART_EXPORT Dart_Handle Dart_HandleMessages() {
   return Api::Success();
 }
 
+DART_EXPORT Dart_Handle Dart_WaitForEventSync(int64_t timeout_millis) {
+  Thread* T = Thread::Current();
+  Isolate* I = T->isolate();
+  CHECK_API_SCOPE(T);
+  CHECK_CALLBACK_STATE(T);
+  API_TIMELINE_BEGIN_END_BASIC;
+  TransitionNativeToVM transition(T);
+  if (I->message_notify_callback() != NULL) {
+    return Api::NewError("waitForEventSync is not supported by this embedder");
+  }
+  // NB: If waitForEventSync() is moved from dart:io to dart:async, then
+  // this call can be made from Dart code instead of from the runtime.
+  Object& result =
+      Object::Handle(Z, DartLibraryCalls::EnsureScheduleImmediate());
+  if (result.IsError()) {
+    return Api::NewHandle(T, result.raw());
+  }
+  result = DartLibraryCalls::DrainMicrotaskQueue();
+  if (result.IsError()) {
+    return Api::NewHandle(T, result.raw());
+  }
+  if (I->message_handler()->PauseAndHandleAllMessages(timeout_millis) !=
+      MessageHandler::kOK) {
+    Dart_Handle error = Api::NewHandle(T, T->sticky_error());
+    T->clear_sticky_error();
+    return error;
+  }
+  return Api::Success();
+}
+
 DART_EXPORT bool Dart_HandleServiceMessages() {
 #if defined(PRODUCT)
   return true;

runtime/vm/dart_entry.cc

@@ -665,6 +665,20 @@ RawObject* DartLibraryCalls::DrainMicrotaskQueue() {
   return result.raw();
 }
 
+RawObject* DartLibraryCalls::EnsureScheduleImmediate() {
+  Zone* zone = Thread::Current()->zone();
+  const Library& async_lib = Library::Handle(zone, Library::AsyncLibrary());
+  ASSERT(!async_lib.IsNull());
+  const Function& function =
+      Function::Handle(zone, async_lib.LookupFunctionAllowPrivate(
+                                 Symbols::_ensureScheduleImmediate()));
+  ASSERT(!function.IsNull());
+  const Object& result = Object::Handle(
+      zone, DartEntry::InvokeFunction(function, Object::empty_array()));
+  ASSERT(result.IsNull() || result.IsError());
+  return result.raw();
+}
+
 RawObject* DartLibraryCalls::MapSetAt(const Instance& map,
                                       const Instance& key,
                                       const Instance& value) {

runtime/vm/dart_entry.h

@@ -282,6 +282,11 @@ class DartLibraryCalls : public AllStatic {
   // Returns null on success, a RawError on failure.
   static RawObject* DrainMicrotaskQueue();
 
+  // Ensures that the isolate's _pendingImmediateCallback is set to
+  // _startMicrotaskLoop from dart:async.
+  // Returns null on success, a RawError on failure.
+  static RawObject* EnsureScheduleImmediate();
+
   // map[key] = value;
   //
   // Returns null on success, a RawError on failure.

runtime/vm/message_handler.cc

@@ -54,6 +54,7 @@ MessageHandler::MessageHandler()
     : queue_(new MessageQueue()),
       oob_queue_(new MessageQueue()),
       oob_message_handling_allowed_(true),
+      paused_for_messages_(false),
       live_ports_(0),
       paused_(0),
 #if !defined(PRODUCT)
@@ -151,6 +152,9 @@ void MessageHandler::PostMessage(Message* message, bool before_events) {
     } else {
       queue_->Enqueue(message, before_events);
     }
+    if (paused_for_messages_) {
+      ml.Notify();
+    }
     message = NULL;  // Do not access message.  May have been deleted.
 
     if ((pool_ != NULL) && (task_ == NULL)) {
@@ -276,6 +280,37 @@ MessageHandler::MessageStatus MessageHandler::HandleAllMessages() {
   return HandleMessages(&ml, true, true);
 }
 
+MessageHandler::MessageStatus MessageHandler::PauseAndHandleAllMessages(
+    int64_t timeout_millis) {
+  MonitorLocker ml(&monitor_);
+  ASSERT(task_ != NULL);
+  ASSERT(!delete_me_);
+#if defined(DEBUG)
+  CheckAccess();
+#endif
+  paused_for_messages_ = true;
+  while (queue_->IsEmpty() && oob_queue_->IsEmpty()) {
+    Monitor::WaitResult wr = ml.Wait(timeout_millis);
+    ASSERT(task_ != NULL);
+    ASSERT(!delete_me_);
+    if (wr == Monitor::kTimedOut) {
+      paused_for_messages_ = false;
+      return kOK;
+    }
+    if (queue_->IsEmpty()) {
+      // There are only OOB messages. Handle them and then continue waiting for
+      // normal messages unless there is an error.
+      MessageStatus status = HandleMessages(&ml, false, false);
+      if (status != kOK) {
+        paused_for_messages_ = false;
+        return status;
+      }
+    }
+  }
+  paused_for_messages_ = false;
+  return HandleMessages(&ml, true, true);
+}
+
 MessageHandler::MessageStatus MessageHandler::HandleOOBMessages() {
   if (!oob_message_handling_allowed_) {
     return kOK;

runtime/vm/message_handler.h

@@ -70,6 +70,10 @@ class MessageHandler {
   // Returns true on success.
   MessageStatus HandleOOBMessages();
 
+  // Blocks the thread on a condition variable until a message arrives, and then
+  // handles all messages.
+  MessageStatus PauseAndHandleAllMessages(int64_t timeout_millis);
+
   // Returns true if there are pending OOB messages for this message
   // handler.
   bool HasOOBMessages();
@@ -232,6 +236,7 @@ class MessageHandler {
   // This flag is not thread safe and can only reliably be accessed on a single
   // thread.
   bool oob_message_handling_allowed_;
+  bool paused_for_messages_;
   intptr_t live_ports_;  // The number of open ports, including control ports.
   intptr_t paused_;      // The number of pause messages received.
 #if !defined(PRODUCT)

runtime/vm/symbols.h

@@ -428,6 +428,7 @@ class ObjectPointerVisitor;
   V(ConstructorStacktracePrefix, "new ")                                       \
   V(_runExtension, "_runExtension")                                            \
   V(_runPendingImmediateCallback, "_runPendingImmediateCallback")              \
+  V(_ensureScheduleImmediate, "_ensureScheduleImmediate")                      \
   V(DartLibrary, "dart.library.")                                              \
   V(DartLibraryMirrors, "dart.library.mirrors")                                \
   V(_name, "_name")                                                            \

sdk/lib/io/annotations.dart

@@ -0,0 +1,49 @@
+// Copyright (c) 2017, 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.
+
+part of dart.io;
+
+/**
+ * The annotation `@Experimental('message')` marks a feature as experimental.
+ *
+ * The annotation `@experimental` is shorthand that uses the default message
+ * "This feature is experimental and not intended for general use".
+ *
+ * The intent of the `@Experimental` annotation is to inform users that a
+ * feature has one ore more of the following issues:
+ * - It is not yet stable,
+ * - It may not work as documented,
+ * - It should only be used by experts,
+ * - It is not available in all environments.
+ *
+ * The documentation for an experimental feature should explain which of these
+ * apply, and ideally give a reasonably safe example usage.
+ *
+ * A tool that processes Dart source code may give reports when experimental
+ * features are used similarly to how they give reports for features marked
+ * with the [Deprecated] annotation.
+ */
+class Experimental {
+  /**
+   * A brief message describing how or why the feature is experimental.
+   */
+  final String message;
+
+  const Experimental({String message}) : this.message = message;
+
+  @override
+  String toString() {
+    if (message == null) {
+      return "This feature is experimental, and not intended for general use.";
+    } else {
+      return "This feature is experimental: $message";
+    }
+  }
+}
+
+/**
+ * Marks a feature as experimental with the message, "Not intended for
+ * general use".
+ */
+const Experimental experimental = const Experimental();

sdk/lib/io/common.dart

@@ -117,3 +117,67 @@ _BufferAndStart _ensureFastAndSerializableByteData(
 class _IOCrypto {
   external static Uint8List getRandomBytes(int count);
 }
+
+// The implementation of waitForEventSync if there is one, and null otherwise.
+void Function(int timeoutMillis) _waitForEventSyncImpl;
+
+/**
+ * Synchronously blocks the calling isolate to wait for asynchronous events to
+ * complete.
+ *
+ * WARNING: EXPERIMENTAL. USE AT YOUR OWN RISK.
+ *
+ * If the [timeout] parameter is supplied, [waitForEventSync] will return after
+ * the specified timeout even if no events have occurred.
+ *
+ * This call does the following:
+ * - suspends the current execution stack,
+ * - runs the microtask queue until it is empty,
+ * - waits until the event queue is not empty,
+ * - handles events on the event queue, plus their associated microtasks,
+ *   until the event queue is empty,
+ * - resumes the original stack.
+ *
+ * This function will synchronously throw the first exception it encounters in
+ * running the microtasks and event handlers, leaving the remaining microtasks
+ * and events on their respective queues.
+ *
+ * Please note that this call is only safe to make in code that is running in
+ * the standalone command-line Dart VM. Further, because it suspends the current
+ * execution stack until all other events have been processed, even when running
+ * in the standalone command-line VM there exists a risk that the current
+ * execution stack will be starved.
+ *
+ * Example:
+ *
+ * ```
+ * main() {
+ *   bool condition = false;
+ *   ...
+ *   // Some asynchronous opertion that eventually sets 'condition' true.
+ *   ...
+ *   print("Waiting for 'condition'");
+ *   Duration timeout = const Duration(seconds: 10);
+ *   Timer.run(() {});  // Ensure that there is at least one event.
+ *   Stopwatch s = new Stopwatch()..start();
+ *   while (!condition) {
+ *     if (s.elapsed > timeout) {
+ *       print("timed out waiting for 'condition'!");
+ *       break;
+ *     }
+ *     print("still waiting...");
+ *     waitForEventSync(timeout: timeout);
+ *   }
+ *   s.stop();
+ * }
+ * ```
+ */
+@Experimental(message: "This feature is only available in the standalone VM")
+void waitForEventSync({Duration timeout}) {
+  if (_waitForEventSyncImpl == null) {
+    throw new UnsupportedError(
+        "waitForEventSync is not supported on this platform");
+  }
+  final int timeout_millis = timeout == null ? 0 : timeout.inMilliseconds;
+  _waitForEventSyncImpl(timeout_millis);
+}

sdk/lib/io/io.dart

@@ -205,6 +205,7 @@ import 'dart:typed_data';
 
 export 'dart:_http';
 
+part 'annotations.dart';
 part 'bytes_builder.dart';
 part 'common.dart';
 part 'data_transformer.dart';

sdk/lib/io/io_sources.gni

@@ -6,6 +6,7 @@ io_sdk_sources = [
   "io.dart",
 
   # The above file needs to be first if additional parts are added to the lib.
+  "annotations.dart",
   "bytes_builder.dart",
   "common.dart",
   "data_transformer.dart",

tests/standalone_2/io/wait_for_event_example_test.dart

@@ -0,0 +1,39 @@
+// Copyright (c) 2017, 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.
+
+import 'dart:async';
+import 'dart:io';
+import 'dart:isolate';
+
+import 'package:async_helper/async_helper.dart';
+import 'package:expect/expect.dart';
+
+// Tests that example code in the doc comment for waitForEventSync is okay.
+
+main() {
+  asyncStart();
+  bool condition = false;
+
+  new Timer(const Duration(seconds: 2), () {
+    asyncEnd();
+    condition = true;
+  });
+
+  Duration timeout = const Duration(milliseconds: 100);
+  Timer.run(() {}); // Ensure that there is at least one event.
+  Stopwatch s = new Stopwatch()..start();
+  while (!condition) {
+    if (s.elapsed > timeout) {
+      break;
+    }
+    waitForEventSync(timeout: timeout);
+  }
+  s.stop();
+
+  Expect.isFalse(condition);
+  while (!condition) {
+    waitForEventSync();
+  }
+  Expect.isTrue(condition);
+}

tests/standalone_2/io/wait_for_event_exception_test.dart

@@ -0,0 +1,22 @@
+// Copyright (c) 2017, 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.
+
+import 'dart:async';
+import 'dart:io';
+import 'dart:isolate';
+
+import 'package:async_helper/async_helper.dart';
+import 'package:expect/expect.dart';
+
+// Tests that exceptions thrown by message handlers running under
+// waitForEventSync() are propagated synchronously.
+
+main() {
+  asyncStart();
+  Timer.run(() {
+    asyncEnd();
+    throw "Exception";
+  });
+  Expect.throws(waitForEventSync, (e) => e is String);
+}

tests/standalone_2/io/wait_for_event_isolate_test.dart

@@ -0,0 +1,41 @@
+// Copyright (c) 2017, 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.
+
+import 'dart:async';
+import 'dart:io';
+import 'dart:isolate';
+
+import 'package:async_helper/async_helper.dart';
+import 'package:expect/expect.dart';
+
+// Tests that waitForEventSync() returns on an Isolate message.
+
+messageSender(SendPort s) {
+  new Timer(const Duration(seconds: 1), () {
+    s.send(true);
+  });
+}
+
+main() {
+  asyncStart();
+  bool flag1 = false;
+  bool flag2 = false;
+  ReceivePort r = new ReceivePort();
+  Isolate.spawn(messageSender, r.sendPort).then((Isolate i) {
+    r.listen((message) {
+      flag1 = true;
+    });
+    Expect.isFalse(flag1);
+    waitForEventSync();
+    Expect.isTrue(flag1);
+    r.close();
+    flag2 = true;
+  });
+  Expect.isFalse(flag1);
+  Expect.isFalse(flag2);
+  waitForEventSync();
+  Expect.isTrue(flag1);
+  Expect.isTrue(flag2);
+  asyncEnd();
+}

tests/standalone_2/io/wait_for_event_microtask_test.dart

@@ -0,0 +1,24 @@
+// Copyright (c) 2017, 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.
+
+import 'dart:async';
+import 'dart:io';
+import 'dart:isolate';
+
+import 'package:async_helper/async_helper.dart';
+import 'package:expect/expect.dart';
+
+// Tests that microtasks are run before waitForEventSync() blocks.
+
+main() {
+  asyncStart();
+  bool flag = false;
+  scheduleMicrotask(() {
+    flag = true;
+    asyncEnd();
+  });
+  Expect.isFalse(flag);
+  waitForEventSync(timeout: const Duration(milliseconds: 10));
+  Expect.isTrue(flag);
+}

tests/standalone_2/io/wait_for_event_nested_microtask_test.dart

@@ -0,0 +1,34 @@
+// Copyright (c) 2017, 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.
+
+import 'dart:async';
+import 'dart:io';
+import 'dart:isolate';
+
+import 'package:async_helper/async_helper.dart';
+import 'package:expect/expect.dart';
+
+// Tests that waitForEventSync() drains microtasks before blocking even when
+// called from a microtask.
+
+main() {
+  asyncStart();
+  bool flag1 = false;
+  bool flag2 = false;
+  scheduleMicrotask(() {
+    scheduleMicrotask(() {
+      flag1 = true;
+      asyncEnd();
+    });
+    Expect.isFalse(flag1);
+    waitForEventSync(timeout: const Duration(milliseconds: 10));
+    Expect.isTrue(flag1);
+    flag2 = true;
+  });
+  Expect.isFalse(flag1);
+  Expect.isFalse(flag2);
+  waitForEventSync(timeout: const Duration(milliseconds: 10));
+  Expect.isTrue(flag1);
+  Expect.isTrue(flag2);
+}

tests/standalone_2/io/wait_for_event_nested_timer_microtask_test.dart

@@ -0,0 +1,27 @@
+// Copyright (c) 2017, 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.
+
+import 'dart:async';
+import 'dart:io';
+import 'dart:isolate';
+
+import 'package:async_helper/async_helper.dart';
+import 'package:expect/expect.dart';
+
+// Tests that the microtasks for a message handler are run before
+// waitForEventSync() returns.
+
+main() {
+  asyncStart();
+  bool flag = false;
+  Timer.run(() {
+    scheduleMicrotask(() {
+      flag = true;
+      asyncEnd();
+    });
+  });
+  Expect.isFalse(flag);
+  waitForEventSync();
+  Expect.isTrue(flag);
+}

tests/standalone_2/io/wait_for_event_nested_timer_test.dart

@@ -0,0 +1,30 @@
+// Copyright (c) 2017, 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.
+
+import 'dart:async';
+import 'dart:io';
+import 'dart:isolate';
+
+import 'package:async_helper/async_helper.dart';
+import 'package:expect/expect.dart';
+
+// Tests that all messages are processed before waitForEventSync() returns.
+
+main() {
+  asyncStart();
+  bool flag = false;
+  Timer.run(() {
+    Timer.run(() {
+      Timer.run(() {
+        Timer.run(() {
+          flag = true;
+          asyncEnd();
+        });
+      });
+    });
+  });
+  Expect.isFalse(flag);
+  waitForEventSync();
+  Expect.isTrue(flag);
+}

tests/standalone_2/io/wait_for_event_nested_waits_test.dart

@@ -0,0 +1,69 @@
+// Copyright (c) 2017, 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.
+
+import 'dart:async';
+import 'dart:io';
+import 'dart:isolate';
+
+import 'package:async_helper/async_helper.dart';
+import 'package:expect/expect.dart';
+
+// Tests that waitForEventSync() works when called from a message handler.
+
+main() {
+  asyncStart();
+  bool flag1 = false;
+  bool flag2 = false;
+  bool flag3 = false;
+  bool flag4 = false;
+  Timer.run(() {
+    Timer.run(() {
+      Timer.run(() {
+        Timer.run(() {
+          flag1 = true;
+          asyncEnd();
+        });
+        Expect.isFalse(flag1);
+        Expect.isFalse(flag2);
+        Expect.isFalse(flag3);
+        Expect.isFalse(flag4);
+        waitForEventSync();
+        Expect.isTrue(flag1);
+        Expect.isFalse(flag2);
+        Expect.isFalse(flag3);
+        Expect.isFalse(flag4);
+        flag2 = true;
+      });
+      Expect.isFalse(flag1);
+      Expect.isFalse(flag2);
+      Expect.isFalse(flag3);
+      Expect.isFalse(flag4);
+      waitForEventSync();
+      Expect.isTrue(flag1);
+      Expect.isTrue(flag2);
+      Expect.isFalse(flag3);
+      Expect.isFalse(flag4);
+      flag3 = true;
+    });
+    Expect.isFalse(flag1);
+    Expect.isFalse(flag2);
+    Expect.isFalse(flag3);
+    Expect.isFalse(flag4);
+    waitForEventSync();
+    Expect.isTrue(flag1);
+    Expect.isTrue(flag2);
+    Expect.isTrue(flag3);
+    Expect.isFalse(flag4);
+    flag4 = true;
+  });
+  Expect.isFalse(flag1);
+  Expect.isFalse(flag2);
+  Expect.isFalse(flag3);
+  Expect.isFalse(flag4);
+  waitForEventSync();
+  Expect.isTrue(flag1);
+  Expect.isTrue(flag2);
+  Expect.isTrue(flag3);
+  Expect.isTrue(flag4);
+}

tests/standalone_2/io/wait_for_event_timeout_test.dart

@@ -0,0 +1,22 @@
+// Copyright (c) 2017, 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.
+
+import 'dart:async';
+import 'dart:io';
+import 'dart:isolate';
+
+import 'package:async_helper/async_helper.dart';
+import 'package:expect/expect.dart';
+
+// Tests that waitForEventSync() respects its 'timeout' parameter.
+
+main() {
+  asyncStart();
+  Duration t = const Duration(seconds: 1);
+  Stopwatch s = new Stopwatch()..start();
+  waitForEventSync(timeout: t);
+  s.stop();
+  Expect.isTrue(s.elapsed > t);
+  asyncEnd();
+}

tests/standalone_2/io/wait_for_event_timer_test.dart

@@ -0,0 +1,24 @@
+// Copyright (c) 2017, 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.
+
+import 'dart:async';
+import 'dart:io';
+import 'dart:isolate';
+
+import 'package:async_helper/async_helper.dart';
+import 'package:expect/expect.dart';
+
+// Tests that waitForEventSync() works in a simple case.
+
+main() {
+  asyncStart();
+  bool flag = false;
+  Timer.run(() {
+    flag = true;
+    asyncEnd();
+  });
+  Expect.isFalse(flag);
+  waitForEventSync();
+  Expect.isTrue(flag);
+}

tests/standalone_2/io/wait_for_event_zone_caught_error_test.dart

@@ -0,0 +1,29 @@
+// Copyright (c) 2017, 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.
+
+import 'dart:async';
+import 'dart:io';
+import 'dart:isolate';
+
+import 'package:async_helper/async_helper.dart';
+import 'package:expect/expect.dart';
+
+// Tests that waitForEventSync() doesn't cause an error to escape a Zone that
+// has an error handler.
+
+main() {
+  asyncStart();
+  bool flag = false;
+  runZoned(() {
+    Timer.run(() {
+      asyncEnd();
+      throw "Exception";
+    });
+  }, onError: (e) {
+    flag = true;
+  });
+  Expect.isFalse(flag);
+  waitForEventSync();
+  Expect.isTrue(flag);
+}

tests/standalone_2/io/wait_for_event_zone_test.dart

@@ -0,0 +1,27 @@
+// Copyright (c) 2017, 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.
+
+import 'dart:async';
+import 'dart:io';
+import 'dart:isolate';
+
+import 'package:async_helper/async_helper.dart';
+import 'package:expect/expect.dart';
+
+// Tests that waitForEventSync() works even when the message handler is run in
+// a different Zone.
+
+main() {
+  asyncStart();
+  bool flag = false;
+  runZoned(() {
+    Timer.run(() {
+      flag = true;
+      asyncEnd();
+    });
+  });
+  Expect.isFalse(flag);
+  waitForEventSync();
+  Expect.isTrue(flag);
+}

tests/standalone_2/io/wait_for_event_zone_uncaught_error_test.dart

@@ -0,0 +1,24 @@
+// Copyright (c) 2017, 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.
+
+import 'dart:async';
+import 'dart:io';
+import 'dart:isolate';
+
+import 'package:async_helper/async_helper.dart';
+import 'package:expect/expect.dart';
+
+// Tests that an exception thrown from a message handler in a Zone without an
+// error handler is propagated synchronously.
+
+main() {
+  asyncStart();
+  runZoned(() {
+    Timer.run(() {
+      asyncEnd();
+      throw "Exception";
+    });
+  });
+  Expect.throws(waitForEventSync, (e) => e is String);
+}