libs/binder/RpcSession.cpp - third_party/android/platform/frameworks/native - Git at Google

 /*
  * Copyright (C) 2020 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.
  */

 #define LOG_TAG "RpcSession"

 #include <binder/RpcSession.h>

 #include <inttypes.h>
 #include <unistd.h>

 #include <string_view>

 #include <binder/Parcel.h>
 #include <binder/RpcServer.h>
 #include <binder/Stability.h>
 #include <utils/String8.h>

 #include "RpcSocketAddress.h"
 #include "RpcState.h"
 #include "RpcWireFormat.h"

 #ifdef __GLIBC__
 extern "C" pid_t gettid();
 #endif

 namespace android {

 using base::unique_fd;

 RpcSession::RpcSession() {
     LOG_RPC_DETAIL("RpcSession created %p", this);

     mState = std::make_unique<RpcState>();
 }
 RpcSession::~RpcSession() {
     LOG_RPC_DETAIL("RpcSession destroyed %p", this);

     std::lock_guard<std::mutex> _l(mMutex);
     LOG_ALWAYS_FATAL_IF(mServerConnections.size() != 0,
                         "Should not be able to destroy a session with servers in use.");
 }

 sp<RpcSession> RpcSession::make() {
     return sp<RpcSession>::make();
 }

 bool RpcSession::setupUnixDomainClient(const char* path) {
     return setupSocketClient(UnixSocketAddress(path));
 }

 bool RpcSession::setupVsockClient(unsigned int cid, unsigned int port) {
     return setupSocketClient(VsockSocketAddress(cid, port));
 }

 bool RpcSession::setupInetClient(const char* addr, unsigned int port) {
     auto aiStart = InetSocketAddress::getAddrInfo(addr, port);
     if (aiStart == nullptr) return false;
     for (auto ai = aiStart.get(); ai != nullptr; ai = ai->ai_next) {
         InetSocketAddress socketAddress(ai->ai_addr, ai->ai_addrlen, addr, port);
         if (setupSocketClient(socketAddress)) return true;
     }
     ALOGE("None of the socket address resolved for %s:%u can be added as inet client.", addr, port);
     return false;
 }

 bool RpcSession::addNullDebuggingClient() {
     unique_fd serverFd(TEMP_FAILURE_RETRY(open("/dev/null", O_WRONLY | O_CLOEXEC)));

     if (serverFd == -1) {
         ALOGE("Could not connect to /dev/null: %s", strerror(errno));
         return false;
     }

     addClientConnection(std::move(serverFd));
     return true;
 }

 sp<IBinder> RpcSession::getRootObject() {
     ExclusiveConnection connection(sp<RpcSession>::fromExisting(this), ConnectionUse::CLIENT);
     return state()->getRootObject(connection.fd(), sp<RpcSession>::fromExisting(this));
 }

 status_t RpcSession::getRemoteMaxThreads(size_t* maxThreads) {
     ExclusiveConnection connection(sp<RpcSession>::fromExisting(this), ConnectionUse::CLIENT);
     return state()->getMaxThreads(connection.fd(), sp<RpcSession>::fromExisting(this), maxThreads);
 }

 status_t RpcSession::transact(const RpcAddress& address, uint32_t code, const Parcel& data,
                               Parcel* reply, uint32_t flags) {
     ExclusiveConnection connection(sp<RpcSession>::fromExisting(this),
                                    (flags & IBinder::FLAG_ONEWAY) ? ConnectionUse::CLIENT_ASYNC
                                                                   : ConnectionUse::CLIENT);
     return state()->transact(connection.fd(), address, code, data,
                              sp<RpcSession>::fromExisting(this), reply, flags);
 }

 status_t RpcSession::sendDecStrong(const RpcAddress& address) {
     ExclusiveConnection connection(sp<RpcSession>::fromExisting(this),
                                    ConnectionUse::CLIENT_REFCOUNT);
     return state()->sendDecStrong(connection.fd(), address);
 }

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

     int32_t id;

     ExclusiveConnection connection(sp<RpcSession>::fromExisting(this), ConnectionUse::CLIENT);
     status_t status =
             state()->getSessionId(connection.fd(), sp<RpcSession>::fromExisting(this), &id);
     if (status != OK) return status;

     LOG_RPC_DETAIL("RpcSession %p has id %d", this, id);
     mId = id;
     return OK;
 }

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

     {
         std::lock_guard<std::mutex> _l(mMutex);
         mThreads[thread.get_id()] = std::move(thread);
     }
 }

 void RpcSession::join(unique_fd client) {
     // must be registered to allow arbitrary client code executing commands to
     // be able to do nested calls (we can't only read from it)
     sp<RpcConnection> connection = assignServerToThisThread(std::move(client));

     while (true) {
         status_t error =
                 state()->getAndExecuteCommand(connection->fd, sp<RpcSession>::fromExisting(this));

         if (error != OK) {
             ALOGI("Binder connection thread closing w/ status %s", statusToString(error).c_str());
             break;
         }
     }

     LOG_ALWAYS_FATAL_IF(!removeServerConnection(connection),
                         "bad state: connection object guaranteed to be in list");

     {
         std::lock_guard<std::mutex> _l(mMutex);
         auto it = mThreads.find(std::this_thread::get_id());
         LOG_ALWAYS_FATAL_IF(it == mThreads.end());
         it->second.detach();
         mThreads.erase(it);
     }
 }

 void RpcSession::terminateLocked() {
     // TODO(b/185167543):
     // - kindly notify other side of the connection of termination (can't be
     // locked)
     // - prevent new client/servers from being added
     // - stop all threads which are currently reading/writing
     // - terminate RpcState?

     if (mTerminated) return;

     sp<RpcServer> server = mForServer.promote();
     if (server) {
         server->onSessionTerminating(sp<RpcSession>::fromExisting(this));
     }
 }

 wp<RpcServer> RpcSession::server() {
     return mForServer;
 }

 bool RpcSession::setupSocketClient(const RpcSocketAddress& addr) {
     {
         std::lock_guard<std::mutex> _l(mMutex);
         LOG_ALWAYS_FATAL_IF(mClientConnections.size() != 0,
                             "Must only setup session once, but already has %zu clients",
                             mClientConnections.size());
     }

     if (!setupOneSocketClient(addr, RPC_SESSION_ID_NEW)) return false;

     // TODO(b/185167543): we should add additional sessions dynamically
     // instead of all at once.
     // TODO(b/186470974): first risk of blocking
     size_t numThreadsAvailable;
     if (status_t status = getRemoteMaxThreads(&numThreadsAvailable); status != OK) {
         ALOGE("Could not get max threads after initial session to %s: %s", addr.toString().c_str(),
               statusToString(status).c_str());
         return false;
     }

     if (status_t status = readId(); status != OK) {
         ALOGE("Could not get session id after initial session to %s; %s", addr.toString().c_str(),
               statusToString(status).c_str());
         return false;
     }

     // we've already setup one client
     for (size_t i = 0; i + 1 < numThreadsAvailable; i++) {
         // TODO(b/185167543): shutdown existing connections?
         if (!setupOneSocketClient(addr, mId.value())) return false;
     }

     return true;
 }

 bool RpcSession::setupOneSocketClient(const RpcSocketAddress& addr, int32_t id) {
     for (size_t tries = 0; tries < 5; tries++) {
         if (tries > 0) usleep(10000);

         unique_fd serverFd(
                 TEMP_FAILURE_RETRY(socket(addr.addr()->sa_family, SOCK_STREAM | SOCK_CLOEXEC, 0)));
         if (serverFd == -1) {
             int savedErrno = errno;
             ALOGE("Could not create socket at %s: %s", addr.toString().c_str(),
                   strerror(savedErrno));
             return false;
         }

         if (0 != TEMP_FAILURE_RETRY(connect(serverFd.get(), addr.addr(), addr.addrSize()))) {
             if (errno == ECONNRESET) {
                 ALOGW("Connection reset on %s", addr.toString().c_str());
                 continue;
             }
             int savedErrno = errno;
             ALOGE("Could not connect socket at %s: %s", addr.toString().c_str(),
                   strerror(savedErrno));
             return false;
         }

         if (sizeof(id) != TEMP_FAILURE_RETRY(write(serverFd.get(), &id, sizeof(id)))) {
             int savedErrno = errno;
             ALOGE("Could not write id to socket at %s: %s", addr.toString().c_str(),
                   strerror(savedErrno));
             return false;
         }

         LOG_RPC_DETAIL("Socket at %s client with fd %d", addr.toString().c_str(), serverFd.get());

         addClientConnection(std::move(serverFd));
         return true;
     }

     ALOGE("Ran out of retries to connect to %s", addr.toString().c_str());
     return false;
 }

 void RpcSession::addClientConnection(unique_fd fd) {
     std::lock_guard<std::mutex> _l(mMutex);
     sp<RpcConnection> session = sp<RpcConnection>::make();
     session->fd = std::move(fd);
     mClientConnections.push_back(session);
 }

 void RpcSession::setForServer(const wp<RpcServer>& server, int32_t sessionId) {
     mId = sessionId;
     mForServer = server;
 }

 sp<RpcSession::RpcConnection> RpcSession::assignServerToThisThread(unique_fd fd) {
     std::lock_guard<std::mutex> _l(mMutex);
     sp<RpcConnection> session = sp<RpcConnection>::make();
     session->fd = std::move(fd);
     session->exclusiveTid = gettid();
     mServerConnections.push_back(session);

     return session;
 }

 bool RpcSession::removeServerConnection(const sp<RpcConnection>& connection) {
     std::lock_guard<std::mutex> _l(mMutex);
     if (auto it = std::find(mServerConnections.begin(), mServerConnections.end(), connection);
         it != mServerConnections.end()) {
         mServerConnections.erase(it);
         if (mServerConnections.size() == 0) {
             terminateLocked();
         }
         return true;
     }
     return false;
 }

 RpcSession::ExclusiveConnection::ExclusiveConnection(const sp<RpcSession>& session,
                                                      ConnectionUse use)
       : mSession(session) {
     pid_t tid = gettid();
     std::unique_lock<std::mutex> _l(mSession->mMutex);

     mSession->mWaitingThreads++;
     while (true) {
         sp<RpcConnection> exclusive;
         sp<RpcConnection> available;

         // CHECK FOR DEDICATED CLIENT SOCKET
         //
         // A server/looper should always use a dedicated session if available
         findConnection(tid, &exclusive, &available, mSession->mClientConnections,
                        mSession->mClientConnectionsOffset);

         // WARNING: this assumes a server cannot request its client to send
         // a transaction, as mServerConnections is excluded below.
         //
         // Imagine we have more than one thread in play, and a single thread
         // sends a synchronous, then an asynchronous command. Imagine the
         // asynchronous command is sent on the first client connection. Then, if
         // we naively send a synchronous command to that same connection, the
         // thread on the far side might be busy processing the asynchronous
         // command. So, we move to considering the second available thread
         // for subsequent calls.
         if (use == ConnectionUse::CLIENT_ASYNC && (exclusive != nullptr || available != nullptr)) {
             mSession->mClientConnectionsOffset =
                     (mSession->mClientConnectionsOffset + 1) % mSession->mClientConnections.size();
         }

         // USE SERVING SOCKET (for nested transaction)
         //
         // asynchronous calls cannot be nested
         if (use != ConnectionUse::CLIENT_ASYNC) {
             // server connections are always assigned to a thread
             findConnection(tid, &exclusive, nullptr /*available*/, mSession->mServerConnections,
                            0 /* index hint */);
         }

         // if our thread is already using a session, prioritize using that
         if (exclusive != nullptr) {
             mConnection = exclusive;
             mReentrant = true;
             break;
         } else if (available != nullptr) {
             mConnection = available;
             mConnection->exclusiveTid = tid;
             break;
         }

         // in regular binder, this would usually be a deadlock :)
         LOG_ALWAYS_FATAL_IF(mSession->mClientConnections.size() == 0,
                             "Not a client of any session. You must create a session to an "
                             "RPC server to make any non-nested (e.g. oneway or on another thread) "
                             "calls.");

         LOG_RPC_DETAIL("No available session (have %zu clients and %zu servers). Waiting...",
                        mSession->mClientConnections.size(), mSession->mServerConnections.size());
         mSession->mAvailableConnectionCv.wait(_l);
     }
     mSession->mWaitingThreads--;
 }

 void RpcSession::ExclusiveConnection::findConnection(pid_t tid, sp<RpcConnection>* exclusive,
                                                      sp<RpcConnection>* available,
                                                      std::vector<sp<RpcConnection>>& sockets,
                                                      size_t socketsIndexHint) {
     LOG_ALWAYS_FATAL_IF(sockets.size() > 0 && socketsIndexHint >= sockets.size(),
                         "Bad index %zu >= %zu", socketsIndexHint, sockets.size());

     if (*exclusive != nullptr) return; // consistent with break below

     for (size_t i = 0; i < sockets.size(); i++) {
         sp<RpcConnection>& socket = sockets[(i + socketsIndexHint) % sockets.size()];

         // take first available session (intuition = caching)
         if (available && *available == nullptr && socket->exclusiveTid == std::nullopt) {
             *available = socket;
             continue;
         }

         // though, prefer to take session which is already inuse by this thread
         // (nested transactions)
         if (exclusive && socket->exclusiveTid == tid) {
             *exclusive = socket;
             break; // consistent with return above
         }
     }
 }

 RpcSession::ExclusiveConnection::~ExclusiveConnection() {
     // reentrant use of a session means something less deep in the call stack
     // is using this fd, and it retains the right to it. So, we don't give up
     // exclusive ownership, and no thread is freed.
     if (!mReentrant) {
         std::unique_lock<std::mutex> _l(mSession->mMutex);
         mConnection->exclusiveTid = std::nullopt;
         if (mSession->mWaitingThreads > 0) {
             _l.unlock();
             mSession->mAvailableConnectionCv.notify_one();
         }
     }
 }

 } // namespace android
	/*
	* Copyright (C) 2020 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.
	*/

	#define LOG_TAG "RpcSession"

	#include <binder/RpcSession.h>

	#include <inttypes.h>
	#include <unistd.h>

	#include <string_view>

	#include <binder/Parcel.h>
	#include <binder/RpcServer.h>
	#include <binder/Stability.h>
	#include <utils/String8.h>

	#include "RpcSocketAddress.h"
	#include "RpcState.h"
	#include "RpcWireFormat.h"

	#ifdef __GLIBC__
	extern "C" pid_t gettid();
	#endif

	namespace android {

	using base::unique_fd;

	RpcSession::RpcSession() {
	LOG_RPC_DETAIL("RpcSession created %p", this);

	mState = std::make_unique<RpcState>();
	}
	RpcSession::~RpcSession() {
	LOG_RPC_DETAIL("RpcSession destroyed %p", this);

	std::lock_guard<std::mutex> _l(mMutex);
	LOG_ALWAYS_FATAL_IF(mServerConnections.size() != 0,
	"Should not be able to destroy a session with servers in use.");
	}

	sp<RpcSession> RpcSession::make() {
	return sp<RpcSession>::make();
	}

	bool RpcSession::setupUnixDomainClient(const char* path) {
	return setupSocketClient(UnixSocketAddress(path));
	}

	bool RpcSession::setupVsockClient(unsigned int cid, unsigned int port) {
	return setupSocketClient(VsockSocketAddress(cid, port));
	}

	bool RpcSession::setupInetClient(const char* addr, unsigned int port) {
	auto aiStart = InetSocketAddress::getAddrInfo(addr, port);
	if (aiStart == nullptr) return false;
	for (auto ai = aiStart.get(); ai != nullptr; ai = ai->ai_next) {
	InetSocketAddress socketAddress(ai->ai_addr, ai->ai_addrlen, addr, port);
	if (setupSocketClient(socketAddress)) return true;
	}
	ALOGE("None of the socket address resolved for %s:%u can be added as inet client.", addr, port);
	return false;
	}

	bool RpcSession::addNullDebuggingClient() {
	unique_fd serverFd(TEMP_FAILURE_RETRY(open("/dev/null", O_WRONLY \| O_CLOEXEC)));

	if (serverFd == -1) {
	ALOGE("Could not connect to /dev/null: %s", strerror(errno));
	return false;
	}

	addClientConnection(std::move(serverFd));
	return true;
	}

	sp<IBinder> RpcSession::getRootObject() {
	ExclusiveConnection connection(sp<RpcSession>::fromExisting(this), ConnectionUse::CLIENT);
	return state()->getRootObject(connection.fd(), sp<RpcSession>::fromExisting(this));
	}

	status_t RpcSession::getRemoteMaxThreads(size_t* maxThreads) {
	ExclusiveConnection connection(sp<RpcSession>::fromExisting(this), ConnectionUse::CLIENT);
	return state()->getMaxThreads(connection.fd(), sp<RpcSession>::fromExisting(this), maxThreads);
	}

	status_t RpcSession::transact(const RpcAddress& address, uint32_t code, const Parcel& data,
	Parcel* reply, uint32_t flags) {
	ExclusiveConnection connection(sp<RpcSession>::fromExisting(this),
	(flags & IBinder::FLAG_ONEWAY) ? ConnectionUse::CLIENT_ASYNC
	: ConnectionUse::CLIENT);
	return state()->transact(connection.fd(), address, code, data,
	sp<RpcSession>::fromExisting(this), reply, flags);
	}

	status_t RpcSession::sendDecStrong(const RpcAddress& address) {
	ExclusiveConnection connection(sp<RpcSession>::fromExisting(this),
	ConnectionUse::CLIENT_REFCOUNT);
	return state()->sendDecStrong(connection.fd(), address);
	}

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

	int32_t id;

	ExclusiveConnection connection(sp<RpcSession>::fromExisting(this), ConnectionUse::CLIENT);
	status_t status =
	state()->getSessionId(connection.fd(), sp<RpcSession>::fromExisting(this), &id);
	if (status != OK) return status;

	LOG_RPC_DETAIL("RpcSession %p has id %d", this, id);
	mId = id;
	return OK;
	}

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

	{
	std::lock_guard<std::mutex> _l(mMutex);
	mThreads[thread.get_id()] = std::move(thread);
	}
	}

	void RpcSession::join(unique_fd client) {
	// must be registered to allow arbitrary client code executing commands to
	// be able to do nested calls (we can't only read from it)
	sp<RpcConnection> connection = assignServerToThisThread(std::move(client));

	while (true) {
	status_t error =
	state()->getAndExecuteCommand(connection->fd, sp<RpcSession>::fromExisting(this));

	if (error != OK) {
	ALOGI("Binder connection thread closing w/ status %s", statusToString(error).c_str());
	break;
	}
	}

	LOG_ALWAYS_FATAL_IF(!removeServerConnection(connection),
	"bad state: connection object guaranteed to be in list");

	{
	std::lock_guard<std::mutex> _l(mMutex);
	auto it = mThreads.find(std::this_thread::get_id());
	LOG_ALWAYS_FATAL_IF(it == mThreads.end());
	it->second.detach();
	mThreads.erase(it);
	}
	}

	void RpcSession::terminateLocked() {
	// TODO(b/185167543):
	// - kindly notify other side of the connection of termination (can't be
	// locked)
	// - prevent new client/servers from being added
	// - stop all threads which are currently reading/writing
	// - terminate RpcState?

	if (mTerminated) return;

	sp<RpcServer> server = mForServer.promote();
	if (server) {
	server->onSessionTerminating(sp<RpcSession>::fromExisting(this));
	}
	}

	wp<RpcServer> RpcSession::server() {
	return mForServer;
	}

	bool RpcSession::setupSocketClient(const RpcSocketAddress& addr) {
	{
	std::lock_guard<std::mutex> _l(mMutex);
	LOG_ALWAYS_FATAL_IF(mClientConnections.size() != 0,
	"Must only setup session once, but already has %zu clients",
	mClientConnections.size());
	}

	if (!setupOneSocketClient(addr, RPC_SESSION_ID_NEW)) return false;

	// TODO(b/185167543): we should add additional sessions dynamically
	// instead of all at once.
	// TODO(b/186470974): first risk of blocking
	size_t numThreadsAvailable;
	if (status_t status = getRemoteMaxThreads(&numThreadsAvailable); status != OK) {
	ALOGE("Could not get max threads after initial session to %s: %s", addr.toString().c_str(),
	statusToString(status).c_str());
	return false;
	}

	if (status_t status = readId(); status != OK) {
	ALOGE("Could not get session id after initial session to %s; %s", addr.toString().c_str(),
	statusToString(status).c_str());
	return false;
	}

	// we've already setup one client
	for (size_t i = 0; i + 1 < numThreadsAvailable; i++) {
	// TODO(b/185167543): shutdown existing connections?
	if (!setupOneSocketClient(addr, mId.value())) return false;
	}

	return true;
	}

	bool RpcSession::setupOneSocketClient(const RpcSocketAddress& addr, int32_t id) {
	for (size_t tries = 0; tries < 5; tries++) {
	if (tries > 0) usleep(10000);

	unique_fd serverFd(
	TEMP_FAILURE_RETRY(socket(addr.addr()->sa_family, SOCK_STREAM \| SOCK_CLOEXEC, 0)));
	if (serverFd == -1) {
	int savedErrno = errno;
	ALOGE("Could not create socket at %s: %s", addr.toString().c_str(),
	strerror(savedErrno));
	return false;
	}

	if (0 != TEMP_FAILURE_RETRY(connect(serverFd.get(), addr.addr(), addr.addrSize()))) {
	if (errno == ECONNRESET) {
	ALOGW("Connection reset on %s", addr.toString().c_str());
	continue;
	}
	int savedErrno = errno;
	ALOGE("Could not connect socket at %s: %s", addr.toString().c_str(),
	strerror(savedErrno));
	return false;
	}

	if (sizeof(id) != TEMP_FAILURE_RETRY(write(serverFd.get(), &id, sizeof(id)))) {
	int savedErrno = errno;
	ALOGE("Could not write id to socket at %s: %s", addr.toString().c_str(),
	strerror(savedErrno));
	return false;
	}

	LOG_RPC_DETAIL("Socket at %s client with fd %d", addr.toString().c_str(), serverFd.get());

	addClientConnection(std::move(serverFd));
	return true;
	}

	ALOGE("Ran out of retries to connect to %s", addr.toString().c_str());
	return false;
	}

	void RpcSession::addClientConnection(unique_fd fd) {
	std::lock_guard<std::mutex> _l(mMutex);
	sp<RpcConnection> session = sp<RpcConnection>::make();
	session->fd = std::move(fd);
	mClientConnections.push_back(session);
	}

	void RpcSession::setForServer(const wp<RpcServer>& server, int32_t sessionId) {
	mId = sessionId;
	mForServer = server;
	}

	sp<RpcSession::RpcConnection> RpcSession::assignServerToThisThread(unique_fd fd) {
	std::lock_guard<std::mutex> _l(mMutex);
	sp<RpcConnection> session = sp<RpcConnection>::make();
	session->fd = std::move(fd);
	session->exclusiveTid = gettid();
	mServerConnections.push_back(session);

	return session;
	}

	bool RpcSession::removeServerConnection(const sp<RpcConnection>& connection) {
	std::lock_guard<std::mutex> _l(mMutex);
	if (auto it = std::find(mServerConnections.begin(), mServerConnections.end(), connection);
	it != mServerConnections.end()) {
	mServerConnections.erase(it);
	if (mServerConnections.size() == 0) {
	terminateLocked();
	}
	return true;
	}
	return false;
	}

	RpcSession::ExclusiveConnection::ExclusiveConnection(const sp<RpcSession>& session,
	ConnectionUse use)
	: mSession(session) {
	pid_t tid = gettid();
	std::unique_lock<std::mutex> _l(mSession->mMutex);

	mSession->mWaitingThreads++;
	while (true) {
	sp<RpcConnection> exclusive;
	sp<RpcConnection> available;

	// CHECK FOR DEDICATED CLIENT SOCKET
	//
	// A server/looper should always use a dedicated session if available
	findConnection(tid, &exclusive, &available, mSession->mClientConnections,
	mSession->mClientConnectionsOffset);

	// WARNING: this assumes a server cannot request its client to send
	// a transaction, as mServerConnections is excluded below.
	//
	// Imagine we have more than one thread in play, and a single thread
	// sends a synchronous, then an asynchronous command. Imagine the
	// asynchronous command is sent on the first client connection. Then, if
	// we naively send a synchronous command to that same connection, the
	// thread on the far side might be busy processing the asynchronous
	// command. So, we move to considering the second available thread
	// for subsequent calls.
	if (use == ConnectionUse::CLIENT_ASYNC && (exclusive != nullptr \|\| available != nullptr)) {
	mSession->mClientConnectionsOffset =
	(mSession->mClientConnectionsOffset + 1) % mSession->mClientConnections.size();
	}

	// USE SERVING SOCKET (for nested transaction)
	//
	// asynchronous calls cannot be nested
	if (use != ConnectionUse::CLIENT_ASYNC) {
	// server connections are always assigned to a thread
	findConnection(tid, &exclusive, nullptr /available/, mSession->mServerConnections,
	0 /* index hint */);
	}

	// if our thread is already using a session, prioritize using that
	if (exclusive != nullptr) {
	mConnection = exclusive;
	mReentrant = true;
	break;
	} else if (available != nullptr) {
	mConnection = available;
	mConnection->exclusiveTid = tid;
	break;
	}

	// in regular binder, this would usually be a deadlock :)
	LOG_ALWAYS_FATAL_IF(mSession->mClientConnections.size() == 0,
	"Not a client of any session. You must create a session to an "
	"RPC server to make any non-nested (e.g. oneway or on another thread) "
	"calls.");

	LOG_RPC_DETAIL("No available session (have %zu clients and %zu servers). Waiting...",
	mSession->mClientConnections.size(), mSession->mServerConnections.size());
	mSession->mAvailableConnectionCv.wait(_l);
	}
	mSession->mWaitingThreads--;
	}

	void RpcSession::ExclusiveConnection::findConnection(pid_t tid, sp<RpcConnection>* exclusive,
	sp<RpcConnection>* available,
	std::vector<sp<RpcConnection>>& sockets,
	size_t socketsIndexHint) {
	LOG_ALWAYS_FATAL_IF(sockets.size() > 0 && socketsIndexHint >= sockets.size(),
	"Bad index %zu >= %zu", socketsIndexHint, sockets.size());

	if (*exclusive != nullptr) return; // consistent with break below

	for (size_t i = 0; i < sockets.size(); i++) {
	sp<RpcConnection>& socket = sockets[(i + socketsIndexHint) % sockets.size()];

	// take first available session (intuition = caching)
	if (available && *available == nullptr && socket->exclusiveTid == std::nullopt) {
	*available = socket;
	continue;
	}

	// though, prefer to take session which is already inuse by this thread
	// (nested transactions)
	if (exclusive && socket->exclusiveTid == tid) {
	*exclusive = socket;
	break; // consistent with return above
	}
	}
	}

	RpcSession::ExclusiveConnection::~ExclusiveConnection() {
	// reentrant use of a session means something less deep in the call stack
	// is using this fd, and it retains the right to it. So, we don't give up
	// exclusive ownership, and no thread is freed.
	if (!mReentrant) {
	std::unique_lock<std::mutex> _l(mSession->mMutex);
	mConnection->exclusiveTid = std::nullopt;
	if (mSession->mWaitingThreads > 0) {
	_l.unlock();
	mSession->mAvailableConnectionCv.notify_one();
	}
	}
	}

	} // namespace android