dart-lang/sdk
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Refactor isolate startup code in preparation for making isolate spawning more truly non-blocking.

Browse Source 
Instead of passing a startup message to the new isolate, pass all
necessary information in to _startIsolate directly.  The new isolate
sends its control port and capabilities back to the parent.

R=iposva@google.com

Review URL: https://codereview.chromium.org//456983002

git-svn-id: https://dart.googlecode.com/svn/branches/bleeding_edge/dart@39423 260f80e4-7a28-3924-810f-c04153c831b5
This commit is contained in:
turnidge@google.com
2014-08-20 19:57:09 +00:00
parent e4a9df7b12
commit c48496b873
13 changed files with 265 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files
runtime/bin/main.cc
+10 -17
View File
@@ -1125,29 +1125,22 @@ void main(int argc, char** argv) {
// Call _startIsolate in the isolate library to enable dispatching the
// initial startup message.
Dart_Handle isolate_args[2];
isolate_args[0] = main_closure;
isolate_args[1] = Dart_True();
const intptr_t kNumIsolateArgs = 7;
Dart_Handle isolate_args[kNumIsolateArgs];
isolate_args[0] = Dart_Null(); // no parent port
isolate_args[1] = main_closure; // entryPoint
isolate_args[2] = CreateRuntimeOptions(&dart_options); // args
isolate_args[3] = Dart_Null(); // no message
isolate_args[4] = Dart_True(); // isSpawnUri
isolate_args[5] = Dart_Null(); // no control port
isolate_args[6] = Dart_Null(); // no capabilities
Dart_Handle isolate_lib = Dart_LookupLibrary(
Dart_NewStringFromCString("dart:isolate"));
result = Dart_Invoke(isolate_lib,
Dart_NewStringFromCString("_startIsolate"),
2, isolate_args);
// Setup the arguments in the initial startup message and leave the
// replyTo and message fields empty.
Dart_Handle initial_startup_msg = Dart_NewList(3);
result = Dart_ListSetAt(initial_startup_msg, 1,
CreateRuntimeOptions(&dart_options));
kNumIsolateArgs, isolate_args);
DartExitOnError(result);
Dart_Port main_port = Dart_GetMainPortId();
bool posted = Dart_Post(main_port, initial_startup_msg);
if (!posted) {
ErrorExit(kErrorExitCode,
"Failed posting startup message to main "
"isolate control port.");
}
// Keep handling messages until the last active receive port is closed.
result = Dart_RunLoop();
runtime/lib/isolate.cc
+14 -28
View File
@@ -38,7 +38,7 @@ DEFINE_NATIVE_ENTRY(RawReceivePortImpl_factory, 1) {
ASSERT(TypeArguments::CheckedHandle(arguments->NativeArgAt(0)).IsNull());
Dart_Port port_id =
PortMap::CreatePort(arguments->isolate()->message_handler());
return ReceivePort::New(port_id);
return ReceivePort::New(port_id, false /* not control port */);
}
@@ -177,20 +177,10 @@ static RawObject* Spawn(Isolate* parent_isolate,
ThrowIsolateSpawnException(msg);
}
// The result of spawning an Isolate is an array with 3 elements:
// [main_port, pause_capability, terminate_capability]
const Array& result = Array::Handle(Array::New(3));
// Create a SendPort for the new isolate.
Isolate* spawned_isolate = state->isolate();
const SendPort& port = SendPort::Handle(
SendPort::New(spawned_isolate->main_port()));
result.SetAt(0, port);
Capability& capability = Capability::Handle();
capability = Capability::New(spawned_isolate->pause_capability());
result.SetAt(1, capability); // pauseCapability
capability = Capability::New(spawned_isolate->terminate_capability());
result.SetAt(2, capability); // terminateCapability
// Start the new isolate if it is already marked as runnable.
MutexLocker ml(spawned_isolate->mutex());
@@ -199,12 +189,14 @@ static RawObject* Spawn(Isolate* parent_isolate,
spawned_isolate->Run();
}
return result.raw();
return port.raw();
}
DEFINE_NATIVE_ENTRY(Isolate_spawnFunction, 1) {
GET_NON_NULL_NATIVE_ARGUMENT(Instance, closure, arguments->NativeArgAt(0));
DEFINE_NATIVE_ENTRY(Isolate_spawnFunction, 3) {
GET_NON_NULL_NATIVE_ARGUMENT(SendPort, port, arguments->NativeArgAt(0));
GET_NON_NULL_NATIVE_ARGUMENT(Instance, closure, arguments->NativeArgAt(1));
GET_NON_NULL_NATIVE_ARGUMENT(Instance, message, arguments->NativeArgAt(2));
if (closure.IsClosure()) {
Function& func = Function::Handle();
func = Closure::function(closure);
@@ -214,7 +206,7 @@ DEFINE_NATIVE_ENTRY(Isolate_spawnFunction, 1) {
ctx = Closure::context(closure);
ASSERT(ctx.num_variables() == 0);
#endif
return Spawn(isolate, new IsolateSpawnState(func));
return Spawn(isolate, new IsolateSpawnState(port.Id(), func, message));
}
}
const String& msg = String::Handle(String::New(
@@ -224,8 +216,11 @@ DEFINE_NATIVE_ENTRY(Isolate_spawnFunction, 1) {
}
DEFINE_NATIVE_ENTRY(Isolate_spawnUri, 1) {
GET_NON_NULL_NATIVE_ARGUMENT(String, uri, arguments->NativeArgAt(0));
DEFINE_NATIVE_ENTRY(Isolate_spawnUri, 4) {
GET_NON_NULL_NATIVE_ARGUMENT(SendPort, port, arguments->NativeArgAt(0));
GET_NON_NULL_NATIVE_ARGUMENT(String, uri, arguments->NativeArgAt(1));
GET_NON_NULL_NATIVE_ARGUMENT(Instance, args, arguments->NativeArgAt(2));
GET_NON_NULL_NATIVE_ARGUMENT(Instance, message, arguments->NativeArgAt(3));
// Canonicalize the uri with respect to the current isolate.
char* error = NULL;
@@ -238,7 +233,8 @@ DEFINE_NATIVE_ENTRY(Isolate_spawnUri, 1) {
ThrowIsolateSpawnException(msg);
}
return Spawn(isolate, new IsolateSpawnState(canonical_uri));
return Spawn(isolate, new IsolateSpawnState(port.Id(), canonical_uri,
args, message));
}
@@ -259,14 +255,4 @@ DEFINE_NATIVE_ENTRY(Isolate_sendOOB, 2) {
return Object::null();
}
DEFINE_NATIVE_ENTRY(Isolate_mainPort, 0) {
// The control port is being accessed as a regular port from Dart code. This
// is most likely due to the _startIsolate code in dart:isolate. Account for
// this by increasing the number of open control ports.
isolate->message_handler()->increment_control_ports();
return ReceivePort::New(isolate->main_port());
}
} // namespace dart
runtime/lib/isolate_patch.dart
+70 -60
View File
@@ -77,6 +77,14 @@ void _isolateScheduleImmediate(void callback()) {
_pendingImmediateCallback = callback;
}
void _runPendingImmediateCallback() {
if (_pendingImmediateCallback != null) {
var callback = _pendingImmediateCallback;
_pendingImmediateCallback = null;
callback();
}
}
/// The embedder can execute this function to get hold of
/// [_isolateScheduleImmediate] above.
Function _getIsolateScheduleImmediateClosure() {
@@ -120,11 +128,7 @@ class _RawReceivePortImpl implements RawReceivePort {
// VM. Once we have non-fatal global exceptions we need to catch errors
// so that we can run the immediate callbacks.
handler(message);
if (_pendingImmediateCallback != null) {
var callback = _pendingImmediateCallback;
_pendingImmediateCallback = null;
callback();
}
_runPendingImmediateCallback();
}
// Call into the VM to close the VM maintained mappings.
@@ -181,69 +185,71 @@ typedef _MainFunctionArgsMessage(args, message);
*
* The initial startup message is received through the control port.
*/
void _startIsolate(Function entryPoint, bool isSpawnUri) {
// This port keeps the isolate alive until the initial startup message has
// been received.
var keepAlivePort = new RawReceivePort();
ignoreHandler(message) {
// Messages on the current Isolate's control port are dropped after the
// initial startup message has been received.
void _startIsolate(SendPort parentPort,
Function entryPoint,
List<String> args,
var message,
bool isSpawnUri,
RawReceivePort controlPort,
List capabilities) {
if (controlPort != null) {
controlPort.handler = (_) {}; // Nobody home on the control port.
}
if (parentPort != null) {
// Build a message to our parent isolate providing access to the
// current isolate's control port and capabilities.
//
// TODO(floitsch): Send an error message if we can't find the entry point.
var readyMessage = new List(2);
readyMessage[0] = controlPort.sendPort;
readyMessage[1] = capabilities;
isolateStartHandler(message) {
// We received the initial startup message. Ignore all further messages and
// close the port which kept this isolate alive.
Isolate._self.handler = ignoreHandler;
keepAlivePort.close();
// Out of an excess of paranoia we clear the capabilities from the
// stack. Not really necessary.
capabilities = null;
parentPort.send(readyMessage);
}
assert(capabilities == null);
SendPort replyTo = message[0];
if (replyTo != null) {
// TODO(floitsch): don't send ok-message if we can't find the entry point.
replyTo.send("started");
}
if (isSpawnUri) {
assert(message.length == 3);
List<String> args = message[1];
var isolateMessage = message[2];
if (entryPoint is _MainFunctionArgsMessage) {
entryPoint(args, isolateMessage);
} else if (entryPoint is _MainFunctionArgs) {
entryPoint(args);
} else {
entryPoint();
}
if (isSpawnUri) {
if (entryPoint is _MainFunctionArgsMessage) {
entryPoint(args, message);
} else if (entryPoint is _MainFunctionArgs) {
entryPoint(args);
} else {
assert(message.length == 2);
var entryMessage = message[1];
entryPoint(entryMessage);
entryPoint();
}
} else {
entryPoint(message);
}
Isolate._self.handler = isolateStartHandler;
_runPendingImmediateCallback();
}
patch class Isolate {
/* patch */ static Future<Isolate> spawn(
void entryPoint(message), var message, { bool paused: false }) {
// `paused` isn't handled yet.
RawReceivePort readyPort;
try {
// The VM will invoke [_startIsolate] with entryPoint as argument.
List spawnData = _spawnFunction(entryPoint);
assert(spawnData.length == 3);
SendPort controlPort = spawnData[0];
RawReceivePort readyPort = new RawReceivePort();
controlPort.send([readyPort.sendPort, message]);
readyPort = new RawReceivePort();
_spawnFunction(readyPort.sendPort, entryPoint, message);
Completer completer = new Completer<Isolate>.sync();
readyPort.handler = (readyMessage) {
assert(readyMessage == 'started');
readyPort.close();
assert(readyMessage is List);
assert(readyMessage.length == 2);
SendPort controlPort = readyMessage[0];
List capabilities = readyMessage[1];
completer.complete(new Isolate(controlPort,
pauseCapability: spawnData[1],
terminateCapability: spawnData[2]));
pauseCapability: capabilities[0],
terminateCapability: capabilities[1]));
};
return completer.future;
} catch (e, st) {
if (readyPort != null) {
readyPort.close();
}
return new Future<Isolate>.error(e, st);
};
}
@@ -251,41 +257,45 @@ patch class Isolate {
/* patch */ static Future<Isolate> spawnUri(
Uri uri, List<String> args, var message, { bool paused: false }) {
// `paused` isn't handled yet.
RawReceivePort readyPort;
try {
// The VM will invoke [_startIsolate] and not `main`.
List spawnData = _spawnUri(uri.toString());
assert(spawnData.length == 3);
SendPort controlPort = spawnData[0];
RawReceivePort readyPort = new RawReceivePort();
controlPort.send([readyPort.sendPort, args, message]);
readyPort = new RawReceivePort();
_spawnUri(readyPort.sendPort, uri.toString(), args, message);
Completer completer = new Completer<Isolate>.sync();
readyPort.handler = (readyMessage) {
assert(readyMessage == 'started');
readyPort.close();
assert(readyMessage is List);
assert(readyMessage.length == 2);
SendPort controlPort = readyMessage[0];
List capabilities = readyMessage[1];
completer.complete(new Isolate(controlPort,
pauseCapability: spawnData[1],
terminateCapability: spawnData[2]));
pauseCapability: capabilities[0],
terminateCapability: capabilities[1]));
};
return completer.future;
} catch (e, st) {
if (readyPort != null) {
readyPort.close();
}
return new Future<Isolate>.error(e, st);
};
return completer.future;
}
static final RawReceivePort _self = _mainPort;
static RawReceivePort get _mainPort native "Isolate_mainPort";
// TODO(iposva): Cleanup to have only one definition.
// These values need to be kept in sync with the class IsolateMessageHandler
// in vm/isolate.cc.
static const _PAUSE = 1;
static const _RESUME = 2;
static List _spawnFunction(Function topLevelFunction)
static SendPort _spawnFunction(SendPort readyPort, Function topLevelFunction,
var message)
native "Isolate_spawnFunction";
static List _spawnUri(String uri) native "Isolate_spawnUri";
static SendPort _spawnUri(SendPort readyPort, String uri,
List<String> args, var message)
native "Isolate_spawnUri";
static void _sendOOB(port, msg) native "Isolate_sendOOB";
runtime/vm/bootstrap_natives.h
+3 -4
View File
@@ -286,10 +286,9 @@ namespace dart {
V(Int32x4_setFlagZ, 2) \
V(Int32x4_setFlagW, 2) \
V(Int32x4_select, 3) \
V(Isolate_mainPort, 0) \
V(Isolate_spawnFunction, 1) \
V(Isolate_spawnUri, 1) \
V(Isolate_sendOOB, 2) \
V(Isolate_spawnFunction, 3) \
V(Isolate_spawnUri, 4) \
V(Isolate_sendOOB, 2) \
V(Mirrors_evalInLibraryWithPrivateKey, 2) \
V(Mirrors_makeLocalClassMirror, 1) \
V(Mirrors_makeLocalTypeMirror, 1) \
runtime/vm/isolate.cc
+82 -11
View File
@@ -771,17 +771,28 @@ static bool RunIsolate(uword parameter) {
func ^= result.raw();
func = func.ImplicitClosureFunction();
const Array& capabilities = Array::Handle(Array::New(2));
Capability& capability = Capability::Handle();
capability = Capability::New(isolate->pause_capability());
capabilities.SetAt(0, capability);
capability = Capability::New(isolate->terminate_capability());
capabilities.SetAt(1, capability);
// Instead of directly invoking the entry point we call '_startIsolate' with
// the entry point as argument. The '_startIsolate' function will
// communicate with the spawner to receive the initial message before it
// executes the real entry point.
// the entry point as argument.
// Since this function ("RunIsolate") is used for both Isolate.spawn and
// Isolate.spawnUri we also send a boolean flag as argument so that the
// "_startIsolate" function can act corresponding to how the isolate was
// created.
const Array& args = Array::Handle(Array::New(2));
args.SetAt(0, Instance::Handle(func.ImplicitStaticClosure()));
args.SetAt(1, is_spawn_uri ? Bool::True() : Bool::False());
const Array& args = Array::Handle(Array::New(7));
args.SetAt(0, SendPort::Handle(SendPort::New(state->parent_port())));
args.SetAt(1, Instance::Handle(func.ImplicitStaticClosure()));
args.SetAt(2, Instance::Handle(state->BuildArgs()));
args.SetAt(3, Instance::Handle(state->BuildMessage()));
args.SetAt(4, is_spawn_uri ? Bool::True() : Bool::False());
args.SetAt(5, ReceivePort::Handle(
ReceivePort::New(isolate->main_port(), true /* control port */)));
args.SetAt(6, capabilities);
const Library& lib = Library::Handle(Library::IsolateLibrary());
const String& entry_name = String::Handle(String::New("_startIsolate"));
@@ -1109,7 +1120,6 @@ void Isolate::PrintJSON(JSONStream* stream, bool ref) {
jsobj.AddProperty("depth", (intptr_t)0);
}
jsobj.AddProperty("livePorts", message_handler()->live_ports());
jsobj.AddProperty("controlPorts", message_handler()->control_ports());
jsobj.AddProperty("pauseOnExit", message_handler()->pause_on_exit());
// TODO(turnidge): Make the debugger support paused_on_start/exit.
@@ -1304,13 +1314,50 @@ T* Isolate::AllocateReusableHandle() {
}
IsolateSpawnState::IsolateSpawnState(const Function& func)
static uint8_t* allocator(uint8_t* ptr, intptr_t old_size, intptr_t new_size) {
void* new_ptr = realloc(reinterpret_cast<void*>(ptr), new_size);
return reinterpret_cast<uint8_t*>(new_ptr);
}
static void SerializeObject(const Instance& obj,
uint8_t** obj_data,
intptr_t* obj_len) {
MessageWriter writer(obj_data, &allocator);
writer.WriteMessage(obj);
*obj_len = writer.BytesWritten();
}
static RawInstance* DeserializeObject(Isolate* isolate,
uint8_t* obj_data,
intptr_t obj_len) {
if (obj_data == NULL) {
return Instance::null();
}
SnapshotReader reader(obj_data, obj_len, Snapshot::kMessage, isolate);
const Object& obj = Object::Handle(isolate, reader.ReadObject());
ASSERT(!obj.IsError());
Instance& instance = Instance::Handle(isolate);
instance ^= obj.raw(); // Can't use Instance::Cast because may be null.
return instance.raw();
}
IsolateSpawnState::IsolateSpawnState(Dart_Port parent_port,
const Function& func,
const Instance& message)
: isolate_(NULL),
parent_port_(parent_port),
script_url_(NULL),
library_url_(NULL),
class_name_(NULL),
function_name_(NULL),
exception_callback_name_(NULL) {
exception_callback_name_(NULL),
serialized_args_(NULL),
serialized_args_len_(0),
serialized_message_(NULL),
serialized_message_len_(0) {
script_url_ = NULL;
const Class& cls = Class::Handle(func.Owner());
const Library& lib = Library::Handle(cls.library());
@@ -1324,19 +1371,30 @@ IsolateSpawnState::IsolateSpawnState(const Function& func)
class_name_ = strdup(class_name.ToCString());
}
exception_callback_name_ = strdup("_unhandledExceptionCallback");
SerializeObject(message, &serialized_message_, &serialized_message_len_);
}
IsolateSpawnState::IsolateSpawnState(const char* script_url)
IsolateSpawnState::IsolateSpawnState(Dart_Port parent_port,
const char* script_url,
const Instance& args,
const Instance& message)
: isolate_(NULL),
parent_port_(parent_port),
library_url_(NULL),
class_name_(NULL),
function_name_(NULL),
exception_callback_name_(NULL) {
exception_callback_name_(NULL),
serialized_args_(NULL),
serialized_args_len_(0),
serialized_message_(NULL),
serialized_message_len_(0) {
script_url_ = strdup(script_url);
library_url_ = NULL;
function_name_ = strdup("main");
exception_callback_name_ = strdup("_unhandledExceptionCallback");
SerializeObject(args, &serialized_args_, &serialized_args_len_);
SerializeObject(message, &serialized_message_, &serialized_message_len_);
}
@@ -1346,6 +1404,8 @@ IsolateSpawnState::~IsolateSpawnState() {
free(function_name_);
free(class_name_);
free(exception_callback_name_);
free(serialized_args_);
free(serialized_message_);
}
@@ -1402,6 +1462,17 @@ RawObject* IsolateSpawnState::ResolveFunction() {
}
RawInstance* IsolateSpawnState::BuildArgs() {
return DeserializeObject(isolate_, serialized_args_, serialized_args_len_);
}
RawInstance* IsolateSpawnState::BuildMessage() {
return DeserializeObject(isolate_,
serialized_message_, serialized_message_len_);
}
void IsolateSpawnState::Cleanup() {
SwitchIsolateScope switch_scope(I);
Dart::ShutdownIsolate();
runtime/vm/isolate.h
+16 -2
View File
@@ -803,12 +803,19 @@ class SwitchIsolateScope {
class IsolateSpawnState {
public:
explicit IsolateSpawnState(const Function& func);
explicit IsolateSpawnState(const char* script_url);
IsolateSpawnState(Dart_Port parent_port,
const Function& func,
const Instance& message);
IsolateSpawnState(Dart_Port parent_port,
const char* script_url,
const Instance& args,
const Instance& message);
~IsolateSpawnState();
Isolate* isolate() const { return isolate_; }
void set_isolate(Isolate* value) { isolate_ = value; }
Dart_Port parent_port() const { return parent_port_; }
char* script_url() const { return script_url_; }
char* library_url() const { return library_url_; }
char* class_name() const { return class_name_; }
@@ -817,15 +824,22 @@ class IsolateSpawnState {
bool is_spawn_uri() const { return library_url_ == NULL; }
RawObject* ResolveFunction();
RawInstance* BuildArgs();
RawInstance* BuildMessage();
void Cleanup();
private:
Isolate* isolate_;
Dart_Port parent_port_;
char* script_url_;
char* library_url_;
char* class_name_;
char* function_name_;
char* exception_callback_name_;
uint8_t* serialized_args_;
intptr_t serialized_args_len_;
uint8_t* serialized_message_;
intptr_t serialized_message_len_;
};
} // namespace dart
runtime/vm/message_handler.cc
+2 -21
View File
@@ -34,7 +34,6 @@ class MessageHandlerTask : public ThreadPool::Task {
MessageHandler::MessageHandler()
: queue_(new MessageQueue()),
oob_queue_(new MessageQueue()),
control_ports_(0),
live_ports_(0),
paused_(0),
pause_on_start_(false),
@@ -272,8 +271,8 @@ void MessageHandler::ClosePort(Dart_Port port) {
OS::Print("[-] Closing port:\n"
"\thandler: %s\n"
"\tport: %" Pd64 "\n"
"\tports: control(%" Pd ") live(%" Pd ")\n",
name(), port, control_ports_, live_ports_);
"\tports: live(%" Pd ")\n",
name(), port, live_ports_);
}
}
@@ -307,22 +306,4 @@ void MessageHandler::decrement_live_ports() {
live_ports_--;
}
void MessageHandler::increment_control_ports() {
MonitorLocker ml(&monitor_);
#if defined(DEBUG)
CheckAccess();
#endif
control_ports_++;
}
void MessageHandler::decrement_control_ports() {
MonitorLocker ml(&monitor_);
#if defined(DEBUG)
CheckAccess();
#endif
control_ports_--;
}
} // namespace dart
runtime/vm/message_handler.h
+1 -13
View File
@@ -54,24 +54,13 @@ class MessageHandler {
// Returns true on success.
bool HandleOOBMessages();
// The number of opened control ports is determined whether the isolate has
// live ports. An isolate is considered not having any live ports if only
// control ports are open.
// Usually either 0 or 1.
void increment_control_ports();
void decrement_control_ports();
// A message handler tracks how many live ports it has.
bool HasLivePorts() const { return live_ports_ > control_ports_; }
bool HasLivePorts() const { return live_ports_ > 0; }
intptr_t live_ports() const {
return live_ports_;
}
intptr_t control_ports() const {
return control_ports_;
}
bool paused() const { return paused_ > 0; }
void increment_paused() { paused_++; }
@@ -168,7 +157,6 @@ class MessageHandler {
Monitor monitor_; // Protects all fields in MessageHandler.
MessageQueue* queue_;
MessageQueue* oob_queue_;
intptr_t control_ports_; // The number of open control ports usually 0 or 1.
intptr_t live_ports_; // The number of open ports, including control ports.
intptr_t paused_; // The number of pause messages received.
bool pause_on_start_;
runtime/vm/object.cc
+8 -2
View File
@@ -18808,7 +18808,9 @@ void Capability::PrintJSONImpl(JSONStream* stream, bool ref) const {
}
RawReceivePort* ReceivePort::New(Dart_Port id, Heap::Space space) {
RawReceivePort* ReceivePort::New(Dart_Port id,
bool is_control_port,
Heap::Space space) {
Isolate* isolate = Isolate::Current();
const SendPort& send_port = SendPort::Handle(isolate, SendPort::New(id));
@@ -18821,7 +18823,11 @@ RawReceivePort* ReceivePort::New(Dart_Port id, Heap::Space space) {
result ^= raw;
result.raw_ptr()->send_port_ = send_port.raw();
}
PortMap::SetLive(id);
if (is_control_port) {
PortMap::SetPortState(id, PortMap::kControlPort);
} else {
PortMap::SetPortState(id, PortMap::kLivePort);
}
return result.raw();
}
runtime/vm/object.h
+3 -1
View File
@@ -6915,7 +6915,9 @@ class ReceivePort : public Instance {
static intptr_t InstanceSize() {
return RoundedAllocationSize(sizeof(RawReceivePort));
}
static RawReceivePort* New(Dart_Port id, Heap::Space space = Heap::kNew);
static RawReceivePort* New(Dart_Port id,
bool is_control_port,
Heap::Space space = Heap::kNew);
private:
FINAL_HEAP_OBJECT_IMPLEMENTATION(ReceivePort, Instance);
runtime/vm/port.cc
+27 -7
View File
@@ -63,6 +63,21 @@ void PortMap::Rehash(intptr_t new_capacity) {
}
const char* PortMap::PortStateString(PortState kind) {
switch (kind) {
case kNewPort:
return "new";
case kLivePort:
return "live";
case kControlPort:
return "control";
default:
UNREACHABLE();
return "UNKNOWN";
}
}
Dart_Port PortMap::AllocatePort() {
const Dart_Port kMASK = 0x3fffffff;
Dart_Port result = prng_->NextUInt32() & kMASK;
@@ -79,16 +94,21 @@ Dart_Port PortMap::AllocatePort() {
}
void PortMap::SetLive(Dart_Port port) {
void PortMap::SetPortState(Dart_Port port, PortState state) {
MutexLocker ml(mutex_);
intptr_t index = FindPort(port);
ASSERT(index >= 0);
map_[index].live = true;
map_[index].handler->increment_live_ports();
PortState old_state = map_[index].state;
ASSERT(old_state == kNewPort);
map_[index].state = state;
if (state == kLivePort) {
map_[index].handler->increment_live_ports();
}
if (FLAG_trace_isolates) {
OS::Print("[^] Live port: \n"
OS::Print("[^] Port (%s) -> (%s): \n"
"\thandler: %s\n"
"\tport: %" Pd64 "\n",
PortStateString(old_state), PortStateString(state),
map_[index].handler->name(), port);
}
}
@@ -117,7 +137,7 @@ Dart_Port PortMap::CreatePort(MessageHandler* handler) {
Entry entry;
entry.port = AllocatePort();
entry.handler = handler;
entry.live = false;
entry.state = kNewPort;
// Search for the first unused slot. Make use of the knowledge that here is
// currently no port with this id in the port map.
@@ -179,7 +199,7 @@ bool PortMap::ClosePort(Dart_Port port) {
// pending messages below.
map_[index].port = 0;
map_[index].handler = deleted_entry_;
if (map_[index].live) {
if (map_[index].state == kLivePort) {
handler->decrement_live_ports();
}
@@ -203,7 +223,7 @@ void PortMap::ClosePorts(MessageHandler* handler) {
// Mark the slot as deleted.
map_[i].port = 0;
map_[i].handler = deleted_entry_;
if (map_[i].live) {
if (map_[i].state == kLivePort) {
handler->decrement_live_ports();
}
used_--;
runtime/vm/port.h
+10 -2
View File
@@ -20,12 +20,18 @@ class PortMapTestPeer;
class PortMap: public AllStatic {
public:
enum PortState {
kNewPort = 0, // a newly allocated port
kLivePort = 1, // a regular port (has a ReceivePort)
kControlPort = 2, // a special control port (has a ReceivePort)
};
// Allocate a port for the provided handler and return its VM-global id.
static Dart_Port CreatePort(MessageHandler* handler);
// Indicates that a port has had a ReceivePort created for it at the
// dart language level. The port remains live until it is closed.
static void SetLive(Dart_Port id);
static void SetPortState(Dart_Port id, PortState kind);
// Close the port with id. All pending messages will be dropped.
//
@@ -59,9 +65,11 @@ class PortMap: public AllStatic {
typedef struct {
Dart_Port port;
MessageHandler* handler;
bool live;
PortState state;
} Entry;
static const char* PortStateString(PortState state);
// Allocate a new unique port.
static Dart_Port AllocatePort();
runtime/vm/port_test.cc
+19 -3
View File
@@ -25,7 +25,7 @@ class PortMapTestPeer {
if (index < 0) {
return false;
}
return PortMap::map_[index].live;
return PortMap::map_[index].state == PortMap::kLivePort;
}
};
@@ -100,20 +100,36 @@ TEST_CASE(PortMap_CreateManyPorts) {
}
TEST_CASE(PortMap_SetLive) {
TEST_CASE(PortMap_SetPortState) {
PortTestMessageHandler handler;
// Regular port.
Dart_Port port = PortMap::CreatePort(&handler);
EXPECT_NE(0, port);
EXPECT(PortMapTestPeer::IsActivePort(port));
EXPECT(!PortMapTestPeer::IsLivePort(port));
PortMap::SetLive(port);
PortMap::SetPortState(port, PortMap::kLivePort);
EXPECT(PortMapTestPeer::IsActivePort(port));
EXPECT(PortMapTestPeer::IsLivePort(port));
PortMap::ClosePort(port);
EXPECT(!PortMapTestPeer::IsActivePort(port));
EXPECT(!PortMapTestPeer::IsLivePort(port));
// Control port.
port = PortMap::CreatePort(&handler);
EXPECT_NE(0, port);
EXPECT(PortMapTestPeer::IsActivePort(port));
EXPECT(!PortMapTestPeer::IsLivePort(port));
PortMap::SetPortState(port, PortMap::kControlPort);
EXPECT(PortMapTestPeer::IsActivePort(port));
EXPECT(!PortMapTestPeer::IsLivePort(port));
PortMap::ClosePort(port);
EXPECT(!PortMapTestPeer::IsActivePort(port));
EXPECT(!PortMapTestPeer::IsLivePort(port));
}