1 /*
2 * Copyright (c) 2024 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 "async_runner_manager.h"
17
18 #include <cinttypes>
19
20 #include "helper/error_helper.h"
21 #include "task_manager.h"
22 #include "tools/log.h"
23
24 namespace Commonlibrary::Concurrent::TaskPoolModule {
25
GetInstance()26 AsyncRunnerManager& AsyncRunnerManager::GetInstance()
27 {
28 static AsyncRunnerManager asyncRunnerManager;
29 return asyncRunnerManager;
30 }
31
CreateOrGetGlobalRunner(napi_env env,napi_value thisVar,const std::string & name,uint32_t runningCapacity,uint32_t waitingCapacity)32 AsyncRunner* AsyncRunnerManager::CreateOrGetGlobalRunner(napi_env env, napi_value thisVar, const std::string& name,
33 uint32_t runningCapacity, uint32_t waitingCapacity)
34 {
35 AsyncRunner *asyncRunner = nullptr;
36 {
37 std::unique_lock<std::mutex> lock(globalAsyncRunnerMutex_);
38 auto iter = globalAsyncRunner_.find(name);
39 if (iter == globalAsyncRunner_.end()) {
40 asyncRunner = AsyncRunner::CreateGlobalRunner(name, runningCapacity, waitingCapacity);
41 globalAsyncRunner_.emplace(name, asyncRunner);
42 } else {
43 asyncRunner = iter->second;
44 bool res = asyncRunner->CheckGlobalRunnerParams(env, runningCapacity, waitingCapacity);
45 if (!res) {
46 return nullptr;
47 }
48 if (!FindRunnerAndRef(asyncRunner->asyncRunnerId_)) {
49 return nullptr;
50 }
51 }
52 }
53
54 return asyncRunner;
55 }
56
StoreAsyncRunner(uint64_t asyncRunnerId,AsyncRunner * asyncRunner)57 void AsyncRunnerManager::StoreAsyncRunner(uint64_t asyncRunnerId, AsyncRunner* asyncRunner)
58 {
59 std::unique_lock<std::mutex> lock(asyncRunnersMutex_);
60 asyncRunners_.emplace(asyncRunnerId, asyncRunner);
61 }
62
RemoveAsyncRunner(uint64_t asyncRunnerId)63 void AsyncRunnerManager::RemoveAsyncRunner(uint64_t asyncRunnerId)
64 {
65 asyncRunners_.erase(asyncRunnerId);
66 }
67
GetAsyncRunner(uint64_t asyncRunnerId)68 AsyncRunner* AsyncRunnerManager::GetAsyncRunner(uint64_t asyncRunnerId)
69 {
70 std::unique_lock<std::mutex> lock(asyncRunnersMutex_);
71 auto iter = asyncRunners_.find(asyncRunnerId);
72 if (iter != asyncRunners_.end()) {
73 return iter->second;
74 }
75 return nullptr;
76 }
77
TriggerAsyncRunner(napi_env env,Task * lastTask)78 bool AsyncRunnerManager::TriggerAsyncRunner(napi_env env, Task* lastTask)
79 {
80 uint64_t asyncRunnerId = lastTask->asyncRunnerId_;
81 AsyncRunner* asyncRunner = GetAsyncRunner(asyncRunnerId);
82 if (asyncRunner == nullptr) {
83 HILOG_ERROR("taskpool:: trigger asyncRunner not exist.");
84 return false;
85 }
86 if (UnrefAndDestroyRunner(asyncRunner)) {
87 HILOG_ERROR("taskpool:: trigger asyncRunner is remove.");
88 return false;
89 }
90 asyncRunner->TriggerWaitingTask();
91 return true;
92 }
93
RemoveGlobalAsyncRunner(const std::string & name)94 void AsyncRunnerManager::RemoveGlobalAsyncRunner(const std::string& name)
95 {
96 std::unique_lock<std::mutex> lock(globalAsyncRunnerMutex_);
97 auto iter = globalAsyncRunner_.find(name);
98 if (iter != globalAsyncRunner_.end()) {
99 globalAsyncRunner_.erase(iter);
100 }
101 }
102
CancelAsyncRunnerTask(napi_env env,Task * task)103 void AsyncRunnerManager::CancelAsyncRunnerTask(napi_env env, Task* task)
104 {
105 std::string errMsg = "";
106 if (task->taskState_ == ExecuteState::FINISHED || task->taskState_ == ExecuteState::ENDING) {
107 errMsg = "AsyncRunner task has been executed.";
108 HILOG_ERROR("taskpool:: %{public}s", errMsg.c_str());
109 ErrorHelper::ThrowError(env, ErrorHelper::ERR_CANCEL_NONEXIST_TASK, errMsg.c_str());
110 return;
111 }
112
113 ExecuteState state = task->taskState_.exchange(ExecuteState::CANCELED);
114 task->CancelPendingTask(env);
115 auto asyncRunner = GetAsyncRunner(task->asyncRunnerId_);
116 if (state == ExecuteState::WAITING && task->currentTaskInfo_ != nullptr &&
117 TaskManager::GetInstance().EraseWaitingTaskId(task->taskId_, task->currentTaskInfo_->priority)) {
118 task->DecreaseTaskRefCount();
119 TaskManager::GetInstance().DecreaseRefCount(task->env_, task->taskId_);
120 if (asyncRunner != nullptr) {
121 asyncRunner->TriggerRejectErrorTimer(task, ErrorHelper::ERR_ASYNCRUNNER_TASK_CANCELED, true);
122 }
123 TriggerAsyncRunner(env, task);
124 }
125
126 if (asyncRunner != nullptr) {
127 asyncRunner->RemoveWaitingTask(task);
128 }
129 }
130
RemoveWaitingTask(Task * task)131 void AsyncRunnerManager::RemoveWaitingTask(Task* task)
132 {
133 auto asyncRunner = GetAsyncRunner(task->asyncRunnerId_);
134 if (asyncRunner != nullptr) {
135 asyncRunner->RemoveWaitingTask(task, false);
136 }
137 }
138
FindRunnerAndRef(uint64_t asyncRunnerId)139 bool AsyncRunnerManager::FindRunnerAndRef(uint64_t asyncRunnerId)
140 {
141 std::unique_lock<std::mutex> lock(asyncRunnersMutex_);
142 auto iter = asyncRunners_.find(asyncRunnerId);
143 if (iter == asyncRunners_.end()) {
144 HILOG_ERROR("taskpool:: asyncRunner not exist.");
145 return false;
146 }
147 iter->second->IncreaseAsyncCount();
148 return true;
149 }
150
UnrefAndDestroyRunner(AsyncRunner * asyncRunner)151 bool AsyncRunnerManager::UnrefAndDestroyRunner(AsyncRunner* asyncRunner)
152 {
153 std::unique_lock<std::mutex> lock(asyncRunnersMutex_);
154 if (asyncRunner->DecreaseAsyncCount() != 0) {
155 return false;
156 }
157 if (asyncRunner->isGlobalRunner_) {
158 RemoveGlobalAsyncRunner(asyncRunner->name_);
159 }
160 RemoveAsyncRunner(asyncRunner->asyncRunnerId_);
161 delete asyncRunner;
162 asyncRunner = nullptr;
163 return true;
164 }
165
DecreaseRunningCount(uint64_t asyncRunnerId)166 void AsyncRunnerManager::DecreaseRunningCount(uint64_t asyncRunnerId)
167 {
168 std::unique_lock<std::mutex> lock(asyncRunnersMutex_);
169 auto iter = asyncRunners_.find(asyncRunnerId);
170 if (iter == asyncRunners_.end()) {
171 return;
172 }
173 iter->second->DecreaseRunningCount();
174 }
175 } // namespace Commonlibrary::Concurrent::TaskPoolModule