1 /*
2 * Copyright (c) Huawei Technologies Co., Ltd. 2021-2023. All rights reserved.
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 "socket_context.h"
17
18 #include <sys/socket.h>
19 #include <sys/types.h>
20 #include <unistd.h>
21 #include <vector>
22 #include "logging.h"
23 #include "securec.h"
24 #include "service_base.h"
25 #include "service_entry.h"
26
27 namespace {
28 const int PROTO_SIZE_MAX = 1024 * 1024;
29 const int MEMORY_BLOCK_UNIT = 4096;
30 const int HEARTBEAT_MSG = -1;
31 } // namespace
32
SocketContext()33 SocketContext::SocketContext()
34 {
35 socketHandle_ = -1;
36 clientState_ = CLIENT_STAT_WORKING;
37 lastProcMS_ = 0;
38 serviceBase_ = nullptr;
39 }
40
~SocketContext()41 SocketContext::~SocketContext()
42 {
43 if (socketHandle_ >= 0) {
44 PROFILER_LOG_ERROR(LOG_CORE, "SocketContext socketHandle_ = %d", socketHandle_);
45
46 int ret = shutdown(socketHandle_, SHUT_RDWR);
47 if (ret < 0) {
48 const int bufSize = 256;
49 char buf[bufSize] = { 0 };
50 strerror_r(errno, buf, bufSize);
51 PROFILER_LOG_ERROR(LOG_CORE, "shutdown socket err = %d %s", errno, buf);
52 }
53 ret = close(socketHandle_);
54 if (ret < 0) {
55 const int bufSize = 256;
56 char buf[bufSize] = { 0 };
57 strerror_r(errno, buf, bufSize);
58 PROFILER_LOG_ERROR(LOG_CORE, "close socket err = %d %s", errno, buf);
59 }
60 socketHandle_ = -1;
61 }
62 if (recvThread_.joinable()) {
63 recvThread_.join();
64 }
65 PROFILER_LOG_ERROR(LOG_CORE, "~SocketContext recvThread join success");
66 }
67
RawProtocolProc(uint32_t pnum,const int8_t * buf,const uint32_t size)68 int SocketContext::RawProtocolProc(uint32_t pnum, const int8_t* buf, const uint32_t size)
69 {
70 return -1;
71 }
72
ReceiveData(int sock,uint8_t * databuf,uint32_t size)73 bool SocketContext::ReceiveData(int sock, uint8_t* databuf, uint32_t size)
74 {
75 uint32_t p = 0;
76 if (sock < 0) {
77 return false;
78 }
79 if (size == 0) {
80 return false;
81 }
82 const uint32_t timesToTry = 2; // try 2 times
83 uint32_t nTimes = 0;
84 while (p < size) {
85 int ret = recv(sock, &databuf[p], size - p, 0);
86 if (ret > 0) {
87 p += ret;
88 } else if (ret == -1 && errno == EAGAIN) {
89 if (++nTimes >= timesToTry) {
90 return false;
91 }
92 } else {
93 return false;
94 }
95 }
96 return true;
97 }
98
UnixSocketRecv(void * pp,void (* callback)())99 void* SocketContext::UnixSocketRecv(void* pp, void (*callback)())
100 {
101 pthread_setname_np(pthread_self(), "UnixSocketRecv");
102
103 uint32_t bufferSize = MEMORY_BLOCK_UNIT;
104
105 SocketContext* pssr = (SocketContext*)pp;
106 std::vector<unsigned char> buf(bufferSize);
107
108 struct ProtocolHead* pph = (struct ProtocolHead*)buf.data();
109 uint32_t head_size = sizeof(struct ProtocolHead);
110
111 CHECK_TRUE(pssr->socketHandle_ != -1, nullptr, "UnixSocketRecv pssr->socketHandle_ ==-1");
112 while (pssr->socketHandle_ >= 0) {
113 if (!ReceiveData(pssr->socketHandle_, buf.data(), head_size)) {
114 PROFILER_LOG_DEBUG(LOG_CORE, "====== IPC LOST CONNECT ======");
115 break;
116 }
117 if (pph->protoSize > bufferSize) {
118 if (pph->protoSize > PROTO_SIZE_MAX) {
119 PROFILER_LOG_ERROR(LOG_CORE, "buffer size out of range %d/%d", pph->protoSize, PROTO_SIZE_MAX);
120 break;
121 }
122 bufferSize = (pph->protoSize / MEMORY_BLOCK_UNIT + 1) * MEMORY_BLOCK_UNIT;
123 buf.resize(bufferSize);
124 pph = (struct ProtocolHead*)buf.data();
125 }
126 if (pph->protoSize < head_size) {
127 PROFILER_LOG_ERROR(LOG_CORE, "pph->protoSize is less than head_size!");
128 break;
129 }
130 if (!ReceiveData(pssr->socketHandle_, buf.data() + head_size, pph->protoSize - head_size)) {
131 PROFILER_LOG_DEBUG(LOG_CORE, "====== IPC LOST CONNECT ======");
132 break;
133 }
134 switch (pph->protoType & PROTOCOL_TYPE_FILTER) {
135 case PROTOCOL_TYPE_RAW: {
136 pssr->RawProtocolProc(pph->protoType & (~PROTOCOL_TYPE_FILTER), pph->datas, pph->protoSize - head_size);
137 break;
138 }
139 case PROTOCOL_TYPE_PROTOBUF:
140 if (pssr->serviceBase_ != nullptr) {
141 pssr->serviceBase_->ProtocolProc(*pssr, pph->protoType & (~PROTOCOL_TYPE_FILTER), pph->datas,
142 pph->protoSize - head_size);
143 }
144 break;
145 default:
146 PROFILER_LOG_ERROR(LOG_CORE, "unknown protocol %d", pph->protoType);
147 break;
148 }
149 }
150 if (callback) {
151 callback();
152 }
153 pssr->clientState_ = CLIENT_STAT_THREAD_EXITED;
154 PROFILER_LOG_DEBUG(LOG_CORE, "UnixSocketRecv thread exit");
155 return nullptr;
156 }
157
CreateRecvThread(void (* callback)())158 bool SocketContext::CreateRecvThread(void (*callback)())
159 {
160 recvThread_ = std::thread([this, callback] { this->UnixSocketRecv(this, callback); });
161 CHECK_TRUE(recvThread_.get_id() != std::thread::id(), false, "CreateRecvThread FAIL");
162 return true;
163 }
164
SendRaw(uint32_t pnum,const int8_t * data,uint32_t size,int sockfd)165 bool SocketContext::SendRaw(uint32_t pnum, const int8_t* data, uint32_t size, int sockfd)
166 {
167 CHECK_NOTNULL(data, false, "SendRaw data null");
168 if (sockfd == -1) {
169 sockfd = socketHandle_;
170 }
171 struct ProtocolHead phead;
172 phead.protoType = PROTOCOL_TYPE_RAW | pnum;
173 phead.protoSize = size + sizeof(struct ProtocolHead);
174 CHECK_TRUE(send(sockfd, reinterpret_cast<int8_t*>(&phead), sizeof(struct ProtocolHead), 0) != -1, false,
175 "SendRaw Send Head ERR :%d", errno);
176
177 CHECK_TRUE(send(sockfd, data, size, 0) != -1, false, "SendRaw Send Data ERR : %d", errno);
178 return true;
179 }
180
181 #ifndef NO_PROTOBUF
SendProtobuf(uint32_t pnum,google::protobuf::Message & pmsg)182 bool SocketContext::SendProtobuf(uint32_t pnum, google::protobuf::Message& pmsg)
183 {
184 int size = pmsg.ByteSizeLong();
185 CHECK_TRUE(size > 0, false, "%s:size less than or equal to 0", __func__);
186 int8_t* data = reinterpret_cast<int8_t*>(malloc(size));
187 if (data == nullptr) {
188 return false;
189 }
190 pmsg.SerializeToArray(data, size);
191 struct ProtocolHead phead;
192 phead.protoType = PROTOCOL_TYPE_PROTOBUF | pnum;
193 phead.protoSize = size + sizeof(struct ProtocolHead);
194 send(socketHandle_, reinterpret_cast<int8_t*>(&phead), sizeof(struct ProtocolHead), 0);
195 send(socketHandle_, data, size, 0);
196
197 free(data);
198 return true;
199 }
200 #endif
201
SendHookConfig(const uint8_t * config,size_t len)202 bool SocketContext::SendHookConfig(const uint8_t* config, size_t len)
203 {
204 if (config == nullptr || len == 0) {
205 return false;
206 }
207
208 struct ProtocolHead phead;
209 phead.protoType = PROTOCOL_TYPE_PROTOBUF;
210 phead.protoSize = len + sizeof(struct ProtocolHead);
211 send(socketHandle_, reinterpret_cast<int8_t*>(&phead), sizeof(struct ProtocolHead), 0);
212 send(socketHandle_, config, len, 0);
213 return true;
214 }
215
SendHeartBeat()216 bool SocketContext::SendHeartBeat()
217 {
218 struct ProtocolHead phead;
219 phead.protoType = PROTOCOL_TYPE_PROTOBUF;
220 phead.protoSize = sizeof(HEARTBEAT_MSG) + sizeof(struct ProtocolHead);
221
222 if (send(socketHandle_, reinterpret_cast<int8_t*>(&phead), sizeof(struct ProtocolHead), 0) == -1) {
223 return false;
224 }
225
226 if (send(socketHandle_, &HEARTBEAT_MSG, sizeof(HEARTBEAT_MSG), 0) == -1) {
227 return false;
228 }
229
230 return true;
231 }
232
IsConnected()233 bool SocketContext::IsConnected()
234 {
235 char buf[1];
236 ssize_t recv_bytes = recv(socketHandle_, buf, sizeof(buf), MSG_DONTWAIT);
237 if (recv_bytes == 0) {
238 PROFILER_LOG_ERROR(LOG_CORE, "Peer Process Exceptionally Disconnected");
239 return false;
240 } else if (recv_bytes > 0) {
241 // This is not supposed to happen because currently native_deomon does not send data to the client.
242 return true;
243 } else {
244 return errno == EAGAIN || errno == EWOULDBLOCK;
245 }
246 }
247
SendFileDescriptor(int fd)248 bool SocketContext::SendFileDescriptor(int fd)
249 {
250 struct msghdr msg = {0};
251 struct cmsghdr* cmsg = nullptr;
252 char buf[CMSG_SPACE(1 * sizeof(int))] = {0};
253 char data;
254 if (memset_s(buf, sizeof(buf), 0, sizeof(buf)) != EOK) {
255 PROFILER_LOG_ERROR(LOG_CORE, "memset_s error!");
256 }
257 struct iovec io = {.iov_base = &data, .iov_len = 1};
258 msg.msg_iov = &io;
259 msg.msg_iovlen = 1;
260 msg.msg_control = buf;
261 msg.msg_controllen = sizeof(buf);
262 cmsg = CMSG_FIRSTHDR(&msg);
263 cmsg->cmsg_level = SOL_SOCKET;
264 cmsg->cmsg_type = SCM_RIGHTS;
265 cmsg->cmsg_len = CMSG_LEN(sizeof(int) * 1);
266 if (memcpy_s(CMSG_DATA(cmsg), 1 * sizeof(int), &fd, 1 * sizeof(int)) != EOK) {
267 PROFILER_LOG_ERROR(LOG_CORE, "memcpy_s error");
268 }
269
270 CHECK_TRUE(sendmsg(socketHandle_, &msg, 0) != -1, false, "SendFileDescriptor FAIL");
271 return true;
272 }
273
ReceiveFileDiscriptor()274 int SocketContext::ReceiveFileDiscriptor()
275 {
276 struct msghdr msg = {0};
277 struct cmsghdr* cmsg = nullptr;
278 char buf[CMSG_SPACE(1 * sizeof(int))] = {0};
279 char data;
280 struct iovec io = {.iov_base = &data, .iov_len = 1};
281 msg.msg_iov = &io;
282 msg.msg_iovlen = 1;
283 msg.msg_control = buf;
284 msg.msg_controllen = sizeof(buf);
285
286 CHECK_TRUE(recvmsg(socketHandle_, &msg, 0) != -1, -1, "ReceiveFileDiscriptor FAIL");
287 cmsg = CMSG_FIRSTHDR(&msg);
288
289 return cmsg ? *(int*)CMSG_DATA(cmsg) : -1;
290 }
291