OpenHarmony开发者论坛

标题: Worker源码走读 [打印本页]

作者: 中移-zhangzhou 时间: 2024-7-4 11:02
标题: Worker源码走读
# 前言

Worker主要作用是为应用程序提供一个多线程的运行环境，可满足应用程序在执行过程中与主线程分离，在后台线程中运行一个脚本操作耗时操作，极大避免类似于计算密集型或高延迟的任务阻塞主线程的运行。在js中也有Worker实现多线程，OpenHarmony中的Worker从原理到用法也是对js中的Worker继承。

# 构造初始化

```
##worker.cpp

//ThreadWorker
napi_value Worker::ThreadWorkerConstructor(napi_env env, napi_callback_info cbinfo)
{
if (CanCreateWorker(env, WorkerVersion::NEW)) {
      return Constructor(env, cbinfo, false, WorkerVersion::NEW);
}
return nullptr;
}

//Worker（deprecated）
napi_value Worker::WorkerConstructor(napi_env env, napi_callback_info cbinfo)
{
if (CanCreateWorker(env, WorkerVersion::OLD)) {
      return Constructor(env, cbinfo, false, WorkerVersion::OLD);
}
return nullptr;
}
```

```
napi_value Worker::Constructor(napi_env env, napi_callback_info cbinfo, bool limitSign, WorkerVersion version)
{
//参数校验
......

HILOG_INFO("worker:: Worker start constructor");
Worker* worker = nullptr;
if (limitSign) {
      //RestrictedWorker的处理逻辑
      ......
} else {
      //最大worker数
      int maxWorkers = (version == WorkerVersion::NEW) ? MAX_THREAD_WORKERS : MAX_WORKERS;
#if defined(OHOS_PLATFORM)
      maxWorkers = OHOS::system::GetIntParameter<int>("persist.commonlibrary.maxworkers", maxWorkers);
#endif
      std::lock_guard<std::mutex> lock(g_workersMutex);
      // 当前worker数达到最大限制worker数，抛出异常
      if (static_cast<int>(g_workers.size()) >= maxWorkers) {
         HILOG_ERROR("worker:: the number of workers exceeds the maximum");
         ErrorHelper::ThrowError(env,
            ErrorHelper::ERR_WORKER_INITIALIZATION, "the number of workers exceeds the maximum.");
         return nullptr;
      }

      // 2. new worker instance
      worker = new Worker(env, nullptr);
   ......
      // worker实例加入list
      g_workers.push_back(worker);
}
......

// 3. execute in thread
char* script = NapiHelper::GetString(env, args[0]);
if (script == nullptr) {
      HILOG_ERROR("worker:: the file path is invaild, maybe path is null");
      ErrorHelper::ThrowError(env,
         ErrorHelper::ERR_WORKER_INVALID_FILEPATH, "the file path is invaild, maybe path is null.");
      return nullptr;
}
napi_wrap(
      env, thisVar, worker,
      // js对象准备gc时，终止worker
      [](napi_env env, void* data, void* hint) {
         Worker* worker = reinterpret_cast<Worker*>(data);
         {
            std::lock_guard<std::recursive_mutex> lock(worker->liveStatusLock_);
            if (worker->UpdateHostState(INACTIVE)) {
#if defined(ENABLE_WORKER_EVENTHANDLER)
                  if (!worker->isMainThreadWorker_) {
                     worker->CloseHostHandle();
                  }
#else
                  worker->CloseHostHandle();
#endif
                  worker->ReleaseHostThreadContent();
            }
            if (!worker->IsRunning()) {
                  HILOG_DEBUG("worker:: worker is not in running");
                  return;
            }
            worker->TerminateInner();
         }
      },
      nullptr, &worker->workerRef_);
// worker线程启动执行
worker->StartExecuteInThread(env, script);
return thisVar;
}

```

启动worker线程

