通过N-API使用Libuv线程池

theanarkh

发布于 2021-07-08 16:03:36

8590

发布于 2021-07-08 16:03:36

文章被收录于专栏：原创分享

Node.js不适合处理耗时操作是一直存在的问题，为此Node.js提供了三种解决方案。

1 子进程

2 子线程 3 Libuv线程池

前两种是开发效率比较高的，因为我们只需要写js。但是也有些缺点

1 执行js的成本

2 虽然可以间接使用Libuv线程池，但是受限于Node.js提供的API。

3 无法利用c/c++层提供的解决方案（内置或业界的）。

这时候我们可以尝试第三种解决方案。直接通过N-API使用Libuv线程池。下面我们看看这么做。N-API提供了几个API。

napi_create_async_work // 创建一个worr，但是还没有执行
napi_delete_async_work // 释放上面创建的work的内存
napi_queue_async_work // 往Libuv提交一个work
napi_cancel_async_work // 取消Libuv中的任务，如果已经在执行则无法取消

接下来我们看看如何通过N-API使用Libuv线程池。首先看看js层。

const { submitWork } = require('./build/Release/test.node');
submitWork((sum) => {
    console.log(sum)
})

js提交一个任务，然后传入一个回调。接着看看N-API的代码。

napi_value Init(napi_env env, napi_value exports) {
  napi_value func;
  napi_create_function(env,
                      NULL,
                      NAPI_AUTO_LENGTH,
                      submitWork,
                      NULL,
                      &func);
  napi_set_named_property(env, exports, "submitWork", func);
  return exports;
}

NAPI_MODULE(NODE_GYP_MODULE_NAME, Init)

首先定义导出的函数，接着看核心逻辑。

1 定义一个结构体保存上下文

struct info
{
  int sum; // 保存计算结果
  napi_ref func; // 保存回调
  napi_async_work worker; // 保存work对象
};

2 提交任务到Libuv

static napi_value submitWork(napi_env env, napi_callback_info info) {
  napi_value resource_name;
  napi_status status;
  
  size_t argc = 1;
  napi_value args[1];
  struct info data = {0, nullptr, nullptr};
  struct info * ptr = &data;
  status = napi_get_cb_info(env, info, &argc, args, NULL, NULL);
  if (status != napi_ok) {
    goto done;
  }
  napi_create_reference(env, args[0], 1, &ptr->func);
  status = napi_create_string_utf8(env,"test", NAPI_AUTO_LENGTH, &resource_name);
  if (status != napi_ok) {
    goto done;
  }
  // 创建一个work，ptr保存的上下文会在work函数和done函数里使用
  status = napi_create_async_work(env, nullptr, resource_name, work, done, (void *) ptr, &ptr->worker);
  if (status != napi_ok) {
    goto done;
  }
  // 提及work到Libuv
  status = napi_queue_async_work(env, ptr->worker);

  done: 
    napi_value ret;
    napi_create_int32(env, status == napi_ok ? 0 : -1, &ret);
    return  ret;
}

执行上面的函数，任务就会被提交到Libuv线程池了。

3 Libuv子线程执行任务

void work(napi_env env, void* data) {
  struct info *arg = (struct info *)data;
  printf("doing...\n");
  int sum = 0;
  for (int i = 0; i < 10; i++) {
    sum += i;
  }
  arg->sum = sum;
}

很简单，计算几个数。并且保存结果。

4 回调js

void done(napi_env env, napi_status status, void* data) {
  struct info *arg = (struct info *)data;
  if (status == napi_cancelled) {
    printf("cancel...");
  } else if (status == napi_ok) {
    printf("done...\n");
    napi_value callback;
    napi_value global;  
    napi_value result;
    napi_value sum;
    // 拿到结果
    napi_create_int32(env, arg->sum, &sum);
    napi_get_reference_value(env, arg->func, &callback);
    napi_get_global(env, &global);
    // 回调js
    napi_call_function(env, global, callback, 1, &sum, &result);
    // 清理
    napi_delete_reference(env, arg->func);
    napi_delete_async_work(env, arg->worker);
  }
}

并且执行后，我们看到输出了45。接下来我们分析大致的过程。首先我呢看看ThreadPoolWork，ThreadPoolWork是对Libuv work的封装。

class ThreadPoolWork {
 public:
  explicit inline ThreadPoolWork(Environment* env) : env_(env) {
    CHECK_NOT_NULL(env);
  }
  inline virtual ~ThreadPoolWork() = default;

  inline void ScheduleWork();
  inline int CancelWork();

  virtual void DoThreadPoolWork() = 0;
  virtual void AfterThreadPoolWork(int status) = 0;

  Environment* env() const { return env_; }

 private:
  Environment* env_;
  uv_work_t work_req_;
};

类的定义很简单，主要是封装了uv_work_t。我们看看每个函数的意义。DoThreadPoolWork和AfterThreadPoolWork是虚函数，由子类实现，我们一会看子类的时候再分析。我们看看ScheduleWork

void ThreadPoolWork::ScheduleWork() {
  env_->IncreaseWaitingRequestCounter();
  int status = uv_queue_work(
      env_->event_loop(),
      &work_req_,
      // Libuv子线程里执行的任务函数
      [](uv_work_t* req) {
        ThreadPoolWork* self = ContainerOf(&ThreadPoolWork::work_req_, req);
        self->DoThreadPoolWork();
      },
      // 任务处理完后的回调
      [](uv_work_t* req, int status) {
        ThreadPoolWork* self = ContainerOf(&ThreadPoolWork::work_req_, req);
        self->env_->DecreaseWaitingRequestCounter();
        self->AfterThreadPoolWork(status);
      });
  CHECK_EQ(status, 0);
}

