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 <unistd.h>
19 #include <sys/types.h>
20
21 #include "ipc_debug.h"
22 #include "log_tags.h"
23
24 namespace OHOS {
25 #ifdef CONFIG_IPC_SINGLE
26 namespace IPC_SINGLE {
27 #endif
28
29 static constexpr OHOS::HiviewDFX::HiLogLabel LOG_LABEL = { LOG_CORE, LOG_ID_IPC_COMMON, "IPCWorkThreadPool" };
30
IPCWorkThreadPool(int maxThreadNum)31 IPCWorkThreadPool::IPCWorkThreadPool(int maxThreadNum)
32 : threadSequence_(0),
33 maxThreadNum_(maxThreadNum + maxThreadNum),
34 idleThreadNum_(maxThreadNum),
35 idleSocketThreadNum_(maxThreadNum)
36 {}
37
~IPCWorkThreadPool()38 IPCWorkThreadPool::~IPCWorkThreadPool()
39 {
40 StopAllThreads();
41 threads_.clear();
42 }
43
StopAllThreads()44 void IPCWorkThreadPool::StopAllThreads()
45 {
46 std::lock_guard<std::mutex> lock(mutex_);
47 for (auto it = threads_.begin(); it != threads_.end(); it++) {
48 it->second->StopWorkThread();
49 }
50 }
51
SpawnThread(int policy,int proto)52 bool IPCWorkThreadPool::SpawnThread(int policy, int proto)
53 {
54 std::lock_guard<std::mutex> lock(mutex_);
55 if (!(proto == IRemoteObject::IF_PROT_DEFAULT && idleThreadNum_ > 0) &&
56 !(proto == IRemoteObject::IF_PROT_DATABUS && idleSocketThreadNum_ > 0)) {
57 return false;
58 }
59 std::string threadName = MakeThreadName(proto);
60 ZLOGD(LOG_LABEL, "name:%{public}s", threadName.c_str());
61
62 if (threads_.find(threadName) == threads_.end()) {
63 auto ipcThread = new (std::nothrow) IPCWorkThread(threadName);
64 if (ipcThread == nullptr) {
65 ZLOGE(LOG_LABEL, "create IPCWorkThread object failed");
66 return false;
67 }
68 sptr<IPCWorkThread> newThread = sptr<IPCWorkThread>(ipcThread);
69 threads_[threadName] = newThread;
70 if (proto == IRemoteObject::IF_PROT_DEFAULT) {
71 idleThreadNum_--;
72 ZLOGD(LOG_LABEL, "now idleThreadNum:%{public}d", idleThreadNum_);
73 }
74 if (proto == IRemoteObject::IF_PROT_DATABUS) {
75 idleSocketThreadNum_--;
76 ZLOGD(LOG_LABEL, "now idleSocketThreadNum:%{public}d", idleSocketThreadNum_);
77 }
78 newThread->Start(policy, proto, threadName);
79 return true;
80 }
81 return false;
82 }
83
MakeThreadName(int proto)84 std::string IPCWorkThreadPool::MakeThreadName(int proto)
85 {
86 int sequence = threadSequence_.fetch_add(1, std::memory_order_relaxed);
87 if (proto == IRemoteObject::IF_PROT_DATABUS) {
88 std::string threadName = "OS_DRPC";
89 return threadName + "_" + std::to_string(sequence);
90 } else {
91 std::string threadName = "OS_IPC";
92 return threadName + "_" + std::to_string(sequence);
93 }
94 }
95
RemoveThread(const std::string & threadName)96 bool IPCWorkThreadPool::RemoveThread(const std::string &threadName)
97 {
98 std::lock_guard<std::mutex> lock(mutex_);
99 auto it = threads_.find(threadName);
100 if (it != threads_.end()) {
101 sptr<IPCWorkThread> workThread = it->second;
102 if (workThread == nullptr) {
103 return false;
104 }
105 if (workThread->proto_ == IRemoteObject::IF_PROT_DEFAULT) {
106 idleThreadNum_++;
107 } else if (workThread->proto_ == IRemoteObject::IF_PROT_DATABUS) {
108 idleSocketThreadNum_++;
109 }
110 threads_.erase(it);
111 ZLOGD(LOG_LABEL, "now idleThreadNum:%{public}d", idleSocketThreadNum_);
112 return true;
113 }
114 return false;
115 }
116
GetSocketIdleThreadNum() const117 int IPCWorkThreadPool::GetSocketIdleThreadNum() const
118 {
119 return idleSocketThreadNum_;
120 }
121
GetSocketTotalThreadNum() const122 int IPCWorkThreadPool::GetSocketTotalThreadNum() const
123 {
124 return maxThreadNum_ / PROTO_NUM;
125 }
126
GetMaxThreadNum() const127 int IPCWorkThreadPool::GetMaxThreadNum() const
128 {
129 return maxThreadNum_ / PROTO_NUM;
130 }
131
UpdateMaxThreadNum(int maxThreadNum)132 void IPCWorkThreadPool::UpdateMaxThreadNum(int maxThreadNum)
133 {
134 /*
135 * not support delete thread, because thread is in using
136 */
137 int totalNum = maxThreadNum + maxThreadNum;
138 std::lock_guard<std::mutex> lock(mutex_);
139 if (totalNum <= maxThreadNum_) {
140 return;
141 }
142 int diff = totalNum - maxThreadNum_;
143 maxThreadNum_ = totalNum;
144 idleThreadNum_ += diff / PROTO_NUM;
145 idleSocketThreadNum_ += diff / PROTO_NUM;
146 }
147 #ifdef CONFIG_IPC_SINGLE
148 } // namespace IPC_SINGLE
149 #endif
150 } // namesapce OHOS
151
152