[vm/isolates] Introduce sendAndExit.

sendAndExit allows for fast data passing from worker isolate back to
parent.

```
                                              | linux x64  | spawnIsolate | sendAndExit |
                                              |us per iter | over sync    | over send   |
                                              +------------+--------------+-------------+
IsolateJson.Decode50KBx1(RunTime):               43,175.000   339.83%
IsolateJson.SendAndExit_Decode50KBx1(RunTime):   22,070.000   124.83%        -48.88%
IsolateJson.SyncDecode50KBx1(RunTime):            9,816.284

IsolateJson.Decode50KBx4(RunTime):               77,630.000   104.56%
IsolateJson.SendAndExit_Decode50KBx4(RunTime):   46,307.000   22.02%         -40.35%
IsolateJson.SyncDecode50KBx4(RunTime):           37,949.528

IsolateJson.Decode100KBx1(RunTime):              71,035.000   270.42%
IsolateJson.SendAndExit_Decode100KBx1(RunTime):  43,056.000   124.52%        -39.39%
IsolateJson.SyncDecode100KBx1(RunTime):          19,176.733

IsolateJson.Decode100KBx4(RunTime):             120,915.000   54.66%
IsolateJson.SendAndExit_Decode100KBx4(RunTime):  67,101.000  -14.17%         -44.51%
IsolateJson.SyncDecode100KBx4(RunTime):          78,179.731

IsolateJson.Decode250KBx1(RunTime):             173,574.000  202.52%
IsolateJson.SendAndExit_Decode250KBx1(RunTime): 103,334.000   80.10%         -40.47%
IsolateJson.SyncDecode250KBx1(RunTime):          57,375.314

IsolateJson.Decode250KBx4(RunTime):             292,118.000   20.30%
IsolateJson.SendAndExit_Decode250KBx4(RunTime): 168,444.000  -30.63%         -42.34%
IsolateJson.SyncDecode250KBx4(RunTime):         242,831.000

IsolateJson.Decode1MBx1(RunTime):               631,578.000  166.34%
IsolateJson.SendAndExit_Decode1MBx1(RunTime):   371,127.000   56.50%         -41.24%
IsolateJson.SyncDecode1MBx1(RunTime):           237,135.778

IsolateJson.Decode1MBx4(RunTime):             1,322,789.000   36.16%
IsolateJson.SendAndExit_Decode1MBx4(RunTime):   657,179.000  -32.35%         -50.32%
IsolateJson.SyncDecode1MBx4(RunTime):           971,473.333

```

Bug: dart-lang#37835
Bug: dart-lang#36097
Change-Id: I386641e1431ed9f2e34fac36f562607a666ee4a8
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/142823
Commit-Queue: Alexander Aprelev <[email protected]>
Reviewed-by: Martin Kustermann <[email protected]>
Reviewed-by: Ryan Macnak <[email protected]>

Author: aam

benchmarks/IsolateJson/dart/IsolateJson.dart

@@ -11,24 +11,32 @@ import 'dart:typed_data';
 import 'package:benchmark_harness/benchmark_harness.dart' show BenchmarkBase;
 import 'package:meta/meta.dart';
 
