深刻理解Isolate

時間  2020-01-22
標籤 深刻 理解 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函數比較重要，主要作了如下事情。

  1. 建立Isolate對象
  2. 建立Isolate中的堆空間，在Isolate僅有一塊堆空間。存在堆空間也就會存在對象分配、垃圾回收等。
  3. 將Isolate對象與一個線程進行關聯，也就是能夠說一個線程對應着一個Isolate對象。
  4. 設置消息處理器（IsolateMessageHandler），主要是對於Isolate中的消息處理。子Isolate能夠經過端口向父Isolate的MessageHandler中添加消息，反之亦然。這也是Isolate間的通訊的實現。
  5. 設置api state，暫時沒搞懂這個是幹啥的。
  6. 設置主端口。
  7. 將當前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函數中主要是作了如下幾件事。

  1. 告訴父Isolate，子Isolate已經建立成功。
  2. 調用子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中主要是作了如下操做。

  1. 根據消息級別將消息加入到不一樣的隊列中。主要有OOB消息及普通消息兩個級別，OOB消息在隊列oob_queue_中，普通消息在隊列queue_中。OOB消息級別高於普通消息，會當即處理。
  2. 將一個消息處理任務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消息的處理，來看普通消息的處理。

  1. 首先調用Dart SDK中_RawReceivePortImpl對象的_lookupHandler函數，返回一個在建立_RawReceivePortImpl對象時註冊的一個自定義函數。
  2. 調用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也不只僅只有上述的一些功能，還有代碼的讀取、解析等，後面再一一梳理。

【參考資料】

Glossary of VM Terms

Dart asynchronous programming: Isolates and event loops

Introduction to Dart VM

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程