1 /*
2 * Copyright (C) 2021 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 "ipc_thread_pool.h"
17
18 #include <dlfcn.h>
19 #include <sys/types.h>
20 #include <unistd.h>
21
22 #include "ipc_debug.h"
23 #include "log_tags.h"
24
25 namespace OHOS {
26 #ifdef CONFIG_IPC_SINGLE
27 namespace IPC_SINGLE {
28 #endif
29
30 static constexpr OHOS::HiviewDFX::HiLogLabel LOG_LABEL = { LOG_CORE, LOG_ID_IPC_COMMON, "IPCWorkThreadPool" };
31
32 static void *g_selfSoHandler = nullptr;
33 static constexpr int32_t IDLE_SPAWN_ACTIVE_NUM = 1;
34
35 // this func is called when ipc_single and ipc_core before loading
InitIpcSo()36 extern "C" __attribute__((constructor)) void InitIpcSo()
37 {
38 if (g_selfSoHandler == nullptr) {
39 Dl_info info;
40 // dladdr func return value description
41 // On success, these functions return a nonzero value.
42 // If the address specified in addr could not be matched to a shared object, then these functions return 0
43 int ret = dladdr(reinterpret_cast<void *>(InitIpcSo), &info);
44 if (ret == 0) {
45 ZLOGE(LOG_LABEL, "dladdr func call failed");
46 return;
47 }
48 g_selfSoHandler = dlopen(info.dli_fname, RTLD_LAZY);
49 if (g_selfSoHandler == nullptr) {
50 const char *error = dlerror();
51 ZLOGE(LOG_LABEL, "dlopen failed, dlerror:%{public}s", error != nullptr ? error : "unknown");
52 }
53 }
54 }
55
IPCWorkThreadPool(int maxThreadNum)56 IPCWorkThreadPool::IPCWorkThreadPool(int maxThreadNum)
57 : threadSequence_(0), maxThreadNum_(maxThreadNum + maxThreadNum),
58 idleThreadNum_(maxThreadNum + IDLE_SPAWN_ACTIVE_NUM), idleSocketThreadNum_(maxThreadNum)
59 {
60 }
61
~IPCWorkThreadPool()62 IPCWorkThreadPool::~IPCWorkThreadPool()
63 {
64 StopAllThreads();
65 }
66
StopAllThreads()67 void IPCWorkThreadPool::StopAllThreads()
68 {
69 std::lock_guard<std::mutex> lock(mutex_);
70 for (auto it = threads_.begin(); it != threads_.end(); it++) {
71 it->second->StopWorkThread();
72 }
73 threads_.clear();
74 }
75
SpawnThread(int policy,int proto)76 bool IPCWorkThreadPool::SpawnThread(int policy, int proto)
77 {
78 std::lock_guard<std::mutex> lock(mutex_);
79 if (((proto == IRemoteObject::IF_PROT_DEFAULT) && (idleThreadNum_ <= 0)) ||
80 ((proto == IRemoteObject::IF_PROT_DATABUS) && (idleSocketThreadNum_ <= 0))) {
81 ZLOGE(LOG_LABEL, "Request thread exceeds online limit, proto:%{public}d", proto);
82 return false;
83 }
84 int threadIndex = 0;
85 std::string threadName = MakeThreadName(proto, threadIndex);
86 ZLOGD(LOG_LABEL, "name:%{public}s", threadName.c_str());
87 if (threads_.find(threadName) != threads_.end()) {
88 ZLOGW(LOG_LABEL, "This thread already exists, threadName:%{public}s", threadName.c_str());
89 return false;
90 }
91
92 sptr<IPCWorkThread> newThread = sptr<IPCWorkThread>::MakeSptr(threadName);
93 if (newThread == nullptr) {
94 ZLOGE(LOG_LABEL, "create IPCWorkThread object failed");
95 return false;
96 }
97 threads_[threadName] = newThread;
98 if (proto == IRemoteObject::IF_PROT_DEFAULT) {
99 idleThreadNum_--;
100 ZLOGD(LOG_LABEL, "now idleThreadNum:%{public}d", idleThreadNum_);
101 } else if (proto == IRemoteObject::IF_PROT_DATABUS) {
102 idleSocketThreadNum_--;
103 ZLOGD(LOG_LABEL, "now idleSocketThreadNum:%{public}d", idleSocketThreadNum_);
104 } else {
105 ZLOGE(LOG_LABEL, "Proto is incorrect:%{public}d", proto);
106 return false;
107 }
108 return newThread->Start(policy, proto, threadIndex);
109 }
110
MakeThreadName(int proto,int & threadIndex)111 std::string IPCWorkThreadPool::MakeThreadName(int proto, int &threadIndex)
112 {
113 int sequence = threadSequence_.fetch_add(1, std::memory_order_relaxed);
114 threadIndex = sequence;
115 return IPCWorkThread::MakeBasicThreadName(proto, sequence);
116 }
117
RemoveThread(const std::string & threadName)118 bool IPCWorkThreadPool::RemoveThread(const std::string &threadName)
119 {
120 std::lock_guard<std::mutex> lock(mutex_);
121 auto it = threads_.find(threadName);
122 if (it != threads_.end()) {
123 sptr<IPCWorkThread> workThread = it->second;
124 if (workThread == nullptr) {
125 return false;
126 }
127 if (workThread->proto_ == IRemoteObject::IF_PROT_DEFAULT) {
128 idleThreadNum_++;
129 } else if (workThread->proto_ == IRemoteObject::IF_PROT_DATABUS) {
130 idleSocketThreadNum_++;
131 }
132 threads_.erase(it);
133 ZLOGD(LOG_LABEL, "now idleThreadNum:%{public}d", idleSocketThreadNum_);
134 return true;
135 }
136 return false;
137 }
138
GetSocketIdleThreadNum() const139 int IPCWorkThreadPool::GetSocketIdleThreadNum() const
140 {
141 return idleSocketThreadNum_;
142 }
143
GetSocketTotalThreadNum() const144 int IPCWorkThreadPool::GetSocketTotalThreadNum() const
145 {
146 return maxThreadNum_ / PROTO_NUM;
147 }
148
GetMaxThreadNum() const149 int IPCWorkThreadPool::GetMaxThreadNum() const
150 {
151 return (maxThreadNum_ / PROTO_NUM) + IDLE_SPAWN_ACTIVE_NUM;
152 }
153
UpdateMaxThreadNum(int maxThreadNum)154 void IPCWorkThreadPool::UpdateMaxThreadNum(int maxThreadNum)
155 {
156 /*
157 * not support delete thread, because thread is in using
158 */
159 int totalNum = maxThreadNum + maxThreadNum;
160 std::lock_guard<std::mutex> lock(mutex_);
161 if (totalNum <= maxThreadNum_) {
162 return;
163 }
164 int diff = totalNum - maxThreadNum_;
165 maxThreadNum_ = totalNum;
166 idleThreadNum_ += diff / PROTO_NUM;
167 idleSocketThreadNum_ += diff / PROTO_NUM;
168 }
169 #ifdef CONFIG_IPC_SINGLE
170 } // namespace IPC_SINGLE
171 #endif
172 } // namespace OHOS
173