+import 'runtime/tests/vm/dart/export_sendAndExit_helper.dart' show sendAndExit;
+
 class JsonDecodingBenchmark {
   JsonDecodingBenchmark(this.name,
-      {@required this.sample, @required this.numTasks});
+      {@required this.sample,
+      @required this.numTasks,
+      @required this.useSendAndExit});
 
   Future<void> report() async {
     final stopwatch = Stopwatch()..start();
-    final decodedFutures = <Future>[];
-    for (int i = 0; i < numTasks; i++) {
-      decodedFutures.add(decodeJson(sample));
+    // Benchmark harness counts 10 iterations as one.
+    for (int i = 0; i < 10; i++) {
+      final decodedFutures = <Future>[];
+      for (int i = 0; i < numTasks; i++) {
+        decodedFutures.add(decodeJson(useSendAndExit, sample));
+      }
+      await Future.wait(decodedFutures);
     }
-    await Future.wait(decodedFutures);
 
     print("$name(RunTime): ${stopwatch.elapsedMicroseconds} us.");
   }
 
   final String name;
   final Uint8List sample;
   final int numTasks;
+  final bool useSendAndExit;
 }
 
 Uint8List createSampleJson(final size) {
@@ -41,27 +49,42 @@ Uint8List createSampleJson(final size) {
 }
 
 class JsonDecodeRequest {
+  final bool useSendAndExit;
   final SendPort sendPort;
   final Uint8List encodedJson;
-  const JsonDecodeRequest(this.sendPort, this.encodedJson);
+  const JsonDecodeRequest(this.useSendAndExit, this.sendPort, this.encodedJson);
 }
 
-Future<Map> decodeJson(Uint8List encodedJson) async {
+Future<Map> decodeJson(bool useSendAndExit, Uint8List encodedJson) async {
   final port = ReceivePort();
   final inbox = StreamIterator<dynamic>(port);
-  final workerExitedPort = ReceivePort();
-  await Isolate.spawn(
-      jsonDecodingIsolate, JsonDecodeRequest(port.sendPort, encodedJson),
-      onExit: workerExitedPort.sendPort);
+  final completer = Completer<bool>();
+  final workerExitedPort = RawReceivePort((v) {
+    completer.complete(true);
+  });
+  final workerErroredPort = RawReceivePort((v) {
+    stderr.writeln('worker errored out $v');
+    completer.completeError(true);
+  });
+  await Isolate.spawn(jsonDecodingIsolate,
+      JsonDecodeRequest(useSendAndExit, port.sendPort, encodedJson),
+      onError: workerErroredPort.sendPort, onExit: workerExitedPort.sendPort);
+  await completer.future;
+  workerExitedPort.close();
+  workerErroredPort.close();
   await inbox.moveNext();
   final decodedJson = inbox.current;
-  await workerExitedPort.first;
   port.close();
   return decodedJson;
 }
 
 Future<void> jsonDecodingIsolate(JsonDecodeRequest request) async {
-  request.sendPort.send(json.decode(utf8.decode(request.encodedJson)));
+  final result = json.decode(utf8.decode(request.encodedJson));
+  if (request.useSendAndExit) {
+    sendAndExit(request.sendPort, result);
+  } else {
+    request.sendPort.send(result);
+  }
 }
 
 class SyncJsonDecodingBenchmark extends BenchmarkBase {
@@ -118,6 +141,13 @@ Future<void> main() async {
     for (final iterations in <int>[1, 4]) {
       await JsonDecodingBenchmark(
               "IsolateJson.Decode${config.suffix}x$iterations",
+              useSendAndExit: false,
+              sample: config.sample,
+              numTasks: iterations)
+          .report();
+      await JsonDecodingBenchmark(
+              "IsolateJson.SendAndExit_Decode${config.suffix}x$iterations",
+              useSendAndExit: true,
               sample: config.sample,
               numTasks: iterations)
           .report();

benchmarks/IsolateJson/dart/runtime/tests/vm/dart/export_sendAndExit_helper.dart

@@ -0,0 +1 @@
+export 'dart:_internal' show sendAndExit;

pkg/vm/bin/kernel_service.dart

@@ -691,6 +691,7 @@ Future _processLoadRequest(request) async {
           prepend = ", ";
           if (sb.length > 256) break;
         }
+        sb.write("]");
         partToString = sb.toString();
       } else {
         partToString = part.toString();

runtime/lib/isolate.cc

@@ -15,6 +15,7 @@
 #include "vm/dart_api_message.h"
 #include "vm/dart_entry.h"
 #include "vm/exceptions.h"
+#include "vm/hash_table.h"
 #include "vm/lockers.h"
 #include "vm/longjump.h"
 #include "vm/message_handler.h"
@@ -108,6 +109,153 @@ DEFINE_NATIVE_ENTRY(SendPortImpl_sendInternal_, 0, 2) {
   return Object::null();
 }
 
+class RawObjectPtrSetTraits {
+ public:
+  static bool ReportStats() { return false; }
+  static const char* Name() { return "RawObjectPtrSetTraits"; }
+
+  static bool IsMatch(const RawObject* a, const RawObject* b) { return a == b; }
+
+  static uword Hash(const RawObject* obj) {
+    return reinterpret_cast<uword>(obj);
+  }
+};
+
+static RawObject* ValidateMessageObject(Zone* zone,
+                                        Isolate* isolate,
+                                        const Object& obj) {
+  TIMELINE_DURATION(Thread::Current(), Isolate, "ValidateMessageObject");
+
+  class SendMessageValidator : public ObjectPointerVisitor {
+   public:
+    SendMessageValidator(IsolateGroup* isolate_group,
+                         WeakTable* visited,
+                         MallocGrowableArray<RawObject*>* const working_set)
+        : ObjectPointerVisitor(isolate_group),
+          visited_(visited),
+          working_set_(working_set) {}
+
+   private:
+    void VisitPointers(RawObject** from, RawObject** to) {
+      for (RawObject** raw = from; raw <= to; raw++) {
+        if (!(*raw)->IsHeapObject() || (*raw)->IsCanonical()) {
+          continue;
+        }
+        if (visited_->GetValueExclusive(*raw) == 1) {
+          continue;
+        }
+        visited_->SetValueExclusive(*raw, 1);
+        working_set_->Add(*raw);
+      }
+    }
+
+    WeakTable* visited_;
+    MallocGrowableArray<RawObject*>* const working_set_;
+  };
+  if (!obj.raw()->IsHeapObject() || obj.raw()->IsCanonical()) {
+    return obj.raw();
+  }
+  ClassTable* class_table = isolate->class_table();
+
+  Class& klass = Class::Handle(zone);
+  Closure& closure = Closure::Handle(zone);
+
+  MallocGrowableArray<RawObject*> working_set;
+  std::unique_ptr<WeakTable> visited(new WeakTable());
+
+  NoSafepointScope no_safepoint;
+  SendMessageValidator visitor(isolate->group(), visited.get(), &working_set);
+
+  visited->SetValueExclusive(obj.raw(), 1);
+  working_set.Add(obj.raw());
+
+  while (!working_set.is_empty()) {
+    RawObject* raw = working_set.RemoveLast();
+
+    if (visited->GetValueExclusive(raw) > 0) {
+      continue;
+    }
+    visited->SetValueExclusive(raw, 1);
+
+    const intptr_t cid = raw->GetClassId();
+    switch (cid) {
+      // List below matches the one in raw_object_snapshot.cc
+#define MESSAGE_SNAPSHOT_ILLEGAL(type)                                         \
+  return Exceptions::CreateUnhandledException(                                 \
+      zone, Exceptions::kArgumentValue,                                        \
+      "Illegal argument in isolate message : (object is a " #type ")");        \
+  break;
+
+      MESSAGE_SNAPSHOT_ILLEGAL(DynamicLibrary);
+      MESSAGE_SNAPSHOT_ILLEGAL(MirrorReference);
+      MESSAGE_SNAPSHOT_ILLEGAL(Pointer);
+      MESSAGE_SNAPSHOT_ILLEGAL(ReceivePort);
+      MESSAGE_SNAPSHOT_ILLEGAL(RegExp);
+      MESSAGE_SNAPSHOT_ILLEGAL(StackTrace);
+      MESSAGE_SNAPSHOT_ILLEGAL(UserTag);
+
+      case kClosureCid: {
+        closure = Closure::RawCast(raw);
+        RawFunction* func = closure.function();
+        // We only allow closure of top level methods or static functions in a
+        // class to be sent in isolate messages.
+        if (!Function::IsImplicitStaticClosureFunction(func)) {
+          return Exceptions::CreateUnhandledException(
+              zone, Exceptions::kArgumentValue, "Closures are not allowed");
+        }
+        break;
+      }
+      default:
+        if (cid >= kNumPredefinedCids) {
+          klass = class_table->At(cid);
+          if (klass.num_native_fields() != 0) {
+            return Exceptions::CreateUnhandledException(
+                zone, Exceptions::kArgumentValue,
+                "Objects that extend NativeWrapper are not allowed");
+          }
+        }
+    }
+    raw->VisitPointers(&visitor);
+  }
+  isolate->set_forward_table_new(nullptr);
+  return obj.raw();
+}
+
+DEFINE_NATIVE_ENTRY(SendPortImpl_sendAndExitInternal_, 0, 2) {
+  GET_NON_NULL_NATIVE_ARGUMENT(SendPort, port, arguments->NativeArgAt(0));
+  if (!PortMap::IsReceiverInThisIsolateGroup(port.Id(), isolate->group())) {
+    const auto& error =
+        String::Handle(String::New("sendAndExit is only supported across "
+                                   "isolates spawned via spawnFunction."));
+    Exceptions::ThrowArgumentError(error);
+    UNREACHABLE();
+  }
+
+  GET_NON_NULL_NATIVE_ARGUMENT(Instance, obj, arguments->NativeArgAt(1));
+
+  Object& validated_result = Object::Handle(zone);
+  Object& msg_obj = Object::Handle(zone, obj.raw());
+  validated_result = ValidateMessageObject(zone, isolate, msg_obj);
+  if (validated_result.IsUnhandledException()) {
+    Exceptions::PropagateError(Error::Cast(validated_result));
+    UNREACHABLE();
+  }
+  PersistentHandle* handle =
+      isolate->group()->api_state()->AllocatePersistentHandle();
+  handle->set_raw(msg_obj);
+  isolate->bequeath(std::unique_ptr<Bequest>(new Bequest(handle, port.Id())));
+  // TODO(aam): Ensure there are no dart api calls after this point as we want
+  // to ensure that validated message won't get tampered with.
+  Isolate::KillIfExists(isolate, Isolate::LibMsgId::kKillMsg);
+  // Drain interrupts before running so any IMMEDIATE operations on the current
+  // isolate happen synchronously.
+  const Error& error = Error::Handle(thread->HandleInterrupts());
+  RELEASE_ASSERT(error.IsUnwindError());
+  Exceptions::PropagateError(error);
+  // We will never execute dart code again in this isolate.
+  return Object::null();
+}
+
 static void ThrowIsolateSpawnException(const String& message) {
   const Array& args = Array::Handle(Array::New(1));
   args.SetAt(0, message);

runtime/tests/vm/dart/sendandexit_test.dart

@@ -0,0 +1,85 @@
+// Copyright (c) 2018, 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.
+//
+// VMOptions=--enable-isolate-groups
+//
+// Validates functionality of sendAndExit.
+
+import 'dart:_internal' show sendAndExit;
+import 'dart:async';
+import 'dart:isolate';
+import 'dart:nativewrappers';
+
+import "package:expect/expect.dart";
+
+doNothingWorker(data) {}
+
+spawnWorker(worker, data) async {
+  Completer completer = Completer();
+  runZoned(() async {
+    final isolate = await Isolate.spawn(worker, [data]);
+    completer.complete(isolate);
+  }, onError: (e, st) => completer.complete(e));
+  return await completer.future;
+}
+
+verifyCantSendAnonymousClosure() async {
+  final result = await spawnWorker(doNothingWorker, () {});
+  Expect.equals(
+      "Invalid argument(s): Illegal argument in isolate message :"
+      " (object is a closure - Function '<anonymous closure>': static.)",
+      result.toString());
+}
+
+class NativeWrapperClass extends NativeFieldWrapperClass1 {}
+
+verifyCantSendNative() async {
+  final result = await spawnWorker(doNothingWorker, NativeWrapperClass());
+  Expect.isTrue(result.toString().startsWith("Invalid argument(s): "
+      "Illegal argument in isolate message : "
+      "(object extends NativeWrapper"));
+}
+
+verifyCantSendRegexp() async {
+  var receivePort = ReceivePort();
+  final result = await spawnWorker(doNothingWorker, receivePort);
+  Expect.equals(
+      "Invalid argument(s): Illegal argument in isolate message : "
+      "(object is a ReceivePort)",
+      result.toString());
+  receivePort.close();
+}
+
+class Message {
+  SendPort sendPort;
+  Function closure;
+
+  Message(this.sendPort, this.closure);
+}
+
+add(a, b) => a + b;
+
+worker(Message message) async {
+  final port = new ReceivePort();
+  final inbox = new StreamIterator<dynamic>(port);
+  message.sendPort.send(message.closure(2, 3));
+  port.close();
+}
+
+verifyCanSendStaticMethod() async {
+  final port = ReceivePort();
+  final inbox = StreamIterator<dynamic>(port);
+  final isolate = await Isolate.spawn(worker, Message(port.sendPort, add));
+
+  await inbox.moveNext();
+  Expect.equals(inbox.current, 5);
+  port.close();
+}
+
+main() async {
+  await verifyCantSendAnonymousClosure();
+  await verifyCantSendNative();
+  await verifyCantSendRegexp();
+  await verifyCanSendStaticMethod();
+}

runtime/vm/bootstrap_natives.h

@@ -58,6 +58,7 @@ namespace dart {
   V(SendPortImpl_get_id, 1)                                                    \
   V(SendPortImpl_get_hashcode, 1)                                              \
   V(SendPortImpl_sendInternal_, 2)                                             \
+  V(SendPortImpl_sendAndExitInternal_, 2)                                      \
   V(Smi_bitAndFromSmi, 2)                                                      \
   V(Smi_bitNegate, 1)                                                          \
   V(Smi_bitLength, 1)                                                          \

runtime/vm/exceptions.cc

@@ -1227,4 +1227,16 @@ RawObject* Exceptions::Create(ExceptionType type, const Array& arguments) {
                                           *constructor_name, arguments);
 }
 
+RawUnhandledException* Exceptions::CreateUnhandledException(Zone* zone,
+                                                            ExceptionType type,
+                                                            const char* msg) {
+  const String& error_str = String::Handle(zone, String::New(msg));
+  const Array& args = Array::Handle(zone, Array::New(1));
+  args.SetAt(0, error_str);
+
+  Object& result = Object::Handle(zone, Exceptions::Create(type, args));
+  const StackTrace& stacktrace = StackTrace::Handle(zone);
+  return UnhandledException::New(Instance::Cast(result), stacktrace);
+}
+
 }  // namespace dart