1/* 2 * Copyright (c) 2023 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 "task_group.h" 17 18#include "helper/error_helper.h" 19#include "helper/napi_helper.h" 20#include "helper/object_helper.h" 21#include "napi/native_api.h" 22#include "tools/log.h" 23 24namespace Commonlibrary::Concurrent::TaskPoolModule { 25using namespace Commonlibrary::Concurrent::Common::Helper; 26 27napi_value TaskGroup::TaskGroupConstructor(napi_env env, napi_callback_info cbinfo) 28{ 29 size_t argc = 1; 30 napi_value args[1]; 31 napi_value thisVar; 32 napi_get_cb_info(env, cbinfo, &argc, args, &thisVar, nullptr); 33 if (argc > 1) { 34 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR, "the number of params must be zero or one."); 35 return nullptr; 36 } 37 napi_value name; 38 if (argc == 1) { 39 // check 1st param is taskGroupName 40 if (!NapiHelper::IsString(env, args[0])) { 41 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR, "the first param must be string."); 42 return nullptr; 43 } 44 name = args[0]; 45 } else { 46 name = NapiHelper::CreateEmptyString(env); 47 } 48 TaskGroup* group = new TaskGroup(); 49 uint64_t groupId = reinterpret_cast<uint64_t>(group); 50 group->groupId_ = groupId; 51 TaskGroupManager::GetInstance().StoreTaskGroup(groupId, group); 52 napi_value napiGroupId = NapiHelper::CreateUint64(env, groupId); 53 napi_property_descriptor properties[] = { 54 DECLARE_NAPI_PROPERTY(GROUP_ID_STR, napiGroupId), 55 DECLARE_NAPI_FUNCTION_WITH_DATA("addTask", AddTask, thisVar), 56 }; 57 napi_set_named_property(env, thisVar, NAME, name); 58 napi_define_properties(env, thisVar, sizeof(properties) / sizeof(properties[0]), properties); 59 napi_wrap(env, thisVar, group, TaskGroupDestructor, nullptr, nullptr); 60 napi_create_reference(env, thisVar, 0, &group->groupRef_); 61 return thisVar; 62} 63 64void TaskGroup::TaskGroupDestructor(napi_env env, void* data, [[maybe_unused]] void* hint) 65{ 66 HILOG_DEBUG("taskpool::TaskGroupDestructor"); 67 TaskGroup* group = static_cast<TaskGroup*>(data); 68 TaskGroupManager::GetInstance().ReleaseTaskGroupData(env, group); 69 napi_delete_reference(env, group->groupRef_); 70 delete group; 71} 72 73napi_value TaskGroup::AddTask(napi_env env, napi_callback_info cbinfo) 74{ 75 size_t argc = NapiHelper::GetCallbackInfoArgc(env, cbinfo); 76 std::string errMessage = ""; 77 if (argc < 1) { 78 errMessage = "taskGroup:: the number of params must be at least one"; 79 HILOG_ERROR("%{public}s", errMessage.c_str()); 80 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR, "the number of params must be at least one."); 81 return nullptr; 82 } 83 napi_value* args = new napi_value[argc]; 84 ObjectScope<napi_value> scope(args, true); 85 napi_value thisVar; 86 napi_get_cb_info(env, cbinfo, &argc, args, &thisVar, nullptr); 87 napi_value napiGroupId = NapiHelper::GetNameProperty(env, thisVar, GROUP_ID_STR); 88 uint64_t groupId = NapiHelper::GetUint64Value(env, napiGroupId); 89 TaskGroup* group = TaskGroupManager::GetInstance().GetTaskGroup(groupId); 90 if (group->groupState_ != ExecuteState::NOT_FOUND) { 91 errMessage = "taskpool:: executed taskGroup cannot addTask"; 92 HILOG_ERROR("%{public}s", errMessage.c_str()); 93 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR, errMessage.c_str()); 94 return nullptr; 95 } 96 napi_valuetype type = napi_undefined; 97 napi_typeof(env, args[0], &type); 98 if (type == napi_object) { 99 Task* task = nullptr; 100 napi_unwrap(env, args[0], reinterpret_cast<void**>(&task)); 101 if (task == nullptr) { 102 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR, "the type of the params must be task."); 103 return nullptr; 104 } 105 if (!task->CanForTaskGroup(env)) { 106 return nullptr; 107 } 108 task->taskType_ = TaskType::GROUP_COMMON_TASK; 109 task->groupId_ = groupId; 110 napi_reference_ref(env, task->taskRef_, nullptr); 111 TaskGroupManager::GetInstance().AddTask(groupId, task->taskRef_, task->taskId_); 112 return nullptr; 113 } else if (type == napi_function) { 114 napi_value napiTask = NapiHelper::CreateObject(env); 115 Task* task = Task::GenerateFunctionTask(env, args[0], args + 1, argc - 1, TaskType::GROUP_FUNCTION_TASK); 116 if (task == nullptr) { 117 return nullptr; 118 } 119 task->groupId_ = groupId; 120 TaskManager::GetInstance().StoreTask(task->taskId_, task); 121 napi_wrap(env, napiTask, task, Task::TaskDestructor, nullptr, nullptr); 122 napi_create_reference(env, napiTask, 1, &task->taskRef_); 123 TaskGroupManager::GetInstance().AddTask(groupId, task->taskRef_, task->taskId_); 124 return nullptr; 125 } 126 ErrorHelper::ThrowError(env, ErrorHelper::TYPE_ERROR, "the type of the first param must be object or function."); 127 return nullptr; 128} 129 130uint32_t TaskGroup::GetTaskIndex(uint32_t taskId) 131{ 132 uint32_t index = 0; 133 for (uint32_t id : taskIds_) { 134 if (taskId == id) { 135 break; 136 } 137 index++; 138 } 139 return index; 140} 141 142void TaskGroup::NotifyGroupTask(napi_env env) 143{ 144 HILOG_DEBUG("taskpool:: NotifyGroupTask"); 145 std::lock_guard<RECURSIVE_MUTEX> lock(taskGroupMutex_); 146 if (pendingGroupInfos_.empty()) { 147 return; 148 } 149 groupState_ = ExecuteState::WAITING; 150 currentGroupInfo_ = pendingGroupInfos_.front(); 151 pendingGroupInfos_.pop_front(); 152 for (auto iter = taskRefs_.begin(); iter != taskRefs_.end(); iter++) { 153 napi_value napiTask = NapiHelper::GetReferenceValue(env, *iter); 154 Task* task = nullptr; 155 napi_unwrap(env, napiTask, reinterpret_cast<void**>(&task)); 156 if (task == nullptr) { 157 HILOG_ERROR("taskpool::ExecuteGroup task is nullptr"); 158 return; 159 } 160 napi_reference_ref(env, task->taskRef_, nullptr); 161 Priority priority = currentGroupInfo_->priority; 162 if (task->IsGroupCommonTask()) { 163 task->GetTaskInfo(env, napiTask, priority); 164 } else { 165 reinterpret_cast<NativeEngine*>(env)->IncreaseSubEnvCounter(); 166 } 167 task->IncreaseRefCount(); 168 TaskManager::GetInstance().IncreaseRefCount(task->taskId_); 169 task->taskState_ = ExecuteState::WAITING; 170 TaskManager::GetInstance().EnqueueTaskId(task->taskId_, priority); 171 } 172} 173 174void TaskGroup::CancelPendingGroup(napi_env env) 175{ 176 HILOG_DEBUG("taskpool:: CancelPendingGroup"); 177 if (pendingGroupInfos_.empty()) { 178 return; 179 } 180 napi_value error = ErrorHelper::NewError(env, 0, "taskpool:: taskGroup has been canceled"); 181 auto pendingIter = pendingGroupInfos_.begin(); 182 auto engine = reinterpret_cast<NativeEngine*>(env); 183 for (; pendingIter != pendingGroupInfos_.end(); ++pendingIter) { 184 for (size_t i = 0; i < taskIds_.size(); i++) { 185 engine->DecreaseSubEnvCounter(); 186 } 187 GroupInfo* info = *pendingIter; 188 napi_reject_deferred(env, info->deferred, error); 189 napi_reference_unref(env, groupRef_, nullptr); 190 delete info; 191 } 192 pendingIter = pendingGroupInfos_.begin(); 193 pendingGroupInfos_.erase(pendingIter, pendingGroupInfos_.end()); 194} 195} // namespace Commonlibrary::Concurrent::TaskPoolModule