1 /*
2 * Copyright (C) 2020 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 #define LOG_TAG "RpcServer"
18
19 #include <inttypes.h>
20 #include <poll.h>
21 #include <sys/socket.h>
22 #include <sys/un.h>
23
24 #include <thread>
25 #include <vector>
26
27 #include <android-base/hex.h>
28 #include <android-base/scopeguard.h>
29 #include <binder/Parcel.h>
30 #include <binder/RpcServer.h>
31 #include <binder/RpcTransportRaw.h>
32 #include <log/log.h>
33 #include <utils/Compat.h>
34
35 #include "BuildFlags.h"
36 #include "FdTrigger.h"
37 #include "OS.h"
38 #include "RpcSocketAddress.h"
39 #include "RpcState.h"
40 #include "RpcTransportUtils.h"
41 #include "RpcWireFormat.h"
42 #include "Utils.h"
43
44 namespace android {
45
46 constexpr size_t kSessionIdBytes = 32;
47
48 using base::ScopeGuard;
49 using base::unique_fd;
50
RpcServer(std::unique_ptr<RpcTransportCtx> ctx)51 RpcServer::RpcServer(std::unique_ptr<RpcTransportCtx> ctx) : mCtx(std::move(ctx)) {}
~RpcServer()52 RpcServer::~RpcServer() {
53 RpcMutexUniqueLock _l(mLock);
54 LOG_ALWAYS_FATAL_IF(mShutdownTrigger != nullptr, "Must call shutdown() before destructor");
55 }
56
make(std::unique_ptr<RpcTransportCtxFactory> rpcTransportCtxFactory)57 sp<RpcServer> RpcServer::make(std::unique_ptr<RpcTransportCtxFactory> rpcTransportCtxFactory) {
58 // Default is without TLS.
59 if (rpcTransportCtxFactory == nullptr)
60 rpcTransportCtxFactory = makeDefaultRpcTransportCtxFactory();
61 auto ctx = rpcTransportCtxFactory->newServerCtx();
62 if (ctx == nullptr) return nullptr;
63 return sp<RpcServer>::make(std::move(ctx));
64 }
65
setupUnixDomainSocketBootstrapServer(unique_fd bootstrapFd)66 status_t RpcServer::setupUnixDomainSocketBootstrapServer(unique_fd bootstrapFd) {
67 return setupExternalServer(std::move(bootstrapFd), &RpcServer::recvmsgSocketConnection);
68 }
69
setupUnixDomainServer(const char * path)70 status_t RpcServer::setupUnixDomainServer(const char* path) {
71 return setupSocketServer(UnixSocketAddress(path));
72 }
73
setupVsockServer(unsigned int bindCid,unsigned int port)74 status_t RpcServer::setupVsockServer(unsigned int bindCid, unsigned int port) {
75 return setupSocketServer(VsockSocketAddress(bindCid, port));
76 }
77
setupInetServer(const char * address,unsigned int port,unsigned int * assignedPort)78 status_t RpcServer::setupInetServer(const char* address, unsigned int port,
79 unsigned int* assignedPort) {
80 if (assignedPort != nullptr) *assignedPort = 0;
81 auto aiStart = InetSocketAddress::getAddrInfo(address, port);
82 if (aiStart == nullptr) return UNKNOWN_ERROR;
83 for (auto ai = aiStart.get(); ai != nullptr; ai = ai->ai_next) {
84 InetSocketAddress socketAddress(ai->ai_addr, ai->ai_addrlen, address, port);
85 if (status_t status = setupSocketServer(socketAddress); status != OK) {
86 continue;
87 }
88
89 LOG_ALWAYS_FATAL_IF(socketAddress.addr()->sa_family != AF_INET, "expecting inet");
90 sockaddr_in addr{};
91 socklen_t len = sizeof(addr);
92 if (0 != getsockname(mServer.fd.get(), reinterpret_cast<sockaddr*>(&addr), &len)) {
93 int savedErrno = errno;
94 ALOGE("Could not getsockname at %s: %s", socketAddress.toString().c_str(),
95 strerror(savedErrno));
96 return -savedErrno;
97 }
98 LOG_ALWAYS_FATAL_IF(len != sizeof(addr), "Wrong socket type: len %zu vs len %zu",
99 static_cast<size_t>(len), sizeof(addr));
100 unsigned int realPort = ntohs(addr.sin_port);
101 LOG_ALWAYS_FATAL_IF(port != 0 && realPort != port,
102 "Requesting inet server on %s but it is set up on %u.",
103 socketAddress.toString().c_str(), realPort);
104
105 if (assignedPort != nullptr) {
106 *assignedPort = realPort;
107 }
108
109 return OK;
110 }
111 ALOGE("None of the socket address resolved for %s:%u can be set up as inet server.", address,
112 port);
113 return UNKNOWN_ERROR;
114 }
115
setMaxThreads(size_t threads)116 void RpcServer::setMaxThreads(size_t threads) {
117 LOG_ALWAYS_FATAL_IF(threads <= 0, "RpcServer is useless without threads");
118 LOG_ALWAYS_FATAL_IF(mJoinThreadRunning, "Cannot set max threads while running");
119 mMaxThreads = threads;
120 }
121
getMaxThreads()122 size_t RpcServer::getMaxThreads() {
123 return mMaxThreads;
124 }
125
setProtocolVersion(uint32_t version)126 void RpcServer::setProtocolVersion(uint32_t version) {
127 mProtocolVersion = version;
128 }
129
setSupportedFileDescriptorTransportModes(const std::vector<RpcSession::FileDescriptorTransportMode> & modes)130 void RpcServer::setSupportedFileDescriptorTransportModes(
131 const std::vector<RpcSession::FileDescriptorTransportMode>& modes) {
132 mSupportedFileDescriptorTransportModes.reset();
133 for (RpcSession::FileDescriptorTransportMode mode : modes) {
134 mSupportedFileDescriptorTransportModes.set(static_cast<size_t>(mode));
135 }
136 }
137
setRootObject(const sp<IBinder> & binder)138 void RpcServer::setRootObject(const sp<IBinder>& binder) {
139 RpcMutexLockGuard _l(mLock);
140 mRootObjectFactory = nullptr;
141 mRootObjectWeak = mRootObject = binder;
142 }
143
setRootObjectWeak(const wp<IBinder> & binder)144 void RpcServer::setRootObjectWeak(const wp<IBinder>& binder) {
145 RpcMutexLockGuard _l(mLock);
146 mRootObject.clear();
147 mRootObjectFactory = nullptr;
148 mRootObjectWeak = binder;
149 }
setPerSessionRootObject(std::function<sp<IBinder> (const void *,size_t)> && makeObject)150 void RpcServer::setPerSessionRootObject(
151 std::function<sp<IBinder>(const void*, size_t)>&& makeObject) {
152 RpcMutexLockGuard _l(mLock);
153 mRootObject.clear();
154 mRootObjectWeak.clear();
155 mRootObjectFactory = std::move(makeObject);
156 }
157
setConnectionFilter(std::function<bool (const void *,size_t)> && filter)158 void RpcServer::setConnectionFilter(std::function<bool(const void*, size_t)>&& filter) {
159 RpcMutexLockGuard _l(mLock);
160 LOG_ALWAYS_FATAL_IF(mShutdownTrigger != nullptr, "Already joined");
161 mConnectionFilter = std::move(filter);
162 }
163
getRootObject()164 sp<IBinder> RpcServer::getRootObject() {
165 RpcMutexLockGuard _l(mLock);
166 bool hasWeak = mRootObjectWeak.unsafe_get();
167 sp<IBinder> ret = mRootObjectWeak.promote();
168 ALOGW_IF(hasWeak && ret == nullptr, "RpcServer root object is freed, returning nullptr");
169 return ret;
170 }
171
getCertificate(RpcCertificateFormat format)172 std::vector<uint8_t> RpcServer::getCertificate(RpcCertificateFormat format) {
173 RpcMutexLockGuard _l(mLock);
174 return mCtx->getCertificate(format);
175 }
176
joinRpcServer(sp<RpcServer> && thiz)177 static void joinRpcServer(sp<RpcServer>&& thiz) {
178 thiz->join();
179 }
180
start()181 void RpcServer::start() {
182 RpcMutexLockGuard _l(mLock);
183 LOG_ALWAYS_FATAL_IF(mJoinThread.get(), "Already started!");
184 mJoinThread =
185 std::make_unique<RpcMaybeThread>(&joinRpcServer, sp<RpcServer>::fromExisting(this));
186 rpcJoinIfSingleThreaded(*mJoinThread);
187 }
188
acceptSocketConnection(const RpcServer & server,RpcTransportFd * out)189 status_t RpcServer::acceptSocketConnection(const RpcServer& server, RpcTransportFd* out) {
190 RpcTransportFd clientSocket(unique_fd(TEMP_FAILURE_RETRY(
191 accept4(server.mServer.fd.get(), nullptr, nullptr, SOCK_CLOEXEC | SOCK_NONBLOCK))));
192 if (clientSocket.fd < 0) {
193 int savedErrno = errno;
194 ALOGE("Could not accept4 socket: %s", strerror(savedErrno));
195 return -savedErrno;
196 }
197
198 *out = std::move(clientSocket);
199 return OK;
200 }
201
recvmsgSocketConnection(const RpcServer & server,RpcTransportFd * out)202 status_t RpcServer::recvmsgSocketConnection(const RpcServer& server, RpcTransportFd* out) {
203 int zero = 0;
204 iovec iov{&zero, sizeof(zero)};
205 std::vector<std::variant<base::unique_fd, base::borrowed_fd>> fds;
206
207 ssize_t num_bytes = receiveMessageFromSocket(server.mServer, &iov, 1, &fds);
208 if (num_bytes < 0) {
209 int savedErrno = errno;
210 ALOGE("Failed recvmsg: %s", strerror(savedErrno));
211 return -savedErrno;
212 }
213 if (num_bytes == 0) {
214 return DEAD_OBJECT;
215 }
216 if (fds.size() != 1) {
217 ALOGE("Expected exactly one fd from recvmsg, got %zu", fds.size());
218 return -EINVAL;
219 }
220
221 unique_fd fd(std::move(std::get<unique_fd>(fds.back())));
222 if (auto res = setNonBlocking(fd); !res.ok()) {
223 ALOGE("Failed setNonBlocking: %s", res.error().message().c_str());
224 return res.error().code() == 0 ? UNKNOWN_ERROR : -res.error().code();
225 }
226
227 *out = RpcTransportFd(std::move(fd));
228 return OK;
229 }
230
join()231 void RpcServer::join() {
232
233 {
234 RpcMutexLockGuard _l(mLock);
235 LOG_ALWAYS_FATAL_IF(!mServer.fd.ok(), "RpcServer must be setup to join.");
236 LOG_ALWAYS_FATAL_IF(mAcceptFn == nullptr, "RpcServer must have an accept() function");
237 LOG_ALWAYS_FATAL_IF(mShutdownTrigger != nullptr, "Already joined");
238 mJoinThreadRunning = true;
239 mShutdownTrigger = FdTrigger::make();
240 LOG_ALWAYS_FATAL_IF(mShutdownTrigger == nullptr, "Cannot create join signaler");
241 }
242
243 status_t status;
244 while ((status = mShutdownTrigger->triggerablePoll(mServer, POLLIN)) == OK) {
245 std::array<uint8_t, kRpcAddressSize> addr;
246 static_assert(addr.size() >= sizeof(sockaddr_storage), "kRpcAddressSize is too small");
247 socklen_t addrLen = addr.size();
248
249 RpcTransportFd clientSocket;
250 if ((status = mAcceptFn(*this, &clientSocket)) != OK) {
251 if (status == DEAD_OBJECT)
252 break;
253 else
254 continue;
255 }
256
257 LOG_RPC_DETAIL("accept on fd %d yields fd %d", mServer.fd.get(), clientSocket.fd.get());
258
259 if (getpeername(clientSocket.fd.get(), reinterpret_cast<sockaddr*>(addr.data()),
260 &addrLen)) {
261 ALOGE("Could not getpeername socket: %s", strerror(errno));
262 continue;
263 }
264
265 if (mConnectionFilter != nullptr && !mConnectionFilter(addr.data(), addrLen)) {
266 ALOGE("Dropped client connection fd %d", clientSocket.fd.get());
267 continue;
268 }
269
270 {
271 RpcMutexLockGuard _l(mLock);
272 RpcMaybeThread thread =
273 RpcMaybeThread(&RpcServer::establishConnection,
274 sp<RpcServer>::fromExisting(this), std::move(clientSocket), addr,
275 addrLen, RpcSession::join);
276
277 auto& threadRef = mConnectingThreads[thread.get_id()];
278 threadRef = std::move(thread);
279 rpcJoinIfSingleThreaded(threadRef);
280 }
281 }
282 LOG_RPC_DETAIL("RpcServer::join exiting with %s", statusToString(status).c_str());
283
284 if constexpr (kEnableRpcThreads) {
285 RpcMutexLockGuard _l(mLock);
286 mJoinThreadRunning = false;
287 } else {
288 // Multi-threaded builds clear this in shutdown(), but we need it valid
289 // so the loop above exits cleanly
290 mShutdownTrigger = nullptr;
291 }
292 mShutdownCv.notify_all();
293 }
294
shutdown()295 bool RpcServer::shutdown() {
296 RpcMutexUniqueLock _l(mLock);
297 if (mShutdownTrigger == nullptr) {
298 LOG_RPC_DETAIL("Cannot shutdown. No shutdown trigger installed (already shutdown, or not "
299 "joined yet?)");
300 return false;
301 }
302
303 mShutdownTrigger->trigger();
304
305 for (auto& [id, session] : mSessions) {
306 (void)id;
307 // server lock is a more general lock
308 RpcMutexLockGuard _lSession(session->mMutex);
309 session->mShutdownTrigger->trigger();
310 }
311
312 if constexpr (!kEnableRpcThreads) {
313 // In single-threaded mode we're done here, everything else that
314 // needs to happen should be at the end of RpcServer::join()
315 return true;
316 }
317
318 while (mJoinThreadRunning || !mConnectingThreads.empty() || !mSessions.empty()) {
319 if (std::cv_status::timeout == mShutdownCv.wait_for(_l, std::chrono::seconds(1))) {
320 ALOGE("Waiting for RpcServer to shut down (1s w/o progress). Join thread running: %d, "
321 "Connecting threads: "
322 "%zu, Sessions: %zu. Is your server deadlocked?",
323 mJoinThreadRunning, mConnectingThreads.size(), mSessions.size());
324 }
325 }
326
327 // At this point, we know join() is about to exit, but the thread that calls
328 // join() may not have exited yet.
329 // If RpcServer owns the join thread (aka start() is called), make sure the thread exits;
330 // otherwise ~thread() may call std::terminate(), which may crash the process.
331 // If RpcServer does not own the join thread (aka join() is called directly),
332 // then the owner of RpcServer is responsible for cleaning up that thread.
333 if (mJoinThread.get()) {
334 mJoinThread->join();
335 mJoinThread.reset();
336 }
337
338 LOG_RPC_DETAIL("Finished waiting on shutdown.");
339
340 mShutdownTrigger = nullptr;
341 return true;
342 }
343
listSessions()344 std::vector<sp<RpcSession>> RpcServer::listSessions() {
345 RpcMutexLockGuard _l(mLock);
346 std::vector<sp<RpcSession>> sessions;
347 for (auto& [id, session] : mSessions) {
348 (void)id;
349 sessions.push_back(session);
350 }
351 return sessions;
352 }
353
numUninitializedSessions()354 size_t RpcServer::numUninitializedSessions() {
355 RpcMutexLockGuard _l(mLock);
356 return mConnectingThreads.size();
357 }
358
establishConnection(sp<RpcServer> && server,RpcTransportFd clientFd,std::array<uint8_t,kRpcAddressSize> addr,size_t addrLen,std::function<void (sp<RpcSession> &&,RpcSession::PreJoinSetupResult &&)> && joinFn)359 void RpcServer::establishConnection(
360 sp<RpcServer>&& server, RpcTransportFd clientFd, std::array<uint8_t, kRpcAddressSize> addr,
361 size_t addrLen,
362 std::function<void(sp<RpcSession>&&, RpcSession::PreJoinSetupResult&&)>&& joinFn) {
363 // mShutdownTrigger can only be cleared once connection threads have joined.
364 // It must be set before this thread is started
365 LOG_ALWAYS_FATAL_IF(server->mShutdownTrigger == nullptr);
366 LOG_ALWAYS_FATAL_IF(server->mCtx == nullptr);
367
368 status_t status = OK;
369
370 int clientFdForLog = clientFd.fd.get();
371 auto client = server->mCtx->newTransport(std::move(clientFd), server->mShutdownTrigger.get());
372 if (client == nullptr) {
373 ALOGE("Dropping accept4()-ed socket because sslAccept fails");
374 status = DEAD_OBJECT;
375 // still need to cleanup before we can return
376 } else {
377 LOG_RPC_DETAIL("Created RpcTransport %p for client fd %d", client.get(), clientFdForLog);
378 }
379
380 RpcConnectionHeader header;
381 if (status == OK) {
382 iovec iov{&header, sizeof(header)};
383 status = client->interruptableReadFully(server->mShutdownTrigger.get(), &iov, 1,
384 std::nullopt, /*ancillaryFds=*/nullptr);
385 if (status != OK) {
386 ALOGE("Failed to read ID for client connecting to RPC server: %s",
387 statusToString(status).c_str());
388 // still need to cleanup before we can return
389 }
390 }
391
392 std::vector<uint8_t> sessionId;
393 if (status == OK) {
394 if (header.sessionIdSize > 0) {
395 if (header.sessionIdSize == kSessionIdBytes) {
396 sessionId.resize(header.sessionIdSize);
397 iovec iov{sessionId.data(), sessionId.size()};
398 status = client->interruptableReadFully(server->mShutdownTrigger.get(), &iov, 1,
399 std::nullopt, /*ancillaryFds=*/nullptr);
400 if (status != OK) {
401 ALOGE("Failed to read session ID for client connecting to RPC server: %s",
402 statusToString(status).c_str());
403 // still need to cleanup before we can return
404 }
405 } else {
406 ALOGE("Malformed session ID. Expecting session ID of size %zu but got %" PRIu16,
407 kSessionIdBytes, header.sessionIdSize);
408 status = BAD_VALUE;
409 }
410 }
411 }
412
413 bool incoming = false;
414 uint32_t protocolVersion = 0;
415 bool requestingNewSession = false;
416
417 if (status == OK) {
418 incoming = header.options & RPC_CONNECTION_OPTION_INCOMING;
419 protocolVersion = std::min(header.version,
420 server->mProtocolVersion.value_or(RPC_WIRE_PROTOCOL_VERSION));
421 requestingNewSession = sessionId.empty();
422
423 if (requestingNewSession) {
424 RpcNewSessionResponse response{
425 .version = protocolVersion,
426 };
427
428 iovec iov{&response, sizeof(response)};
429 status = client->interruptableWriteFully(server->mShutdownTrigger.get(), &iov, 1,
430 std::nullopt, nullptr);
431 if (status != OK) {
432 ALOGE("Failed to send new session response: %s", statusToString(status).c_str());
433 // still need to cleanup before we can return
434 }
435 }
436 }
437
438 RpcMaybeThread thisThread;
439 sp<RpcSession> session;
440 {
441 RpcMutexUniqueLock _l(server->mLock);
442
443 auto threadId = server->mConnectingThreads.find(rpc_this_thread::get_id());
444 LOG_ALWAYS_FATAL_IF(threadId == server->mConnectingThreads.end(),
445 "Must establish connection on owned thread");
446 thisThread = std::move(threadId->second);
447 ScopeGuard detachGuard = [&]() {
448 thisThread.detach();
449 _l.unlock();
450 server->mShutdownCv.notify_all();
451 };
452 server->mConnectingThreads.erase(threadId);
453
454 if (status != OK || server->mShutdownTrigger->isTriggered()) {
455 return;
456 }
457
458 if (requestingNewSession) {
459 if (incoming) {
460 ALOGE("Cannot create a new session with an incoming connection, would leak");
461 return;
462 }
463
464 // Uniquely identify session at the application layer. Even if a
465 // client/server use the same certificates, if they create multiple
466 // sessions, we still want to distinguish between them.
467 sessionId.resize(kSessionIdBytes);
468 size_t tries = 0;
469 do {
470 // don't block if there is some entropy issue
471 if (tries++ > 5) {
472 ALOGE("Cannot find new address: %s",
473 base::HexString(sessionId.data(), sessionId.size()).c_str());
474 return;
475 }
476
477 auto status = getRandomBytes(sessionId.data(), sessionId.size());
478 if (status != OK) {
479 ALOGE("Failed to read random session ID: %s", strerror(-status));
480 return;
481 }
482 } while (server->mSessions.end() != server->mSessions.find(sessionId));
483
484 session = sp<RpcSession>::make(nullptr);
485 session->setMaxIncomingThreads(server->mMaxThreads);
486 if (!session->setProtocolVersion(protocolVersion)) return;
487
488 if (header.fileDescriptorTransportMode <
489 server->mSupportedFileDescriptorTransportModes.size() &&
490 server->mSupportedFileDescriptorTransportModes.test(
491 header.fileDescriptorTransportMode)) {
492 session->setFileDescriptorTransportMode(
493 static_cast<RpcSession::FileDescriptorTransportMode>(
494 header.fileDescriptorTransportMode));
495 } else {
496 ALOGE("Rejecting connection: FileDescriptorTransportMode is not supported: %hhu",
497 header.fileDescriptorTransportMode);
498 return;
499 }
500
501 // if null, falls back to server root
502 sp<IBinder> sessionSpecificRoot;
503 if (server->mRootObjectFactory != nullptr) {
504 sessionSpecificRoot = server->mRootObjectFactory(addr.data(), addrLen);
505 if (sessionSpecificRoot == nullptr) {
506 ALOGE("Warning: server returned null from root object factory");
507 }
508 }
509
510 if (!session->setForServer(server,
511 sp<RpcServer::EventListener>::fromExisting(
512 static_cast<RpcServer::EventListener*>(
513 server.get())),
514 sessionId, sessionSpecificRoot)) {
515 ALOGE("Failed to attach server to session");
516 return;
517 }
518
519 server->mSessions[sessionId] = session;
520 } else {
521 auto it = server->mSessions.find(sessionId);
522 if (it == server->mSessions.end()) {
523 ALOGE("Cannot add thread, no record of session with ID %s",
524 base::HexString(sessionId.data(), sessionId.size()).c_str());
525 return;
526 }
527 session = it->second;
528 }
529
530 if (incoming) {
531 LOG_ALWAYS_FATAL_IF(OK != session->addOutgoingConnection(std::move(client), true),
532 "server state must already be initialized");
533 return;
534 }
535
536 detachGuard.Disable();
537 session->preJoinThreadOwnership(std::move(thisThread));
538 }
539
540 auto setupResult = session->preJoinSetup(std::move(client));
541
542 // avoid strong cycle
543 server = nullptr;
544
545 joinFn(std::move(session), std::move(setupResult));
546 }
547
setupSocketServer(const RpcSocketAddress & addr)548 status_t RpcServer::setupSocketServer(const RpcSocketAddress& addr) {
549 LOG_RPC_DETAIL("Setting up socket server %s", addr.toString().c_str());
550 LOG_ALWAYS_FATAL_IF(hasServer(), "Each RpcServer can only have one server.");
551
552 unique_fd socket_fd(TEMP_FAILURE_RETRY(
553 socket(addr.addr()->sa_family, SOCK_STREAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0)));
554 if (!socket_fd.ok()) {
555 int savedErrno = errno;
556 ALOGE("Could not create socket at %s: %s", addr.toString().c_str(), strerror(savedErrno));
557 return -savedErrno;
558 }
559 if (0 != TEMP_FAILURE_RETRY(bind(socket_fd.get(), addr.addr(), addr.addrSize()))) {
560 int savedErrno = errno;
561 ALOGE("Could not bind socket at %s: %s", addr.toString().c_str(), strerror(savedErrno));
562 return -savedErrno;
563 }
564
565 return setupRawSocketServer(std::move(socket_fd));
566 }
567
setupRawSocketServer(unique_fd socket_fd)568 status_t RpcServer::setupRawSocketServer(unique_fd socket_fd) {
569 LOG_ALWAYS_FATAL_IF(!socket_fd.ok(), "Socket must be setup to listen.");
570
571 // Right now, we create all threads at once, making accept4 slow. To avoid hanging the client,
572 // the backlog is increased to a large number.
573 // TODO(b/189955605): Once we create threads dynamically & lazily, the backlog can be reduced
574 // to 1.
575 if (0 != TEMP_FAILURE_RETRY(listen(socket_fd.get(), 50 /*backlog*/))) {
576 int savedErrno = errno;
577 ALOGE("Could not listen initialized Unix socket: %s", strerror(savedErrno));
578 return -savedErrno;
579 }
580 if (status_t status = setupExternalServer(std::move(socket_fd)); status != OK) {
581 ALOGE("Another thread has set up server while calling setupSocketServer. Race?");
582 return status;
583 }
584 return OK;
585 }
586
onSessionAllIncomingThreadsEnded(const sp<RpcSession> & session)587 void RpcServer::onSessionAllIncomingThreadsEnded(const sp<RpcSession>& session) {
588 const std::vector<uint8_t>& id = session->mId;
589 LOG_ALWAYS_FATAL_IF(id.empty(), "Server sessions must be initialized with ID");
590 LOG_RPC_DETAIL("Dropping session with address %s",
591 base::HexString(id.data(), id.size()).c_str());
592
593 RpcMutexLockGuard _l(mLock);
594 auto it = mSessions.find(id);
595 LOG_ALWAYS_FATAL_IF(it == mSessions.end(), "Bad state, unknown session id %s",
596 base::HexString(id.data(), id.size()).c_str());
597 LOG_ALWAYS_FATAL_IF(it->second != session, "Bad state, session has id mismatch %s",
598 base::HexString(id.data(), id.size()).c_str());
599 (void)mSessions.erase(it);
600 }
601
onSessionIncomingThreadEnded()602 void RpcServer::onSessionIncomingThreadEnded() {
603 mShutdownCv.notify_all();
604 }
605
hasServer()606 bool RpcServer::hasServer() {
607 RpcMutexLockGuard _l(mLock);
608 return mServer.fd.ok();
609 }
610
releaseServer()611 unique_fd RpcServer::releaseServer() {
612 RpcMutexLockGuard _l(mLock);
613 return std::move(mServer.fd);
614 }
615
setupExternalServer(base::unique_fd serverFd,std::function<status_t (const RpcServer &,RpcTransportFd *)> && acceptFn)616 status_t RpcServer::setupExternalServer(
617 base::unique_fd serverFd,
618 std::function<status_t(const RpcServer&, RpcTransportFd*)>&& acceptFn) {
619 RpcMutexLockGuard _l(mLock);
620 if (mServer.fd.ok()) {
621 ALOGE("Each RpcServer can only have one server.");
622 return INVALID_OPERATION;
623 }
624 mServer = std::move(serverFd);
625 mAcceptFn = std::move(acceptFn);
626 return OK;
627 }
628
setupExternalServer(base::unique_fd serverFd)629 status_t RpcServer::setupExternalServer(base::unique_fd serverFd) {
630 return setupExternalServer(std::move(serverFd), &RpcServer::acceptSocketConnection);
631 }
632
hasActiveRequests()633 bool RpcServer::hasActiveRequests() {
634 RpcMutexLockGuard _l(mLock);
635 for (const auto& [_, session] : mSessions) {
636 if (session->hasActiveRequests()) {
637 return true;
638 }
639 }
640 return !mServer.isInPollingState();
641 }
642
643 } // namespace android
644