```
void Worker::StartExecuteInThread(napi_env env, const char* script)
{
HILOG_INFO("worker:: Start execute in the thread!");
// 1. init hostHandle in host loop
uv_loop_t* loop = NapiHelper::GetLibUV(env);
if (loop == nullptr) {
      ErrorHelper::ThrowError(env, ErrorHelper::ERR_WORKER_NOT_RUNNING, "engine loop is null");
      CloseHelp:

eletePointer(script, true);
      return;
}
GetContainerScopeId(env);
//初始化uv句柄，宿主的loop handle
#if defined(ENABLE_WORKER_EVENTHANDLER)
if (!OHOS::AppExecFwk::EventRunner::IsAppMainThread()) {
      isMainThreadWorker_ = false;
      InitHostHandle(loop);
} else {
      HILOG_DEBUG("worker:: eventrunner should be nullptr if the current thread is not the main thread");
}
#else
InitHostHandle(loop);
#endif

// 2. copy the script
//寻找worker线程文件
script_ = std::string(script);
// isBundle : FA mode and BundlePack.
bool isBundle = reinterpret_cast<NativeEngine*>(env)->GetIsBundle();
// if worker file is packed in har, need find moduleName in hostVM, and concat new recordName.
bool isHar = script_.find_first_of(PathHelper::NAME_SPACE_TAG) == 0;
if ((isHar && script_.find(PathHelper:

REFIX_BUNDLE) == std::string::npos) ||
(!isBundle && script_.find_first_of(PathHelper:

OINT_TAG) == 0)) {
      PathHelper::ConcatFileNameForWorker(env, script_, fileName_, isRelativePath_);
      HILOG_DEBUG("worker:: Concated worker recordName: %{public}s, fileName: %{public}s",
                  script_.c_str(), fileName_.c_str());
}
// check the path is vaild.
bool isNormalizedOhmUrlPack = reinterpret_cast<NativeEngine*>(env)->GetIsNormalizedOhmUrlPack();
if (!isNormalizedOhmUrlPack && !isBundle) {
      if (!PathHelper::CheckWorkerPath(env, script_, isHar, isRelativePath_)) {
         HILOG_ERROR("worker:: the file path is invaild, can't find the file : %{public}s.", script);
         CloseHelp:

eletePointer(script, true);
         ErrorHelper::ThrowError(env, ErrorHelper::ERR_WORKER_INVALID_FILEPATH,
            "the file path is invaild, can't find the file.");
         return;
      }
}

// 3. create WorkerRunner to Execute
if (!runner_) {
      // 创建WorkerRunner线程，线程启动后回调ExecuteThread
      runner_ = std::make_unique<WorkerRunner>(WorkerStartCallback(ExecuteInThread, this));
}
if (runner_) {
      //Worker线程启动
      runner_->Execute(); // start a new thread
} else {
      HILOG_ERROR("runner_ is nullptr");
}
CloseHelp:

eletePointer(script, true);
}
```

Worker线程创建后执行Executethread

