From 17654b70d720917cfd3c222dba85e21ac67a2353 Mon Sep 17 00:00:00 2001 From: Alexander Aprelev Date: Wed, 22 Apr 2020 17:34:09 +0000 Subject: [PATCH] [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: https://github.com/dart-lang/sdk/issues/37835 Bug: https://github.com/dart-lang/sdk/issues/36097 Change-Id: I386641e1431ed9f2e34fac36f562607a666ee4a8 Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/142823 Commit-Queue: Alexander Aprelev Reviewed-by: Martin Kustermann Reviewed-by: Ryan Macnak --- benchmarks/IsolateJson/dart/IsolateJson.dart | 56 +++++-- .../vm/dart/export_sendAndExit_helper.dart | 1 + pkg/vm/bin/kernel_service.dart | 1 + runtime/lib/isolate.cc | 148 ++++++++++++++++++ runtime/tests/vm/dart/sendandexit_test.dart | 85 ++++++++++ runtime/vm/bootstrap_natives.h | 1 + runtime/vm/exceptions.cc | 12 ++ runtime/vm/exceptions.h | 7 + runtime/vm/heap/heap.cc | 2 + runtime/vm/heap/heap.h | 19 +-- runtime/vm/heap/heap_test.cc | 133 ++++++++++++++++ runtime/vm/isolate.cc | 19 +++ runtime/vm/isolate.h | 24 +++ runtime/vm/message.cc | 29 +++- runtime/vm/message.h | 37 ++++- runtime/vm/port.cc | 8 + runtime/vm/port.h | 3 + sdk/lib/_internal/vm/lib/internal_patch.dart | 4 + .../lib/_internal/vm/lib/internal_patch.dart | 4 + 19 files changed, 562 insertions(+), 31 deletions(-) create mode 100644 benchmarks/IsolateJson/dart/runtime/tests/vm/dart/export_sendAndExit_helper.dart create mode 100644 runtime/tests/vm/dart/sendandexit_test.dart diff --git a/benchmarks/IsolateJson/dart/IsolateJson.dart b/benchmarks/IsolateJson/dart/IsolateJson.dart index bdc79889d86..1fa491950bd 100644 --- a/benchmarks/IsolateJson/dart/IsolateJson.dart +++ b/benchmarks/IsolateJson/dart/IsolateJson.dart @@ -11,17 +11,24 @@ 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 report() async { final stopwatch = Stopwatch()..start(); - final decodedFutures = []; - 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 = []; + 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."); } @@ -29,6 +36,7 @@ class JsonDecodingBenchmark { 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 decodeJson(Uint8List encodedJson) async { +Future decodeJson(bool useSendAndExit, Uint8List encodedJson) async { final port = ReceivePort(); final inbox = StreamIterator(port); - final workerExitedPort = ReceivePort(); - await Isolate.spawn( - jsonDecodingIsolate, JsonDecodeRequest(port.sendPort, encodedJson), - onExit: workerExitedPort.sendPort); + final completer = Completer(); + 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 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 main() async { for (final iterations in [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(); diff --git a/benchmarks/IsolateJson/dart/runtime/tests/vm/dart/export_sendAndExit_helper.dart b/benchmarks/IsolateJson/dart/runtime/tests/vm/dart/export_sendAndExit_helper.dart new file mode 100644 index 00000000000..75628d69d78 --- /dev/null +++ b/benchmarks/IsolateJson/dart/runtime/tests/vm/dart/export_sendAndExit_helper.dart @@ -0,0 +1 @@ +export 'dart:_internal' show sendAndExit; diff --git a/pkg/vm/bin/kernel_service.dart b/pkg/vm/bin/kernel_service.dart index 050407d0516..99638ada233 100644 --- a/pkg/vm/bin/kernel_service.dart +++ b/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(); diff --git a/runtime/lib/isolate.cc b/runtime/lib/isolate.cc index 1855954fb36..5ccc9d6a1f2 100644 --- a/runtime/lib/isolate.cc +++ b/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(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* 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* 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 working_set; + std::unique_ptr 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(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); diff --git a/runtime/tests/vm/dart/sendandexit_test.dart b/runtime/tests/vm/dart/sendandexit_test.dart new file mode 100644 index 00000000000..cbc617e3655 --- /dev/null +++ b/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 '': 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(port); + message.sendPort.send(message.closure(2, 3)); + port.close(); +} + +verifyCanSendStaticMethod() async { + final port = ReceivePort(); + final inbox = StreamIterator(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(); +} diff --git a/runtime/vm/bootstrap_natives.h b/runtime/vm/bootstrap_natives.h index 8f3b76243cd..5a91d3fdfdb 100644 --- a/runtime/vm/bootstrap_natives.h +++ b/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) \ diff --git a/runtime/vm/exceptions.cc b/runtime/vm/exceptions.cc index 76d009742a7..47502476aa0 100644 --- a/runtime/vm/exceptions.cc +++ b/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 diff --git a/runtime/vm/exceptions.h b/runtime/vm/exceptions.h index 1ec8e121b91..82807eae968 100644 --- a/runtime/vm/exceptions.h +++ b/runtime/vm/exceptions.h @@ -23,6 +23,7 @@ class RawInstance; class RawObject; class RawScript; class RawStackTrace; +class RawUnhandledException; class ReadStream; class WriteStream; class String; @@ -92,6 +93,12 @@ class Exceptions : AllStatic { // otherwise returns a RawError. static RawObject* Create(ExceptionType type, const Array& arguments); + // Returns RawUnhandledException that wraps exception of type [type] with + // [msg] as a single argument. + static RawUnhandledException* CreateUnhandledException(Zone* zone, + ExceptionType type, + const char* msg); + DART_NORETURN static void JumpToFrame(Thread* thread, uword program_counter, uword stack_pointer, diff --git a/runtime/vm/heap/heap.cc b/runtime/vm/heap/heap.cc index 1ff35e188aa..81019b08808 100644 --- a/runtime/vm/heap/heap.cc +++ b/runtime/vm/heap/heap.cc @@ -871,6 +871,8 @@ const char* Heap::GCReasonToString(GCReason gc_reason) { return "low memory"; case kDebugging: return "debugging"; + case kSendAndExit: + return "send_and_exit"; default: UNREACHABLE(); return ""; diff --git a/runtime/vm/heap/heap.h b/runtime/vm/heap/heap.h index 09f7584459c..f3ae81b408c 100644 --- a/runtime/vm/heap/heap.h +++ b/runtime/vm/heap/heap.h @@ -55,15 +55,16 @@ class Heap { }; enum GCReason { - kNewSpace, // New space is full. - kPromotion, // Old space limit crossed after a scavenge. - kOldSpace, // Old space limit crossed. - kFinalize, // Concurrent marking finished. - kFull, // Heap::CollectAllGarbage - kExternal, // Dart_NewWeakPersistentHandle - kIdle, // Dart_NotifyIdle - kLowMemory, // Dart_NotifyLowMemory - kDebugging, // service request, etc. + kNewSpace, // New space is full. + kPromotion, // Old space limit crossed after a scavenge. + kOldSpace, // Old space limit crossed. + kFinalize, // Concurrent marking finished. + kFull, // Heap::CollectAllGarbage + kExternal, // Dart_NewWeakPersistentHandle + kIdle, // Dart_NotifyIdle + kLowMemory, // Dart_NotifyLowMemory + kDebugging, // service request, etc. + kSendAndExit, // SendPort.sendAndExit }; // Pattern for unused new space and swept old space. diff --git a/runtime/vm/heap/heap_test.cc b/runtime/vm/heap/heap_test.cc index 948d6f3f057..4bedc777bb4 100644 --- a/runtime/vm/heap/heap_test.cc +++ b/runtime/vm/heap/heap_test.cc @@ -2,14 +2,22 @@ // 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 +#include +#include +#include + #include "platform/globals.h" #include "platform/assert.h" +#include "vm/class_finalizer.h" #include "vm/dart_api_impl.h" #include "vm/globals.h" #include "vm/heap/become.h" #include "vm/heap/heap.h" +#include "vm/message_handler.h" #include "vm/object_graph.h" +#include "vm/port.h" #include "vm/symbols.h" #include "vm/unit_test.h" @@ -607,6 +615,131 @@ class HeapTestHelper { } }; +class MergeIsolatesHeapsHandler : public MessageHandler { + public: + explicit MergeIsolatesHeapsHandler(Isolate* owner) + : msg_(Utils::CreateCStringUniquePtr(nullptr)), owner_(owner) {} + + const char* name() const { return "merge-isolates-heaps-handler"; } + + ~MergeIsolatesHeapsHandler() { PortMap::ClosePorts(this); } + + MessageStatus HandleMessage(std::unique_ptr message) { + // Parse the message. + Object& response_obj = Object::Handle(); + if (message->IsRaw()) { + response_obj = message->raw_obj(); + } else if (message->IsBequest()) { + Bequest* bequest = message->bequest(); + PersistentHandle* handle = bequest->handle(); + // Object in the receiving isolate's heap. + EXPECT(isolate()->heap()->Contains(RawObject::ToAddr(handle->raw()))); + response_obj = handle->raw(); + isolate()->group()->api_state()->FreePersistentHandle(handle); + } else { + Thread* thread = Thread::Current(); + MessageSnapshotReader reader(message.get(), thread); + response_obj = reader.ReadObject(); + } + if (response_obj.IsString()) { + String& response = String::Handle(); + response ^= response_obj.raw(); + msg_.reset(strdup(response.ToCString())); + } else { + ASSERT(response_obj.IsArray()); + Array& response_array = Array::Handle(); + response_array ^= response_obj.raw(); + ASSERT(response_array.Length() == 1); + ExternalTypedData& response = ExternalTypedData::Handle(); + response ^= response_array.At(0); + msg_.reset(strdup(reinterpret_cast(response.DataAddr(0)))); + } + + return kOK; + } + + const char* msg() const { return msg_.get(); } + + virtual Isolate* isolate() const { return owner_; } + + private: + Utils::CStringUniquePtr msg_; + Isolate* owner_; +}; + +VM_UNIT_TEST_CASE(CleanupBequestNeverReceived) { + const char* TEST_MESSAGE = "hello, world"; + Dart_Isolate parent = TestCase::CreateTestIsolate("parent"); + EXPECT_EQ(parent, Dart_CurrentIsolate()); + { + MergeIsolatesHeapsHandler handler(Isolate::Current()); + Dart_Port port_id = PortMap::CreatePort(&handler); + EXPECT_EQ(PortMap::GetIsolate(port_id), Isolate::Current()); + Dart_ExitIsolate(); + + Dart_Isolate worker = TestCase::CreateTestIsolateInGroup("worker", parent); + EXPECT_EQ(worker, Dart_CurrentIsolate()); + { + Thread* thread = Thread::Current(); + TransitionNativeToVM transition(thread); + StackZone zone(thread); + HANDLESCOPE(thread); + + String& string = String::Handle(String::New(TEST_MESSAGE)); + PersistentHandle* handle = + Isolate::Current()->group()->api_state()->AllocatePersistentHandle(); + handle->set_raw(string.raw()); + + reinterpret_cast(worker)->bequeath( + std::unique_ptr(new Bequest(handle, port_id))); + } + } + Dart_ShutdownIsolate(); + Dart_EnterIsolate(parent); + Dart_ShutdownIsolate(); +} + +VM_UNIT_TEST_CASE(ReceivesSendAndExitMessage) { + const char* TEST_MESSAGE = "hello, world"; + Dart_Isolate parent = TestCase::CreateTestIsolate("parent"); + EXPECT_EQ(parent, Dart_CurrentIsolate()); + MergeIsolatesHeapsHandler handler(Isolate::Current()); + Dart_Port port_id = PortMap::CreatePort(&handler); + EXPECT_EQ(PortMap::GetIsolate(port_id), Isolate::Current()); + Dart_ExitIsolate(); + + Dart_Isolate worker = TestCase::CreateTestIsolateInGroup("worker", parent); + EXPECT_EQ(worker, Dart_CurrentIsolate()); + { + Thread* thread = Thread::Current(); + TransitionNativeToVM transition(thread); + StackZone zone(thread); + HANDLESCOPE(thread); + + String& string = String::Handle(String::New(TEST_MESSAGE)); + + PersistentHandle* handle = + Isolate::Current()->group()->api_state()->AllocatePersistentHandle(); + handle->set_raw(string.raw()); + + reinterpret_cast(worker)->bequeath( + std::unique_ptr(new Bequest(handle, port_id))); + } + + Dart_ShutdownIsolate(); + Dart_EnterIsolate(parent); + { + Thread* thread = Thread::Current(); + TransitionNativeToVM transition(thread); + StackZone zone(thread); + HANDLESCOPE(thread); + + EXPECT_EQ(MessageHandler::kOK, handler.HandleNextMessage()); + } + EXPECT_STREQ(handler.msg(), TEST_MESSAGE); + Dart_ShutdownIsolate(); +} + ISOLATE_UNIT_TEST_CASE(ExternalAllocationStats) { Isolate* isolate = thread->isolate(); Heap* heap = thread->heap(); diff --git a/runtime/vm/isolate.cc b/runtime/vm/isolate.cc index 6da9a6af96d..c150aef3725 100644 --- a/runtime/vm/isolate.cc +++ b/runtime/vm/isolate.cc @@ -648,6 +648,15 @@ NoReloadScope::~NoReloadScope() { #endif // !defined(PRODUCT) && !defined(DART_PRECOMPILED_RUNTIME) } +Bequest::~Bequest() { + IsolateGroup* isolate_group = IsolateGroup::Current(); + CHECK_ISOLATE_GROUP(isolate_group); + NoSafepointScope no_safepoint_scope; + ApiState* state = isolate_group->api_state(); + ASSERT(state != nullptr); + state->FreePersistentHandle(handle_); +} + void Isolate::RegisterClass(const Class& cls) { #if !defined(PRODUCT) && !defined(DART_PRECOMPILED_RUNTIME) if (group()->IsReloading()) { @@ -1042,6 +1051,11 @@ MessageHandler::MessageStatus IsolateMessageHandler::HandleMessage( msg_obj = message->raw_obj(); // We should only be sending RawObjects that can be converted to CObjects. ASSERT(ApiObjectConverter::CanConvert(msg_obj.raw())); + } else if (message->IsBequest()) { + Bequest* bequest = message->bequest(); + PersistentHandle* handle = bequest->handle(); + const Object& obj = Object::Handle(zone, handle->raw()); + msg_obj = obj.raw(); } else { MessageSnapshotReader reader(message.get(), thread); msg_obj = reader.ReadObject(); @@ -2375,6 +2389,11 @@ void Isolate::Shutdown() { Isolate::UnMarkIsolateReady(this); LowLevelShutdown(); + if (bequest_.get() != nullptr) { + PortMap::PostMessage(Message::New( + bequest_->beneficiary(), bequest_.release(), Message::kNormalPriority)); + } + // Now we can unregister from the thread, invoke cleanup callback, delete the // isolate (and possibly the isolate group). Isolate::LowLevelCleanup(this); diff --git a/runtime/vm/isolate.h b/runtime/vm/isolate.h index 6633e3e28fc..4d7f5edd9d7 100644 --- a/runtime/vm/isolate.h +++ b/runtime/vm/isolate.h @@ -71,6 +71,7 @@ class Object; class ObjectIdRing; class ObjectPointerVisitor; class ObjectStore; +class PersistentHandle; class RawInstance; class RawArray; class RawContext; @@ -603,6 +604,22 @@ class IsolateGroup : public IntrusiveDListEntry { uint32_t isolate_group_flags_ = 0; }; +// When an isolate sends-and-exits this class represent things that it passed +// to the beneficiary. +class Bequest { + public: + Bequest(PersistentHandle* handle, Dart_Port beneficiary) + : handle_(handle), beneficiary_(beneficiary) {} + ~Bequest(); + + PersistentHandle* handle() { return handle_; } + Dart_Port beneficiary() { return beneficiary_; } + + private: + PersistentHandle* handle_; + Dart_Port beneficiary_; +}; + class Isolate : public BaseIsolate, public IntrusiveDListEntry { public: // Keep both these enums in sync with isolate_patch.dart. @@ -720,6 +737,10 @@ class Isolate : public BaseIsolate, public IntrusiveDListEntry { message_notify_callback_ = value; } + void bequeath(std::unique_ptr bequest) { + bequest_ = std::move(bequest); + } + IsolateGroupSource* source() const { return isolate_group_->source(); } IsolateGroup* group() const { return isolate_group_; } @@ -1436,6 +1457,9 @@ class Isolate : public BaseIsolate, public IntrusiveDListEntry { RawError* sticky_error_; + std::unique_ptr bequest_; + Dart_Port beneficiary_ = 0; + // Protect access to boxed_field_list_. Mutex field_list_mutex_; // List of fields that became boxed and that trigger deoptimization. diff --git a/runtime/vm/message.cc b/runtime/vm/message.cc index 59c1917deca..02b714db888 100644 --- a/runtime/vm/message.cc +++ b/runtime/vm/message.cc @@ -24,13 +24,13 @@ Message::Message(Dart_Port dest_port, : next_(NULL), dest_port_(dest_port), delivery_failure_port_(delivery_failure_port), - snapshot_(snapshot), + payload_(snapshot), snapshot_length_(snapshot_length), finalizable_data_(finalizable_data), priority_(priority) { ASSERT((priority == kNormalPriority) || (delivery_failure_port == kIllegalPort)); - ASSERT(!IsRaw()); + ASSERT(IsSnapshot()); } Message::Message(Dart_Port dest_port, @@ -40,7 +40,7 @@ Message::Message(Dart_Port dest_port, : next_(NULL), dest_port_(dest_port), delivery_failure_port_(delivery_failure_port), - snapshot_(reinterpret_cast(raw_obj)), + payload_(raw_obj), snapshot_length_(0), finalizable_data_(NULL), priority_(priority) { @@ -50,12 +50,31 @@ Message::Message(Dart_Port dest_port, ASSERT(IsRaw()); } +Message::Message(Dart_Port dest_port, + Bequest* bequest, + Priority priority, + Dart_Port delivery_failure_port) + : next_(nullptr), + dest_port_(dest_port), + delivery_failure_port_(delivery_failure_port), + payload_(bequest), + snapshot_length_(-1), + finalizable_data_(nullptr), + priority_(priority) { + ASSERT((priority == kNormalPriority) || + (delivery_failure_port == kIllegalPort)); + ASSERT(IsBequest()); +} + Message::~Message() { ASSERT(delivery_failure_port_ == kIllegalPort); - if (!IsRaw()) { - free(snapshot_); + if (IsSnapshot()) { + free(payload_.snapshot_); } delete finalizable_data_; + if (IsBequest()) { + delete (payload_.bequest_); + } } bool Message::RedirectToDeliveryFailurePort() { diff --git a/runtime/vm/message.h b/runtime/vm/message.h index 66949407656..db45469493b 100644 --- a/runtime/vm/message.h +++ b/runtime/vm/message.h @@ -18,8 +18,13 @@ typedef int64_t Dart_Port; namespace dart { +class Bequest; class JSONStream; class RawObject; +class PersistentHandle; +class HeapPage; +class WeakTable; +class FreeList; class Message { public: @@ -62,6 +67,11 @@ class Message { Priority priority, Dart_Port delivery_failure_port = kIllegalPort); + Message(Dart_Port dest_port, + Bequest* bequest, + Priority priority, + Dart_Port delivery_failure_port = kIllegalPort); + ~Message(); template @@ -72,8 +82,8 @@ class Message { Dart_Port dest_port() const { return dest_port_; } uint8_t* snapshot() const { - ASSERT(!IsRaw()); - return snapshot_; + ASSERT(IsSnapshot()); + return payload_.snapshot_; } intptr_t snapshot_length() const { return snapshot_length_; } @@ -89,12 +99,23 @@ class Message { RawObject* raw_obj() const { ASSERT(IsRaw()); - return reinterpret_cast(snapshot_); + return payload_.raw_obj_; + } + Bequest* bequest() const { + ASSERT(IsBequest()); + return payload_.bequest_; } Priority priority() const { return priority_; } + // A message processed at any interrupt point (stack overflow check) instead + // of at the top of the message loop. Control messages from dart:isolate or + // vm-service requests. bool IsOOB() const { return priority_ == Message::kOOBPriority; } + bool IsSnapshot() const { return !IsRaw() && !IsBequest(); } + // A message whose object is an immortal object from the vm-isolate's heap. bool IsRaw() const { return snapshot_length_ == 0; } + // A message sent from sendAndExit. + bool IsBequest() const { return snapshot_length_ == -1; } bool RedirectToDeliveryFailurePort(); @@ -114,7 +135,15 @@ class Message { Message* next_; Dart_Port dest_port_; Dart_Port delivery_failure_port_; - uint8_t* snapshot_; + union Payload { + Payload(uint8_t* snapshot) : snapshot_(snapshot) {} + Payload(RawObject* raw_obj) : raw_obj_(raw_obj) {} + Payload(Bequest* bequest) : bequest_(bequest) {} + + uint8_t* snapshot_; + RawObject* raw_obj_; + Bequest* bequest_; + } payload_; intptr_t snapshot_length_; MessageFinalizableData* finalizable_data_; Priority priority_; diff --git a/runtime/vm/port.cc b/runtime/vm/port.cc index 140121d6fc4..520de8691b1 100644 --- a/runtime/vm/port.cc +++ b/runtime/vm/port.cc @@ -223,6 +223,14 @@ Isolate* PortMap::GetIsolate(Dart_Port id) { return handler->isolate(); } +bool PortMap::IsReceiverInThisIsolateGroup(Dart_Port receiver, + IsolateGroup* group) { + MutexLocker ml(mutex_); + auto it = ports_->TryLookup(receiver); + if (it == ports_->end()) return false; + return (*it).handler->isolate()->group() == group; +} + void PortMap::Init() { // TODO(bkonyi): don't keep ports_ after Dart_Cleanup. if (mutex_ == NULL) { diff --git a/runtime/vm/port.h b/runtime/vm/port.h index 1d9c88e1db4..0297f1d0cb3 100644 --- a/runtime/vm/port.h +++ b/runtime/vm/port.h @@ -58,6 +58,9 @@ class PortMap : public AllStatic { // Returns the owning Isolate for port 'id'. static Isolate* GetIsolate(Dart_Port id); + static bool IsReceiverInThisIsolateGroup(Dart_Port receiver, + IsolateGroup* group); + static void Init(); static void Cleanup(); diff --git a/sdk/lib/_internal/vm/lib/internal_patch.dart b/sdk/lib/_internal/vm/lib/internal_patch.dart index 515ad8b90a9..6accfc3d598 100644 --- a/sdk/lib/_internal/vm/lib/internal_patch.dart +++ b/sdk/lib/_internal/vm/lib/internal_patch.dart @@ -11,6 +11,7 @@ import "dart:core" hide Symbol; +import "dart:isolate" show SendPort; import "dart:typed_data" show Int32List; /// These are the additional parts of this patch library: @@ -130,3 +131,6 @@ T unsafeCast(Object v) native "Internal_unsafeCast"; // This is implemented by a recognized method, but in bytecode through a native. @pragma('vm:prefer-inline') void reachabilityFence(Object object) native "Internal_reachabilityFence"; + +void sendAndExit(SendPort sendPort, var message) + native "SendPortImpl_sendAndExitInternal_"; diff --git a/sdk_nnbd/lib/_internal/vm/lib/internal_patch.dart b/sdk_nnbd/lib/_internal/vm/lib/internal_patch.dart index ddb399216ee..d3aef4d9f7c 100644 --- a/sdk_nnbd/lib/_internal/vm/lib/internal_patch.dart +++ b/sdk_nnbd/lib/_internal/vm/lib/internal_patch.dart @@ -9,6 +9,7 @@ import "dart:core" hide Symbol; +import "dart:isolate" show SendPort; import "dart:typed_data" show Int32List; /// These are the additional parts of this patch library: @@ -128,3 +129,6 @@ T unsafeCast(Object? v) native "Internal_unsafeCast"; // This is implemented by a recognized method, but in bytecode through a native. @pragma('vm:prefer-inline') void reachabilityFence(Object object) native "Internal_reachabilityFence"; + +void sendAndExit(SendPort sendPort, var message) + native "SendPortImpl_sendAndExitInternal_";