深刻理解Isolate
因为Dart是一种单线程模型语言，因此避免了多线程环境下产生的一系列下降运行效率的问题。但单线程模型却有一个很是严重的问题，就是耗时任务的执行。当执行耗时任务时，会致使当前线程会被阻塞，从而没法继续执行。这时候就须要异步任务，而Dart提供了Isolate来执行异步任务。c++
在刚开始学习Isolate时，觉得它相似Java、C/CPP中的线程。但随着学习的深刻，发现Isolate远比Java、C/CPP中的线程复杂。下面就来一窥究竟。git
一、Isolate的使用
在Dart中，Isolate的使用及通讯都较为复杂，主要是经过 Isolate.spawn及Isolate.spawnUri来建立Isolate，ReceivePort来进行Isolate间通讯。下面就来看如何使用Isolate。github
1.一、Isolate单向通讯
先来看Isolate间的单向通讯，代码以下。api
//在父Isolate中调用
Isolate isolate;
start() async {
ReceivePort receivePort = ReceivePort();
//建立子Isolate对象
isolate = await Isolate.spawn(getMsg, receivePort.sendPort);
//监听子Isolate的返回数据
receivePort.listen((data) {
print('data：$data');
receivePort.close();
//关闭Isolate对象
isolate?.kill(priority: Isolate.immediate);
isolate = null;
});
}
//子Isolate对象的入口函数，能够在该函数中作耗时操做
getMsg(sendPort) => sendPort.send("hello");
复制代码
运行代码后，就会输出新建立Isolate对象返回的数据，以下。 markdown
1.二、Isolate双向通讯
再来看多个Isolate之间的通讯实现，代码以下。多线程
//当前函数在父Isolate中
Future<dynamic> asyncFactoriali(n) async {
//父Isolate对应的ReceivePort对象
final response = ReceivePort();
//建立一个子Isolate对象
await Isolate.spawn(_isolate, response.sendPort);
final sendPort = await response.first as SendPort;
final answer = ReceivePort();
//给子Isolate发送数据
sendPort.send([n, answer.sendPort]);
return answer.first;
}
//子Isolate的入口函数，能够在该函数中作耗时操做
_isolate(SendPort initialReplyTo) async {
//子Isolate对应的ReceivePort对象
final port = ReceivePort();
initialReplyTo.send(port.sendPort);
final message = await port.first as List;
final data = message[0] as int;
final send = message[1] as SendPort;
//给父Isolate的返回数据
send.send(syncFactorial(data));
}
//运行代码
start() async {
print("计算结果：${await asyncFactoriali(4)}");
}
start();
复制代码
经过在新建立的Isolate中计算并返回数据后，获得以下返回结果。并发
经过上面代码，咱们就能够可以经过Isolate来执行异步任务。下面再来看其具体实现原理。异步
二、isolate的建立与运行
先从下面的时序图来看isolate是如何建立及运行的。 async
isolate的建立及运行两方面来对上图进行详细介绍。
2.一、isolate的建立
首先来看isolate的建立，在上面例子中是经过Isolate.spawn来建立Isolate对象。ide
class Isolate {
//声明外部实现
external static Future<Isolate> spawn<T>(
void entryPoint(T message), T message,
{bool paused: false,
bool errorsAreFatal,
SendPort onExit,
SendPort onError,
@Since("2.3") String debugName});
}
复制代码
这里的external关键字主要是声明spawn这个函数，具体实现由外部提供。在Dart中，该函数的具体实现是在isolate_patch.dart中。先来看spawn的具体实现。
@patch
class Isolate {
@patch
static Future<Isolate> spawn<T>(void entryPoint(T message), T message,
{bool paused: false,
bool errorsAreFatal,
SendPort onExit,
SendPort onError,
String debugName}) async {
// `paused` isn't handled yet.
RawReceivePort readyPort;
try {
//该函数执行是异步的
_spawnFunction(
readyPort.sendPort,
script.toString(),
entryPoint,
message,
paused,
errorsAreFatal,
onExit,
onError,
null,
packageConfig,
debugName);
return await _spawnCommon(readyPort);
} catch (e, st) {
...
}
}
static Future<Isolate> _spawnCommon(RawReceivePort readyPort) {
Completer completer = new Completer<Isolate>.sync();
//监听Isolate是否建立完毕，当子Isolate建立完毕后会通知父Isolate
readyPort.handler = (readyMessage) {
//关闭端口
readyPort.close();
if (readyMessage is List && readyMessage.length == 2) {//子Isolate建立成功
SendPort controlPort = readyMessage[0];
List capabilities = readyMessage[1];
completer.complete(new Isolate(controlPort,
pauseCapability: capabilities[0],
terminateCapability: capabilities[1]));
} else if (readyMessage is String) {...} else {...}
};
return completer.future;
}
......
//调用虚拟机中的Isolate_spawnFunction函数
static void _spawnFunction(
SendPort readyPort,
String uri,
Function topLevelFunction,
var message,
bool paused,
bool errorsAreFatal,
SendPort onExit,
SendPort onError,
String packageRoot,
String packageConfig,
String debugName) native "Isolate_spawnFunction";
......
}
复制代码
这里的_spawnFunction调用的是Dart VM中的Isolate_spawnFunction函数，该函数就是把Isolate对象的建立交给线程池执行，因此Isolate对象的建立是异步的。这里的线程池是在Dart VM初始化的时候建立的。
[->third_party/dart/runtime/lib/isolate.cc]
DEFINE_NATIVE_ENTRY(Isolate_spawnFunction, 0, 11) {
...
if (closure.IsClosure()) {
...
//异步执行，thread_pool是一个线程池
Dart::thread_pool()->Run<SpawnIsolateTask>(isolate, std::move(state));
return Object::null();
}
}
...
return Object::null();
}
复制代码
SpawnIsolateTask是一个相似Java中实现了Runable接口的类，在该类中主要是进行子Isolate对象的建立及运行，来看其具体实现。
[->third_party/dart/runtime/lib/isolate.cc]
//在子线程中执行
class SpawnIsolateTask : public ThreadPool::Task {
void Run() override {
...
// initialize_callback对应[->third_party/dart/runtime/bin/main.cc]中的OnIsolateInitialize函数
// OnIsolateInitialize是在Dart VM初始化时设置的
//[见小节2.2]
Dart_InitializeIsolateCallback initialize_callback =
Isolate::InitializeCallback();
...
// Create a new isolate.
char* error = nullptr;
Isolate* isolate = nullptr;
//在AOT编译环境下，FLAG_enable_isolate_groups为true，不然为false
//group及initialize_callback都是在虚拟机初始化的时候设置的
if (!FLAG_enable_isolate_groups || group == nullptr ||
initialize_callback == nullptr) {
...
} else {
...
//先直接看AOT编译下的Isolate建立
isolate = CreateWithinExistingIsolateGroup(group, name, &error);
...
void* child_isolate_data = nullptr;
//将Isolate设置为可运行
//[见2.2小节]
bool success = initialize_callback(&child_isolate_data, &error);
}
//建立失败
if (isolate == nullptr) {
FailedSpawn(error);
free(error);
return;
}
...
// isolate是不是可运行的
// 是在OnIsolateInitialize中设置的
if (isolate->is_runnable()) {
//运行isolate
//[见2.2小节]
isolate->Run();
}
}
};
复制代码
AOT编译下，在子线程中调用CreateWithinExistingIsolateGroup函数来建立Isolate对象。
[->third_party/dart/runtime/vm/dart_api_impl.cc]
Isolate* CreateWithinExistingIsolateGroup(IsolateGroup* group, const char* name, char** error) {
//建立Isolate对象
Isolate* isolate = reinterpret_cast<Isolate*>(
CreateIsolate(group, name, /*isolate_data=*/nullptr, error));
...
return isolate;
}
...
static Dart_Isolate CreateIsolate(IsolateGroup* group, const char* name, void* isolate_data, char** error) {
auto source = group->source();
Isolate* I = Dart::CreateIsolate(name, source->flags, group);
...
Dart::ShutdownIsolate();
return reinterpret_cast<Dart_Isolate>(NULL);
}
复制代码
通过一系列调用，最终调用dart.cc中的CreateIsolate函数，该函数很简单，就是建立一个新的Isolate对象。
[->third_party/dart/runtime/vm/dart.cc]
Isolate* Dart::CreateIsolate(const char* name_prefix,
const Dart_IsolateFlags& api_flags,
IsolateGroup* isolate_group) {
// Create a new isolate.
Isolate* isolate =
Isolate::InitIsolate(name_prefix, isolate_group, api_flags);
return isolate;
}
复制代码
[->third_party/dart/runtime/vm/isolate.cc]
//初始化Isolate
Isolate* Isolate::InitIsolate(const char* name_prefix,
IsolateGroup* isolate_group,
const Dart_IsolateFlags& api_flags,
bool is_vm_isolate) {
//一、建立一个Isolate对象
Isolate* result = new Isolate(isolate_group, api_flags);
...
//二、建立Isolate对应的堆空间，在该堆空间中，存在对象的分配，垃圾回收等。
Heap::Init(result,
is_vm_isolate
? 0 // New gen size 0; VM isolate should only allocate in old.
: FLAG_new_gen_semi_max_size * MBInWords,//MBInWords值是128kb，
(is_service_or_kernel_isolate ? kDefaultMaxOldGenHeapSize
: FLAG_old_gen_heap_size) *
MBInWords);
//三、将Isolate与Thread相关联
if (!Thread::EnterIsolate(result)) {
// We failed to enter the isolate, it is possible the VM is shutting down,
// return back a NULL so that CreateIsolate reports back an error.
if (KernelIsolate::IsKernelIsolate(result)) {
KernelIsolate::SetKernelIsolate(nullptr);
}
if (ServiceIsolate::IsServiceIsolate(result)) {
ServiceIsolate::SetServiceIsolate(nullptr);
}
delete result;
return nullptr;
}
// Setup the isolate message handler.
//四、设置isolate的消息处理器
MessageHandler* handler = new IsolateMessageHandler(result);
result->set_message_handler(handler);
// Setup the Dart API state.
//五、启动Dart API状态
ApiState* state = new ApiState();
result->set_api_state(state);
//六、设置主端口
result->set_main_port(PortMap::CreatePort(result->message_handler()));
// Add to isolate list. Shutdown and delete the isolate on failure.
//七、将当前的Isolate添加到链表中（一个单链表）
if (!AddIsolateToList(result)) {
//添加失败，销毁该Isolate
result->LowLevelShutdown();
//取消线程与Isolate的关联
Thread::ExitIsolate();
//若是是虚拟机内部的Isolate
if (KernelIsolate::IsKernelIsolate(result)) {
KernelIsolate::SetKernelIsolate(nullptr);
}
//若是是Service Isolate
if (ServiceIsolate::IsServiceIsolate(result)) {
ServiceIsolate::SetServiceIsolate(nullptr);
}
//删除当前Isolate对象
delete result;
return nullptr;
}
return result;
}
复制代码
InitIsolate函数比较重要，主要作了如下事情。
- 建立
Isolate对象 - 建立
Isolate中的堆空间，在Isolate仅有一块堆空间。存在堆空间也就会存在对象分配、垃圾回收等。 - 将
Isolate对象与一个线程进行关联，也就是能够说一个线程对应着一个Isolate对象。 - 设置消息处理器（
IsolateMessageHandler），主要是对于Isolate中的消息处理。子Isolate能够经过端口向父Isolate的MessageHandler中添加消息，反之亦然。这也是Isolate间的通讯的实现。 - 设置api state，暂时没搞懂这个是干啥的。
- 设置主端口。
- 将当前
Isolate添加到链表中。
当上面的一些操做执行完毕后，一个Isolate对象就建立成功了。
2.二、isolate的运行
再回到SpawnIsolateTask类中，当调用CreateWithinExistingIsolateGroup建立Isolate成功后，也仅是建立了一个Isolate对象。这时候的Isolate并未运行，也不能执行该Isolate中的任何代码。因此还得主动调用Isolate的Run函数，使Isolate可以运行其中的代码并执行相应的消息。
首先须要经过initialize_callback函数来将Isolate设置为可运行。initialize_callback函数在Dart VM初始化的时候设置，对应着[->third_party/dart/runtime/bin/main.cc]中的OnIsolateInitialize函数。把Isolate设置为可运行后，才能够运行Isolate。
[->third_party/dart/runtime/vm/isolate.cc]
void Isolate::Run() {
//向消息处理器中添加的第一个消息
//记住该RunIsolate函数，在后面会说到
message_handler()->Run(Dart::thread_pool(), RunIsolate, ShutdownIsolate,
reinterpret_cast<uword>(this));
}
复制代码
[->third_party/dart/runtime/vm/MessageHandler.cc]
void MessageHandler::Run(ThreadPool* pool,
StartCallback start_callback,
EndCallback end_callback,
CallbackData data) {
MonitorLocker ml(&monitor_);
pool_ = pool;
start_callback_ = start_callback;
end_callback_ = end_callback;
callback_data_ = data;
task_running_ = true;
//在线程池中执行任务
const bool launched_successfully = pool_->Run<MessageHandlerTask>(this);
}
复制代码
而后继续异步执行，但此次是在子Isolate中执行的。下面再来看MessageHandlerTask，在MessageHandlerTask的run函数中执行的是TaskCallback函数。
[->third_party/dart/runtime/vm/message_handler.cc]
void MessageHandler::TaskCallback() {
MessageStatus status = kOK;
bool run_end_callback = false;
bool delete_me = false;
EndCallback end_callback = NULL;
CallbackData callback_data = 0;
{
...
if (status == kOK) {
//仅当子Isolate第一次运行时，start_callback_才不为null
if (start_callback_ != nullptr) {
ml.Exit();
//调用Isolate的第一个函数（容许多线程并发执行）
status = start_callback_(callback_data_);
ASSERT(Isolate::Current() == NULL);
start_callback_ = NULL;
ml.Enter();
}
...
}
...
}
...
}
复制代码
先无论消息处理[见小结3]，这里重点来看start_callback_，它对应着RunIsolate这个函数。
[->third_party/dart/runtime/vm/isolate.cc]
//运行Isolate
static MessageHandler::MessageStatus RunIsolate(uword parameter) {
...
{
...
//args是调用Dart层_startIsolate函数所需的参数集合
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(thread)));
args.SetAt(3, Instance::Handle(state->BuildMessage(thread)));
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);
//调用Dart层的_startIsolate函数，该函数在isolate_patch.dart文件中
const Library& lib = Library::Handle(Library::IsolateLibrary());
const String& entry_name = String::Handle(String::New("_startIsolate"));
const Function& entry_point =
Function::Handle(lib.LookupLocalFunction(entry_name));
ASSERT(entry_point.IsFunction() && !entry_point.IsNull());
result = DartEntry::InvokeFunction(entry_point, args);
if (result.IsError()) {
return StoreError(thread, Error::Cast(result));
}
}
return MessageHandler::kOK;
}
复制代码
在RunIsolate中，会调用isolate_patch.dart中的_startIsolate函数，从而调用建立Isolate对象时传递的初始化函数。
@pragma("vm:entry-point", "call")
void _startIsolate(
SendPort parentPort,
Function entryPoint,
List<String> args,
var message,
bool isSpawnUri,
RawReceivePort controlPort,
List capabilities) {
// The control port (aka the main isolate port) does not handle any messages.
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;
// Out of an excess of paranoia we clear the capabilities from the
// stack. Not really necessary.
capabilities = null;
//告诉父Isolate，当前`Isolate`已经建立成功
parentPort.send(readyMessage);
}
// Delay all user code handling to the next run of the message loop. This
// allows us to intercept certain conditions in the event dispatch, such as
// starting in paused state.
RawReceivePort port = new RawReceivePort();
port.handler = (_) {
port.close();
if (isSpawnUri) {
if (entryPoint is _BinaryFunction) {
(entryPoint as dynamic)(args, message);
} else if (entryPoint is _UnaryFunction) {
(entryPoint as dynamic)(args);
} else {
entryPoint();
}
} else {
//初始化函数
entryPoint(message);
}
};
// Make sure the message handler is triggered.
port.sendPort.send(null);
}
复制代码
在_startIsolate函数中主要是作了如下几件事。
- 告诉父
Isolate，子Isolate已经建立成功。 - 调用子
Isolate的初始化函数，也就是入口函数。
到此，一个新的Isolate就已经建立完毕。在建立过程当中，会从Dart SDK调用虚拟机函数，而后在新的Isolate对象中经过异步的方式调用入口函数。
注意：主Isolate的入口函数就是熟悉的main函数。
三、isolate之间的通讯原理
经过前面一节，知道了Dart是如何建立一个新的Isolate对象的。但也仍是省略了不少东西的，好比子Isolate通知父Isolate的原理，也就是Isolate间的通讯原理。
3.一、ReceivePort与SendPort
在Isolate给另一个Isolate发送消息以前，须要先来熟悉ReceivePort及SendPort。代码以下。
abstract class ReceivePort implements Stream {
//声明外部实现
external factory ReceivePort();
}
//在isolate_patch.dart中
@patch
class ReceivePort {
@patch
factory ReceivePort() => new _ReceivePortImpl();
@patch
factory ReceivePort.fromRawReceivePort(RawReceivePort rawPort) {
return new _ReceivePortImpl.fromRawReceivePort(rawPort);
}
}
class _ReceivePortImpl extends Stream implements ReceivePort {
_ReceivePortImpl() : this.fromRawReceivePort(new RawReceivePort());
_ReceivePortImpl.fromRawReceivePort(this._rawPort) {
_controller = new StreamController(onCancel: close, sync: true);
_rawPort.handler = _controller.add;
}
//返回一个SendPort对象
SendPort get sendPort {
return _rawPort.sendPort;
}
//监听发送的消息
StreamSubscription listen(void onData(var message),
{Function onError, void onDone(), bool cancelOnError}) {
return _controller.stream.listen(onData,
onError: onError, onDone: onDone, cancelOnError: cancelOnError);
}
...
}
@patch
class RawReceivePort {
@patch
factory RawReceivePort([Function handler]) {
_RawReceivePortImpl result = new _RawReceivePortImpl();
result.handler = handler;
return result;
}
}
@pragma("vm:entry-point")
class _RawReceivePortImpl implements RawReceivePort {
factory _RawReceivePortImpl() native "RawReceivePortImpl_factory";
...
SendPort get sendPort {
return _get_sendport();
}
...
/**** Internal implementation details ****/
_get_id() native "RawReceivePortImpl_get_id";
_get_sendport() native "RawReceivePortImpl_get_sendport";
...
}
复制代码
在代码中，一个ReceivePort对象包含一个RawReceivePort对象及SendPort对象。其中RawReceivePort对象是在虚拟机中建立的，它对应着虚拟机中的ReceivePort类。代码以下。
[->third_party/dart/runtime/lib.isolate.cc]
DEFINE_NATIVE_ENTRY(RawReceivePortImpl_factory, 0, 1) {
ASSERT(
TypeArguments::CheckedHandle(zone, arguments->NativeArgAt(0)).IsNull());
//建立一个Entry对象并返回一个端口号。
Dart_Port port_id = PortMap::CreatePort(isolate->message_handler());
//建立ReceivePort对象
return ReceivePort::New(port_id, false /* not control port */);
}
复制代码
在建立ReceivePort对象对象以前，首先会将当前Isolate中的MessageHandler对象添加到map中。这里是一个全局的map，在Dart VM初始化的时候建立，每一个元素都是一个Entry对象，在Entry中，有一个MessageHandler对象，一个端口号及该端口的状态。
typedef struct {
//端口号
Dart_Port port;
//消息处理器
MessageHandler* handler;
//端口号状态
PortState state;
} Entry;
复制代码
[->third_party/dart/runtime/vm/port.cc]
Dart_Port PortMap::CreatePort(MessageHandler* handler) {
...
Entry entry;
//分配一个端口号
entry.port = AllocatePort();
//设置消息处理器
entry.handler = handler;
//端口号状态
entry.state = kNewPort;
//查找当前entry的位置
intptr_t index = entry.port % capacity_;
Entry cur = map_[index];
// Stop the search at the first found unused (free or deleted) slot.
//找到空闲或将要被删除的Entry。
while (cur.port != 0) {
index = (index + 1) % capacity_;
cur = map_[index];
}
if (map_[index].handler == deleted_entry_) {
// Consuming a deleted entry.
deleted_--;
}
//插入到map中
map_[index] = entry;
// Increment number of used slots and grow if necessary.
used_++;
//检查是否须要扩容
MaintainInvariants();
...
//返回端口号
return entry.port;
}
复制代码
注意： 这里的map的初始容量是8，当达到容量的3/4时，会进行扩容，新的容量是旧的容量2倍。熟悉Java的就知道，这跟HashMap相似，初始容量为8，加载因子为0.75，扩容是指数级增加。
再来看ReceivePort对象的建立。
[->third_party/dart/runtime/vm/object.cc]
RawReceivePort* ReceivePort::New(Dart_Port id,
bool is_control_port,
Heap::Space space) {
Thread* thread = Thread::Current();
Zone* zone = thread->zone();
const SendPort& send_port =
//建立SendPort对象
SendPort::Handle(zone, SendPort::New(id, thread->isolate()->origin_id()));
ReceivePort& result = ReceivePort::Handle(zone);
{
//建立ReceivePort对象
RawObject* raw = Object::Allocate(ReceivePort::kClassId,//classId
ReceivePort::InstanceSize(),//对象大小
space);
NoSafepointScope no_safepoint;
result ^= raw;
result.StorePointer(&result.raw_ptr()->send_port_, send_port.raw());
}
if (is_control_port) {
//更新端口的状态，设为kControlPort
PortMap::SetPortState(id, PortMap::kControlPort);
} else {
//更新端口的状态，设为kLivePort
PortMap::SetPortState(id, PortMap::kLivePort);
}
return result.raw();
}
复制代码
[->third_party/dart/runtime/vm/object.cc]
RawSendPort* SendPort::New(Dart_Port id,
Dart_Port origin_id,
Heap::Space space) {
SendPort& result = SendPort::Handle();
{
//建立SendPort对象
RawObject* raw =
Object::Allocate(SendPort::kClassId, //classId
SendPort::InstanceSize(), //对象ID
space);
NoSafepointScope no_safepoint;
result ^= raw;
result.StoreNonPointer(&result.raw_ptr()->id_, id);
result.StoreNonPointer(&result.raw_ptr()->origin_id_, origin_id);
}
return result.raw();
}
复制代码
这里建立对象时传递的classId是在Isolate对象初始化时注册的，而后根据该classId来建立相应的对象。在这里，ReceivePort对应着Dart SDK中的_RawReceivePortImpl对象，SendPort对应着Dart SDK中的_SendPortImpl对象。
也就是当建立ReceivePort对象时，会经过Dart VM来建立对应的_RawReceivePortImpl对象及SendPort对应的_SendPortImpl对象。
3.二、isolate间通讯
当ReceivePort建立成功后，就能够经过调用_SendPortImpl的send函数来发送消息。
@pragma("vm:entry-point")
class _SendPortImpl implements SendPort {
...
/*--- public interface ---*/
@pragma("vm:entry-point", "call")
void send(var message) {
_sendInternal(message);
}
...
// Forward the implementation of sending messages to the VM.
void _sendInternal(var message) native "SendPortImpl_sendInternal_";
}
复制代码
_sendInternal的具体实如今Dart VM中。
[->third_party/dart/runtime/lib/isolate.cc]
DEFINE_NATIVE_ENTRY(SendPortImpl_sendInternal_, 0, 2) {
...
//目标Isolate所对应端口号
const Dart_Port destination_port_id = port.Id();
const bool can_send_any_object = isolate->origin_id() == port.origin_id();
if (ApiObjectConverter::CanConvert(obj.raw())) {//若是发送消息为null或者发送消息不是堆对象
PortMap::PostMessage(
Message::New(destination_port_id, obj.raw(), Message::kNormalPriority));
} else {
//建立一个MessageWriter对象——writer
MessageWriter writer(can_send_any_object);
// TODO(turnidge): Throw an exception when the return value is false?
PortMap::PostMessage(writer.WriteMessage(obj, destination_port_id,
Message::kNormalPriority));
}
return Object::null();
}
复制代码
[->third_party/dart/runtime/vm/port.cc]
bool PortMap::PostMessage(std::unique_ptr<Message> message,
bool before_events) {
MutexLocker ml(mutex_);
//在map中根据目标端口号寻找Entry所在的位置
intptr_t index = FindPort(message->dest_port());
if (index < 0) {
return false;
}
//从map中拿到Entry对象并取出MessageHandler对象
MessageHandler* handler = map_[index].handler;
//这里的handler是目标Isolate中的MessageHandler
handler->PostMessage(std::move(message), before_events);
return true;
}
复制代码
到这里就已经成功将消息加入到了目标Isolate的MessageHandler中，成功完成了Isolate间消息的传递，但还还没有对消息进行处理。
再来看Isolate对于消息的处理。
[->third_party/dart/runtime/vm/message_handler.cc]
void MessageHandler::PostMessage(std::unique_ptr<Message> message,
bool before_events) {
Message::Priority saved_priority;
{
MonitorLocker ml(&monitor_);
...
saved_priority = message->priority();
if (message->IsOOB()) {
//加入到OOB类型消息的队列中
oob_queue_->Enqueue(std::move(message), before_events);
} else {
//加入到普通消息队列中
queue_->Enqueue(std::move(message), before_events);
}
if (paused_for_messages_) {
ml.Notify();
}
if (pool_ != nullptr && !task_running_) {
task_running_ = true;
//异步处理
const bool launched_successfully = pool_->Run<MessageHandlerTask>(this);
}
}
// Invoke any custom message notification.
//若是自定义了消息通知函数，那么在消息处理完毕后会调用该函数
MessageNotify(saved_priority);
}
复制代码
在PostMessage中主要是作了如下操做。
- 根据消息级别将消息加入到不一样的队列中。主要有OOB消息及普通消息两个级别，OOB消息在队列
oob_queue_中，普通消息在队列queue_中。OOB消息级别高于普通消息，会当即处理。 - 将一个消息处理任务
MessageHandlerTask加入到线程中。
这里的线程池是在Dart VM建立的时候建立的，在Isolate运行时传递给MessageHandler的。
下面再来看MessageHandlerTask，在MessageHandlerTask的run函数中执行的是TaskCallback函数。
[->third_party/dart/runtime/vm/message_handler.cc]
void MessageHandler::TaskCallback() {
MessageStatus status = kOK;
bool run_end_callback = false;
bool delete_me = false;
EndCallback end_callback = NULL;
CallbackData callback_data = 0;
{
...
if (status == kOK) {
...
bool handle_messages = true;
while (handle_messages) {
handle_messages = false;
// Handle any pending messages for this message handler.
if (status != kShutdown) {
//处理消息
status = HandleMessages(&ml, (status == kOK), true);
}
if (status == kOK && HasLivePorts()) {
handle_messages = CheckIfIdleLocked(&ml);
}
}
}
...
}
...
}
复制代码
消息的处理是在HandleMessages函数中进行的。
[->third_party/dart/runtime/vm/message_handler.cc]
MessageHandler::MessageStatus MessageHandler::HandleMessages(
MonitorLocker* ml,
bool allow_normal_messages,
bool allow_multiple_normal_messages) {
...
//从队列中获取一个消息，优先OOB消息
std::unique_ptr<Message> message = DequeueMessage(min_priority);
//没有消息时退出循环，中止消息的处理
while (message != nullptr) {
//获取消息的长度
intptr_t message_len = message->Size();
...
//获取消息级别
Message::Priority saved_priority = message->priority();
Dart_Port saved_dest_port = message->dest_port();
MessageStatus status = kOK;
{
DisableIdleTimerScope disable_idle_timer(idle_time_handler);
//消息的处理
status = HandleMessage(std::move(message));
}
...
//若是是已关闭状态，将清除OOB类型消息
if (status == kShutdown) {
ClearOOBQueue();
break;
}
...
//继续从队列中获取消息
message = DequeueMessage(min_priority);
}
return max_status;
}
复制代码
在HandleMessages函数中会根据消息的优先级别来遍历全部消息并一一处理，直至处理完毕。具体消息处理是在HandleMessage函数中进行的。该函数在其子类IsolateMessageHandler中实现。
[->third_party/dart/runtime/vm/isolate.cc]
MessageHandler::MessageStatus IsolateMessageHandler::HandleMessage(
std::unique_ptr<Message> message) {
...
//若是是普通消息
if (!message->IsOOB() && (message->dest_port() != Message::kIllegalPort)) {
//调用Dart层的_lookupHandler函数，返回该函数在isolate_patch.dart中
msg_handler = DartLibraryCalls::LookupHandler(message->dest_port());
...
}
...
MessageStatus status = kOK;
if (message->IsOOB()) {//处理OOB消息
...
} else if (message->dest_port() == Message::kIllegalPort) {//处理OOB消息，主要是处理延迟OOB消息
...
} else {//处理普通消息
...
//调用Dart层的_RawReceivePortImpl对象中的_handleMessage函数，该函数在isolate_patch.dart中
const Object& result =
Object::Handle(zone, DartLibraryCalls::HandleMessage(msg_handler, msg));
if (result.IsError()) {
status = ProcessUnhandledException(Error::Cast(result));
} else {
...
}
}
return status;
}
复制代码
在这里先暂时无论OOB消息的处理，来看普通消息的处理。
- 首先调用Dart SDK中
_RawReceivePortImpl对象的_lookupHandler函数，返回一个在建立_RawReceivePortImpl对象时注册的一个自定义函数。 - 调用Dart SDK中
_RawReceivePortImpl对象的_handleMessage函数并传入1中返回的自定义函数，经过该自定义函数将消息分发出去。
至此，一个Isolate就已经成功的向另一个Isolate成功发送并接收消息。而双向通讯也很简单，在父Isolate中建立一个端口，并在建立子Isolate时，将这个端口传递给子Isolate。而后在子Isolate调用其入口函数时也建立一个新端口，并经过父Isolate传递过来的端口把子Isolate建立的端口传递给父Isolate，这样父Isolate与子Isolate分别拥有对方的一个端口号，从而实现了通讯。具体代码[见小节1.2]。
四、总结
主要是梳理了纯Dart环境中Isolate的建立、运行及通讯的实现，内容比较多且枯燥。能够发现，Isolate比Java、c/c++中的线程复杂多了，好比有本身的堆。固然，Isolate也不只仅只有上述的一些功能，还有代码的读取、解析等，后面再一一梳理。
【参考资料】
相关文章
- 1. 深刻了解Flutter的isolate(2) --- 建立本身的isolate
- 2. 深刻了解Flutter的isolate(3) --- Flutter的thread model（线程模型）
- 3. 深刻了解Flutter的isolate(4) --- 使用Compute写isolates
- 4. 理解v8的Isolate调度
- 5. 深刻理解JVM
- 6. 深刻理解volatile
- 7. 深刻理解DefaultMessageListenerContainer
- 8. 深刻理解vue
- 9. 深刻理解Volatile
- 10. 深刻理解RunLoop
- 更多相关文章...
- • SQLite Explain（解释） - SQLite教程
- • 错误处理 - RUST 教程
- • 算法总结-深度优先算法
- • Docker 清理命令
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
欢迎关注本站公众号,获取更多信息