```
void Worker::ExecuteInThread(const void* data)
{
HITRACE_HELPER_START_TRACE(__PRETTY_FUNCTION__);
HILOG_INFO("worker:: Execute in the thread!");
auto worker = reinterpret_cast<Worker*>(const_cast<void*>(data));
// 1. create a runtime, nativeengine
napi_env workerEnv = nullptr;
{
      std::lock_guard<std::recursive_mutex> lock(worker->liveStatusLock_);
      if (worker->HostIsStop()) {
         HILOG_ERROR("worker:: host thread is stop");
         CloseHelp:

eletePointer(worker, false);
         return;
      }
      napi_env env = worker->GetHostEnv();
      //创建runtime，为worker创建ArkNativeEngine
      napi_create_runtime(env, &workerEnv);
      if (workerEnv == nullptr) {
         HILOG_ERROR("worker:: Worker create runtime error");
         ErrorHelper::ThrowError(env, ErrorHelper::ERR_WORKER_NOT_RUNNING, "Worker create runtime error");
         return;
      }
         // mark worker env is workerThread
reinterpret_cast<NativeEngine*>(workerEnv)->MarkWorkerThread();
      // for load balance in taskpool
      reinterpret_cast<NativeEngine*>(env)->IncreaseSubEnvCounter();

      worker->SetWorkerEnv(workerEnv);
}
//根据workerEnv得到worker线程的loop
uv_loop_t* loop = worker->GetWorkerLoop();
if (loop == nullptr) {
      HILOG_ERROR("worker:: Worker loop is nullptr");
      return;
}

// 2. add some preparation for the worker
//执行worker初始化操作
if (worker->

repareForWorkerInstance()) {
      worker->workerOnMessageSignal_ = new uv_async_t;
      // worker loop，初始化workerOnMessageSignal_ handle，handle回调执行WorkerOnMessage
      uv_async_init(loop, worker->workerOnMessageSignal_, reinterpret_cast<uv_async_cb>(Worker::WorkerOnMessage));
      worker->workerOnMessageSignal_->data = worker;
#if !defined(WINDOWS_PLATFORM) && !defined(MAC_PLATFORM)
      uv_async_init(loop, &worker->debuggerOnPostTaskSignal_, reinterpret_cast<uv_async_cb>(
         Worker::HandleDebuggerTask));
#endif
      //worker状态置为RUNNING
      worker->UpdateWorkerState(RUNNING);
      // in order to invoke worker send before subThread start
      uv_async_send(worker->workerOnMessageSignal_);
      HITRACE_HELPER_FINISH_TRACE;
      //运行事件循环
      // 3. start worker loop
      worker->Loop();
} else {
      HILOG_ERROR("worker:: worker PrepareForWorkerInstance fail");
      worker->UpdateWorkerState(TERMINATED);
      HITRACE_HELPER_FINISH_TRACE;
}
// worker初始化失败或worker的loop结束，释放worker线程资源：删除监听、端口，清空worker消息队列
worker->ReleaseWorkerThreadContent();
std::lock_guard<std::recursive_mutex> lock(worker->liveStatusLock_);
//如果宿主stop，直接释放worker；否则向宿主消息队列添加nullptr，宿主接收信号后关闭回调
if (worker->HostIsStop()) {
      HILOG_INFO("worker:: host is stopped");
      CloseHelp:

eletePointer(worker, false);
} else {
worker->

ublishWorkerOverSignal();
}
}

```

执行worker初始化操作

```
bool Worker:

repareForWorkerInstance()
{
std::string rawFileName = script_;
uint8_t* scriptContent = nullptr;
size_t scriptContentSize = 0;
std::vector<uint8_t> content;
std::string workerAmi;
{
      std::lock_guard<std::recursive_mutex> lock(liveStatusLock_);
      if (HostIsStop()) {
         HILOG_INFO("worker:: host is stopped");
         return false;
      }
      // 1. init worker async func
      auto workerEngine = reinterpret_cast<NativeEngine*>(workerEnv_);

      auto hostEngine = reinterpret_cast<NativeEngine*>(hostEnv_);
      // 2. init worker environment
#if !defined(WINDOWS_PLATFORM) && !defined(MAC_PLATFORM)
      workerEngine->SetDebuggerPostTaskFunc(
         std::bind(&Worker:

ebuggerOnPostTask, this, std::placeholders::_1));
#endif
      if (!hostEngine->CallInitWorkerFunc(workerEngine)) {
         HILOG_ERROR("worker:: CallInitWorkerFunc error");
         return false;
      }
      // 3. get uril content
      if (isRelativePath_) {
         rawFileName = fileName_;
      }
      //worker线程文件
      if (!hostEngine->GetAbcBuffer(rawFileName.c_str(), &scriptContent, &scriptContentSize, content, workerAmi)) {
         HILOG_ERROR("worker:: GetAbcBuffer error");
         return false;
      }
}
// add timer interface
Timer::RegisterTime(workerEnv_);
HILOG_DEBUG("worker:: stringContent size is %{public}zu", scriptContentSize);
napi_value execScriptResult = nullptr;
//run actor，JSNApi执行worker线程文件
napi_status status = napi_run_actor(workerEnv_, scriptContent, scriptContentSize,
      workerAmi.c_str(), &execScriptResult, const_cast<char*>(script_.c_str()));
if (status != napi_ok || execScriptResult == nullptr) {
      // An exception occurred when running the script.
      HILOG_ERROR("worker:: run script exception occurs, will handle exception");
      HandleException();
      return false;
}

// 4. register worker name in DedicatedWorkerGlobalScope
if (!uv_async_init_.empty()) {
      napi_value nameValue = nullptr;
      napi_create_string_utf8(workerEnv_, name_.c_str(), name_.length(), &nameValue);
      NapiHelper::SetNamePropertyInGlobal(workerEnv_, "name", nameValue);
}
return true;
}

```

