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