blob: 65f6bc68c9e077f2308d055ea14d1ebe3466eefc [file] [log] [blame]
Steven Morelandbdb53ab2021-05-05 17:57:41 +00001/*
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 "RpcSession"
18
19#include <binder/RpcSession.h>
20
Yifan Hong194acf22021-06-29 18:44:56 -070021#include <dlfcn.h>
Steven Morelandbdb53ab2021-05-05 17:57:41 +000022#include <inttypes.h>
Steven Moreland4ec3c432021-05-20 00:32:47 +000023#include <poll.h>
Yifan Hong194acf22021-06-29 18:44:56 -070024#include <pthread.h>
Steven Morelandbdb53ab2021-05-05 17:57:41 +000025#include <unistd.h>
26
27#include <string_view>
28
Steven Moreland826367f2021-09-10 14:05:31 -070029#include <android-base/hex.h>
Steven Moreland4ec3c432021-05-20 00:32:47 +000030#include <android-base/macros.h>
Yifan Hong194acf22021-06-29 18:44:56 -070031#include <android_runtime/vm.h>
Steven Moreland4f622fe2021-09-13 17:38:09 -070032#include <binder/BpBinder.h>
Steven Morelandbdb53ab2021-05-05 17:57:41 +000033#include <binder/Parcel.h>
Steven Morelandee78e762021-05-05 21:12:51 +000034#include <binder/RpcServer.h>
Yifan Hong702115c2021-06-24 15:39:18 -070035#include <binder/RpcTransportRaw.h>
Steven Morelandbdb53ab2021-05-05 17:57:41 +000036#include <binder/Stability.h>
Yifan Hong194acf22021-06-29 18:44:56 -070037#include <jni.h>
Steven Morelandbdb53ab2021-05-05 17:57:41 +000038#include <utils/String8.h>
39
Yifan Hong8c950422021-08-05 17:13:55 -070040#include "FdTrigger.h"
Steven Morelandbdb53ab2021-05-05 17:57:41 +000041#include "RpcSocketAddress.h"
42#include "RpcState.h"
43#include "RpcWireFormat.h"
Yifan Hongb675ffe2021-08-05 16:37:17 -070044#include "Utils.h"
Steven Morelandbdb53ab2021-05-05 17:57:41 +000045
46#ifdef __GLIBC__
47extern "C" pid_t gettid();
48#endif
49
50namespace android {
51
52using base::unique_fd;
53
Yifan Hongecf937d2021-08-11 17:29:28 -070054RpcSession::RpcSession(std::unique_ptr<RpcTransportCtx> ctx) : mCtx(std::move(ctx)) {
Steven Morelandbdb53ab2021-05-05 17:57:41 +000055 LOG_RPC_DETAIL("RpcSession created %p", this);
56
57 mState = std::make_unique<RpcState>();
58}
59RpcSession::~RpcSession() {
60 LOG_RPC_DETAIL("RpcSession destroyed %p", this);
61
62 std::lock_guard<std::mutex> _l(mMutex);
Steven Moreland19fc9f72021-06-10 03:57:30 +000063 LOG_ALWAYS_FATAL_IF(mIncomingConnections.size() != 0,
Steven Morelandbdb53ab2021-05-05 17:57:41 +000064 "Should not be able to destroy a session with servers in use.");
65}
66
Yifan Hongecf937d2021-08-11 17:29:28 -070067sp<RpcSession> RpcSession::make() {
Yifan Hong702115c2021-06-24 15:39:18 -070068 // Default is without TLS.
Yifan Hongfdd9f692021-09-09 15:12:52 -070069 return make(RpcTransportCtxFactoryRaw::make());
Yifan Hongecf937d2021-08-11 17:29:28 -070070}
71
Yifan Hongfdd9f692021-09-09 15:12:52 -070072sp<RpcSession> RpcSession::make(std::unique_ptr<RpcTransportCtxFactory> rpcTransportCtxFactory) {
Yifan Hongecf937d2021-08-11 17:29:28 -070073 auto ctx = rpcTransportCtxFactory->newClientCtx();
74 if (ctx == nullptr) return nullptr;
Yifan Hongecf937d2021-08-11 17:29:28 -070075 return sp<RpcSession>::make(std::move(ctx));
Steven Morelandbdb53ab2021-05-05 17:57:41 +000076}
77
Steven Moreland103424e2021-06-02 18:16:19 +000078void RpcSession::setMaxThreads(size_t threads) {
79 std::lock_guard<std::mutex> _l(mMutex);
Steven Moreland19fc9f72021-06-10 03:57:30 +000080 LOG_ALWAYS_FATAL_IF(!mOutgoingConnections.empty() || !mIncomingConnections.empty(),
Steven Moreland103424e2021-06-02 18:16:19 +000081 "Must set max threads before setting up connections, but has %zu client(s) "
82 "and %zu server(s)",
Steven Moreland19fc9f72021-06-10 03:57:30 +000083 mOutgoingConnections.size(), mIncomingConnections.size());
Steven Moreland103424e2021-06-02 18:16:19 +000084 mMaxThreads = threads;
85}
86
87size_t RpcSession::getMaxThreads() {
88 std::lock_guard<std::mutex> _l(mMutex);
89 return mMaxThreads;
Steven Moreland659416d2021-05-11 00:47:50 +000090}
91
Steven Morelandbf57bce2021-07-26 15:26:12 -070092bool RpcSession::setProtocolVersion(uint32_t version) {
93 if (version >= RPC_WIRE_PROTOCOL_VERSION_NEXT &&
94 version != RPC_WIRE_PROTOCOL_VERSION_EXPERIMENTAL) {
95 ALOGE("Cannot start RPC session with version %u which is unknown (current protocol version "
96 "is %u).",
97 version, RPC_WIRE_PROTOCOL_VERSION);
98 return false;
99 }
100
101 std::lock_guard<std::mutex> _l(mMutex);
Steven Moreland40b736e2021-07-30 14:37:10 -0700102 if (mProtocolVersion && version > *mProtocolVersion) {
103 ALOGE("Cannot upgrade explicitly capped protocol version %u to newer version %u",
104 *mProtocolVersion, version);
105 return false;
106 }
107
Steven Morelandbf57bce2021-07-26 15:26:12 -0700108 mProtocolVersion = version;
109 return true;
110}
111
112std::optional<uint32_t> RpcSession::getProtocolVersion() {
113 std::lock_guard<std::mutex> _l(mMutex);
114 return mProtocolVersion;
115}
116
Steven Moreland2372f9d2021-08-05 15:42:01 -0700117status_t RpcSession::setupUnixDomainClient(const char* path) {
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000118 return setupSocketClient(UnixSocketAddress(path));
119}
120
Steven Moreland2372f9d2021-08-05 15:42:01 -0700121status_t RpcSession::setupVsockClient(unsigned int cid, unsigned int port) {
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000122 return setupSocketClient(VsockSocketAddress(cid, port));
123}
124
Steven Moreland2372f9d2021-08-05 15:42:01 -0700125status_t RpcSession::setupInetClient(const char* addr, unsigned int port) {
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000126 auto aiStart = InetSocketAddress::getAddrInfo(addr, port);
Steven Moreland2372f9d2021-08-05 15:42:01 -0700127 if (aiStart == nullptr) return UNKNOWN_ERROR;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000128 for (auto ai = aiStart.get(); ai != nullptr; ai = ai->ai_next) {
129 InetSocketAddress socketAddress(ai->ai_addr, ai->ai_addrlen, addr, port);
Steven Moreland2372f9d2021-08-05 15:42:01 -0700130 if (status_t status = setupSocketClient(socketAddress); status == OK) return OK;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000131 }
132 ALOGE("None of the socket address resolved for %s:%u can be added as inet client.", addr, port);
Steven Moreland2372f9d2021-08-05 15:42:01 -0700133 return NAME_NOT_FOUND;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000134}
135
Steven Moreland2372f9d2021-08-05 15:42:01 -0700136status_t RpcSession::setupPreconnectedClient(unique_fd fd, std::function<unique_fd()>&& request) {
Steven Moreland826367f2021-09-10 14:05:31 -0700137 return setupClient([&](const std::vector<uint8_t>& sessionId, bool incoming) -> status_t {
Steven Moreland4198a122021-08-03 17:37:58 -0700138 // std::move'd from fd becomes -1 (!ok())
139 if (!fd.ok()) {
140 fd = request();
Steven Moreland2372f9d2021-08-05 15:42:01 -0700141 if (!fd.ok()) return BAD_VALUE;
Steven Moreland4198a122021-08-03 17:37:58 -0700142 }
Yifan Hongb675ffe2021-08-05 16:37:17 -0700143 if (auto res = setNonBlocking(fd); !res.ok()) {
144 ALOGE("setupPreconnectedClient: %s", res.error().message().c_str());
145 return res.error().code() == 0 ? UNKNOWN_ERROR : -res.error().code();
146 }
Steven Moreland4198a122021-08-03 17:37:58 -0700147 return initAndAddConnection(std::move(fd), sessionId, incoming);
148 });
149}
150
Steven Moreland2372f9d2021-08-05 15:42:01 -0700151status_t RpcSession::addNullDebuggingClient() {
Yifan Hong702115c2021-06-24 15:39:18 -0700152 // Note: only works on raw sockets.
Yifan Hong832521e2021-08-05 14:55:40 -0700153 if (auto status = initShutdownTrigger(); status != OK) return status;
154
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000155 unique_fd serverFd(TEMP_FAILURE_RETRY(open("/dev/null", O_WRONLY | O_CLOEXEC)));
156
157 if (serverFd == -1) {
Steven Moreland2372f9d2021-08-05 15:42:01 -0700158 int savedErrno = errno;
159 ALOGE("Could not connect to /dev/null: %s", strerror(savedErrno));
160 return -savedErrno;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000161 }
162
Yifan Hongecf937d2021-08-11 17:29:28 -0700163 auto server = mCtx->newTransport(std::move(serverFd), mShutdownTrigger.get());
Yifan Hong702115c2021-06-24 15:39:18 -0700164 if (server == nullptr) {
165 ALOGE("Unable to set up RpcTransport");
Steven Moreland2372f9d2021-08-05 15:42:01 -0700166 return UNKNOWN_ERROR;
Yifan Hong702115c2021-06-24 15:39:18 -0700167 }
168 return addOutgoingConnection(std::move(server), false);
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000169}
170
171sp<IBinder> RpcSession::getRootObject() {
Steven Moreland195edb82021-06-08 02:44:39 +0000172 ExclusiveConnection connection;
173 status_t status = ExclusiveConnection::find(sp<RpcSession>::fromExisting(this),
174 ConnectionUse::CLIENT, &connection);
175 if (status != OK) return nullptr;
Steven Moreland5ae62562021-06-10 03:21:42 +0000176 return state()->getRootObject(connection.get(), sp<RpcSession>::fromExisting(this));
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000177}
178
Steven Moreland1be91352021-05-11 22:12:15 +0000179status_t RpcSession::getRemoteMaxThreads(size_t* maxThreads) {
Steven Moreland195edb82021-06-08 02:44:39 +0000180 ExclusiveConnection connection;
181 status_t status = ExclusiveConnection::find(sp<RpcSession>::fromExisting(this),
182 ConnectionUse::CLIENT, &connection);
183 if (status != OK) return status;
Steven Moreland5ae62562021-06-10 03:21:42 +0000184 return state()->getMaxThreads(connection.get(), sp<RpcSession>::fromExisting(this), maxThreads);
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000185}
186
Steven Morelandc9d7b532021-06-04 20:57:41 +0000187bool RpcSession::shutdownAndWait(bool wait) {
Steven Moreland659416d2021-05-11 00:47:50 +0000188 std::unique_lock<std::mutex> _l(mMutex);
Steven Moreland659416d2021-05-11 00:47:50 +0000189 LOG_ALWAYS_FATAL_IF(mShutdownTrigger == nullptr, "Shutdown trigger not installed");
Steven Moreland659416d2021-05-11 00:47:50 +0000190
191 mShutdownTrigger->trigger();
Steven Moreland659416d2021-05-11 00:47:50 +0000192
Steven Morelandc9d7b532021-06-04 20:57:41 +0000193 if (wait) {
194 LOG_ALWAYS_FATAL_IF(mShutdownListener == nullptr, "Shutdown listener not installed");
Steven Moreland791e4662021-09-13 15:22:58 -0700195 mShutdownListener->waitForShutdown(_l, sp<RpcSession>::fromExisting(this));
Steven Morelanddd67b942021-07-23 17:15:41 -0700196
Steven Morelandc9d7b532021-06-04 20:57:41 +0000197 LOG_ALWAYS_FATAL_IF(!mThreads.empty(), "Shutdown failed");
198 }
199
200 _l.unlock();
201 mState->clear();
202
Steven Moreland659416d2021-05-11 00:47:50 +0000203 return true;
204}
205
Steven Morelandf5174272021-05-25 00:39:28 +0000206status_t RpcSession::transact(const sp<IBinder>& binder, uint32_t code, const Parcel& data,
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000207 Parcel* reply, uint32_t flags) {
Steven Moreland195edb82021-06-08 02:44:39 +0000208 ExclusiveConnection connection;
209 status_t status =
210 ExclusiveConnection::find(sp<RpcSession>::fromExisting(this),
211 (flags & IBinder::FLAG_ONEWAY) ? ConnectionUse::CLIENT_ASYNC
212 : ConnectionUse::CLIENT,
213 &connection);
214 if (status != OK) return status;
Steven Moreland5ae62562021-06-10 03:21:42 +0000215 return state()->transact(connection.get(), binder, code, data,
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000216 sp<RpcSession>::fromExisting(this), reply, flags);
217}
218
Steven Moreland4f622fe2021-09-13 17:38:09 -0700219status_t RpcSession::sendDecStrong(const BpBinder* binder) {
Steven Morelandfd1e8a02021-07-21 23:30:29 +0000220 // target is 0 because this is used to free BpBinder objects
221 return sendDecStrongToTarget(binder->getPrivateAccessor().rpcAddress(), 0 /*target*/);
Steven Moreland4f622fe2021-09-13 17:38:09 -0700222}
223
Steven Morelandfd1e8a02021-07-21 23:30:29 +0000224status_t RpcSession::sendDecStrongToTarget(uint64_t address, size_t target) {
Steven Moreland195edb82021-06-08 02:44:39 +0000225 ExclusiveConnection connection;
226 status_t status = ExclusiveConnection::find(sp<RpcSession>::fromExisting(this),
227 ConnectionUse::CLIENT_REFCOUNT, &connection);
228 if (status != OK) return status;
Steven Morelandfd1e8a02021-07-21 23:30:29 +0000229 return state()->sendDecStrongToTarget(connection.get(), sp<RpcSession>::fromExisting(this),
230 address, target);
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000231}
232
233status_t RpcSession::readId() {
234 {
235 std::lock_guard<std::mutex> _l(mMutex);
236 LOG_ALWAYS_FATAL_IF(mForServer != nullptr, "Can only update ID for client.");
237 }
238
Steven Moreland195edb82021-06-08 02:44:39 +0000239 ExclusiveConnection connection;
240 status_t status = ExclusiveConnection::find(sp<RpcSession>::fromExisting(this),
241 ConnectionUse::CLIENT, &connection);
242 if (status != OK) return status;
243
Steven Moreland826367f2021-09-10 14:05:31 -0700244 status = state()->getSessionId(connection.get(), sp<RpcSession>::fromExisting(this), &mId);
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000245 if (status != OK) return status;
246
Steven Moreland826367f2021-09-10 14:05:31 -0700247 LOG_RPC_DETAIL("RpcSession %p has id %s", this,
248 base::HexString(mId.data(), mId.size()).c_str());
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000249 return OK;
250}
251
Steven Morelanddd67b942021-07-23 17:15:41 -0700252void RpcSession::WaitForShutdownListener::onSessionAllIncomingThreadsEnded(
Steven Moreland659416d2021-05-11 00:47:50 +0000253 const sp<RpcSession>& session) {
254 (void)session;
Steven Moreland659416d2021-05-11 00:47:50 +0000255}
256
Steven Moreland19fc9f72021-06-10 03:57:30 +0000257void RpcSession::WaitForShutdownListener::onSessionIncomingThreadEnded() {
Steven Moreland659416d2021-05-11 00:47:50 +0000258 mCv.notify_all();
259}
260
Steven Moreland791e4662021-09-13 15:22:58 -0700261void RpcSession::WaitForShutdownListener::waitForShutdown(std::unique_lock<std::mutex>& lock,
262 const sp<RpcSession>& session) {
263 while (session->mIncomingConnections.size() > 0) {
Steven Moreland659416d2021-05-11 00:47:50 +0000264 if (std::cv_status::timeout == mCv.wait_for(lock, std::chrono::seconds(1))) {
Steven Moreland791e4662021-09-13 15:22:58 -0700265 ALOGE("Waiting for RpcSession to shut down (1s w/o progress): %zu incoming connections "
266 "still.",
267 session->mIncomingConnections.size());
Steven Moreland659416d2021-05-11 00:47:50 +0000268 }
269 }
270}
271
Steven Morelandc88b7fc2021-06-10 00:40:39 +0000272void RpcSession::preJoinThreadOwnership(std::thread thread) {
Steven Morelanda63ff932021-05-12 00:03:15 +0000273 LOG_ALWAYS_FATAL_IF(thread.get_id() != std::this_thread::get_id(), "Must own this thread");
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000274
Steven Morelanda63ff932021-05-12 00:03:15 +0000275 {
276 std::lock_guard<std::mutex> _l(mMutex);
277 mThreads[thread.get_id()] = std::move(thread);
278 }
Steven Moreland5802c2b2021-05-12 20:13:04 +0000279}
Steven Morelanda63ff932021-05-12 00:03:15 +0000280
Yifan Hong702115c2021-06-24 15:39:18 -0700281RpcSession::PreJoinSetupResult RpcSession::preJoinSetup(
282 std::unique_ptr<RpcTransport> rpcTransport) {
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000283 // must be registered to allow arbitrary client code executing commands to
284 // be able to do nested calls (we can't only read from it)
Yifan Hong702115c2021-06-24 15:39:18 -0700285 sp<RpcConnection> connection = assignIncomingConnectionToThisThread(std::move(rpcTransport));
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000286
Steven Morelanddd67b942021-07-23 17:15:41 -0700287 status_t status;
288
289 if (connection == nullptr) {
290 status = DEAD_OBJECT;
291 } else {
292 status = mState->readConnectionInit(connection, sp<RpcSession>::fromExisting(this));
293 }
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000294
Steven Morelandc88b7fc2021-06-10 00:40:39 +0000295 return PreJoinSetupResult{
296 .connection = std::move(connection),
297 .status = status,
298 };
299}
300
Yifan Hong194acf22021-06-29 18:44:56 -0700301namespace {
302// RAII object for attaching / detaching current thread to JVM if Android Runtime exists. If
303// Android Runtime doesn't exist, no-op.
304class JavaThreadAttacher {
305public:
306 JavaThreadAttacher() {
307 // Use dlsym to find androidJavaAttachThread because libandroid_runtime is loaded after
308 // libbinder.
309 auto vm = getJavaVM();
310 if (vm == nullptr) return;
311
312 char threadName[16];
313 if (0 != pthread_getname_np(pthread_self(), threadName, sizeof(threadName))) {
314 constexpr const char* defaultThreadName = "UnknownRpcSessionThread";
315 memcpy(threadName, defaultThreadName,
316 std::min<size_t>(sizeof(threadName), strlen(defaultThreadName) + 1));
317 }
318 LOG_RPC_DETAIL("Attaching current thread %s to JVM", threadName);
319 JavaVMAttachArgs args;
320 args.version = JNI_VERSION_1_2;
321 args.name = threadName;
322 args.group = nullptr;
323 JNIEnv* env;
324
325 LOG_ALWAYS_FATAL_IF(vm->AttachCurrentThread(&env, &args) != JNI_OK,
326 "Cannot attach thread %s to JVM", threadName);
327 mAttached = true;
328 }
329 ~JavaThreadAttacher() {
330 if (!mAttached) return;
331 auto vm = getJavaVM();
332 LOG_ALWAYS_FATAL_IF(vm == nullptr,
333 "Unable to detach thread. No JavaVM, but it was present before!");
334
335 LOG_RPC_DETAIL("Detaching current thread from JVM");
336 if (vm->DetachCurrentThread() != JNI_OK) {
337 mAttached = false;
338 } else {
339 ALOGW("Unable to detach current thread from JVM");
340 }
341 }
342
343private:
344 DISALLOW_COPY_AND_ASSIGN(JavaThreadAttacher);
345 bool mAttached = false;
346
347 static JavaVM* getJavaVM() {
348 static auto fn = reinterpret_cast<decltype(&AndroidRuntimeGetJavaVM)>(
349 dlsym(RTLD_DEFAULT, "AndroidRuntimeGetJavaVM"));
350 if (fn == nullptr) return nullptr;
351 return fn();
352 }
353};
354} // namespace
355
Steven Morelandc88b7fc2021-06-10 00:40:39 +0000356void RpcSession::join(sp<RpcSession>&& session, PreJoinSetupResult&& setupResult) {
357 sp<RpcConnection>& connection = setupResult.connection;
358
359 if (setupResult.status == OK) {
Steven Morelanddd67b942021-07-23 17:15:41 -0700360 LOG_ALWAYS_FATAL_IF(!connection, "must have connection if setup succeeded");
Yifan Hong194acf22021-06-29 18:44:56 -0700361 JavaThreadAttacher javaThreadAttacher;
Steven Morelandc88b7fc2021-06-10 00:40:39 +0000362 while (true) {
Steven Moreland5ae62562021-06-10 03:21:42 +0000363 status_t status = session->state()->getAndExecuteCommand(connection, session,
Steven Morelandc88b7fc2021-06-10 00:40:39 +0000364 RpcState::CommandType::ANY);
365 if (status != OK) {
366 LOG_RPC_DETAIL("Binder connection thread closing w/ status %s",
367 statusToString(status).c_str());
368 break;
369 }
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000370 }
Steven Morelandc88b7fc2021-06-10 00:40:39 +0000371 } else {
372 ALOGE("Connection failed to init, closing with status %s",
373 statusToString(setupResult.status).c_str());
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000374 }
375
Steven Moreland659416d2021-05-11 00:47:50 +0000376 sp<RpcSession::EventListener> listener;
Steven Morelanda63ff932021-05-12 00:03:15 +0000377 {
Steven Moreland659416d2021-05-11 00:47:50 +0000378 std::lock_guard<std::mutex> _l(session->mMutex);
379 auto it = session->mThreads.find(std::this_thread::get_id());
380 LOG_ALWAYS_FATAL_IF(it == session->mThreads.end());
Steven Morelanda63ff932021-05-12 00:03:15 +0000381 it->second.detach();
Steven Moreland659416d2021-05-11 00:47:50 +0000382 session->mThreads.erase(it);
Steven Morelandee3f4662021-05-22 01:07:33 +0000383
Steven Moreland659416d2021-05-11 00:47:50 +0000384 listener = session->mEventListener.promote();
Steven Morelandee3f4662021-05-22 01:07:33 +0000385 }
386
Steven Morelanddd67b942021-07-23 17:15:41 -0700387 // done after all cleanup, since session shutdown progresses via callbacks here
388 if (connection != nullptr) {
389 LOG_ALWAYS_FATAL_IF(!session->removeIncomingConnection(connection),
390 "bad state: connection object guaranteed to be in list");
391 }
392
Steven Moreland659416d2021-05-11 00:47:50 +0000393 session = nullptr;
394
395 if (listener != nullptr) {
Steven Moreland19fc9f72021-06-10 03:57:30 +0000396 listener->onSessionIncomingThreadEnded();
Steven Morelandee78e762021-05-05 21:12:51 +0000397 }
398}
399
Steven Moreland7b8bc4c2021-06-10 22:50:27 +0000400sp<RpcServer> RpcSession::server() {
401 RpcServer* unsafeServer = mForServer.unsafe_get();
402 sp<RpcServer> server = mForServer.promote();
403
404 LOG_ALWAYS_FATAL_IF((unsafeServer == nullptr) != (server == nullptr),
405 "wp<> is to avoid strong cycle only");
406 return server;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000407}
408
Steven Moreland826367f2021-09-10 14:05:31 -0700409status_t RpcSession::setupClient(const std::function<status_t(const std::vector<uint8_t>& sessionId,
410 bool incoming)>& connectAndInit) {
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000411 {
412 std::lock_guard<std::mutex> _l(mMutex);
Steven Moreland19fc9f72021-06-10 03:57:30 +0000413 LOG_ALWAYS_FATAL_IF(mOutgoingConnections.size() != 0,
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000414 "Must only setup session once, but already has %zu clients",
Steven Moreland19fc9f72021-06-10 03:57:30 +0000415 mOutgoingConnections.size());
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000416 }
Yifan Hong832521e2021-08-05 14:55:40 -0700417 if (auto status = initShutdownTrigger(); status != OK) return status;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000418
Steven Moreland826367f2021-09-10 14:05:31 -0700419 if (status_t status = connectAndInit({}, false /*incoming*/); status != OK) return status;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000420
Steven Morelandbf57bce2021-07-26 15:26:12 -0700421 {
422 ExclusiveConnection connection;
Steven Moreland2372f9d2021-08-05 15:42:01 -0700423 if (status_t status = ExclusiveConnection::find(sp<RpcSession>::fromExisting(this),
424 ConnectionUse::CLIENT, &connection);
425 status != OK)
426 return status;
Steven Morelandbf57bce2021-07-26 15:26:12 -0700427
428 uint32_t version;
Steven Moreland2372f9d2021-08-05 15:42:01 -0700429 if (status_t status =
430 state()->readNewSessionResponse(connection.get(),
431 sp<RpcSession>::fromExisting(this), &version);
432 status != OK)
433 return status;
434 if (!setProtocolVersion(version)) return BAD_VALUE;
Steven Morelandbf57bce2021-07-26 15:26:12 -0700435 }
436
Steven Morelanda5036f02021-06-08 02:26:57 +0000437 // TODO(b/189955605): we should add additional sessions dynamically
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000438 // instead of all at once.
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000439 size_t numThreadsAvailable;
Steven Moreland1be91352021-05-11 22:12:15 +0000440 if (status_t status = getRemoteMaxThreads(&numThreadsAvailable); status != OK) {
Steven Moreland4198a122021-08-03 17:37:58 -0700441 ALOGE("Could not get max threads after initial session setup: %s",
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000442 statusToString(status).c_str());
Steven Moreland2372f9d2021-08-05 15:42:01 -0700443 return status;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000444 }
445
446 if (status_t status = readId(); status != OK) {
Steven Moreland4198a122021-08-03 17:37:58 -0700447 ALOGE("Could not get session id after initial session setup: %s",
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000448 statusToString(status).c_str());
Steven Moreland2372f9d2021-08-05 15:42:01 -0700449 return status;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000450 }
451
Steven Morelanda5036f02021-06-08 02:26:57 +0000452 // TODO(b/189955605): we should add additional sessions dynamically
Steven Moreland659416d2021-05-11 00:47:50 +0000453 // instead of all at once - the other side should be responsible for setting
454 // up additional connections. We need to create at least one (unless 0 are
455 // requested to be set) in order to allow the other side to reliably make
456 // any requests at all.
457
Steven Moreland4198a122021-08-03 17:37:58 -0700458 // we've already setup one client
459 for (size_t i = 0; i + 1 < numThreadsAvailable; i++) {
Steven Moreland826367f2021-09-10 14:05:31 -0700460 if (status_t status = connectAndInit(mId, false /*incoming*/); status != OK) return status;
Steven Moreland4198a122021-08-03 17:37:58 -0700461 }
462
Steven Moreland103424e2021-06-02 18:16:19 +0000463 for (size_t i = 0; i < mMaxThreads; i++) {
Steven Moreland826367f2021-09-10 14:05:31 -0700464 if (status_t status = connectAndInit(mId, true /*incoming*/); status != OK) return status;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000465 }
466
Steven Moreland2372f9d2021-08-05 15:42:01 -0700467 return OK;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000468}
469
Steven Moreland2372f9d2021-08-05 15:42:01 -0700470status_t RpcSession::setupSocketClient(const RpcSocketAddress& addr) {
Steven Moreland826367f2021-09-10 14:05:31 -0700471 return setupClient([&](const std::vector<uint8_t>& sessionId, bool incoming) {
Steven Moreland4198a122021-08-03 17:37:58 -0700472 return setupOneSocketConnection(addr, sessionId, incoming);
473 });
474}
475
Steven Moreland2372f9d2021-08-05 15:42:01 -0700476status_t RpcSession::setupOneSocketConnection(const RpcSocketAddress& addr,
Steven Moreland826367f2021-09-10 14:05:31 -0700477 const std::vector<uint8_t>& sessionId,
478 bool incoming) {
Steven Moreland76d2c1f2021-05-05 20:28:58 +0000479 for (size_t tries = 0; tries < 5; tries++) {
480 if (tries > 0) usleep(10000);
481
Yifan Hongb675ffe2021-08-05 16:37:17 -0700482 unique_fd serverFd(TEMP_FAILURE_RETRY(
483 socket(addr.addr()->sa_family, SOCK_STREAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0)));
Steven Moreland76d2c1f2021-05-05 20:28:58 +0000484 if (serverFd == -1) {
485 int savedErrno = errno;
486 ALOGE("Could not create socket at %s: %s", addr.toString().c_str(),
487 strerror(savedErrno));
Steven Moreland2372f9d2021-08-05 15:42:01 -0700488 return -savedErrno;
Steven Moreland76d2c1f2021-05-05 20:28:58 +0000489 }
490
491 if (0 != TEMP_FAILURE_RETRY(connect(serverFd.get(), addr.addr(), addr.addrSize()))) {
Yifan Hong95d15e52021-08-25 17:15:15 -0700492 int connErrno = errno;
493 if (connErrno == EAGAIN || connErrno == EINPROGRESS) {
494 // For non-blocking sockets, connect() may return EAGAIN (for unix domain socket) or
495 // EINPROGRESS (for others). Call poll() and getsockopt() to get the error.
496 status_t pollStatus = mShutdownTrigger->triggerablePoll(serverFd, POLLOUT);
497 if (pollStatus != OK) {
498 ALOGE("Could not POLLOUT after connect() on non-blocking socket: %s",
499 statusToString(pollStatus).c_str());
500 return pollStatus;
501 }
502 // Set connErrno to the errno that connect() would have set if the fd were blocking.
503 socklen_t connErrnoLen = sizeof(connErrno);
504 int ret =
505 getsockopt(serverFd.get(), SOL_SOCKET, SO_ERROR, &connErrno, &connErrnoLen);
506 if (ret == -1) {
507 int savedErrno = errno;
508 ALOGE("Could not getsockopt() after connect() on non-blocking socket: %s. "
509 "(Original error from connect() is: %s)",
510 strerror(savedErrno), strerror(connErrno));
511 return -savedErrno;
512 }
513 // Retrieved the real connErrno as if connect() was called with a blocking socket
514 // fd. Continue checking connErrno.
515 }
516 if (connErrno == ECONNRESET) {
Steven Moreland76d2c1f2021-05-05 20:28:58 +0000517 ALOGW("Connection reset on %s", addr.toString().c_str());
518 continue;
519 }
Yifan Hong95d15e52021-08-25 17:15:15 -0700520 // connErrno could be zero if getsockopt determines so. Hence zero-check again.
521 if (connErrno != 0) {
Yifan Hongd9f8cef2021-08-05 15:17:31 -0700522 ALOGE("Could not connect socket at %s: %s", addr.toString().c_str(),
Yifan Hong95d15e52021-08-25 17:15:15 -0700523 strerror(connErrno));
524 return -connErrno;
Yifan Hongd9f8cef2021-08-05 15:17:31 -0700525 }
Steven Moreland76d2c1f2021-05-05 20:28:58 +0000526 }
Yifan Hong702115c2021-06-24 15:39:18 -0700527 LOG_RPC_DETAIL("Socket at %s client with fd %d", addr.toString().c_str(), serverFd.get());
528
Steven Moreland4198a122021-08-03 17:37:58 -0700529 return initAndAddConnection(std::move(serverFd), sessionId, incoming);
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000530 }
531
Steven Moreland76d2c1f2021-05-05 20:28:58 +0000532 ALOGE("Ran out of retries to connect to %s", addr.toString().c_str());
Steven Moreland2372f9d2021-08-05 15:42:01 -0700533 return UNKNOWN_ERROR;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000534}
535
Steven Moreland826367f2021-09-10 14:05:31 -0700536status_t RpcSession::initAndAddConnection(unique_fd fd, const std::vector<uint8_t>& sessionId,
Steven Moreland2372f9d2021-08-05 15:42:01 -0700537 bool incoming) {
Yifan Hong8c950422021-08-05 17:13:55 -0700538 LOG_ALWAYS_FATAL_IF(mShutdownTrigger == nullptr);
Yifan Hongecf937d2021-08-11 17:29:28 -0700539 auto server = mCtx->newTransport(std::move(fd), mShutdownTrigger.get());
Steven Moreland4198a122021-08-03 17:37:58 -0700540 if (server == nullptr) {
Yifan Hongecf937d2021-08-11 17:29:28 -0700541 ALOGE("%s: Unable to set up RpcTransport", __PRETTY_FUNCTION__);
Steven Moreland2372f9d2021-08-05 15:42:01 -0700542 return UNKNOWN_ERROR;
Steven Moreland4198a122021-08-03 17:37:58 -0700543 }
544
545 LOG_RPC_DETAIL("Socket at client with RpcTransport %p", server.get());
546
Steven Moreland826367f2021-09-10 14:05:31 -0700547 if (sessionId.size() > std::numeric_limits<uint16_t>::max()) {
548 ALOGE("Session ID too big %zu", sessionId.size());
549 return BAD_VALUE;
550 }
551
Steven Moreland4198a122021-08-03 17:37:58 -0700552 RpcConnectionHeader header{
553 .version = mProtocolVersion.value_or(RPC_WIRE_PROTOCOL_VERSION),
554 .options = 0,
Steven Moreland826367f2021-09-10 14:05:31 -0700555 .sessionIdSize = static_cast<uint16_t>(sessionId.size()),
Steven Moreland4198a122021-08-03 17:37:58 -0700556 };
Steven Moreland4198a122021-08-03 17:37:58 -0700557
Steven Moreland826367f2021-09-10 14:05:31 -0700558 if (incoming) {
559 header.options |= RPC_CONNECTION_OPTION_INCOMING;
560 }
Steven Moreland4198a122021-08-03 17:37:58 -0700561
Yifan Hong8c950422021-08-05 17:13:55 -0700562 auto sendHeaderStatus =
Steven Moreland43921d52021-09-27 17:15:56 -0700563 server->interruptableWriteFully(mShutdownTrigger.get(), &header, sizeof(header), {});
Yifan Hong8c950422021-08-05 17:13:55 -0700564 if (sendHeaderStatus != OK) {
Steven Moreland4198a122021-08-03 17:37:58 -0700565 ALOGE("Could not write connection header to socket: %s",
Yifan Hong8c950422021-08-05 17:13:55 -0700566 statusToString(sendHeaderStatus).c_str());
567 return sendHeaderStatus;
Steven Moreland4198a122021-08-03 17:37:58 -0700568 }
569
Steven Moreland826367f2021-09-10 14:05:31 -0700570 if (sessionId.size() > 0) {
571 auto sendSessionIdStatus =
572 server->interruptableWriteFully(mShutdownTrigger.get(), sessionId.data(),
Steven Moreland43921d52021-09-27 17:15:56 -0700573 sessionId.size(), {});
Steven Moreland826367f2021-09-10 14:05:31 -0700574 if (sendSessionIdStatus != OK) {
575 ALOGE("Could not write session ID ('%s') to socket: %s",
576 base::HexString(sessionId.data(), sessionId.size()).c_str(),
577 statusToString(sendSessionIdStatus).c_str());
578 return sendSessionIdStatus;
579 }
580 }
581
Steven Moreland4198a122021-08-03 17:37:58 -0700582 LOG_RPC_DETAIL("Socket at client: header sent");
583
584 if (incoming) {
585 return addIncomingConnection(std::move(server));
586 } else {
587 return addOutgoingConnection(std::move(server), true /*init*/);
588 }
589}
590
Steven Moreland2372f9d2021-08-05 15:42:01 -0700591status_t RpcSession::addIncomingConnection(std::unique_ptr<RpcTransport> rpcTransport) {
Steven Morelandfba6f772021-07-15 22:45:09 +0000592 std::mutex mutex;
593 std::condition_variable joinCv;
594 std::unique_lock<std::mutex> lock(mutex);
595 std::thread thread;
596 sp<RpcSession> thiz = sp<RpcSession>::fromExisting(this);
597 bool ownershipTransferred = false;
598 thread = std::thread([&]() {
599 std::unique_lock<std::mutex> threadLock(mutex);
Yifan Hong702115c2021-06-24 15:39:18 -0700600 std::unique_ptr<RpcTransport> movedRpcTransport = std::move(rpcTransport);
Steven Morelandfba6f772021-07-15 22:45:09 +0000601 // NOLINTNEXTLINE(performance-unnecessary-copy-initialization)
602 sp<RpcSession> session = thiz;
603 session->preJoinThreadOwnership(std::move(thread));
604
605 // only continue once we have a response or the connection fails
Yifan Hong702115c2021-06-24 15:39:18 -0700606 auto setupResult = session->preJoinSetup(std::move(movedRpcTransport));
Steven Morelandfba6f772021-07-15 22:45:09 +0000607
608 ownershipTransferred = true;
609 threadLock.unlock();
610 joinCv.notify_one();
611 // do not use & vars below
612
613 RpcSession::join(std::move(session), std::move(setupResult));
614 });
615 joinCv.wait(lock, [&] { return ownershipTransferred; });
616 LOG_ALWAYS_FATAL_IF(!ownershipTransferred);
Steven Moreland2372f9d2021-08-05 15:42:01 -0700617 return OK;
Steven Morelandfba6f772021-07-15 22:45:09 +0000618}
619
Yifan Hong832521e2021-08-05 14:55:40 -0700620status_t RpcSession::initShutdownTrigger() {
621 // first client connection added, but setForServer not called, so
622 // initializaing for a client.
623 if (mShutdownTrigger == nullptr) {
624 mShutdownTrigger = FdTrigger::make();
625 mEventListener = mShutdownListener = sp<WaitForShutdownListener>::make();
626 if (mShutdownTrigger == nullptr) return INVALID_OPERATION;
627 }
628 return OK;
629}
630
Steven Moreland2372f9d2021-08-05 15:42:01 -0700631status_t RpcSession::addOutgoingConnection(std::unique_ptr<RpcTransport> rpcTransport, bool init) {
Steven Morelandc88b7fc2021-06-10 00:40:39 +0000632 sp<RpcConnection> connection = sp<RpcConnection>::make();
633 {
634 std::lock_guard<std::mutex> _l(mMutex);
Yifan Hong702115c2021-06-24 15:39:18 -0700635 connection->rpcTransport = std::move(rpcTransport);
Steven Morelandc88b7fc2021-06-10 00:40:39 +0000636 connection->exclusiveTid = gettid();
Steven Moreland19fc9f72021-06-10 03:57:30 +0000637 mOutgoingConnections.push_back(connection);
Steven Morelandee3f4662021-05-22 01:07:33 +0000638 }
639
Steven Morelandb86e26b2021-06-12 00:35:58 +0000640 status_t status = OK;
641 if (init) {
642 mState->sendConnectionInit(connection, sp<RpcSession>::fromExisting(this));
643 }
Steven Morelandc88b7fc2021-06-10 00:40:39 +0000644
645 {
646 std::lock_guard<std::mutex> _l(mMutex);
647 connection->exclusiveTid = std::nullopt;
648 }
649
Steven Moreland2372f9d2021-08-05 15:42:01 -0700650 return status;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000651}
652
Steven Morelanda8b44292021-06-08 01:27:53 +0000653bool RpcSession::setForServer(const wp<RpcServer>& server, const wp<EventListener>& eventListener,
Steven Moreland826367f2021-09-10 14:05:31 -0700654 const std::vector<uint8_t>& sessionId) {
Steven Moreland659416d2021-05-11 00:47:50 +0000655 LOG_ALWAYS_FATAL_IF(mForServer != nullptr);
656 LOG_ALWAYS_FATAL_IF(server == nullptr);
657 LOG_ALWAYS_FATAL_IF(mEventListener != nullptr);
658 LOG_ALWAYS_FATAL_IF(eventListener == nullptr);
Steven Morelandee3f4662021-05-22 01:07:33 +0000659 LOG_ALWAYS_FATAL_IF(mShutdownTrigger != nullptr);
Steven Morelanda8b44292021-06-08 01:27:53 +0000660
661 mShutdownTrigger = FdTrigger::make();
662 if (mShutdownTrigger == nullptr) return false;
Steven Morelandee3f4662021-05-22 01:07:33 +0000663
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000664 mId = sessionId;
665 mForServer = server;
Steven Moreland659416d2021-05-11 00:47:50 +0000666 mEventListener = eventListener;
Steven Morelanda8b44292021-06-08 01:27:53 +0000667 return true;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000668}
669
Yifan Hong702115c2021-06-24 15:39:18 -0700670sp<RpcSession::RpcConnection> RpcSession::assignIncomingConnectionToThisThread(
671 std::unique_ptr<RpcTransport> rpcTransport) {
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000672 std::lock_guard<std::mutex> _l(mMutex);
Steven Morelanddd67b942021-07-23 17:15:41 -0700673
Steven Moreland132d5bf2021-08-03 16:13:24 -0700674 if (mIncomingConnections.size() >= mMaxThreads) {
675 ALOGE("Cannot add thread to session with %zu threads (max is set to %zu)",
676 mIncomingConnections.size(), mMaxThreads);
677 return nullptr;
678 }
679
Steven Morelanddd67b942021-07-23 17:15:41 -0700680 // Don't accept any more connections, some have shutdown. Usually this
681 // happens when new connections are still being established as part of a
682 // very short-lived session which shuts down after it already started
683 // accepting new connections.
684 if (mIncomingConnections.size() < mMaxIncomingConnections) {
685 return nullptr;
686 }
687
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000688 sp<RpcConnection> session = sp<RpcConnection>::make();
Yifan Hong702115c2021-06-24 15:39:18 -0700689 session->rpcTransport = std::move(rpcTransport);
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000690 session->exclusiveTid = gettid();
Steven Morelanddd67b942021-07-23 17:15:41 -0700691
Steven Moreland19fc9f72021-06-10 03:57:30 +0000692 mIncomingConnections.push_back(session);
Steven Morelanddd67b942021-07-23 17:15:41 -0700693 mMaxIncomingConnections = mIncomingConnections.size();
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000694
695 return session;
696}
697
Steven Moreland19fc9f72021-06-10 03:57:30 +0000698bool RpcSession::removeIncomingConnection(const sp<RpcConnection>& connection) {
Steven Morelanddd67b942021-07-23 17:15:41 -0700699 std::unique_lock<std::mutex> _l(mMutex);
Steven Moreland19fc9f72021-06-10 03:57:30 +0000700 if (auto it = std::find(mIncomingConnections.begin(), mIncomingConnections.end(), connection);
701 it != mIncomingConnections.end()) {
702 mIncomingConnections.erase(it);
703 if (mIncomingConnections.size() == 0) {
Steven Moreland659416d2021-05-11 00:47:50 +0000704 sp<EventListener> listener = mEventListener.promote();
705 if (listener) {
Steven Morelanddd67b942021-07-23 17:15:41 -0700706 _l.unlock();
707 listener->onSessionAllIncomingThreadsEnded(sp<RpcSession>::fromExisting(this));
Steven Morelanda86e8fe2021-05-26 22:52:35 +0000708 }
Steven Morelandee78e762021-05-05 21:12:51 +0000709 }
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000710 return true;
711 }
712 return false;
713}
714
Yifan Hong9734cfc2021-09-13 16:14:09 -0700715std::vector<uint8_t> RpcSession::getCertificate(RpcCertificateFormat format) {
Yifan Hongecf937d2021-08-11 17:29:28 -0700716 return mCtx->getCertificate(format);
717}
718
Steven Moreland195edb82021-06-08 02:44:39 +0000719status_t RpcSession::ExclusiveConnection::find(const sp<RpcSession>& session, ConnectionUse use,
720 ExclusiveConnection* connection) {
721 connection->mSession = session;
722 connection->mConnection = nullptr;
723 connection->mReentrant = false;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000724
Steven Moreland195edb82021-06-08 02:44:39 +0000725 pid_t tid = gettid();
726 std::unique_lock<std::mutex> _l(session->mMutex);
727
728 session->mWaitingThreads++;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000729 while (true) {
730 sp<RpcConnection> exclusive;
731 sp<RpcConnection> available;
732
733 // CHECK FOR DEDICATED CLIENT SOCKET
734 //
Steven Moreland85e067b2021-05-26 17:43:53 +0000735 // A server/looper should always use a dedicated connection if available
Steven Moreland19fc9f72021-06-10 03:57:30 +0000736 findConnection(tid, &exclusive, &available, session->mOutgoingConnections,
737 session->mOutgoingConnectionsOffset);
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000738
739 // WARNING: this assumes a server cannot request its client to send
Steven Moreland19fc9f72021-06-10 03:57:30 +0000740 // a transaction, as mIncomingConnections is excluded below.
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000741 //
742 // Imagine we have more than one thread in play, and a single thread
743 // sends a synchronous, then an asynchronous command. Imagine the
744 // asynchronous command is sent on the first client connection. Then, if
745 // we naively send a synchronous command to that same connection, the
746 // thread on the far side might be busy processing the asynchronous
747 // command. So, we move to considering the second available thread
748 // for subsequent calls.
749 if (use == ConnectionUse::CLIENT_ASYNC && (exclusive != nullptr || available != nullptr)) {
Steven Moreland19fc9f72021-06-10 03:57:30 +0000750 session->mOutgoingConnectionsOffset = (session->mOutgoingConnectionsOffset + 1) %
751 session->mOutgoingConnections.size();
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000752 }
753
Steven Morelandc7d40132021-06-10 03:42:11 +0000754 // USE SERVING SOCKET (e.g. nested transaction)
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000755 if (use != ConnectionUse::CLIENT_ASYNC) {
Steven Moreland19fc9f72021-06-10 03:57:30 +0000756 sp<RpcConnection> exclusiveIncoming;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000757 // server connections are always assigned to a thread
Steven Moreland19fc9f72021-06-10 03:57:30 +0000758 findConnection(tid, &exclusiveIncoming, nullptr /*available*/,
759 session->mIncomingConnections, 0 /* index hint */);
Steven Morelandc7d40132021-06-10 03:42:11 +0000760
761 // asynchronous calls cannot be nested, we currently allow ref count
762 // calls to be nested (so that you can use this without having extra
763 // threads). Note 'drainCommands' is used so that these ref counts can't
764 // build up.
Steven Moreland19fc9f72021-06-10 03:57:30 +0000765 if (exclusiveIncoming != nullptr) {
766 if (exclusiveIncoming->allowNested) {
Steven Morelandc7d40132021-06-10 03:42:11 +0000767 // guaranteed to be processed as nested command
Steven Moreland19fc9f72021-06-10 03:57:30 +0000768 exclusive = exclusiveIncoming;
Steven Morelandc7d40132021-06-10 03:42:11 +0000769 } else if (use == ConnectionUse::CLIENT_REFCOUNT && available == nullptr) {
770 // prefer available socket, but if we don't have one, don't
771 // wait for one
Steven Moreland19fc9f72021-06-10 03:57:30 +0000772 exclusive = exclusiveIncoming;
Steven Morelandc7d40132021-06-10 03:42:11 +0000773 }
774 }
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000775 }
776
Steven Moreland85e067b2021-05-26 17:43:53 +0000777 // if our thread is already using a connection, prioritize using that
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000778 if (exclusive != nullptr) {
Steven Moreland195edb82021-06-08 02:44:39 +0000779 connection->mConnection = exclusive;
780 connection->mReentrant = true;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000781 break;
782 } else if (available != nullptr) {
Steven Moreland195edb82021-06-08 02:44:39 +0000783 connection->mConnection = available;
784 connection->mConnection->exclusiveTid = tid;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000785 break;
786 }
787
Steven Moreland19fc9f72021-06-10 03:57:30 +0000788 if (session->mOutgoingConnections.size() == 0) {
Steven Moreland195edb82021-06-08 02:44:39 +0000789 ALOGE("Session has no client connections. This is required for an RPC server to make "
790 "any non-nested (e.g. oneway or on another thread) calls. Use: %d. Server "
791 "connections: %zu",
Steven Moreland19fc9f72021-06-10 03:57:30 +0000792 static_cast<int>(use), session->mIncomingConnections.size());
Steven Moreland195edb82021-06-08 02:44:39 +0000793 return WOULD_BLOCK;
794 }
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000795
Steven Moreland85e067b2021-05-26 17:43:53 +0000796 LOG_RPC_DETAIL("No available connections (have %zu clients and %zu servers). Waiting...",
Steven Moreland19fc9f72021-06-10 03:57:30 +0000797 session->mOutgoingConnections.size(), session->mIncomingConnections.size());
Steven Moreland195edb82021-06-08 02:44:39 +0000798 session->mAvailableConnectionCv.wait(_l);
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000799 }
Steven Moreland195edb82021-06-08 02:44:39 +0000800 session->mWaitingThreads--;
801
802 return OK;
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000803}
804
805void RpcSession::ExclusiveConnection::findConnection(pid_t tid, sp<RpcConnection>* exclusive,
806 sp<RpcConnection>* available,
807 std::vector<sp<RpcConnection>>& sockets,
808 size_t socketsIndexHint) {
809 LOG_ALWAYS_FATAL_IF(sockets.size() > 0 && socketsIndexHint >= sockets.size(),
810 "Bad index %zu >= %zu", socketsIndexHint, sockets.size());
811
812 if (*exclusive != nullptr) return; // consistent with break below
813
814 for (size_t i = 0; i < sockets.size(); i++) {
815 sp<RpcConnection>& socket = sockets[(i + socketsIndexHint) % sockets.size()];
816
Steven Moreland85e067b2021-05-26 17:43:53 +0000817 // take first available connection (intuition = caching)
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000818 if (available && *available == nullptr && socket->exclusiveTid == std::nullopt) {
819 *available = socket;
820 continue;
821 }
822
Steven Moreland85e067b2021-05-26 17:43:53 +0000823 // though, prefer to take connection which is already inuse by this thread
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000824 // (nested transactions)
825 if (exclusive && socket->exclusiveTid == tid) {
826 *exclusive = socket;
827 break; // consistent with return above
828 }
829 }
830}
831
832RpcSession::ExclusiveConnection::~ExclusiveConnection() {
Steven Moreland85e067b2021-05-26 17:43:53 +0000833 // reentrant use of a connection means something less deep in the call stack
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000834 // is using this fd, and it retains the right to it. So, we don't give up
835 // exclusive ownership, and no thread is freed.
Steven Moreland195edb82021-06-08 02:44:39 +0000836 if (!mReentrant && mConnection != nullptr) {
Steven Morelandbdb53ab2021-05-05 17:57:41 +0000837 std::unique_lock<std::mutex> _l(mSession->mMutex);
838 mConnection->exclusiveTid = std::nullopt;
839 if (mSession->mWaitingThreads > 0) {
840 _l.unlock();
841 mSession->mAvailableConnectionCv.notify_one();
842 }
843 }
844}
845
846} // namespace android