# 消息传递

宿主线程向worker线程发送消息PostMessage

```
napi_value Worker:

ostMessage(napi_env env, napi_callback_info cbinfo)
{
HITRACE_HELPER_METER_NAME(__PRETTY_FUNCTION__);
return CommonPostMessage(env, cbinfo, true);
}

napi_value Worker::CommonPostMessage(napi_env env, napi_callback_info cbinfo, bool cloneSendable)
{
......
if (worker == nullptr || worker->IsTerminated() || worker->IsTerminating()) {
      HILOG_ERROR("worker:: worker is nullptr when PostMessage, maybe worker is terminated");
      WorkerThrowError(env, ErrorHelper::ERR_WORKER_NOT_RUNNING, "maybe worker is terminated when PostMessage");
      return nullptr;
}
//序列化数据
MessageDataType data = nullptr;
napi_status serializeStatus = napi_ok;
bool defaultClone = cloneSendable ? true : false;
napi_value undefined = NapiHelper::GetUndefinedValue(env);
if (argc >= NUM_WORKER_ARGS) {
      if (!NapiHelper::IsArray(env, argv[1])) {
         ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR, "transfer list must be an Array");
         return nullptr;
      }
      serializeStatus = napi_serialize_inner(env, argv[0], argv[1], undefined, false, defaultClone, &data);
} else {
      serializeStatus = napi_serialize_inner(env, argv[0], undefined, undefined, false, defaultClone, &data);
}
if (serializeStatus != napi_ok || data == nullptr) {
      worker->HostOnMessageErrorInner();
      WorkerThrowError(env, ErrorHelper::ERR_WORKER_SERIALIZATION, "failed to serialize message.");
      return nullptr;
}
worker->

ostMessageInner(data);
return NapiHelper::GetUndefinedValue(env);
}

void Worker:

ostMessageInner(MessageDataType data)
{
if (IsTerminated()) {
      HILOG_DEBUG("worker:: worker has been terminated when PostMessageInner.");
      return;
}
// 加锁数据加入worker消息队列
workerMessageQueue_.EnQueue(data);
std::lock_guard<std::mutex> lock(workerOnmessageMutex_);
if (workerOnMessageSignal_ != nullptr && uv_is_active((uv_handle_t*)workerOnMessageSignal_)) {
      //uv_async_send通知worker线程执行WorkerOnMessage
      uv_async_send(workerOnMessageSignal_);
}
}

```

worker loop执行WorkerOnMessage