ScheduleWork是负责给Libuv提交任务的函数。接着看看CancelWork。

int ThreadPoolWork::CancelWork() {
  return uv_cancel(reinterpret_cast<uv_req_t*>(&work_req_));
}

直接调用Libuv的函数取消任务。看完父类，我们看看子类的定义，子类在N-API里实现。

class Work : public node::AsyncResource, public node::ThreadPoolWork {
 private:
  explicit Work(node_napi_env env,
                v8::Local<v8::Object> async_resource,
                v8::Local<v8::String> async_resource_name,
                napi_async_execute_callback execute,
                napi_async_complete_callback complete = nullptr,
                void* data = nullptr)
    : AsyncResource(env->isolate,
                    async_resource,
                    *v8::String::Utf8Value(env->isolate, async_resource_name)),
      ThreadPoolWork(env->node_env()),
      _env(env),
      _data(data),
      _execute(execute),
      _complete(complete) {
  }

  ~Work() override = default;

 public:
  static Work* New(node_napi_env env,
                   v8::Local<v8::Object> async_resource,
                   v8::Local<v8::String> async_resource_name,
                   napi_async_execute_callback execute,
                   napi_async_complete_callback complete,
                   void* data) {
    return new Work(env, async_resource, async_resource_name,
                    execute, complete, data);
  }
  // 释放该类对象的内存
  static void Delete(Work* work) {
    delete work;
  }
  // 执行用户设置的函数
  void DoThreadPoolWork() override {
    _execute(_env, _data);
  }

  void AfterThreadPoolWork(int status) override {
   // 执行用户设置的回调
    _complete(env, ConvertUVErrorCode(status), _data);
  }

 private:
  node_napi_env _env;
  // 用户设置的数据，用于保存执行结果等
  void* _data;
  // 执行任务的函数
  napi_async_execute_callback _execute;
  // 任务处理完的回调
  napi_async_complete_callback _complete;
};

在Work类我们看到了虚函数DoThreadPoolWork和AfterThreadPoolWork的实现，没有太多逻辑。最后我们看看N-API提供的API的实现。

1 napi_create_async_work

napi_status napi_create_async_work(napi_env env,
                                   napi_value async_resource,
                                   napi_value async_resource_name,
                                   napi_async_execute_callback execute,
                                   napi_async_complete_callback complete,
                                   void* data,
                                   napi_async_work* result) {
  v8::Local<v8::Context> context = env->context();

  v8::Local<v8::Object> resource;
  if (async_resource != nullptr) {
    CHECK_TO_OBJECT(env, context, resource, async_resource);
  } else {
    resource = v8::Object::New(env->isolate);
  }

  v8::Local<v8::String> resource_name;
  CHECK_TO_STRING(env, context, resource_name, async_resource_name);

  uvimpl::Work* work = uvimpl::Work::New(reinterpret_cast<node_napi_env>(env),
                                         resource,
                                         resource_name,
                                         execute,
                                         complete,
                                         data);

  *result = reinterpret_cast<napi_async_work>(work);

  return napi_clear_last_error(env);
}

napi_create_async_work本质上是对Work的简单封装，创建一个Work并返回给用户。

2 napi_delete_async_work

napi_status napi_delete_async_work(napi_env env, napi_async_work work) {
  CHECK_ENV(env);
  CHECK_ARG(env, work);

  uvimpl::Work::Delete(reinterpret_cast<uvimpl::Work*>(work));

  return napi_clear_last_error(env);
}

napi_delete_async_work用于任务执行完后释放Work对应的内存。

3 napi_queue_async_work

napi_status napi_queue_async_work(napi_env env, napi_async_work work) {
  CHECK_ENV(env);
  CHECK_ARG(env, work);

  napi_status status;
  uv_loop_t* event_loop = nullptr;
  status = napi_get_uv_event_loop(env, &event_loop);
  if (status != napi_ok)
    return napi_set_last_error(env, status);

  uvimpl::Work* w = reinterpret_cast<uvimpl::Work*>(work);

  w->ScheduleWork();

  return napi_clear_last_error(env);
}

napi_queue_async_work是对ScheduleWork的封装，作用是给Libuv线程池提交任务。

4 napi_cancel_async_work

napi_status napi_cancel_async_work(napi_env env, napi_async_work work) {
  CHECK_ENV(env);
  CHECK_ARG(env, work);

  uvimpl::Work* w = reinterpret_cast<uvimpl::Work*>(work);

  CALL_UV(env, w->CancelWork());

  return napi_clear_last_error(env);
}

napi_cancel_async_work是对CancelWork的封装，即取消Libuv线程池的任务。我们看到一层层套，没有太多逻辑，主要是要符合N-API的规范。

总结：通过N-API提供的API，使得我们不再受限于Nod.js本身提供的一些异步接口（使用Libuv线程池的接口），而是直接使用Libuv线程池，这样我们不仅可以自己写c/c++，还可以复用业界的一些解决方案解决Node.js里的一些耗时任务。

仓库：https://github.com/theanarkh/learn-to-write-nodejs-addons

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2021-06-05，如有侵权请联系 cloudcommunity@tencent.com 删除

api