1 /*
2 * Copyright (C) 2017 The Android Open Source Project
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 * http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17 #include "chre_host/socket_client.h"
18
19 #include <inttypes.h>
20
21 #include <string.h>
22 #include <unistd.h>
23
24 #include <chrono>
25
26 #include <cutils/sockets.h>
27 #include <sys/epoll.h>
28 #include <sys/socket.h>
29 #include <utils/RefBase.h>
30 #include <utils/StrongPointer.h>
31
32 #include "chre_host/log.h"
33
34 namespace android {
35 namespace chre {
36
SocketClient()37 SocketClient::SocketClient() {
38 std::atomic_init(&mSockFd, INVALID_SOCKET);
39 }
40
~SocketClient()41 SocketClient::~SocketClient() {
42 disconnect();
43 }
44
connect(const char * socketName,const sp<ICallbacks> & callbacks)45 bool SocketClient::connect(const char *socketName,
46 const sp<ICallbacks> &callbacks) {
47 return doConnect(socketName, callbacks, false /* connectInBackground */);
48 }
49
connectInBackground(const char * socketName,const sp<ICallbacks> & callbacks)50 bool SocketClient::connectInBackground(const char *socketName,
51 const sp<ICallbacks> &callbacks) {
52 return doConnect(socketName, callbacks, true /* connectInBackground */);
53 }
54
disconnect()55 void SocketClient::disconnect() {
56 if (inReceiveThread()) {
57 LOGE("disconnect() can't be called from a receive thread callback");
58 } else if (receiveThreadRunning()) {
59 // Inform the RX thread that we're requesting a shutdown, breaking it out of
60 // the retry wait if it's currently blocked there
61 {
62 std::lock_guard<std::mutex> lock(mShutdownMutex);
63 mGracefulShutdown = true;
64 }
65 mShutdownCond.notify_all();
66
67 // Invalidate the socket (will kick the RX thread out of recv if it's
68 // currently blocked there)
69 if (mSockFd != INVALID_SOCKET && shutdown(mSockFd, SHUT_RDWR) != 0) {
70 LOG_ERROR("Couldn't shut down socket", errno);
71 }
72
73 if (mRxThread.joinable()) {
74 LOGD("Waiting for RX thread to exit");
75 mRxThread.join();
76 }
77 }
78 }
79
isConnected() const80 bool SocketClient::isConnected() const {
81 return (mSockFd != INVALID_SOCKET);
82 }
83
sendMessage(const void * data,size_t length)84 bool SocketClient::sendMessage(const void *data, size_t length) {
85 bool success = false;
86
87 if (mSockFd == INVALID_SOCKET) {
88 LOGW("Tried sending a message, but don't have a valid socket handle");
89 } else {
90 ssize_t bytesSent = send(mSockFd, data, length, 0);
91 if (bytesSent < 0) {
92 LOGE("Failed to send %zu bytes of data: %s", length, strerror(errno));
93 } else if (bytesSent == 0) {
94 LOGW("Failed to send data; remote side disconnected");
95 } else if (static_cast<size_t>(bytesSent) != length) {
96 LOGW("Truncated packet, tried sending %zu bytes, only %zd went through",
97 length, bytesSent);
98 } else {
99 success = true;
100 }
101 }
102
103 return success;
104 }
105
doConnect(const char * socketName,const sp<ICallbacks> & callbacks,bool connectInBackground)106 bool SocketClient::doConnect(const char *socketName,
107 const sp<ICallbacks> &callbacks,
108 bool connectInBackground) {
109 bool success = false;
110 if (inReceiveThread()) {
111 LOGE("Can't attempt to connect from a receive thread callback");
112 } else {
113 if (receiveThreadRunning()) {
114 LOGW("Re-connecting socket with implicit disconnect");
115 disconnect();
116 }
117
118 size_t socketNameLen =
119 strlcpy(mSocketName, socketName, sizeof(mSocketName));
120 if (socketNameLen >= sizeof(mSocketName)) {
121 LOGE("Socket name length parameter is too long (%zu, max %zu)",
122 socketNameLen, sizeof(mSocketName));
123 } else if (callbacks == nullptr) {
124 LOGE("Callbacks parameter must be provided");
125 } else if (connectInBackground || tryConnect()) {
126 mGracefulShutdown = false;
127 mCallbacks = callbacks;
128 mRxThread = std::thread([this]() { receiveThread(); });
129 success = true;
130 }
131 }
132
133 return success;
134 }
135
inReceiveThread() const136 bool SocketClient::inReceiveThread() const {
137 return (std::this_thread::get_id() == mRxThread.get_id());
138 }
139
receiveThread()140 void SocketClient::receiveThread() {
141 constexpr size_t kReceiveBufferSize = 4096;
142 uint8_t buffer[kReceiveBufferSize];
143
144 LOGV("Receive thread started");
145 while (!mGracefulShutdown && (mSockFd != INVALID_SOCKET || reconnect())) {
146 struct epoll_event requestedEvent;
147 requestedEvent.data.fd = mSockFd;
148 requestedEvent.events = EPOLLIN | EPOLLWAKEUP;
149
150 int epollFd = TEMP_FAILURE_RETRY(epoll_create1(0));
151 if (epollFd < 0) {
152 LOG_ERROR("Error creating epoll fd", errno);
153 break;
154 }
155
156 if (TEMP_FAILURE_RETRY(epoll_ctl(epollFd, EPOLL_CTL_ADD,
157 requestedEvent.data.fd, &requestedEvent)) <
158 0) {
159 LOG_ERROR("Error adding socket fd to epoll", errno);
160 close(epollFd);
161 break;
162 }
163
164 while (!mGracefulShutdown) {
165 struct epoll_event returnedEvent;
166 // Blockingly wait for the next epoll event. The implicit wakelock will be
167 // held until the next call to epoll_wait on the same epoll file
168 // descriptor
169 int eventsReady = TEMP_FAILURE_RETRY(epoll_wait(epollFd, &returnedEvent,
170 /* event_count= */ 1,
171 /* timeout_ms= */ -1));
172 if (eventsReady < 0) {
173 LOG_ERROR("Poll error", errno);
174 break;
175 }
176
177 ssize_t bytesReceived = recv(mSockFd, buffer, sizeof(buffer), 0);
178 if (bytesReceived < 0) {
179 LOG_ERROR("Exiting RX thread", errno);
180 break;
181 } else if (bytesReceived == 0) {
182 if (!mGracefulShutdown) {
183 LOGI("Socket disconnected on remote end");
184 mCallbacks->onDisconnected();
185 }
186 break;
187 }
188
189 mCallbacks->onMessageReceived(buffer, bytesReceived);
190 }
191
192 if (close(mSockFd) != 0) {
193 LOG_ERROR("Couldn't close socket", errno);
194 }
195 mSockFd = INVALID_SOCKET;
196 close(epollFd);
197 }
198
199 if (!mGracefulShutdown) {
200 mCallbacks->onConnectionAborted();
201 }
202
203 mCallbacks.clear();
204 LOGV("Exiting receive thread");
205 }
206
receiveThreadRunning() const207 bool SocketClient::receiveThreadRunning() const {
208 return mRxThread.joinable();
209 }
210
reconnect()211 bool SocketClient::reconnect() {
212 constexpr auto kMinDelay = std::chrono::duration<int32_t, std::milli>(250);
213 constexpr auto kMaxDelay = std::chrono::minutes(5);
214 // Try reconnecting at initial delay this many times before backing off
215 constexpr unsigned int kExponentialBackoffDelay =
216 std::chrono::seconds(10) / kMinDelay;
217 // Give up after this many tries (~2.5 hours)
218 constexpr unsigned int kRetryLimit = kExponentialBackoffDelay + 40;
219 auto delay = kMinDelay;
220 unsigned int retryCount = 0;
221
222 while (retryCount++ < kRetryLimit) {
223 {
224 std::unique_lock<std::mutex> lock(mShutdownMutex);
225 mShutdownCond.wait_for(lock, delay,
226 [this]() { return mGracefulShutdown.load(); });
227 if (mGracefulShutdown) {
228 break;
229 }
230 }
231
232 bool suppressErrorLogs = (delay == kMinDelay);
233 if (!tryConnect(suppressErrorLogs)) {
234 if (!suppressErrorLogs) {
235 LOGW("Failed to (re)connect, next try in %" PRId32 " ms",
236 delay.count());
237 }
238 if (retryCount > kExponentialBackoffDelay) {
239 delay *= 2;
240 }
241 if (delay > kMaxDelay) {
242 delay = kMaxDelay;
243 }
244 } else {
245 LOGD("Successfully (re)connected");
246 mCallbacks->onConnected();
247 return true;
248 }
249 }
250
251 return false;
252 }
253
tryConnect(bool suppressErrorLogs)254 bool SocketClient::tryConnect(bool suppressErrorLogs) {
255 bool success = false;
256
257 errno = 0;
258 int sockFd = socket(AF_LOCAL, SOCK_SEQPACKET, 0);
259 if (sockFd >= 0) {
260 // Set the send buffer size to 2MB to allow plenty of room for nanoapp
261 // loading
262 int sndbuf = 2 * 1024 * 1024;
263 // Normally, send() should effectively return immediately, but in the event
264 // that we get blocked due to flow control, don't stay blocked for more than
265 // 3 seconds
266 struct timeval timeout = {
267 .tv_sec = 3,
268 .tv_usec = 0,
269 };
270 int ret;
271
272 if ((ret = setsockopt(sockFd, SOL_SOCKET, SO_SNDBUF, &sndbuf,
273 sizeof(sndbuf))) != 0) {
274 if (!suppressErrorLogs) {
275 LOGE("Failed to set SO_SNDBUF to %d: %s", sndbuf, strerror(errno));
276 }
277 } else if ((ret = setsockopt(sockFd, SOL_SOCKET, SO_SNDTIMEO, &timeout,
278 sizeof(timeout))) != 0) {
279 if (!suppressErrorLogs) {
280 LOGE("Failed to set SO_SNDTIMEO: %s", strerror(errno));
281 }
282 } else {
283 mSockFd = socket_local_client_connect(sockFd, mSocketName,
284 ANDROID_SOCKET_NAMESPACE_RESERVED,
285 SOCK_SEQPACKET);
286 if (mSockFd != INVALID_SOCKET) {
287 success = true;
288 } else if (!suppressErrorLogs) {
289 LOGE("Couldn't connect client socket to '%s': %s", mSocketName,
290 strerror(errno));
291 }
292 }
293
294 if (!success) {
295 close(sockFd);
296 }
297 } else if (!suppressErrorLogs) {
298 LOGE("Couldn't create local socket: %s", strerror(errno));
299 }
300
301 return success;
302 }
303
304 } // namespace chre
305 } // namespace android
306