```
void Worker::WorkerOnMessage(const uv_async_t* req)
{
HITRACE_HELPER_METER_NAME(__PRETTY_FUNCTION__);
Worker* worker = static_cast<Worker*>(req->data);
if (worker == nullptr) {
      HILOG_ERROR("worker::worker is null");
      return;
}
worker->WorkerOnMessageInner();
}

void Worker::WorkerOnMessageInner()
{
if (IsTerminated()) {
      return;
}
napi_status status;
napi_handle_scope scope = nullptr;
status = napi_open_handle_scope(workerEnv_, &scope);
if (status != napi_ok || scope == nullptr) {
      HILOG_ERROR("worker:: WorkerOnMessage open handle scope failed.");
      return;
}
MessageDataType data = nullptr;
//worker线程没有终结且worker消息队列有数据，处理出列数据
while (!IsTerminated() && workerMessageQueue_.DeQueue(&data)) {
      // nullptr表明宿主发送的是终止信号
      if (data == nullptr) {
         HILOG_DEBUG("worker:: worker reveive terminate signal");
         // Close handlescope need before TerminateWorker
         napi_close_handle_scope(workerEnv_, scope);
         TerminateWorker();
         return;
      }
      //反序列化数据
      napi_value result = nullptr;
      status = napi_deserialize(workerEnv_, data, &result);
      napi_delete_serialization_data(workerEnv_, data);
      if (status != napi_ok || result == nullptr) {
         WorkerOnMessageErrorInner();
         continue;
      }

      napi_value event = nullptr;
      napi_create_object(workerEnv_, &event);
      napi_set_named_property(workerEnv_, event, "data", result);
      napi_value argv[1] = { event };
      //调用worker线程文件中定义的onmessagee方法
      CallWorkerFunction(1, argv, "onmessage", true);

      napi_value obj = NapiHelper::GetReferenceValue(workerEnv_, this->workerPort_);
      ParentPortHandleEventListeners(workerEnv_, obj, 1, argv, "message", true);
}
napi_close_handle_scope(workerEnv_, scope);
}

```

# 关闭worker

```
napi_value Worker::Terminate(napi_env env, napi_callback_info cbinfo)
{
HITRACE_HELPER_METER_NAME(__PRETTY_FUNCTION__);
napi_value thisVar = nullptr;
napi_get_cb_info(env, cbinfo, nullptr, nullptr, &thisVar, nullptr);
Worker* worker = nullptr;
napi_unwrap(env, thisVar, reinterpret_cast<void**>(&worker));
if (worker == nullptr) {
      HILOG_ERROR("worker:: worker is nullptr when Terminate, maybe worker is terminated");
      WorkerThrowError(env, ErrorHelper::ERR_WORKER_NOT_RUNNING, "worker is nullptr when Terminate");
      return nullptr;
}
if (worker->IsTerminated() || worker->IsTerminating()) {
      HILOG_DEBUG("worker:: worker is not in running when Terminate");
      return nullptr;
}
worker->TerminateInner();
return NapiHelper::GetUndefinedValue(env);
}

void Worker::TerminateInner()
{
if (IsTerminated() || IsTerminating()) {
      HILOG_INFO("worker:: worker is not in running when TerminateInner");
      return;
}
//worker状态置为TERMINATEING
// 1. Update State
UpdateWorkerState(TERMINATEING);
//发送nullptr信号，worker线程接收到后执行TerminateWorker
// 2. send null signal
PostMessageInner(nullptr);
}

```

# 总结

简单的worker的创建、消息向worker线程传递、销毁流程：

1.通过NAPI，ts层的worker.ThreadWorker、worker.workertPort和c++层的worker相互调用；

2.构造时创建worker实例，初始化宿主线程loop uv句柄；加载worker线程文件；

3‘创建并启动worker线程，worker状态置为RUNNING，运行worker loop事件循环；

4.ts层宿主调用ThreadWorker向PostMessage向worker线程传递消息，数据包装后加入worker线程消息队列，uv发送消息；

5.worker线程接收消息，从消息队列取数据，调用worker线程文件中的onmessage方法执行；

6.宿主调用terminate方法终结worker，worker状态置为TERMINATEING，nullptr加入消息队列，uv发送消息；

7.worker线程接收消息，从消息队列取数据为nullptr，worker线程关闭事件循环，worker状态置为TERMINATED。

欢迎光临 OpenHarmony开发者论坛 (https://forums.openharmony.cn/)