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 numWaitingForThreads_(0)
35 {}
36
~IPCWorkThreadPool()37 IPCWorkThreadPool::~IPCWorkThreadPool()
38 {
39 StopAllThreads();
40 threads_.clear();
41 }
42
StopAllThreads()43 void IPCWorkThreadPool::StopAllThreads()
44 {
45 std::lock_guard<std::mutex> lock(mutex_);
46 for (auto it = threads_.begin(); it != threads_.end(); it++) {
47 it->second->StopWorkThread();
48 }
49 }
50
SpawnThread(int policy,int proto)51 bool IPCWorkThreadPool::SpawnThread(int policy, int proto)
52 {
53 std::lock_guard<std::mutex> lock(mutex_);
54 if (!(proto == IRemoteObject::IF_PROT_DEFAULT && idleThreadNum_ > 0) &&
55 !(proto == IRemoteObject::IF_PROT_DATABUS && idleSocketThreadNum_ > 0)) {
56 return false;
57 }
58 std::string threadName = MakeThreadName(proto);
59 ZLOGD(LOG_LABEL, "SpawnThread Name= %{public}s", threadName.c_str());
60
61 if (threads_.find(threadName) == threads_.end()) {
62 auto ipcThread = new (std::nothrow) IPCWorkThread(threadName);
63 if (ipcThread == nullptr) {
64 ZLOGE(LOG_LABEL, "create IPCWorkThread object failed");
65 return false;
66 }
67 sptr<IPCWorkThread> newThread = sptr<IPCWorkThread>(ipcThread);
68 threads_[threadName] = newThread;
69 if (proto == IRemoteObject::IF_PROT_DEFAULT) {
70 idleThreadNum_--;
71 ZLOGD(LOG_LABEL, "SpawnThread, now idleThreadNum_ =%d", idleThreadNum_);
72 }
73 if (proto == IRemoteObject::IF_PROT_DATABUS) {
74 idleSocketThreadNum_--;
75 ZLOGD(LOG_LABEL, "SpawnThread, now idleSocketThreadNum_ =%d", idleSocketThreadNum_);
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 if (numWaitingForThreads_ > 0) {
107 cv_.notify_all();
108 }
109 } else if (workThread->proto_ == IRemoteObject::IF_PROT_DATABUS) {
110 idleSocketThreadNum_++;
111 }
112 threads_.erase(it);
113 ZLOGD(LOG_LABEL, "SpawnThread, now idleThreadNum_ =%d", idleSocketThreadNum_);
114 return true;
115 }
116 return false;
117 }
118
GetSocketIdleThreadNum() const119 int IPCWorkThreadPool::GetSocketIdleThreadNum() const
120 {
121 return idleSocketThreadNum_;
122 }
123
GetSocketTotalThreadNum() const124 int IPCWorkThreadPool::GetSocketTotalThreadNum() const
125 {
126 return maxThreadNum_ / PROTO_NUM;
127 }
128
GetMaxThreadNum() const129 int IPCWorkThreadPool::GetMaxThreadNum() const
130 {
131 return maxThreadNum_ / PROTO_NUM;
132 }
133
UpdateMaxThreadNum(int maxThreadNum)134 void IPCWorkThreadPool::UpdateMaxThreadNum(int maxThreadNum)
135 {
136 /*
137 * not support delete thread, because thread is in using
138 */
139 int totalNum = maxThreadNum + maxThreadNum;
140 std::lock_guard<std::mutex> lock(mutex_);
141 if (totalNum <= maxThreadNum_) {
142 return;
143 }
144 int diff = totalNum - maxThreadNum_;
145 maxThreadNum_ = totalNum;
146 idleThreadNum_ += diff / PROTO_NUM;
147 idleSocketThreadNum_ += diff / PROTO_NUM;
148 }
149
BlockUntilThreadAvailable()150 void IPCWorkThreadPool::BlockUntilThreadAvailable()
151 {
152 std::unique_lock<std::mutex> lock(mutex_);
153 numWaitingForThreads_++;
154 while (idleThreadNum_ <= 0) {
155 cv_.wait(lock);
156 }
157 numWaitingForThreads_--;
158 }
159 #ifdef CONFIG_IPC_SINGLE
160 } // namespace IPC_SINGLE
161 #endif
162 } // namesapce OHOS
163
164