Merge "libbinder: reverse connections" am: e81fe8297f (e8663b69) · Commits · e / os / platform_frameworks_native

libs/binder/RpcServer.cpp

+24 −11

Original line number	Diff line number	Diff line
		@@ -239,15 +239,16 @@ void RpcServer::establishConnection(sp<RpcServer>&& server, base::unique_fd clie
		// It must be set before this thread is started
		LOG_ALWAYS_FATAL_IF(server->mShutdownTrigger == nullptr);

		int32_t id;
		status_t status =
		server->mShutdownTrigger->interruptableReadFully(clientFd.get(), &id, sizeof(id));
		RpcConnectionHeader header;
		status_t status = server->mShutdownTrigger->interruptableReadFully(clientFd.get(), &header,
		sizeof(header));
		bool idValid = status == OK;
		if (!idValid) {
		ALOGE("Failed to read ID for client connecting to RPC server: %s",
		statusToString(status).c_str());
		// still need to cleanup before we can return
		}
		bool reverse = header.options & RPC_CONNECTION_OPTION_REVERSE;

		std::thread thisThread;
		sp<RpcSession> session;
		@@ -269,24 +270,37 @@ void RpcServer::establishConnection(sp<RpcServer>&& server, base::unique_fd clie
		return;
		}

		if (id == RPC_SESSION_ID_NEW) {
		if (header.sessionId == RPC_SESSION_ID_NEW) {
		if (reverse) {
		ALOGE("Cannot create a new session with a reverse connection, would leak");
		return;
		}

		LOG_ALWAYS_FATAL_IF(server->mSessionIdCounter >= INT32_MAX, "Out of session IDs");
		server->mSessionIdCounter++;

		session = RpcSession::make();
		session->setForServer(wp<RpcServer>(server), server->mSessionIdCounter,
		server->mShutdownTrigger);
		session->setForServer(server,
		sp<RpcServer::EventListener>::fromExisting(
		static_cast<RpcServer::EventListener*>(server.get())),
		server->mSessionIdCounter, server->mShutdownTrigger);

		server->mSessions[server->mSessionIdCounter] = session;
		} else {
		auto it = server->mSessions.find(id);
		auto it = server->mSessions.find(header.sessionId);
		if (it == server->mSessions.end()) {
		ALOGE("Cannot add thread, no record of session with ID %d", id);
		ALOGE("Cannot add thread, no record of session with ID %d", header.sessionId);
		return;
		}
		session = it->second;
		}

		if (reverse) {
		LOG_ALWAYS_FATAL_IF(!session->addClientConnection(std::move(clientFd)),
		"server state must already be initialized");
		return;
		}

		detachGuard.Disable();
		session->preJoin(std::move(thisThread));
		}
		@@ -294,7 +308,7 @@ void RpcServer::establishConnection(sp<RpcServer>&& server, base::unique_fd clie
		// avoid strong cycle
		server = nullptr;

		session->join(std::move(clientFd));
		RpcSession::join(std::move(session), std::move(clientFd));
		}

		bool RpcServer::setupSocketServer(const RpcSocketAddress& addr) {
		@@ -341,8 +355,7 @@ void RpcServer::onSessionLockedAllServerThreadsEnded(const sp<RpcSession>& sessi
		(void)mSessions.erase(it);
		}

		void RpcServer::onSessionServerThreadEnded(const sp<RpcSession>& session) {
		(void)session;
		void RpcServer::onSessionServerThreadEnded() {
		mShutdownCv.notify_all();
		}

libs/binder/RpcSession.cpp

+119 −24

Original line number	Diff line number	Diff line
		@@ -59,6 +59,17 @@ sp<RpcSession> RpcSession::make() {
		return sp<RpcSession>::make();
		}

		void RpcSession::setMaxReverseConnections(size_t connections) {
		{
		std::lock_guard<std::mutex> _l(mMutex);
		LOG_ALWAYS_FATAL_IF(mClientConnections.size() != 0,
		"Must setup reverse connections before setting up client connections, "
		"but already has %zu clients",
		mClientConnections.size());
		}
		mMaxReverseConnections = connections;
		}

		bool RpcSession::setupUnixDomainClient(const char* path) {
		return setupSocketClient(UnixSocketAddress(path));
		}
		@@ -99,6 +110,20 @@ status_t RpcSession::getRemoteMaxThreads(size_t* maxThreads) {
		return state()->getMaxThreads(connection.fd(), sp<RpcSession>::fromExisting(this), maxThreads);
		}

		bool RpcSession::shutdown() {
		std::unique_lock<std::mutex> _l(mMutex);
		LOG_ALWAYS_FATAL_IF(mForServer.promote() != nullptr, "Can only shut down client session");
		LOG_ALWAYS_FATAL_IF(mShutdownTrigger == nullptr, "Shutdown trigger not installed");
		LOG_ALWAYS_FATAL_IF(mShutdownListener == nullptr, "Shutdown listener not installed");

		mShutdownTrigger->trigger();
		mShutdownListener->waitForShutdown(_l);
		mState->terminate();

		LOG_ALWAYS_FATAL_IF(!mThreads.empty(), "Shutdown failed");
		return true;
		}

		status_t RpcSession::transact(const sp<IBinder>& binder, uint32_t code, const Parcel& data,
		Parcel* reply, uint32_t flags) {
		ExclusiveConnection connection(sp<RpcSession>::fromExisting(this),
		@@ -179,6 +204,24 @@ status_t RpcSession::readId() {
		return OK;
		}

		void RpcSession::WaitForShutdownListener::onSessionLockedAllServerThreadsEnded(
		const sp<RpcSession>& session) {
		(void)session;
		mShutdown = true;
		}

		void RpcSession::WaitForShutdownListener::onSessionServerThreadEnded() {
		mCv.notify_all();
		}

		void RpcSession::WaitForShutdownListener::waitForShutdown(std::unique_lock<std::mutex>& lock) {
		while (!mShutdown) {
		if (std::cv_status::timeout == mCv.wait_for(lock, std::chrono::seconds(1))) {
		ALOGE("Waiting for RpcSession to shut down (1s w/o progress).");
		}
		}
		}

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

		@@ -188,14 +231,13 @@ void RpcSession::preJoin(std::thread thread) {
		}
		}

		void RpcSession::join(unique_fd client) {
		void RpcSession::join(sp<RpcSession>&& session, 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));
		sp<RpcConnection> connection = session->assignServerToThisThread(std::move(client));

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

		if (error != OK) {
		LOG_RPC_DETAIL("Binder connection thread closing w/ status %s",
		@@ -204,22 +246,24 @@ void RpcSession::join(unique_fd client) {
		}
		}

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

		sp<RpcServer> server;
		sp<RpcSession::EventListener> listener;
		{
		std::lock_guard<std::mutex> _l(mMutex);
		auto it = mThreads.find(std::this_thread::get_id());
		LOG_ALWAYS_FATAL_IF(it == mThreads.end());
		std::lock_guard<std::mutex> _l(session->mMutex);
		auto it = session->mThreads.find(std::this_thread::get_id());
		LOG_ALWAYS_FATAL_IF(it == session->mThreads.end());
		it->second.detach();
		mThreads.erase(it);
		session->mThreads.erase(it);

		server = mForServer.promote();
		listener = session->mEventListener.promote();
		}

		if (server != nullptr) {
		server->onSessionServerThreadEnded(sp<RpcSession>::fromExisting(this));
		session = nullptr;

		if (listener != nullptr) {
		listener->onSessionServerThreadEnded();
		}
		}

		@@ -235,7 +279,7 @@ bool RpcSession::setupSocketClient(const RpcSocketAddress& addr) {
		mClientConnections.size());
		}

		if (!setupOneSocketClient(addr, RPC_SESSION_ID_NEW)) return false;
		if (!setupOneSocketConnection(addr, RPC_SESSION_ID_NEW, false /reverse/)) return false;

		// TODO(b/185167543): we should add additional sessions dynamically
		// instead of all at once.
		@@ -256,13 +300,23 @@ bool RpcSession::setupSocketClient(const RpcSocketAddress& addr) {
		// 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;
		if (!setupOneSocketConnection(addr, mId.value(), false /reverse/)) return false;
		}

		// TODO(b/185167543): we should add additional sessions dynamically
		// instead of all at once - the other side should be responsible for setting
		// up additional connections. We need to create at least one (unless 0 are
		// requested to be set) in order to allow the other side to reliably make
		// any requests at all.

		for (size_t i = 0; i < mMaxReverseConnections; i++) {
		if (!setupOneSocketConnection(addr, mId.value(), true /reverse/)) return false;
		}

		return true;
		}

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

		@@ -286,17 +340,48 @@ bool RpcSession::setupOneSocketClient(const RpcSocketAddress& addr, int32_t id)
		return false;
		}

		if (sizeof(id) != TEMP_FAILURE_RETRY(write(serverFd.get(), &id, sizeof(id)))) {
		RpcConnectionHeader header{
		.sessionId = id,
		};
		if (reverse) header.options \|= RPC_CONNECTION_OPTION_REVERSE;

		if (sizeof(header) != TEMP_FAILURE_RETRY(write(serverFd.get(), &header, sizeof(header)))) {
		int savedErrno = errno;
		ALOGE("Could not write id to socket at %s: %s", addr.toString().c_str(),
		ALOGE("Could not write connection header 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());

		if (reverse) {
		std::mutex mutex;
		std::condition_variable joinCv;
		std::unique_lock<std::mutex> lock(mutex);
		std::thread thread;
		sp<RpcSession> thiz = sp<RpcSession>::fromExisting(this);
		bool ownershipTransferred = false;
		thread = std::thread([&]() {
		std::unique_lock<std::mutex> threadLock(mutex);
		unique_fd fd = std::move(serverFd);
		// NOLINTNEXTLINE(performance-unnecessary-copy-initialization)
		sp<RpcSession> session = thiz;
		session->preJoin(std::move(thread));
		ownershipTransferred = true;
		joinCv.notify_one();

		threadLock.unlock();
		// do not use & vars below

		RpcSession::join(std::move(session), std::move(fd));
		});
		joinCv.wait(lock, [&] { return ownershipTransferred; });
		LOG_ALWAYS_FATAL_IF(!ownershipTransferred);
		return true;
		} else {
		return addClientConnection(std::move(serverFd));
		}
		}

		ALOGE("Ran out of retries to connect to %s", addr.toString().c_str());
		return false;
		@@ -305,8 +390,11 @@ bool RpcSession::setupOneSocketClient(const RpcSocketAddress& addr, int32_t id)
		bool RpcSession::addClientConnection(unique_fd fd) {
		std::lock_guard<std::mutex> _l(mMutex);

		// first client connection added, but setForServer not called, so
		// initializaing for a client.
		if (mShutdownTrigger == nullptr) {
		mShutdownTrigger = FdTrigger::make();
		mEventListener = mShutdownListener = sp<WaitForShutdownListener>::make();
		if (mShutdownTrigger == nullptr) return false;
		}

		@@ -316,14 +404,19 @@ bool RpcSession::addClientConnection(unique_fd fd) {
		return true;
		}

		void RpcSession::setForServer(const wp<RpcServer>& server, int32_t sessionId,
		void RpcSession::setForServer(const wp<RpcServer>& server, const wp<EventListener>& eventListener,
		int32_t sessionId,
		const std::shared_ptr<FdTrigger>& shutdownTrigger) {
		LOG_ALWAYS_FATAL_IF(mForServer.unsafe_get() != nullptr);
		LOG_ALWAYS_FATAL_IF(mForServer != nullptr);
		LOG_ALWAYS_FATAL_IF(server == nullptr);
		LOG_ALWAYS_FATAL_IF(mEventListener != nullptr);
		LOG_ALWAYS_FATAL_IF(eventListener == nullptr);
		LOG_ALWAYS_FATAL_IF(mShutdownTrigger != nullptr);
		LOG_ALWAYS_FATAL_IF(shutdownTrigger == nullptr);

		mId = sessionId;
		mForServer = server;
		mEventListener = eventListener;
		mShutdownTrigger = shutdownTrigger;
		}

		@@ -343,9 +436,9 @@ bool RpcSession::removeServerConnection(const sp<RpcConnection>& connection) {
		it != mServerConnections.end()) {
		mServerConnections.erase(it);
		if (mServerConnections.size() == 0) {
		sp<RpcServer> server = mForServer.promote();
		if (server) {
		server->onSessionLockedAllServerThreadsEnded(sp<RpcSession>::fromExisting(this));
		sp<EventListener> listener = mEventListener.promote();
		if (listener) {
		listener->onSessionLockedAllServerThreadsEnded(sp<RpcSession>::fromExisting(this));
		}
		}
		return true;
		@@ -405,6 +498,8 @@ RpcSession::ExclusiveConnection::ExclusiveConnection(const sp<RpcSession>& sessi
		break;
		}

		// TODO(b/185167543): this should return an error, rather than crash a
		// server
		// in regular binder, this would usually be a deadlock :)
		LOG_ALWAYS_FATAL_IF(mSession->mClientConnections.size() == 0,
		"Session has no client connections. This is required for an RPC server "

libs/binder/RpcState.cpp

+1 −0

Original line number	Diff line number	Diff line
		@@ -383,6 +383,7 @@ status_t RpcState::transactAddress(const base::unique_fd& fd, const RpcAddress&
		return status;

		if (flags & IBinder::FLAG_ONEWAY) {
		LOG_RPC_DETAIL("Oneway command, so no longer waiting on %d", fd.get());
		return OK; // do not wait for result
		}

libs/binder/RpcState.h

+1 −1

Original line number	Diff line number	Diff line
		@@ -86,7 +86,6 @@ public:
		size_t countBinders();
		void dump();

		private:
		/**
		* Called when reading or writing data to a session fails to clean up
		* data associated with the session in order to cleanup binders.
		@@ -105,6 +104,7 @@ private:
		*/
		void terminate();

		private:
		// Alternative to std::vector<uint8_t> that doesn't abort on allocation failure and caps
		// large allocations to avoid being requested from allocating too much data.
		struct CommandData {

libs/binder/RpcWireFormat.h

+12 −2

Original line number	Diff line number	Diff line
		@@ -20,6 +20,18 @@ namespace android {
		#pragma clang diagnostic push
		#pragma clang diagnostic error "-Wpadded"

		constexpr int32_t RPC_SESSION_ID_NEW = -1;

		enum : uint8_t {
		RPC_CONNECTION_OPTION_REVERSE = 0x1,
		};

		struct RpcConnectionHeader {
		int32_t sessionId;
		uint8_t options;
		uint8_t reserved[3];
		};

		enum : uint32_t {
		/**
		* follows is RpcWireTransaction, if flags != oneway, reply w/ RPC_COMMAND_REPLY expected
		@@ -51,8 +63,6 @@ enum : uint32_t {
		RPC_SPECIAL_TRANSACT_GET_SESSION_ID = 2,
		};

		constexpr int32_t RPC_SESSION_ID_NEW = -1;

		// serialization is like:
		// \|RpcWireHeader\|struct desginated by 'command'\| (over and over again)