1 /*
2 * Copyright (c) 2022 Huawei Device Co., Ltd.
3 * Licensed under the Apache License, Version 2.0 (the "License");
4 * you may not use this file except in compliance with the License.
5 * You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software
10 * distributed under the License is distributed on an "AS IS" BASIS,
11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 * See the License for the specific language governing permissions and
13 * limitations under the License.
14 */
15
16 #include "taskpool.h"
17
18 #include <cinttypes>
19
20 #include "helper/error_helper.h"
21 #include "helper/hitrace_helper.h"
22 #include "helper/napi_helper.h"
23 #include "helper/object_helper.h"
24 #include "message_queue.h"
25 #include "task_manager.h"
26 #include "tools/log.h"
27 #include "worker.h"
28
29 namespace Commonlibrary::Concurrent::TaskPoolModule {
30 using namespace Commonlibrary::Concurrent::Common::Helper;
31
InitTaskPool(napi_env env,napi_value exports)32 napi_value TaskPool::InitTaskPool(napi_env env, napi_value exports)
33 {
34 HILOG_INFO("taskpool:: Import taskpool");
35 HITRACE_HELPER_METER_NAME(__PRETTY_FUNCTION__);
36 napi_value taskClass = nullptr;
37 napi_define_class(env, "Task", NAPI_AUTO_LENGTH, Task::TaskConstructor, nullptr, 0, nullptr, &taskClass);
38 napi_value longTaskClass = nullptr;
39 napi_define_class(env, "LongTask", NAPI_AUTO_LENGTH, Task::LongTaskConstructor,
40 nullptr, 0, nullptr, &longTaskClass);
41 napi_value genericsTaskClass = nullptr;
42 napi_define_class(env, "GenericsTask", NAPI_AUTO_LENGTH, Task::TaskConstructor,
43 nullptr, 0, nullptr, &genericsTaskClass);
44 napi_value isCanceledFunc = nullptr;
45 napi_create_function(env, "isCanceled", NAPI_AUTO_LENGTH, Task::IsCanceled, NULL, &isCanceledFunc);
46 napi_set_named_property(env, taskClass, "isCanceled", isCanceledFunc);
47 napi_set_named_property(env, longTaskClass, "isCanceled", isCanceledFunc);
48 napi_value sendDataFunc = nullptr;
49 napi_create_function(env, "sendData", NAPI_AUTO_LENGTH, Task::SendData, NULL, &sendDataFunc);
50 napi_set_named_property(env, taskClass, "sendData", sendDataFunc);
51 napi_set_named_property(env, longTaskClass, "sendData", sendDataFunc);
52 napi_value taskGroupClass = nullptr;
53 napi_define_class(env, "TaskGroup", NAPI_AUTO_LENGTH, TaskGroup::TaskGroupConstructor, nullptr, 0, nullptr,
54 &taskGroupClass);
55 napi_value seqRunnerClass = nullptr;
56 napi_define_class(env, "SequenceRunner", NAPI_AUTO_LENGTH, SequenceRunner::SeqRunnerConstructor,
57 nullptr, 0, nullptr, &seqRunnerClass);
58
59 // define priority
60 napi_value priorityObj = NapiHelper::CreateObject(env);
61 napi_value highPriority = NapiHelper::CreateUint32(env, Priority::HIGH);
62 napi_value mediumPriority = NapiHelper::CreateUint32(env, Priority::MEDIUM);
63 napi_value lowPriority = NapiHelper::CreateUint32(env, Priority::LOW);
64 napi_value idlePriority = NapiHelper::CreateUint32(env, Priority::IDLE);
65 napi_property_descriptor exportPriority[] = {
66 DECLARE_NAPI_PROPERTY("HIGH", highPriority),
67 DECLARE_NAPI_PROPERTY("MEDIUM", mediumPriority),
68 DECLARE_NAPI_PROPERTY("LOW", lowPriority),
69 DECLARE_NAPI_PROPERTY("IDLE", idlePriority),
70 };
71 napi_define_properties(env, priorityObj, sizeof(exportPriority) / sizeof(exportPriority[0]), exportPriority);
72
73 // define State
74 napi_value stateObj = NapiHelper::CreateObject(env);
75 napi_value waitingState = NapiHelper::CreateUint32(env, ExecuteState::WAITING);
76 napi_value runningState = NapiHelper::CreateUint32(env, ExecuteState::RUNNING);
77 napi_value canceledState = NapiHelper::CreateUint32(env, ExecuteState::CANCELED);
78 napi_property_descriptor exportState[] = {
79 DECLARE_NAPI_PROPERTY("WAITING", waitingState),
80 DECLARE_NAPI_PROPERTY("RUNNING", runningState),
81 DECLARE_NAPI_PROPERTY("CANCELED", canceledState),
82 };
83 napi_define_properties(env, stateObj, sizeof(exportState) / sizeof(exportState[0]), exportState);
84
85 napi_property_descriptor properties[] = {
86 DECLARE_NAPI_PROPERTY("Task", taskClass),
87 DECLARE_NAPI_PROPERTY("LongTask", longTaskClass),
88 DECLARE_NAPI_PROPERTY("GenericsTask", genericsTaskClass),
89 DECLARE_NAPI_PROPERTY("TaskGroup", taskGroupClass),
90 DECLARE_NAPI_PROPERTY("SequenceRunner", seqRunnerClass),
91 DECLARE_NAPI_PROPERTY("Priority", priorityObj),
92 DECLARE_NAPI_PROPERTY("State", stateObj),
93 DECLARE_NAPI_FUNCTION("execute", Execute),
94 DECLARE_NAPI_FUNCTION("executeDelayed", ExecuteDelayed),
95 DECLARE_NAPI_FUNCTION("cancel", Cancel),
96 DECLARE_NAPI_FUNCTION("getTaskPoolInfo", GetTaskPoolInfo),
97 DECLARE_NAPI_FUNCTION("terminateTask", TerminateTask),
98 DECLARE_NAPI_FUNCTION("isConcurrent", IsConcurrent),
99 DECLARE_NAPI_FUNCTION("executePeriodically", ExecutePeriodically),
100 };
101 napi_define_properties(env, exports, sizeof(properties) / sizeof(properties[0]), properties);
102
103 TaskManager::GetInstance().InitTaskManager(env);
104 return exports;
105 }
106
107 // ---------------------------------- SendData ---------------------------------------
ExecuteCallback(const uv_async_t * req)108 void TaskPool::ExecuteCallback(const uv_async_t* req)
109 {
110 auto* msgQueue = TaskManager::GetInstance().GetMessageQueue(req);
111 if (msgQueue == nullptr) {
112 HILOG_WARN("taskpool:: msgQueue is nullptr");
113 return;
114 }
115 ExecuteCallbackInner(*msgQueue);
116 }
117
ExecuteCallbackTask(CallbackInfo * callbackInfo)118 void TaskPool::ExecuteCallbackTask(CallbackInfo* callbackInfo)
119 {
120 auto* msgQueue = TaskManager::GetInstance().GetMessageQueueFromCallbackInfo(callbackInfo);
121 if (msgQueue == nullptr) {
122 HILOG_WARN("taskpool:: msgQueue is nullptr");
123 return;
124 }
125 ExecuteCallbackInner(*msgQueue);
126 }
127
ExecuteCallbackInner(MsgQueue & msgQueue)128 void TaskPool::ExecuteCallbackInner(MsgQueue& msgQueue)
129 {
130 while (!msgQueue.IsEmpty()) {
131 auto resultInfo = msgQueue.DeQueue();
132 if (resultInfo == nullptr) {
133 HILOG_DEBUG("taskpool:: resultInfo is nullptr");
134 continue;
135 }
136 ObjectScope<TaskResultInfo> resultInfoScope(resultInfo, false);
137 napi_status status = napi_ok;
138 CallbackScope callbackScope(resultInfo->hostEnv, resultInfo->workerEnv, resultInfo->taskId, status);
139 if (status != napi_ok) {
140 HILOG_ERROR("napi_open_handle_scope failed");
141 return;
142 }
143 auto env = resultInfo->hostEnv;
144 auto callbackInfo = TaskManager::GetInstance().GetCallbackInfo(resultInfo->taskId);
145 if (callbackInfo == nullptr) {
146 HILOG_ERROR("taskpool:: the callback in SendData is not registered on the host side");
147 ErrorHelper::ThrowError(env, ErrorHelper::ERR_NOT_REGISTERED);
148 continue;
149 }
150 TaskManager::GetInstance().ResetCallbackInfoWorker(callbackInfo);
151 auto func = NapiHelper::GetReferenceValue(env, callbackInfo->callbackRef);
152 napi_value args;
153 napi_value result;
154 status = napi_deserialize(env, resultInfo->serializationArgs, &args);
155 napi_delete_serialization_data(env, resultInfo->serializationArgs);
156 if (status != napi_ok || args == nullptr) {
157 std::string errMessage = "taskpool:: failed to serialize function";
158 HILOG_ERROR("%{public}s in SendData", errMessage.c_str());
159 ErrorHelper::ThrowError(env, ErrorHelper::ERR_WORKER_SERIALIZATION, errMessage.c_str());
160 continue;
161 }
162 uint32_t argsNum = NapiHelper::GetArrayLength(env, args);
163 napi_value argsArray[argsNum];
164 for (size_t i = 0; i < argsNum; i++) {
165 argsArray[i] = NapiHelper::GetElement(env, args, i);
166 }
167 napi_call_function(env, NapiHelper::GetGlobalObject(env), func, argsNum, argsArray, &result);
168 if (NapiHelper::IsExceptionPending(env)) {
169 napi_value exception = nullptr;
170 napi_get_and_clear_last_exception(env, &exception);
171 HILOG_ERROR("taskpool:: an exception has occurred in napi_call_function");
172 }
173 }
174 }
175 // ---------------------------------- SendData ---------------------------------------
176
GetTaskPoolInfo(napi_env env,napi_callback_info cbinfo)177 napi_value TaskPool::GetTaskPoolInfo(napi_env env, [[maybe_unused]] napi_callback_info cbinfo)
178 {
179 napi_value result = nullptr;
180 napi_create_object(env, &result);
181 napi_value threadInfos = TaskManager::GetInstance().GetThreadInfos(env);
182 napi_value taskInfos = TaskManager::GetInstance().GetTaskInfos(env);
183 napi_set_named_property(env, result, "threadInfos", threadInfos);
184 napi_set_named_property(env, result, "taskInfos", taskInfos);
185 return result;
186 }
187
TerminateTask(napi_env env,napi_callback_info cbinfo)188 napi_value TaskPool::TerminateTask(napi_env env, napi_callback_info cbinfo)
189 {
190 HITRACE_HELPER_METER_NAME(__PRETTY_FUNCTION__);
191 size_t argc = 1; // 1: long task
192 napi_value args[1];
193 napi_get_cb_info(env, cbinfo, &argc, args, nullptr, nullptr);
194 if (argc < 1) {
195 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR,
196 "the number of the params must be one.");
197 return nullptr;
198 }
199 if (!NapiHelper::IsObject(env, args[0])) {
200 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR,
201 "the type of the params must be object.");
202 return nullptr;
203 }
204 napi_value napiTaskId = NapiHelper::GetNameProperty(env, args[0], TASKID_STR);
205 uint64_t taskId = NapiHelper::GetUint64Value(env, napiTaskId);
206 auto task = TaskManager::GetInstance().GetTask(taskId);
207 if (task == nullptr || !task->IsLongTask()) {
208 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR,
209 "the type of the params must be long task.");
210 return nullptr;
211 }
212 TaskManager::GetInstance().TerminateTask(taskId);
213 return nullptr;
214 }
215
Execute(napi_env env,napi_callback_info cbinfo)216 napi_value TaskPool::Execute(napi_env env, napi_callback_info cbinfo)
217 {
218 HITRACE_HELPER_METER_NAME(__PRETTY_FUNCTION__);
219 size_t argc = NapiHelper::GetCallbackInfoArgc(env, cbinfo);
220 if (argc < 1) {
221 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR,
222 "the number of params must be at least one.");
223 return nullptr;
224 }
225 napi_value* args = new napi_value[argc];
226 ObjectScope<napi_value> scope(args, true);
227 napi_get_cb_info(env, cbinfo, &argc, args, nullptr, nullptr);
228 napi_valuetype type = napi_undefined;
229 napi_typeof(env, args[0], &type);
230 if (type == napi_object) {
231 uint32_t priority = Priority::DEFAULT; // DEFAULT priority is MEDIUM
232 if (argc > 1) {
233 if (!NapiHelper::IsNumber(env, args[1])) {
234 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR,
235 "the type of the second param must be number.");
236 return nullptr;
237 }
238 priority = NapiHelper::GetUint32Value(env, args[1]);
239 if (priority >= Priority::NUMBER) {
240 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR, "priority value is error");
241 return nullptr;
242 }
243 }
244 if (NapiHelper::HasNameProperty(env, args[0], GROUP_ID_STR)) {
245 return ExecuteGroup(env, args[0], static_cast<Priority>(priority));
246 }
247 Task* task = nullptr;
248 napi_unwrap(env, args[0], reinterpret_cast<void**>(&task));
249 if (task == nullptr) {
250 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR,
251 "the type of the first param must be task.");
252 return nullptr;
253 }
254 if (!task->CanExecute(env)) {
255 return nullptr;
256 }
257 napi_value promise = task->GetTaskInfoPromise(env, args[0], TaskType::COMMON_TASK,
258 static_cast<Priority>(priority));
259 if (promise == nullptr) {
260 return nullptr;
261 }
262 ExecuteTask(env, task, static_cast<Priority>(priority));
263 return promise;
264 }
265 if (type != napi_function) {
266 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR,
267 "the type of the first param must be object or function.");
268 return nullptr;
269 }
270 Task* task = Task::GenerateFunctionTask(env, args[0], args + 1, argc - 1, TaskType::FUNCTION_TASK);
271 if (task == nullptr) {
272 HILOG_ERROR("taskpool:: GenerateFunctionTask failed");
273 return nullptr;
274 }
275 TaskManager::GetInstance().StoreTask(task->taskId_, task);
276 napi_value promise = NapiHelper::CreatePromise(env, &task->currentTaskInfo_->deferred);
277 ExecuteTask(env, task);
278 return promise;
279 }
280
DelayTask(uv_timer_t * handle)281 void TaskPool::DelayTask(uv_timer_t* handle)
282 {
283 TaskMessage *taskMessage = static_cast<TaskMessage *>(handle->data);
284 auto task = TaskManager::GetInstance().GetTask(taskMessage->taskId);
285 napi_status status = napi_ok;
286 if (task == nullptr) {
287 HILOG_DEBUG("taskpool:: task is nullptr");
288 } else if (task->taskState_ == ExecuteState::CANCELED) {
289 HILOG_DEBUG("taskpool:: DelayTask task has been canceled");
290 HandleScope scope(task->env_, status);
291 if (status != napi_ok) {
292 HILOG_ERROR("taskpool:: napi_open_handle_scope failed");
293 return;
294 }
295 napi_value error = ErrorHelper::NewError(task->env_, 0, "taskpool:: task has been canceled");
296 napi_reject_deferred(task->env_, taskMessage->deferred, error);
297 } else {
298 HILOG_INFO("taskpool:: DelayTask taskId %{public}s", std::to_string(taskMessage->taskId).c_str());
299 HandleScope scope(task->env_, status);
300 if (status != napi_ok) {
301 HILOG_ERROR("taskpool:: napi_open_handle_scope failed");
302 return;
303 }
304 TaskManager::GetInstance().IncreaseRefCount(taskMessage->taskId);
305 task->IncreaseRefCount();
306 napi_value napiTask = NapiHelper::GetReferenceValue(task->env_, task->taskRef_);
307 TaskInfo* taskInfo = task->GetTaskInfo(task->env_, napiTask, taskMessage->priority);
308 if (taskInfo != nullptr) {
309 taskInfo->deferred = taskMessage->deferred;
310 if (task->taskState_ == ExecuteState::DELAYED || task->taskState_ == ExecuteState::FINISHED) {
311 task->taskState_ = ExecuteState::WAITING;
312 TaskManager::GetInstance().EnqueueTaskId(taskMessage->taskId, Priority(taskMessage->priority));
313 }
314 } else {
315 napi_value execption = nullptr;
316 napi_get_and_clear_last_exception(task->env_, &execption);
317 if (execption != nullptr) {
318 napi_reject_deferred(task->env_, taskMessage->deferred, execption);
319 }
320 }
321 }
322 if (task != nullptr) {
323 std::lock_guard<RECURSIVE_MUTEX> lock(task->taskMutex_);
324 task->delayedTimers_.erase(handle);
325 }
326 uv_timer_stop(handle);
327 ConcurrentHelper::UvHandleClose(handle);
328 delete taskMessage;
329 taskMessage = nullptr;
330 }
331
ExecuteDelayed(napi_env env,napi_callback_info cbinfo)332 napi_value TaskPool::ExecuteDelayed(napi_env env, napi_callback_info cbinfo)
333 {
334 HITRACE_HELPER_METER_NAME(__PRETTY_FUNCTION__);
335 uint32_t priority = Priority::DEFAULT; // DEFAULT priority is MEDIUM
336 int32_t delayTime = 0;
337 Task* task = nullptr;
338 if (!CheckDelayedParams(env, cbinfo, priority, delayTime, task)) {
339 return nullptr;
340 }
341
342 if (!task->IsExecuted() || task->taskState_ == ExecuteState::CANCELED ||
343 task->taskState_ == ExecuteState::FINISHED) {
344 task->taskState_ = ExecuteState::DELAYED;
345 }
346 task->UpdateTaskType(TaskType::COMMON_TASK);
347
348 uv_loop_t* loop = NapiHelper::GetLibUV(env);
349 uv_update_time(loop);
350 uv_timer_t* timer = new uv_timer_t;
351 uv_timer_init(loop, timer);
352 TaskMessage *taskMessage = new TaskMessage();
353 taskMessage->priority = static_cast<Priority>(priority);
354 taskMessage->taskId = task->taskId_;
355 napi_value promise = NapiHelper::CreatePromise(env, &taskMessage->deferred);
356 timer->data = taskMessage;
357
358 std::string strTrace = "ExecuteDelayed: taskId: " + std::to_string(task->taskId_);
359 strTrace += ", priority: " + std::to_string(priority);
360 strTrace += ", delayTime " + std::to_string(delayTime);
361 HITRACE_HELPER_METER_NAME(strTrace);
362 HILOG_INFO("taskpool:: %{public}s", strTrace.c_str());
363
364 uv_timer_start(timer, reinterpret_cast<uv_timer_cb>(DelayTask), delayTime, 0);
365 {
366 std::lock_guard<RECURSIVE_MUTEX> lock(task->taskMutex_);
367 task->delayedTimers_.insert(timer);
368 }
369 NativeEngine* engine = reinterpret_cast<NativeEngine*>(env);
370 if (engine->IsMainThread()) {
371 uv_async_send(&loop->wq_async);
372 } else {
373 uv_work_t *work = new uv_work_t;
374 uv_queue_work_with_qos(loop, work, [](uv_work_t *) {},
375 [](uv_work_t *work, int32_t) {delete work; }, uv_qos_user_initiated);
376 }
377 return promise;
378 }
379
ExecuteGroup(napi_env env,napi_value napiTaskGroup,Priority priority)380 napi_value TaskPool::ExecuteGroup(napi_env env, napi_value napiTaskGroup, Priority priority)
381 {
382 napi_value napiGroupId = NapiHelper::GetNameProperty(env, napiTaskGroup, GROUP_ID_STR);
383 uint64_t groupId = NapiHelper::GetUint64Value(env, napiGroupId);
384 HILOG_INFO("taskpool::ExecuteGroup groupId %{public}s", std::to_string(groupId).c_str());
385 auto taskGroup = TaskGroupManager::GetInstance().GetTaskGroup(groupId);
386 napi_reference_ref(env, taskGroup->groupRef_, nullptr);
387 if (taskGroup->groupState_ == ExecuteState::NOT_FOUND || taskGroup->groupState_ == ExecuteState::FINISHED ||
388 taskGroup->groupState_ == ExecuteState::CANCELED) {
389 taskGroup->groupState_ = ExecuteState::WAITING;
390 }
391 GroupInfo* groupInfo = new GroupInfo();
392 groupInfo->priority = priority;
393 napi_value resArr;
394 napi_create_array_with_length(env, taskGroup->taskIds_.size(), &resArr);
395 napi_ref arrRef = NapiHelper::CreateReference(env, resArr, 1);
396 groupInfo->resArr = arrRef;
397 napi_value promise = NapiHelper::CreatePromise(env, &groupInfo->deferred);
398 {
399 std::lock_guard<RECURSIVE_MUTEX> lock(taskGroup->taskGroupMutex_);
400 if (taskGroup->currentGroupInfo_ == nullptr) {
401 taskGroup->currentGroupInfo_ = groupInfo;
402 for (auto iter = taskGroup->taskRefs_.begin(); iter != taskGroup->taskRefs_.end(); iter++) {
403 napi_value napiTask = NapiHelper::GetReferenceValue(env, *iter);
404 Task* task = nullptr;
405 napi_unwrap(env, napiTask, reinterpret_cast<void**>(&task));
406 if (task == nullptr) {
407 HILOG_ERROR("taskpool::ExecuteGroup task is nullptr");
408 return nullptr;
409 }
410 napi_reference_ref(env, task->taskRef_, nullptr);
411 if (task->IsGroupCommonTask()) {
412 task->GetTaskInfo(env, napiTask, static_cast<Priority>(priority));
413 }
414 ExecuteTask(env, task, static_cast<Priority>(priority));
415 }
416 } else {
417 taskGroup->pendingGroupInfos_.push_back(groupInfo);
418 }
419 }
420 return promise;
421 }
422
HandleTaskResult(const uv_async_t * req)423 void TaskPool::HandleTaskResult(const uv_async_t* req)
424 {
425 HILOG_DEBUG("taskpool:: HandleTaskResult task");
426 HITRACE_HELPER_METER_NAME(__PRETTY_FUNCTION__);
427 auto task = static_cast<Task*>(req->data);
428 if (task == nullptr) { // LCOV_EXCL_BR_LINE
429 HILOG_FATAL("taskpool:: HandleTaskResult task is null");
430 return;
431 }
432 if (!task->IsMainThreadTask()) {
433 if (task->ShouldDeleteTask(false)) {
434 delete task;
435 return;
436 }
437 if (task->IsFunctionTask()) {
438 napi_remove_env_cleanup_hook(task->env_, Task::CleanupHookFunc, task);
439 }
440 }
441 task->DecreaseTaskRefCount();
442 HandleTaskResultCallback(task);
443 }
444
HandleTaskResultCallback(Task * task)445 void TaskPool::HandleTaskResultCallback(Task* task)
446 {
447 napi_handle_scope scope = nullptr;
448 NAPI_CALL_RETURN_VOID(task->env_, napi_open_handle_scope(task->env_, &scope));
449 napi_value napiTaskResult = nullptr;
450 napi_status status = napi_deserialize(task->env_, task->result_, &napiTaskResult);
451 napi_delete_serialization_data(task->env_, task->result_);
452
453 // tag for trace parse: Task PerformTask End
454 std::string strTrace = "Task PerformTask End: taskId : " + std::to_string(task->taskId_);
455 if (task->taskState_ == ExecuteState::CANCELED) {
456 strTrace += ", performResult : IsCanceled";
457 napiTaskResult = ErrorHelper::NewError(task->env_, 0, "taskpool:: task has been canceled");
458 } else if (status != napi_ok) {
459 HILOG_ERROR("taskpool: failed to deserialize result");
460 strTrace += ", performResult : DeserializeFailed";
461 } else if (task->success_) {
462 strTrace += ", performResult : Successful";
463 } else {
464 strTrace += ", performResult : Unsuccessful";
465 }
466 HITRACE_HELPER_METER_NAME(strTrace);
467 HILOG_INFO("taskpool:: %{public}s", strTrace.c_str());
468 if (napiTaskResult == nullptr) {
469 napi_get_undefined(task->env_, &napiTaskResult);
470 }
471 reinterpret_cast<NativeEngine*>(task->env_)->DecreaseSubEnvCounter();
472 bool success = ((status == napi_ok) && (task->taskState_ != ExecuteState::CANCELED)) && (task->success_);
473 task->taskState_ = ExecuteState::ENDING;
474 if (task->IsGroupTask()) {
475 UpdateGroupInfoByResult(task->env_, task, napiTaskResult, success);
476 } else if (!task->IsPeriodicTask()) {
477 if (success) {
478 napi_resolve_deferred(task->env_, task->currentTaskInfo_->deferred, napiTaskResult);
479 if (task->onExecutionSucceededCallBackInfo_ != nullptr) {
480 task->ExecuteListenerCallback(task->onExecutionSucceededCallBackInfo_);
481 }
482 } else {
483 napi_reject_deferred(task->env_, task->currentTaskInfo_->deferred, napiTaskResult);
484 if (task->onExecutionFailedCallBackInfo_ != nullptr) {
485 task->onExecutionFailedCallBackInfo_->taskError_ = napiTaskResult;
486 task->ExecuteListenerCallback(task->onExecutionFailedCallBackInfo_);
487 }
488 }
489 }
490 NAPI_CALL_RETURN_VOID(task->env_, napi_close_handle_scope(task->env_, scope));
491 TriggerTask(task);
492 HILOG_DEBUG("taskpool:: %{public}s", strTrace.c_str());
493 }
494
TriggerTask(Task * task)495 void TaskPool::TriggerTask(Task* task)
496 {
497 HILOG_DEBUG("taskpool:: task:%{public}s TriggerTask", std::to_string(task->taskId_).c_str());
498 if (task->IsGroupTask()) {
499 return;
500 }
501 TaskManager::GetInstance().DecreaseRefCount(task->env_, task->taskId_);
502 task->taskState_ = ExecuteState::FINISHED;
503 // seqRunnerTask will trigger the next
504 if (task->IsSeqRunnerTask()) {
505 if (!TaskGroupManager::GetInstance().TriggerSeqRunner(task->env_, task)) {
506 HILOG_ERROR("seqRunner:: task %{public}s trigger in seqRunner %{public}s failed",
507 std::to_string(task->taskId_).c_str(), std::to_string(task->seqRunnerId_).c_str());
508 }
509 } else if (task->IsCommonTask()) {
510 task->NotifyPendingTask();
511 }
512 if (task->IsPeriodicTask()) {
513 return;
514 }
515 if (!task->IsFunctionTask()) {
516 napi_reference_unref(task->env_, task->taskRef_, nullptr);
517 return;
518 }
519 TaskManager::GetInstance().RemoveTask(task->taskId_);
520 delete task;
521 }
522
UpdateGroupInfoByResult(napi_env env,Task * task,napi_value res,bool success)523 void TaskPool::UpdateGroupInfoByResult(napi_env env, Task* task, napi_value res, bool success)
524 {
525 HILOG_DEBUG("taskpool:: task:%{public}s UpdateGroupInfoByResult", std::to_string(task->taskId_).c_str());
526 TaskManager::GetInstance().DecreaseRefCount(task->env_, task->taskId_);
527 task->taskState_ = ExecuteState::FINISHED;
528 napi_reference_unref(env, task->taskRef_, nullptr);
529 if (task->IsGroupCommonTask()) {
530 delete task->currentTaskInfo_;
531 task->currentTaskInfo_ = nullptr;
532 }
533 TaskGroup* taskGroup = TaskGroupManager::GetInstance().GetTaskGroup(task->groupId_);
534 if (taskGroup == nullptr || taskGroup->currentGroupInfo_ == nullptr) {
535 HILOG_DEBUG("taskpool:: taskGroup may have been released or canceled");
536 return;
537 }
538 // store the result
539 uint32_t index = taskGroup->GetTaskIndex(task->taskId_);
540 auto groupInfo = taskGroup->currentGroupInfo_;
541 napi_ref arrRef = groupInfo->resArr;
542 napi_value resArr = NapiHelper::GetReferenceValue(env, arrRef);
543 napi_set_element(env, resArr, index, res);
544 groupInfo->finishedTaskNum++;
545 // store the index when the first exception occurs
546 if (!success && !groupInfo->HasException()) {
547 groupInfo->SetFailedIndex(index);
548 }
549 // we will not handle the result until all tasks are finished
550 if (groupInfo->finishedTaskNum < taskGroup->taskNum_) {
551 return;
552 }
553 // if there is no exception, just resolve
554 if (!groupInfo->HasException()) {
555 HILOG_INFO("taskpool:: taskGroup perform end, taskGroupId %{public}s", std::to_string(task->groupId_).c_str());
556 napi_resolve_deferred(env, groupInfo->deferred, resArr);
557 for (uint64_t taskId : taskGroup->taskIds_) {
558 auto task = TaskManager::GetInstance().GetTask(taskId);
559 if (task->onExecutionSucceededCallBackInfo_ != nullptr) {
560 task->ExecuteListenerCallback(task->onExecutionSucceededCallBackInfo_);
561 }
562 }
563 } else {
564 napi_value res = nullptr;
565 napi_get_element(env, resArr, groupInfo->GetFailedIndex(), &res);
566 napi_reject_deferred(env, groupInfo->deferred, res);
567 auto iter = taskGroup->taskIds_.begin();
568 std::advance(iter, groupInfo->GetFailedIndex());
569 auto task = iter != taskGroup->taskIds_.end() ? TaskManager::GetInstance().GetTask(*iter) : nullptr;
570 if (task != nullptr && task->onExecutionFailedCallBackInfo_ != nullptr) {
571 task->onExecutionFailedCallBackInfo_->taskError_ = res;
572 task->ExecuteListenerCallback(task->onExecutionFailedCallBackInfo_);
573 }
574 }
575 taskGroup->groupState_ = ExecuteState::FINISHED;
576 napi_delete_reference(env, groupInfo->resArr);
577 napi_reference_unref(env, taskGroup->groupRef_, nullptr);
578 delete groupInfo;
579 taskGroup->currentGroupInfo_ = nullptr;
580 taskGroup->NotifyGroupTask(env);
581 }
582
ExecuteTask(napi_env env,Task * task,Priority priority)583 void TaskPool::ExecuteTask(napi_env env, Task* task, Priority priority)
584 {
585 // tag for trace parse: Task Allocation
586 std::string strTrace = "Task Allocation: taskId : " + std::to_string(task->taskId_)
587 + ", priority : " + std::to_string(priority)
588 + ", executeState : " + std::to_string(ExecuteState::WAITING);
589 HITRACE_HELPER_METER_NAME(strTrace);
590 HILOG_INFO("taskpool:: %{public}s", strTrace.c_str());
591 task->IncreaseRefCount();
592 TaskManager::GetInstance().IncreaseRefCount(task->taskId_);
593 if (task->IsFunctionTask() || (task->taskState_ != ExecuteState::WAITING &&
594 task->taskState_ != ExecuteState::RUNNING && task->taskState_ != ExecuteState::ENDING)) {
595 task->taskState_ = ExecuteState::WAITING;
596 TaskManager::GetInstance().EnqueueTaskId(task->taskId_, priority);
597 }
598 }
599
Cancel(napi_env env,napi_callback_info cbinfo)600 napi_value TaskPool::Cancel(napi_env env, napi_callback_info cbinfo)
601 {
602 HITRACE_HELPER_METER_NAME(__PRETTY_FUNCTION__);
603 size_t argc = 1;
604 napi_value args[1];
605 napi_get_cb_info(env, cbinfo, &argc, args, nullptr, nullptr);
606 if (argc < 1) {
607 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR,
608 "the number of the params must be 1.");
609 return nullptr;
610 }
611
612 if (!NapiHelper::IsObject(env, args[0])) {
613 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR,
614 "the type of the params must be object.");
615 return nullptr;
616 }
617
618 if (!NapiHelper::HasNameProperty(env, args[0], GROUP_ID_STR)) {
619 napi_value napiTaskId = NapiHelper::GetNameProperty(env, args[0], TASKID_STR);
620 if (napiTaskId == nullptr) {
621 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR,
622 "the type of the params must be task.");
623 return nullptr;
624 }
625 uint64_t taskId = NapiHelper::GetUint64Value(env, napiTaskId);
626 TaskManager::GetInstance().CancelTask(env, taskId);
627 } else {
628 napi_value napiGroupId = NapiHelper::GetNameProperty(env, args[0], GROUP_ID_STR);
629 if (napiGroupId == nullptr) {
630 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR,
631 "the type of the params must be taskGroup.");
632 return nullptr;
633 }
634 uint64_t groupId = NapiHelper::GetUint64Value(env, napiGroupId);
635 TaskGroupManager::GetInstance().CancelGroup(env, groupId);
636 }
637 return nullptr;
638 }
639
IsConcurrent(napi_env env,napi_callback_info cbinfo)640 napi_value TaskPool::IsConcurrent(napi_env env, napi_callback_info cbinfo)
641 {
642 size_t argc = 1;
643 napi_value args[1];
644 napi_get_cb_info(env, cbinfo, &argc, args, nullptr, nullptr);
645 if (argc != 1) {
646 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR,
647 "the number of the params must be 1.");
648 return nullptr;
649 }
650
651 if (!NapiHelper::IsFunction(env, args[0])) {
652 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR,
653 "the type of the first param must be function.");
654 return nullptr;
655 }
656
657 bool isConcurrent = NapiHelper::IsConcurrentFunction(env, args[0]);
658 return NapiHelper::CreateBooleanValue(env, isConcurrent);
659 }
660
PeriodicTaskCallback(uv_timer_t * handle)661 void TaskPool::PeriodicTaskCallback(uv_timer_t* handle)
662 {
663 Task* task = reinterpret_cast<Task*>(handle->data);
664 if (task == nullptr) {
665 HILOG_DEBUG("taskpool:: the task is nullptr");
666 return;
667 } else if (!task->IsPeriodicTask()) {
668 HILOG_DEBUG("taskpool:: the current task is not a periodic task");
669 return;
670 } else if (task->taskState_ == ExecuteState::CANCELED) {
671 HILOG_DEBUG("taskpool:: the periodic task has been canceled");
672 return;
673 }
674 TaskManager::GetInstance().IncreaseRefCount(task->taskId_);
675
676 if (!task->isFirstTaskInfo_) {
677 napi_status status = napi_ok;
678 HandleScope scope(task->env_, status);
679 if (status != napi_ok) {
680 HILOG_ERROR("taskpool:: napi_open_handle_scope failed");
681 return;
682 }
683 napi_value napiTask = NapiHelper::GetReferenceValue(task->env_, task->taskRef_);
684 TaskInfo* taskInfo = task->GetTaskInfo(task->env_, napiTask, task->periodicTaskPriority_);
685 if (taskInfo == nullptr) {
686 HILOG_DEBUG("taskpool:: the periodic task taskInfo is nullptr");
687 return;
688 }
689 }
690 task->isFirstTaskInfo_ = false;
691
692 task->IncreaseRefCount();
693 HILOG_INFO("taskpool:: PeriodicTaskCallback taskId %{public}s", std::to_string(task->taskId_).c_str());
694 if (task->taskState_ == ExecuteState::NOT_FOUND || task->taskState_ == ExecuteState::FINISHED) {
695 task->taskState_ = ExecuteState::WAITING;
696 TaskManager::GetInstance().EnqueueTaskId(task->taskId_, task->periodicTaskPriority_);
697 }
698 }
699
ExecutePeriodically(napi_env env,napi_callback_info cbinfo)700 napi_value TaskPool::ExecutePeriodically(napi_env env, napi_callback_info cbinfo)
701 {
702 int32_t period = 0;
703 uint32_t priority = Priority::DEFAULT;
704 Task* periodicTask = nullptr;
705 if (!CheckPeriodicallyParams(env, cbinfo, period, priority, periodicTask)) {
706 return nullptr;
707 }
708
709 if (!periodicTask->CanExecutePeriodically(env)) {
710 return nullptr;
711 }
712 periodicTask->UpdatePeriodicTask();
713
714 periodicTask->periodicTaskPriority_ = static_cast<Priority>(priority);
715 napi_value napiTask = NapiHelper::GetReferenceValue(env, periodicTask->taskRef_);
716 TaskInfo* taskInfo = periodicTask->GetTaskInfo(env, napiTask, periodicTask->periodicTaskPriority_);
717 if (taskInfo == nullptr) {
718 return nullptr;
719 }
720
721 periodicTask->isFirstTaskInfo_ = true; // periodic task first Generate TaskInfo
722
723 TriggerTimer(env, periodicTask, period);
724 return nullptr;
725 }
726
TriggerTimer(napi_env env,Task * task,int32_t period)727 void TaskPool::TriggerTimer(napi_env env, Task* task, int32_t period)
728 {
729 HILOG_INFO("taskpool::TriggerTimer taskId %{public}s", std::to_string(task->taskId_).c_str());
730 uv_loop_t* loop = NapiHelper::GetLibUV(env);
731 task->timer_ = new uv_timer_t;
732 uv_timer_init(loop, task->timer_);
733 task->timer_->data = task;
734 uv_update_time(loop);
735 uv_timer_start(task->timer_, PeriodicTaskCallback, period, period);
736 NativeEngine* engine = reinterpret_cast<NativeEngine*>(env);
737 if (engine->IsMainThread()) {
738 uv_async_send(&loop->wq_async);
739 } else {
740 uv_work_t* work = new uv_work_t;
741 uv_queue_work_with_qos(loop, work, [](uv_work_t*) {},
742 [](uv_work_t* work, int32_t) { delete work; }, uv_qos_user_initiated);
743 }
744 }
745
CheckDelayedParams(napi_env env,napi_callback_info cbinfo,uint32_t & priority,int32_t & delayTime,Task * & task)746 bool TaskPool::CheckDelayedParams(napi_env env, napi_callback_info cbinfo, uint32_t &priority, int32_t &delayTime,
747 Task* &task)
748 {
749 size_t argc = 3; // 3: delayTime, task and priority
750 napi_value args[3]; // 3: delayTime, task and priority
751 napi_get_cb_info(env, cbinfo, &argc, args, nullptr, nullptr);
752 if (argc < 2 || argc > 3) { // 2: delayTime and task 3: delayTime, task and priority
753 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR,
754 "the number of params must be two or three.");
755 return false;
756 }
757
758 if (!NapiHelper::IsNumber(env, args[0])) {
759 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR,
760 "the type of the first param must be number.");
761 return false;
762 }
763
764 if (!NapiHelper::IsObject(env, args[1])) {
765 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR,
766 "the type of the second param must be object.");
767 return false;
768 }
769
770 delayTime = NapiHelper::GetInt32Value(env, args[0]);
771 if (delayTime < 0) {
772 ErrorHelper::ThrowError(env, ErrorHelper::ERR_DELAY_TIME_ERROR, "The delayTime is less than zero");
773 return false;
774 }
775
776 if (argc > 2) { // 2: the params might have priority
777 if (!NapiHelper::IsNumber(env, args[2])) {
778 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR,
779 "the type of the third param must be number.");
780 return false;
781 }
782 priority = NapiHelper::GetUint32Value(env, args[2]); // 2: get task priority
783 if (priority >= Priority::NUMBER) {
784 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR, "priority value is error.");
785 return false;
786 }
787 }
788
789 napi_unwrap(env, args[1], reinterpret_cast<void**>(&task));
790 if (task == nullptr) {
791 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR,
792 "the type of second param must be task");
793 return false;
794 }
795 if (!task->CanExecuteDelayed(env)) {
796 return false;
797 }
798 return true;
799 }
800
CheckPeriodicallyParams(napi_env env,napi_callback_info cbinfo,int32_t & period,uint32_t & priority,Task * & periodicTask)801 bool TaskPool::CheckPeriodicallyParams(napi_env env, napi_callback_info cbinfo, int32_t &period,
802 uint32_t &priority, Task* &periodicTask)
803 {
804 size_t argc = 3; // 3 : period, task, priority
805 napi_value args[3]; // 3 : period, task, priority
806 napi_get_cb_info(env, cbinfo, &argc, args, nullptr, nullptr);
807 if (argc < 2 || argc > 3) { // 2 : period, task and 3 : period, task, priority
808 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR, "the number of params must be two or three.");
809 return false;
810 }
811 if (!NapiHelper::IsNumber(env, args[0])) {
812 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR, "the type of the first param must be number.");
813 return false;
814 }
815 period = NapiHelper::GetInt32Value(env, args[0]);
816 if (period < 0) {
817 ErrorHelper::ThrowError(env, ErrorHelper::ERR_DELAY_TIME_ERROR, "The period value is less than zero.");
818 return false;
819 }
820 if (!NapiHelper::IsObject(env, args[1])) {
821 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR, "the type of the second param must be task.");
822 return false;
823 }
824
825 if (argc >= 3) { // 3 : third param maybe priority
826 if (!NapiHelper::IsNumber(env, args[2])) { // 2 : priority
827 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR, "the third param must be priority.");
828 return false;
829 }
830 priority = NapiHelper::GetUint32Value(env, args[2]); // 2 : priority
831 if (priority >= Priority::NUMBER) {
832 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR, "the value of the priority is invalid.");
833 return false;
834 }
835 }
836
837 napi_unwrap(env, args[1], reinterpret_cast<void**>(&periodicTask));
838 if (periodicTask == nullptr) {
839 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR, "the type of the second param must be task.");
840 return false;
841 }
842
843 return true;
844 }
845 } // namespace Commonlibrary::Concurrent::TaskPoolModule
846