Merge "libbinder: RPC mThreadState -> mConnections" am: a25b012056 (53ad1b32) · Commits · e / os / android_frameworks_native-old

libs/binder/RpcSession.cpp

+37 −40

Original line number	Original line	Diff line number	Diff line
	@@ -61,7 +61,7 @@ RpcSession::~RpcSession() {
	LOG_RPC_DETAIL("RpcSession destroyed %p", this);		LOG_RPC_DETAIL("RpcSession destroyed %p", this);

	std::lock_guard<std::mutex> _l(mMutex);		std::lock_guard<std::mutex> _l(mMutex);
	LOG_ALWAYS_FATAL_IF(mThreadState.mIncomingConnections.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.");
	}		}

	@@ -78,12 +78,10 @@ sp<RpcSession> RpcSession::make(std::unique_ptr<RpcTransportCtxFactory> rpcTrans

	void RpcSession::setMaxThreads(size_t threads) {		void RpcSession::setMaxThreads(size_t threads) {
	std::lock_guard<std::mutex> _l(mMutex);		std::lock_guard<std::mutex> _l(mMutex);
	LOG_ALWAYS_FATAL_IF(!mThreadState.mOutgoingConnections.empty() \|\|		LOG_ALWAYS_FATAL_IF(!mConnections.mOutgoing.empty() \|\| !mConnections.mIncoming.empty(),
	!mThreadState.mIncomingConnections.empty(),
	"Must set max threads before setting up connections, but has %zu client(s) "		"Must set max threads before setting up connections, but has %zu client(s) "
	"and %zu server(s)",		"and %zu server(s)",
	mThreadState.mOutgoingConnections.size(),		mConnections.mOutgoing.size(), mConnections.mIncoming.size());
	mThreadState.mIncomingConnections.size());
	mMaxThreads = threads;		mMaxThreads = threads;
	}		}

	@@ -197,7 +195,7 @@ bool RpcSession::shutdownAndWait(bool wait) {
	LOG_ALWAYS_FATAL_IF(mShutdownListener == nullptr, "Shutdown listener not installed");		LOG_ALWAYS_FATAL_IF(mShutdownListener == nullptr, "Shutdown listener not installed");
	mShutdownListener->waitForShutdown(_l, sp<RpcSession>::fromExisting(this));		mShutdownListener->waitForShutdown(_l, sp<RpcSession>::fromExisting(this));

	LOG_ALWAYS_FATAL_IF(!mThreadState.mThreads.empty(), "Shutdown failed");		LOG_ALWAYS_FATAL_IF(!mConnections.mThreads.empty(), "Shutdown failed");
	}		}

	_l.unlock();		_l.unlock();
	@@ -263,11 +261,11 @@ void RpcSession::WaitForShutdownListener::onSessionIncomingThreadEnded() {

	void RpcSession::WaitForShutdownListener::waitForShutdown(std::unique_lock<std::mutex>& lock,		void RpcSession::WaitForShutdownListener::waitForShutdown(std::unique_lock<std::mutex>& lock,
	const sp<RpcSession>& session) {		const sp<RpcSession>& session) {
	while (session->mThreadState.mIncomingConnections.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))) {
	ALOGE("Waiting for RpcSession to shut down (1s w/o progress): %zu incoming connections "		ALOGE("Waiting for RpcSession to shut down (1s w/o progress): %zu incoming connections "
	"still.",		"still.",
	session->mThreadState.mIncomingConnections.size());		session->mConnections.mIncoming.size());
	}		}
	}		}
	}		}
	@@ -277,7 +275,7 @@ void RpcSession::preJoinThreadOwnership(std::thread thread) {

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

	@@ -380,10 +378,10 @@ void RpcSession::join(sp<RpcSession>&& session, PreJoinSetupResult&& setupResult
	sp<RpcSession::EventListener> listener;		sp<RpcSession::EventListener> listener;
	{		{
	std::lock_guard<std::mutex> _l(session->mMutex);		std::lock_guard<std::mutex> _l(session->mMutex);
	auto it = session->mThreadState.mThreads.find(std::this_thread::get_id());		auto it = session->mConnections.mThreads.find(std::this_thread::get_id());
	LOG_ALWAYS_FATAL_IF(it == session->mThreadState.mThreads.end());		LOG_ALWAYS_FATAL_IF(it == session->mConnections.mThreads.end());
	it->second.detach();		it->second.detach();
	session->mThreadState.mThreads.erase(it);		session->mConnections.mThreads.erase(it);

	listener = session->mEventListener.promote();		listener = session->mEventListener.promote();
	}		}
	@@ -414,9 +412,9 @@ status_t RpcSession::setupClient(const std::function<status_t(const std::vector<
	bool incoming)>& connectAndInit) {		bool incoming)>& connectAndInit) {
	{		{
	std::lock_guard<std::mutex> _l(mMutex);		std::lock_guard<std::mutex> _l(mMutex);
	LOG_ALWAYS_FATAL_IF(mThreadState.mOutgoingConnections.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",
	mThreadState.mOutgoingConnections.size());		mConnections.mOutgoing.size());
	}		}

	if (auto status = initShutdownTrigger(); status != OK) return status;		if (auto status = initShutdownTrigger(); status != OK) return status;
	@@ -439,7 +437,7 @@ status_t RpcSession::setupClient(const std::function<status_t(const std::vector<
	// downgrade again		// downgrade again
	mProtocolVersion = oldProtocolVersion;		mProtocolVersion = oldProtocolVersion;

	mThreadState = {};		mConnections = {};
	});		});

	if (status_t status = connectAndInit({}, false /incoming/); status != OK) return status;		if (status_t status = connectAndInit({}, false /incoming/); status != OK) return status;
	@@ -662,7 +660,7 @@ status_t RpcSession::addOutgoingConnection(std::unique_ptr<RpcTransport> rpcTran
	std::lock_guard<std::mutex> _l(mMutex);		std::lock_guard<std::mutex> _l(mMutex);
	connection->rpcTransport = std::move(rpcTransport);		connection->rpcTransport = std::move(rpcTransport);
	connection->exclusiveTid = gettid();		connection->exclusiveTid = gettid();
	mThreadState.mOutgoingConnections.push_back(connection);		mConnections.mOutgoing.push_back(connection);
	}		}

	status_t status = OK;		status_t status = OK;
	@@ -699,9 +697,9 @@ sp<RpcSession::RpcConnection> RpcSession::assignIncomingConnectionToThisThread(
	std::unique_ptr<RpcTransport> rpcTransport) {		std::unique_ptr<RpcTransport> rpcTransport) {
	std::lock_guard<std::mutex> _l(mMutex);		std::lock_guard<std::mutex> _l(mMutex);

	if (mThreadState.mIncomingConnections.size() >= mMaxThreads) {		if (mConnections.mIncoming.size() >= mMaxThreads) {
	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)",
	mThreadState.mIncomingConnections.size(), mMaxThreads);		mConnections.mIncoming.size(), mMaxThreads);
	return nullptr;		return nullptr;
	}		}

	@@ -709,7 +707,7 @@ sp<RpcSession::RpcConnection> RpcSession::assignIncomingConnectionToThisThread(
	// happens when new connections are still being established as part of a		// happens when new connections are still being established as part of a
	// very short-lived session which shuts down after it already started		// very short-lived session which shuts down after it already started
	// accepting new connections.		// accepting new connections.
	if (mThreadState.mIncomingConnections.size() < mThreadState.mMaxIncomingConnections) {		if (mConnections.mIncoming.size() < mConnections.mMaxIncoming) {
	return nullptr;		return nullptr;
	}		}

	@@ -717,19 +715,19 @@ sp<RpcSession::RpcConnection> RpcSession::assignIncomingConnectionToThisThread(
	session->rpcTransport = std::move(rpcTransport);		session->rpcTransport = std::move(rpcTransport);
	session->exclusiveTid = gettid();		session->exclusiveTid = gettid();

	mThreadState.mIncomingConnections.push_back(session);		mConnections.mIncoming.push_back(session);
	mThreadState.mMaxIncomingConnections = mThreadState.mIncomingConnections.size();		mConnections.mMaxIncoming = mConnections.mIncoming.size();

	return session;		return session;
	}		}

	bool RpcSession::removeIncomingConnection(const sp<RpcConnection>& connection) {		bool RpcSession::removeIncomingConnection(const sp<RpcConnection>& connection) {
	std::unique_lock<std::mutex> _l(mMutex);		std::unique_lock<std::mutex> _l(mMutex);
	if (auto it = std::find(mThreadState.mIncomingConnections.begin(),		if (auto it =
	mThreadState.mIncomingConnections.end(), connection);		std::find(mConnections.mIncoming.begin(), mConnections.mIncoming.end(), connection);
	it != mThreadState.mIncomingConnections.end()) {		it != mConnections.mIncoming.end()) {
	mThreadState.mIncomingConnections.erase(it);		mConnections.mIncoming.erase(it);
	if (mThreadState.mIncomingConnections.size() == 0) {		if (mConnections.mIncoming.size() == 0) {
	sp<EventListener> listener = mEventListener.promote();		sp<EventListener> listener = mEventListener.promote();
	if (listener) {		if (listener) {
	_l.unlock();		_l.unlock();
	@@ -754,7 +752,7 @@ status_t RpcSession::ExclusiveConnection::find(const sp<RpcSession>& session, Co
	pid_t tid = gettid();		pid_t tid = gettid();
	std::unique_lock<std::mutex> _l(session->mMutex);		std::unique_lock<std::mutex> _l(session->mMutex);

	session->mThreadState.mWaitingThreads++;		session->mConnections.mWaitingThreads++;
	while (true) {		while (true) {
	sp<RpcConnection> exclusive;		sp<RpcConnection> exclusive;
	sp<RpcConnection> available;		sp<RpcConnection> available;
	@@ -762,11 +760,11 @@ status_t RpcSession::ExclusiveConnection::find(const sp<RpcSession>& session, Co
	// CHECK FOR DEDICATED CLIENT SOCKET		// CHECK FOR DEDICATED CLIENT SOCKET
	//		//
	// A server/looper should always use a dedicated connection if available		// A server/looper should always use a dedicated connection if available
	findConnection(tid, &exclusive, &available, session->mThreadState.mOutgoingConnections,		findConnection(tid, &exclusive, &available, session->mConnections.mOutgoing,
	session->mThreadState.mOutgoingConnectionsOffset);		session->mConnections.mOutgoingOffset);

	// WARNING: this assumes a server cannot request its client to send		// WARNING: this assumes a server cannot request its client to send
	// a transaction, as mIncomingConnections is excluded below.		// a transaction, as mIncoming is excluded below.
	//		//
	// Imagine we have more than one thread in play, and a single thread		// Imagine we have more than one thread in play, and a single thread
	// sends a synchronous, then an asynchronous command. Imagine the		// sends a synchronous, then an asynchronous command. Imagine the
	@@ -776,9 +774,8 @@ status_t RpcSession::ExclusiveConnection::find(const sp<RpcSession>& session, Co
	// command. So, we move to considering the second available thread		// command. So, we move to considering the second available thread
	// for subsequent calls.		// for subsequent calls.
	if (use == ConnectionUse::CLIENT_ASYNC && (exclusive != nullptr \|\| available != nullptr)) {		if (use == ConnectionUse::CLIENT_ASYNC && (exclusive != nullptr \|\| available != nullptr)) {
	session->mThreadState.mOutgoingConnectionsOffset =		session->mConnections.mOutgoingOffset = (session->mConnections.mOutgoingOffset + 1) %
	(session->mThreadState.mOutgoingConnectionsOffset + 1) %		session->mConnections.mOutgoing.size();
	session->mThreadState.mOutgoingConnections.size();
	}		}

	// USE SERVING SOCKET (e.g. nested transaction)		// USE SERVING SOCKET (e.g. nested transaction)
	@@ -786,7 +783,7 @@ status_t RpcSession::ExclusiveConnection::find(const sp<RpcSession>& session, Co
	sp<RpcConnection> exclusiveIncoming;		sp<RpcConnection> exclusiveIncoming;
	// server connections are always assigned to a thread		// server connections are always assigned to a thread
	findConnection(tid, &exclusiveIncoming, nullptr /available/,		findConnection(tid, &exclusiveIncoming, nullptr /available/,
	session->mThreadState.mIncomingConnections, 0 /* index hint */);		session->mConnections.mIncoming, 0 /* index hint */);

	// asynchronous calls cannot be nested, we currently allow ref count		// asynchronous calls cannot be nested, we currently allow ref count
	// calls to be nested (so that you can use this without having extra		// calls to be nested (so that you can use this without having extra
	@@ -815,20 +812,20 @@ status_t RpcSession::ExclusiveConnection::find(const sp<RpcSession>& session, Co
	break;		break;
	}		}

	if (session->mThreadState.mOutgoingConnections.size() == 0) {		if (session->mConnections.mOutgoing.size() == 0) {
	ALOGE("Session has no client connections. This is required for an RPC server to make "		ALOGE("Session has no client connections. This is required for an RPC server to make "
	"any non-nested (e.g. oneway or on another thread) calls. Use: %d. Server "		"any non-nested (e.g. oneway or on another thread) calls. Use: %d. Server "
	"connections: %zu",		"connections: %zu",
	static_cast<int>(use), session->mThreadState.mIncomingConnections.size());		static_cast<int>(use), session->mConnections.mIncoming.size());
	return WOULD_BLOCK;		return WOULD_BLOCK;
	}		}

	LOG_RPC_DETAIL("No available connections (have %zu clients and %zu servers). Waiting...",		LOG_RPC_DETAIL("No available connections (have %zu clients and %zu servers). Waiting...",
	session->mThreadState.mOutgoingConnections.size(),		session->mConnections.mOutgoing.size(),
	session->mThreadState.mIncomingConnections.size());		session->mConnections.mIncoming.size());
	session->mAvailableConnectionCv.wait(_l);		session->mAvailableConnectionCv.wait(_l);
	}		}
	session->mThreadState.mWaitingThreads--;		session->mConnections.mWaitingThreads--;

	return OK;		return OK;
	}		}
	@@ -867,7 +864,7 @@ RpcSession::ExclusiveConnection::~ExclusiveConnection() {
	if (!mReentrant && mConnection != nullptr) {		if (!mReentrant && mConnection != nullptr) {
	std::unique_lock<std::mutex> _l(mSession->mMutex);		std::unique_lock<std::mutex> _l(mSession->mMutex);
	mConnection->exclusiveTid = std::nullopt;		mConnection->exclusiveTid = std::nullopt;
	if (mSession->mThreadState.mWaitingThreads > 0) {		if (mSession->mConnections.mWaitingThreads > 0) {
	_l.unlock();		_l.unlock();
	mSession->mAvailableConnectionCv.notify_one();		mSession->mAvailableConnectionCv.notify_one();
	}		}

libs/binder/include/binder/RpcSession.h

+9 −9

Original line number	Original line	Diff line number	Diff line
	@@ -281,13 +281,13 @@ private:

	const std::unique_ptr<RpcTransportCtx> mCtx;		const std::unique_ptr<RpcTransportCtx> mCtx;

	// On the other side of a session, for each of mOutgoingConnections here, there should		// On the other side of a session, for each of mOutgoing here, there should
	// be one of mIncomingConnections on the other side (and vice versa).		// be one of mIncoming on the other side (and vice versa).
	//		//
	// For the simplest session, a single server with one client, you would		// For the simplest session, a single server with one client, you would
	// have:		// have:
	// - the server has a single 'mIncomingConnections' and a thread listening on this		// - the server has a single 'mIncoming' and a thread listening on this
	// - the client has a single 'mOutgoingConnections' and makes calls to this		// - the client has a single 'mOutgoing' and makes calls to this
	// - here, when the client makes a call, the server can call back into it		// - here, when the client makes a call, the server can call back into it
	// (nested calls), but outside of this, the client will only ever read		// (nested calls), but outside of this, the client will only ever read
	// calls from the server when it makes a call itself.		// calls from the server when it makes a call itself.
	@@ -315,12 +315,12 @@ private:
	struct ThreadState {		struct ThreadState {
	size_t mWaitingThreads = 0;		size_t mWaitingThreads = 0;
	// hint index into clients, ++ when sending an async transaction		// hint index into clients, ++ when sending an async transaction
	size_t mOutgoingConnectionsOffset = 0;		size_t mOutgoingOffset = 0;
	std::vector<sp<RpcConnection>> mOutgoingConnections;		std::vector<sp<RpcConnection>> mOutgoing;
	size_t mMaxIncomingConnections = 0;		size_t mMaxIncoming = 0;
	std::vector<sp<RpcConnection>> mIncomingConnections;		std::vector<sp<RpcConnection>> mIncoming;
	std::map<std::thread::id, std::thread> mThreads;		std::map<std::thread::id, std::thread> mThreads;
	} mThreadState;		} mConnections;
	};		};

	} // namespace android		} // namespace android