[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 <aam@google.com>
Reviewed-by: Martin Kustermann <kustermann@google.com>
Reviewed-by: Ryan Macnak <rmacnak@google.com>
This commit is contained in:
Alexander Aprelev
committed by
commit-bot@chromium.org
commit-bot@chromium.org
parent
23cbe39355
commit
17654b70d7
@@ -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<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.");
|
||||
}
|
||||
@@ -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<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();
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
export 'dart:_internal' show sendAndExit;
|
||||
@@ -691,6 +691,7 @@ Future _processLoadRequest(request) async {
|
||||
prepend = ", ";
|
||||
if (sb.length > 256) break;
|
||||
}
|
||||
sb.write("]");
|
||||
partToString = sb.toString();
|
||||
} else {
|
||||
partToString = part.toString();
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
@@ -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) \
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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 "";
|
||||
|
||||
+10
-9
@@ -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.
|
||||
|
||||
@@ -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 <map>
|
||||
#include <memory>
|
||||
#include <set>
|
||||
#include <string>
|
||||
|
||||
#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> 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<char*>(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<Isolate*>(worker)->bequeath(
|
||||
std::unique_ptr<Bequest>(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<Isolate*>(worker)->bequeath(
|
||||
std::unique_ptr<Bequest>(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();
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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<IsolateGroup> {
|
||||
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<Isolate> {
|
||||
public:
|
||||
// Keep both these enums in sync with isolate_patch.dart.
|
||||
@@ -720,6 +737,10 @@ class Isolate : public BaseIsolate, public IntrusiveDListEntry<Isolate> {
|
||||
message_notify_callback_ = value;
|
||||
}
|
||||
|
||||
void bequeath(std::unique_ptr<Bequest> 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<Isolate> {
|
||||
|
||||
RawError* sticky_error_;
|
||||
|
||||
std::unique_ptr<Bequest> 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.
|
||||
|
||||
+24
-5
@@ -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<uint8_t*>(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() {
|
||||
|
||||
+33
-4
@@ -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 <typename... Args>
|
||||
@@ -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<RawObject*>(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_;
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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();
|
||||
|
||||
|
||||
@@ -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<T>(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_";
|
||||
|
||||
@@ -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<T>(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_";
|
||||
|
||||
Reference in New Issue
Block a user