Merge "libbinder: add build option for single-threaded RPC" am: 8021d31371 (2d6e35a2) · Commits · e / os / platform_frameworks_native

libs/binder/BuildFlags.h

0 → 100644

+25 −0

Original line number	Original line	Diff line number	Diff line
			/*
			* Copyright (C) 2022 The Android Open Source Project
			*
			* Licensed under the Apache License, Version 2.0 (the "License");
			* you may not use this file except in compliance with the License.
			* You may obtain a copy of the License at
			*
			* http://www.apache.org/licenses/LICENSE-2.0
			*
			* Unless required by applicable law or agreed to in writing, software
			* distributed under the License is distributed on an "AS IS" BASIS,
			* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
			* See the License for the specific language governing permissions and
			* limitations under the License.
			*/

			namespace android {

			#ifdef BINDER_RPC_SINGLE_THREADED
			constexpr bool kEnableRpcThreads = false;
			#else
			constexpr bool kEnableRpcThreads = true;
			#endif

			} // namespace android

libs/binder/FdTrigger.cpp

+24 −2

Original line number	Original line	Diff line number	Diff line
	@@ -28,25 +28,45 @@ namespace android {

	std::unique_ptr<FdTrigger> FdTrigger::make() {		std::unique_ptr<FdTrigger> FdTrigger::make() {
	auto ret = std::make_unique<FdTrigger>();		auto ret = std::make_unique<FdTrigger>();
			#ifndef BINDER_RPC_SINGLE_THREADED
	if (!android::base::Pipe(&ret->mRead, &ret->mWrite)) {		if (!android::base::Pipe(&ret->mRead, &ret->mWrite)) {
	ALOGE("Could not create pipe %s", strerror(errno));		ALOGE("Could not create pipe %s", strerror(errno));
	return nullptr;		return nullptr;
	}		}
			#endif
	return ret;		return ret;
	}		}

	void FdTrigger::trigger() {		void FdTrigger::trigger() {
			#ifdef BINDER_RPC_SINGLE_THREADED
			mTriggered = true;
			#else
	mWrite.reset();		mWrite.reset();
			#endif
	}		}

	bool FdTrigger::isTriggered() {		bool FdTrigger::isTriggered() {
			#ifdef BINDER_RPC_SINGLE_THREADED
			return mTriggered;
			#else
	return mWrite == -1;		return mWrite == -1;
			#endif
	}		}

	status_t FdTrigger::triggerablePoll(base::borrowed_fd fd, int16_t event) {		status_t FdTrigger::triggerablePoll(base::borrowed_fd fd, int16_t event) {
			#ifdef BINDER_RPC_SINGLE_THREADED
			if (mTriggered) {
			return DEAD_OBJECT;
			}
			#endif

	LOG_ALWAYS_FATAL_IF(event == 0, "triggerablePoll %d with event 0 is not allowed", fd.get());		LOG_ALWAYS_FATAL_IF(event == 0, "triggerablePoll %d with event 0 is not allowed", fd.get());
	pollfd pfd[]{{.fd = fd.get(), .events = static_cast<int16_t>(event), .revents = 0},		pollfd pfd[]{
	{.fd = mRead.get(), .events = 0, .revents = 0}};		{.fd = fd.get(), .events = static_cast<int16_t>(event), .revents = 0},
			#ifndef BINDER_RPC_SINGLE_THREADED
			{.fd = mRead.get(), .events = 0, .revents = 0},
			#endif
			};
	int ret = TEMP_FAILURE_RETRY(poll(pfd, arraysize(pfd), -1));		int ret = TEMP_FAILURE_RETRY(poll(pfd, arraysize(pfd), -1));
	if (ret < 0) {		if (ret < 0) {
	return -errno;		return -errno;
	@@ -55,6 +75,7 @@ status_t FdTrigger::triggerablePoll(base::borrowed_fd fd, int16_t event) {

	// At least one FD has events. Check them.		// At least one FD has events. Check them.

			#ifndef BINDER_RPC_SINGLE_THREADED
	// Detect explicit trigger(): DEAD_OBJECT		// Detect explicit trigger(): DEAD_OBJECT
	if (pfd[1].revents & POLLHUP) {		if (pfd[1].revents & POLLHUP) {
	return DEAD_OBJECT;		return DEAD_OBJECT;
	@@ -68,6 +89,7 @@ status_t FdTrigger::triggerablePoll(base::borrowed_fd fd, int16_t event) {

	// pfd[1].revents is 0, hence pfd[0].revents must be set, and only possible values are		// pfd[1].revents is 0, hence pfd[0].revents must be set, and only possible values are
	// a subset of event \| POLLHUP \| POLLERR \| POLLNVAL.		// a subset of event \| POLLHUP \| POLLERR \| POLLNVAL.
			#endif

	// POLLNVAL: invalid FD number, e.g. not opened.		// POLLNVAL: invalid FD number, e.g. not opened.
	if (pfd[0].revents & POLLNVAL) {		if (pfd[0].revents & POLLNVAL) {

libs/binder/FdTrigger.h

+4 −0

Original line number	Original line	Diff line number	Diff line
	@@ -55,7 +55,11 @@ public:
	[[nodiscard]] status_t triggerablePoll(base::borrowed_fd fd, int16_t event);		[[nodiscard]] status_t triggerablePoll(base::borrowed_fd fd, int16_t event);

	private:		private:
			#ifdef BINDER_RPC_SINGLE_THREADED
			bool mTriggered = false;
			#else
	base::unique_fd mWrite;		base::unique_fd mWrite;
	base::unique_fd mRead;		base::unique_fd mRead;
			#endif
	};		};
	} // namespace android		} // namespace android

libs/binder/RpcServer.cpp

+42 −26

Original line number	Original line	Diff line number	Diff line
	@@ -32,6 +32,7 @@
	#include <log/log.h>		#include <log/log.h>
	#include <utils/Compat.h>		#include <utils/Compat.h>

			#include "BuildFlags.h"
	#include "FdTrigger.h"		#include "FdTrigger.h"
	#include "RpcSocketAddress.h"		#include "RpcSocketAddress.h"
	#include "RpcState.h"		#include "RpcState.h"
	@@ -131,27 +132,27 @@ void RpcServer::setSupportedFileDescriptorTransportModes(
	}		}

	void RpcServer::setRootObject(const sp<IBinder>& binder) {		void RpcServer::setRootObject(const sp<IBinder>& binder) {
	std::lock_guard<std::mutex> _l(mLock);		RpcMutexLockGuard _l(mLock);
	mRootObjectFactory = nullptr;		mRootObjectFactory = nullptr;
	mRootObjectWeak = mRootObject = binder;		mRootObjectWeak = mRootObject = binder;
	}		}

	void RpcServer::setRootObjectWeak(const wp<IBinder>& binder) {		void RpcServer::setRootObjectWeak(const wp<IBinder>& binder) {
	std::lock_guard<std::mutex> _l(mLock);		RpcMutexLockGuard _l(mLock);
	mRootObject.clear();		mRootObject.clear();
	mRootObjectFactory = nullptr;		mRootObjectFactory = nullptr;
	mRootObjectWeak = binder;		mRootObjectWeak = binder;
	}		}
	void RpcServer::setPerSessionRootObject(		void RpcServer::setPerSessionRootObject(
	std::function<sp<IBinder>(const void*, size_t)>&& makeObject) {		std::function<sp<IBinder>(const void*, size_t)>&& makeObject) {
	std::lock_guard<std::mutex> _l(mLock);		RpcMutexLockGuard _l(mLock);
	mRootObject.clear();		mRootObject.clear();
	mRootObjectWeak.clear();		mRootObjectWeak.clear();
	mRootObjectFactory = std::move(makeObject);		mRootObjectFactory = std::move(makeObject);
	}		}

	sp<IBinder> RpcServer::getRootObject() {		sp<IBinder> RpcServer::getRootObject() {
	std::lock_guard<std::mutex> _l(mLock);		RpcMutexLockGuard _l(mLock);
	bool hasWeak = mRootObjectWeak.unsafe_get();		bool hasWeak = mRootObjectWeak.unsafe_get();
	sp<IBinder> ret = mRootObjectWeak.promote();		sp<IBinder> ret = mRootObjectWeak.promote();
	ALOGW_IF(hasWeak && ret == nullptr, "RpcServer root object is freed, returning nullptr");		ALOGW_IF(hasWeak && ret == nullptr, "RpcServer root object is freed, returning nullptr");
	@@ -159,7 +160,7 @@ sp<IBinder> RpcServer::getRootObject() {
	}		}

	std::vector<uint8_t> RpcServer::getCertificate(RpcCertificateFormat format) {		std::vector<uint8_t> RpcServer::getCertificate(RpcCertificateFormat format) {
	std::lock_guard<std::mutex> _l(mLock);		RpcMutexLockGuard _l(mLock);
	return mCtx->getCertificate(format);		return mCtx->getCertificate(format);
	}		}

	@@ -168,15 +169,17 @@ static void joinRpcServer(sp<RpcServer>&& thiz) {
	}		}

	void RpcServer::start() {		void RpcServer::start() {
	std::lock_guard<std::mutex> _l(mLock);		RpcMutexLockGuard _l(mLock);
	LOG_ALWAYS_FATAL_IF(mJoinThread.get(), "Already started!");		LOG_ALWAYS_FATAL_IF(mJoinThread.get(), "Already started!");
	mJoinThread = std::make_unique<std::thread>(&joinRpcServer, sp<RpcServer>::fromExisting(this));		mJoinThread =
			std::make_unique<RpcMaybeThread>(&joinRpcServer, sp<RpcServer>::fromExisting(this));
			rpcJoinIfSingleThreaded(*mJoinThread);
	}		}

	void RpcServer::join() {		void RpcServer::join() {

	{		{
	std::lock_guard<std::mutex> _l(mLock);		RpcMutexLockGuard _l(mLock);
	LOG_ALWAYS_FATAL_IF(!mServer.ok(), "RpcServer must be setup to join.");		LOG_ALWAYS_FATAL_IF(!mServer.ok(), "RpcServer must be setup to join.");
	LOG_ALWAYS_FATAL_IF(mShutdownTrigger != nullptr, "Already joined");		LOG_ALWAYS_FATAL_IF(mShutdownTrigger != nullptr, "Already joined");
	mJoinThreadRunning = true;		mJoinThreadRunning = true;
	@@ -204,24 +207,31 @@ void RpcServer::join() {
	LOG_RPC_DETAIL("accept4 on fd %d yields fd %d", mServer.get(), clientFd.get());		LOG_RPC_DETAIL("accept4 on fd %d yields fd %d", mServer.get(), clientFd.get());

	{		{
	std::lock_guard<std::mutex> _l(mLock);		RpcMutexLockGuard _l(mLock);
	std::thread thread =		RpcMaybeThread thread = RpcMaybeThread(&RpcServer::establishConnection,
	std::thread(&RpcServer::establishConnection, sp<RpcServer>::fromExisting(this),		sp<RpcServer>::fromExisting(this),
	std::move(clientFd), addr, addrLen);		std::move(clientFd), addr, addrLen);
	mConnectingThreads[thread.get_id()] = std::move(thread);
			auto& threadRef = mConnectingThreads[thread.get_id()];
			threadRef = std::move(thread);
			rpcJoinIfSingleThreaded(threadRef);
	}		}
	}		}
	LOG_RPC_DETAIL("RpcServer::join exiting with %s", statusToString(status).c_str());		LOG_RPC_DETAIL("RpcServer::join exiting with %s", statusToString(status).c_str());

	{		if constexpr (kEnableRpcThreads) {
	std::lock_guard<std::mutex> _l(mLock);		RpcMutexLockGuard _l(mLock);
	mJoinThreadRunning = false;		mJoinThreadRunning = false;
			} else {
			// Multi-threaded builds clear this in shutdown(), but we need it valid
			// so the loop above exits cleanly
			mShutdownTrigger = nullptr;
	}		}
	mShutdownCv.notify_all();		mShutdownCv.notify_all();
	}		}

	bool RpcServer::shutdown() {		bool RpcServer::shutdown() {
	std::unique_lock<std::mutex> _l(mLock);		RpcMutexUniqueLock _l(mLock);
	if (mShutdownTrigger == nullptr) {		if (mShutdownTrigger == nullptr) {
	LOG_RPC_DETAIL("Cannot shutdown. No shutdown trigger installed (already shutdown?)");		LOG_RPC_DETAIL("Cannot shutdown. No shutdown trigger installed (already shutdown?)");
	return false;		return false;
	@@ -232,10 +242,16 @@ bool RpcServer::shutdown() {
	for (auto& [id, session] : mSessions) {		for (auto& [id, session] : mSessions) {
	(void)id;		(void)id;
	// server lock is a more general lock		// server lock is a more general lock
	std::lock_guard<std::mutex> _lSession(session->mMutex);		RpcMutexLockGuard _lSession(session->mMutex);
	session->mShutdownTrigger->trigger();		session->mShutdownTrigger->trigger();
	}		}

			if constexpr (!kEnableRpcThreads) {
			// In single-threaded mode we're done here, everything else that
			// needs to happen should be at the end of RpcServer::join()
			return true;
			}

	while (mJoinThreadRunning \|\| !mConnectingThreads.empty() \|\| !mSessions.empty()) {		while (mJoinThreadRunning \|\| !mConnectingThreads.empty() \|\| !mSessions.empty()) {
	if (std::cv_status::timeout == mShutdownCv.wait_for(_l, std::chrono::seconds(1))) {		if (std::cv_status::timeout == mShutdownCv.wait_for(_l, std::chrono::seconds(1))) {
	ALOGE("Waiting for RpcServer to shut down (1s w/o progress). Join thread running: %d, "		ALOGE("Waiting for RpcServer to shut down (1s w/o progress). Join thread running: %d, "
	@@ -263,7 +279,7 @@ bool RpcServer::shutdown() {
	}		}

	std::vector<sp<RpcSession>> RpcServer::listSessions() {		std::vector<sp<RpcSession>> RpcServer::listSessions() {
	std::lock_guard<std::mutex> _l(mLock);		RpcMutexLockGuard _l(mLock);
	std::vector<sp<RpcSession>> sessions;		std::vector<sp<RpcSession>> sessions;
	for (auto& [id, session] : mSessions) {		for (auto& [id, session] : mSessions) {
	(void)id;		(void)id;
	@@ -273,7 +289,7 @@ std::vector<sp<RpcSession>> RpcServer::listSessions() {
	}		}

	size_t RpcServer::numUninitializedSessions() {		size_t RpcServer::numUninitializedSessions() {
	std::lock_guard<std::mutex> _l(mLock);		RpcMutexLockGuard _l(mLock);
	return mConnectingThreads.size();		return mConnectingThreads.size();
	}		}

	@@ -354,12 +370,12 @@ void RpcServer::establishConnection(sp<RpcServer>&& server, base::unique_fd clie
	}		}
	}		}

	std::thread thisThread;		RpcMaybeThread thisThread;
	sp<RpcSession> session;		sp<RpcSession> session;
	{		{
	std::unique_lock<std::mutex> _l(server->mLock);		RpcMutexUniqueLock _l(server->mLock);

	auto threadId = server->mConnectingThreads.find(std::this_thread::get_id());		auto threadId = server->mConnectingThreads.find(rpc_this_thread::get_id());
	LOG_ALWAYS_FATAL_IF(threadId == server->mConnectingThreads.end(),		LOG_ALWAYS_FATAL_IF(threadId == server->mConnectingThreads.end(),
	"Must establish connection on owned thread");		"Must establish connection on owned thread");
	thisThread = std::move(threadId->second);		thisThread = std::move(threadId->second);
	@@ -505,7 +521,7 @@ void RpcServer::onSessionAllIncomingThreadsEnded(const sp<RpcSession>& session)
	LOG_RPC_DETAIL("Dropping session with address %s",		LOG_RPC_DETAIL("Dropping session with address %s",
	base::HexString(id.data(), id.size()).c_str());		base::HexString(id.data(), id.size()).c_str());

	std::lock_guard<std::mutex> _l(mLock);		RpcMutexLockGuard _l(mLock);
	auto it = mSessions.find(id);		auto it = mSessions.find(id);
	LOG_ALWAYS_FATAL_IF(it == mSessions.end(), "Bad state, unknown session id %s",		LOG_ALWAYS_FATAL_IF(it == mSessions.end(), "Bad state, unknown session id %s",
	base::HexString(id.data(), id.size()).c_str());		base::HexString(id.data(), id.size()).c_str());
	@@ -519,17 +535,17 @@ void RpcServer::onSessionIncomingThreadEnded() {
	}		}

	bool RpcServer::hasServer() {		bool RpcServer::hasServer() {
	std::lock_guard<std::mutex> _l(mLock);		RpcMutexLockGuard _l(mLock);
	return mServer.ok();		return mServer.ok();
	}		}

	unique_fd RpcServer::releaseServer() {		unique_fd RpcServer::releaseServer() {
	std::lock_guard<std::mutex> _l(mLock);		RpcMutexLockGuard _l(mLock);
	return std::move(mServer);		return std::move(mServer);
	}		}

	status_t RpcServer::setupExternalServer(base::unique_fd serverFd) {		status_t RpcServer::setupExternalServer(base::unique_fd serverFd) {
	std::lock_guard<std::mutex> _l(mLock);		RpcMutexLockGuard _l(mLock);
	if (mServer.ok()) {		if (mServer.ok()) {
	ALOGE("Each RpcServer can only have one server.");		ALOGE("Each RpcServer can only have one server.");
	return INVALID_OPERATION;		return INVALID_OPERATION;

libs/binder/RpcSession.cpp

+36 −31

Original line number	Original line	Diff line number	Diff line
	@@ -21,7 +21,6 @@
	#include <dlfcn.h>		#include <dlfcn.h>
	#include <inttypes.h>		#include <inttypes.h>
	#include <poll.h>		#include <poll.h>
	#include <pthread.h>
	#include <unistd.h>		#include <unistd.h>

	#include <string_view>		#include <string_view>
	@@ -60,7 +59,7 @@ RpcSession::RpcSession(std::unique_ptr<RpcTransportCtx> ctx) : mCtx(std::move(ct
	RpcSession::~RpcSession() {		RpcSession::~RpcSession() {
	LOG_RPC_DETAIL("RpcSession destroyed %p", this);		LOG_RPC_DETAIL("RpcSession destroyed %p", this);

	std::lock_guard<std::mutex> _l(mMutex);		RpcMutexLockGuard _l(mMutex);
	LOG_ALWAYS_FATAL_IF(mConnections.mIncoming.size() != 0,		LOG_ALWAYS_FATAL_IF(mConnections.mIncoming.size() != 0,
	"Should not be able to destroy a session with servers in use.");		"Should not be able to destroy a session with servers in use.");
	}		}
	@@ -77,7 +76,7 @@ sp<RpcSession> RpcSession::make(std::unique_ptr<RpcTransportCtxFactory> rpcTrans
	}		}

	void RpcSession::setMaxIncomingThreads(size_t threads) {		void RpcSession::setMaxIncomingThreads(size_t threads) {
	std::lock_guard<std::mutex> _l(mMutex);		RpcMutexLockGuard _l(mMutex);
	LOG_ALWAYS_FATAL_IF(!mConnections.mOutgoing.empty() \|\| !mConnections.mIncoming.empty(),		LOG_ALWAYS_FATAL_IF(!mConnections.mOutgoing.empty() \|\| !mConnections.mIncoming.empty(),
	"Must set max incoming threads before setting up connections, but has %zu "		"Must set max incoming threads before setting up connections, but has %zu "
	"client(s) and %zu server(s)",		"client(s) and %zu server(s)",
	@@ -86,12 +85,12 @@ void RpcSession::setMaxIncomingThreads(size_t threads) {
	}		}

	size_t RpcSession::getMaxIncomingThreads() {		size_t RpcSession::getMaxIncomingThreads() {
	std::lock_guard<std::mutex> _l(mMutex);		RpcMutexLockGuard _l(mMutex);
	return mMaxIncomingThreads;		return mMaxIncomingThreads;
	}		}

	void RpcSession::setMaxOutgoingThreads(size_t threads) {		void RpcSession::setMaxOutgoingThreads(size_t threads) {
	std::lock_guard<std::mutex> _l(mMutex);		RpcMutexLockGuard _l(mMutex);
	LOG_ALWAYS_FATAL_IF(!mConnections.mOutgoing.empty() \|\| !mConnections.mIncoming.empty(),		LOG_ALWAYS_FATAL_IF(!mConnections.mOutgoing.empty() \|\| !mConnections.mIncoming.empty(),
	"Must set max outgoing threads before setting up connections, but has %zu "		"Must set max outgoing threads before setting up connections, but has %zu "
	"client(s) and %zu server(s)",		"client(s) and %zu server(s)",
	@@ -100,7 +99,7 @@ void RpcSession::setMaxOutgoingThreads(size_t threads) {
	}		}

	size_t RpcSession::getMaxOutgoingThreads() {		size_t RpcSession::getMaxOutgoingThreads() {
	std::lock_guard<std::mutex> _l(mMutex);		RpcMutexLockGuard _l(mMutex);
	return mMaxOutgoingThreads;		return mMaxOutgoingThreads;
	}		}

	@@ -113,7 +112,7 @@ bool RpcSession::setProtocolVersion(uint32_t version) {
	return false;		return false;
	}		}

	std::lock_guard<std::mutex> _l(mMutex);		RpcMutexLockGuard _l(mMutex);
	if (mProtocolVersion && version > *mProtocolVersion) {		if (mProtocolVersion && version > *mProtocolVersion) {
	ALOGE("Cannot upgrade explicitly capped protocol version %u to newer version %u",		ALOGE("Cannot upgrade explicitly capped protocol version %u to newer version %u",
	*mProtocolVersion, version);		*mProtocolVersion, version);
	@@ -125,7 +124,7 @@ bool RpcSession::setProtocolVersion(uint32_t version) {
	}		}

	std::optional<uint32_t> RpcSession::getProtocolVersion() {		std::optional<uint32_t> RpcSession::getProtocolVersion() {
	std::lock_guard<std::mutex> _l(mMutex);		RpcMutexLockGuard _l(mMutex);
	return mProtocolVersion;		return mProtocolVersion;
	}		}

	@@ -209,7 +208,7 @@ status_t RpcSession::getRemoteMaxThreads(size_t* maxThreads) {
	}		}

	bool RpcSession::shutdownAndWait(bool wait) {		bool RpcSession::shutdownAndWait(bool wait) {
	std::unique_lock<std::mutex> _l(mMutex);		RpcMutexUniqueLock _l(mMutex);
	LOG_ALWAYS_FATAL_IF(mShutdownTrigger == nullptr, "Shutdown trigger not installed");		LOG_ALWAYS_FATAL_IF(mShutdownTrigger == nullptr, "Shutdown trigger not installed");

	mShutdownTrigger->trigger();		mShutdownTrigger->trigger();
	@@ -222,6 +221,7 @@ bool RpcSession::shutdownAndWait(bool wait) {
	}		}

	_l.unlock();		_l.unlock();

	mRpcBinderState->clear();		mRpcBinderState->clear();

	return true;		return true;
	@@ -256,7 +256,7 @@ status_t RpcSession::sendDecStrongToTarget(uint64_t address, size_t target) {

	status_t RpcSession::readId() {		status_t RpcSession::readId() {
	{		{
	std::lock_guard<std::mutex> _l(mMutex);		RpcMutexLockGuard _l(mMutex);
	LOG_ALWAYS_FATAL_IF(mForServer != nullptr, "Can only update ID for client.");		LOG_ALWAYS_FATAL_IF(mForServer != nullptr, "Can only update ID for client.");
	}		}

	@@ -282,7 +282,7 @@ void RpcSession::WaitForShutdownListener::onSessionIncomingThreadEnded() {
	mCv.notify_all();		mCv.notify_all();
	}		}

	void RpcSession::WaitForShutdownListener::waitForShutdown(std::unique_lock<std::mutex>& lock,		void RpcSession::WaitForShutdownListener::waitForShutdown(RpcMutexUniqueLock& lock,
	const sp<RpcSession>& session) {		const sp<RpcSession>& session) {
	while (session->mConnections.mIncoming.size() > 0) {		while (session->mConnections.mIncoming.size() > 0) {
	if (std::cv_status::timeout == mCv.wait_for(lock, std::chrono::seconds(1))) {		if (std::cv_status::timeout == mCv.wait_for(lock, std::chrono::seconds(1))) {
	@@ -293,11 +293,11 @@ void RpcSession::WaitForShutdownListener::waitForShutdown(std::unique_lock<std::
	}		}
	}		}

	void RpcSession::preJoinThreadOwnership(std::thread thread) {		void RpcSession::preJoinThreadOwnership(RpcMaybeThread thread) {
	LOG_ALWAYS_FATAL_IF(thread.get_id() != std::this_thread::get_id(), "Must own this thread");		LOG_ALWAYS_FATAL_IF(thread.get_id() != rpc_this_thread::get_id(), "Must own this thread");

	{		{
	std::lock_guard<std::mutex> _l(mMutex);		RpcMutexLockGuard _l(mMutex);
	mConnections.mThreads[thread.get_id()] = std::move(thread);		mConnections.mThreads[thread.get_id()] = std::move(thread);
	}		}
	}		}
	@@ -404,8 +404,8 @@ void RpcSession::join(sp<RpcSession>&& session, PreJoinSetupResult&& setupResult

	sp<RpcSession::EventListener> listener;		sp<RpcSession::EventListener> listener;
	{		{
	std::lock_guard<std::mutex> _l(session->mMutex);		RpcMutexLockGuard _l(session->mMutex);
	auto it = session->mConnections.mThreads.find(std::this_thread::get_id());		auto it = session->mConnections.mThreads.find(rpc_this_thread::get_id());
	LOG_ALWAYS_FATAL_IF(it == session->mConnections.mThreads.end());		LOG_ALWAYS_FATAL_IF(it == session->mConnections.mThreads.end());
	it->second.detach();		it->second.detach();
	session->mConnections.mThreads.erase(it);		session->mConnections.mThreads.erase(it);
	@@ -438,7 +438,7 @@ sp<RpcServer> RpcSession::server() {
	status_t RpcSession::setupClient(const std::function<status_t(const std::vector<uint8_t>& sessionId,		status_t RpcSession::setupClient(const std::function<status_t(const std::vector<uint8_t>& sessionId,
	bool incoming)>& connectAndInit) {		bool incoming)>& connectAndInit) {
	{		{
	std::lock_guard<std::mutex> _l(mMutex);		RpcMutexLockGuard _l(mMutex);
	LOG_ALWAYS_FATAL_IF(mConnections.mOutgoing.size() != 0,		LOG_ALWAYS_FATAL_IF(mConnections.mOutgoing.size() != 0,
	"Must only setup session once, but already has %zu clients",		"Must only setup session once, but already has %zu clients",
	mConnections.mOutgoing.size());		mConnections.mOutgoing.size());
	@@ -500,7 +500,11 @@ status_t RpcSession::setupClient(const std::function<status_t(const std::vector<
	return status;		return status;
	}		}

			#ifdef BINDER_RPC_SINGLE_THREADED
			constexpr size_t outgoingThreads = 1;
			#else // BINDER_RPC_SINGLE_THREADED
	size_t outgoingThreads = std::min(numThreadsAvailable, mMaxOutgoingThreads);		size_t outgoingThreads = std::min(numThreadsAvailable, mMaxOutgoingThreads);
			#endif // BINDER_RPC_SINGLE_THREADED
	ALOGI_IF(outgoingThreads != numThreadsAvailable,		ALOGI_IF(outgoingThreads != numThreadsAvailable,
	"Server hints client to start %zu outgoing threads, but client will only start %zu "		"Server hints client to start %zu outgoing threads, but client will only start %zu "
	"because it is preconfigured to start at most %zu outgoing threads.",		"because it is preconfigured to start at most %zu outgoing threads.",
	@@ -655,14 +659,14 @@ status_t RpcSession::initAndAddConnection(unique_fd fd, const std::vector<uint8_
	}		}

	status_t RpcSession::addIncomingConnection(std::unique_ptr<RpcTransport> rpcTransport) {		status_t RpcSession::addIncomingConnection(std::unique_ptr<RpcTransport> rpcTransport) {
	std::mutex mutex;		RpcMutex mutex;
	std::condition_variable joinCv;		RpcConditionVariable joinCv;
	std::unique_lock<std::mutex> lock(mutex);		RpcMutexUniqueLock lock(mutex);
	std::thread thread;		RpcMaybeThread thread;
	sp<RpcSession> thiz = sp<RpcSession>::fromExisting(this);		sp<RpcSession> thiz = sp<RpcSession>::fromExisting(this);
	bool ownershipTransferred = false;		bool ownershipTransferred = false;
	thread = std::thread([&]() {		thread = RpcMaybeThread([&]() {
	std::unique_lock<std::mutex> threadLock(mutex);		RpcMutexUniqueLock threadLock(mutex);
	std::unique_ptr<RpcTransport> movedRpcTransport = std::move(rpcTransport);		std::unique_ptr<RpcTransport> movedRpcTransport = std::move(rpcTransport);
	// NOLINTNEXTLINE(performance-unnecessary-copy-initialization)		// NOLINTNEXTLINE(performance-unnecessary-copy-initialization)
	sp<RpcSession> session = thiz;		sp<RpcSession> session = thiz;
	@@ -678,6 +682,7 @@ status_t RpcSession::addIncomingConnection(std::unique_ptr<RpcTransport> rpcTran

	RpcSession::join(std::move(session), std::move(setupResult));		RpcSession::join(std::move(session), std::move(setupResult));
	});		});
			rpcJoinIfSingleThreaded(thread);
	joinCv.wait(lock, [&] { return ownershipTransferred; });		joinCv.wait(lock, [&] { return ownershipTransferred; });
	LOG_ALWAYS_FATAL_IF(!ownershipTransferred);		LOG_ALWAYS_FATAL_IF(!ownershipTransferred);
	return OK;		return OK;
	@@ -697,9 +702,9 @@ status_t RpcSession::initShutdownTrigger() {
	status_t RpcSession::addOutgoingConnection(std::unique_ptr<RpcTransport> rpcTransport, bool init) {		status_t RpcSession::addOutgoingConnection(std::unique_ptr<RpcTransport> rpcTransport, bool init) {
	sp<RpcConnection> connection = sp<RpcConnection>::make();		sp<RpcConnection> connection = sp<RpcConnection>::make();
	{		{
	std::lock_guard<std::mutex> _l(mMutex);		RpcMutexLockGuard _l(mMutex);
	connection->rpcTransport = std::move(rpcTransport);		connection->rpcTransport = std::move(rpcTransport);
	connection->exclusiveTid = base::GetThreadId();		connection->exclusiveTid = rpcGetThreadId();
	mConnections.mOutgoing.push_back(connection);		mConnections.mOutgoing.push_back(connection);
	}		}

	@@ -736,7 +741,7 @@ bool RpcSession::setForServer(const wp<RpcServer>& server, const wp<EventListene

	sp<RpcSession::RpcConnection> RpcSession::assignIncomingConnectionToThisThread(		sp<RpcSession::RpcConnection> RpcSession::assignIncomingConnectionToThisThread(
	std::unique_ptr<RpcTransport> rpcTransport) {		std::unique_ptr<RpcTransport> rpcTransport) {
	std::lock_guard<std::mutex> _l(mMutex);		RpcMutexLockGuard _l(mMutex);

	if (mConnections.mIncoming.size() >= mMaxIncomingThreads) {		if (mConnections.mIncoming.size() >= mMaxIncomingThreads) {
	ALOGE("Cannot add thread to session with %zu threads (max is set to %zu)",		ALOGE("Cannot add thread to session with %zu threads (max is set to %zu)",
	@@ -754,7 +759,7 @@ sp<RpcSession::RpcConnection> RpcSession::assignIncomingConnectionToThisThread(

	sp<RpcConnection> session = sp<RpcConnection>::make();		sp<RpcConnection> session = sp<RpcConnection>::make();
	session->rpcTransport = std::move(rpcTransport);		session->rpcTransport = std::move(rpcTransport);
	session->exclusiveTid = base::GetThreadId();		session->exclusiveTid = rpcGetThreadId();

	mConnections.mIncoming.push_back(session);		mConnections.mIncoming.push_back(session);
	mConnections.mMaxIncoming = mConnections.mIncoming.size();		mConnections.mMaxIncoming = mConnections.mIncoming.size();
	@@ -763,7 +768,7 @@ sp<RpcSession::RpcConnection> RpcSession::assignIncomingConnectionToThisThread(
	}		}

	bool RpcSession::removeIncomingConnection(const sp<RpcConnection>& connection) {		bool RpcSession::removeIncomingConnection(const sp<RpcConnection>& connection) {
	std::unique_lock<std::mutex> _l(mMutex);		RpcMutexUniqueLock _l(mMutex);
	if (auto it =		if (auto it =
	std::find(mConnections.mIncoming.begin(), mConnections.mIncoming.end(), connection);		std::find(mConnections.mIncoming.begin(), mConnections.mIncoming.end(), connection);
	it != mConnections.mIncoming.end()) {		it != mConnections.mIncoming.end()) {
	@@ -781,7 +786,7 @@ bool RpcSession::removeIncomingConnection(const sp<RpcConnection>& connection) {
	}		}

	void RpcSession::clearConnectionTid(const sp<RpcConnection>& connection) {		void RpcSession::clearConnectionTid(const sp<RpcConnection>& connection) {
	std::unique_lock<std::mutex> _l(mMutex);		RpcMutexUniqueLock _l(mMutex);
	connection->exclusiveTid = std::nullopt;		connection->exclusiveTid = std::nullopt;
	if (mConnections.mWaitingThreads > 0) {		if (mConnections.mWaitingThreads > 0) {
	_l.unlock();		_l.unlock();
	@@ -799,8 +804,8 @@ status_t RpcSession::ExclusiveConnection::find(const sp<RpcSession>& session, Co
	connection->mConnection = nullptr;		connection->mConnection = nullptr;
	connection->mReentrant = false;		connection->mReentrant = false;

	uint64_t tid = base::GetThreadId();		uint64_t tid = rpcGetThreadId();
	std::unique_lock<std::mutex> _l(session->mMutex);		RpcMutexUniqueLock _l(session->mMutex);

	session->mConnections.mWaitingThreads++;		session->mConnections.mWaitingThreads++;
	while (true) {		while (true) {