• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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