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