Merge changes Ie1fc2d92,Ie66e92cc,I09a4520a,Ia13d0dc1 am: 1bbd2eef58 am: 0c008e79c1 (7bae0b32) · Commits · e / os / android_frameworks_native

libs/binder/RpcServer.cpp

+6 −5

Original line number	Diff line number	Diff line
		@@ -16,6 +16,7 @@

		#define LOG_TAG "RpcServer"

		#include <poll.h>
		#include <sys/socket.h>
		#include <sys/un.h>

		@@ -152,7 +153,7 @@ void RpcServer::join() {
		}

		status_t status;
		while ((status = mShutdownTrigger->triggerablePollRead(mServer)) == OK) {
		while ((status = mShutdownTrigger->triggerablePoll(mServer, POLLIN)) == OK) {
		unique_fd clientFd(TEMP_FAILURE_RETRY(
		accept4(mServer.get(), nullptr, nullptr /length/, SOCK_CLOEXEC)));

		@@ -250,7 +251,7 @@ void RpcServer::establishConnection(sp<RpcServer>&& server, base::unique_fd clie
		statusToString(status).c_str());
		// still need to cleanup before we can return
		}
		bool reverse = header.options & RPC_CONNECTION_OPTION_REVERSE;
		bool incoming = header.options & RPC_CONNECTION_OPTION_INCOMING;

		std::thread thisThread;
		sp<RpcSession> session;
		@@ -275,8 +276,8 @@ void RpcServer::establishConnection(sp<RpcServer>&& server, base::unique_fd clie
		RpcAddress sessionId = RpcAddress::fromRawEmbedded(&header.sessionId);

		if (sessionId.isZero()) {
		if (reverse) {
		ALOGE("Cannot create a new session with a reverse connection, would leak");
		if (incoming) {
		ALOGE("Cannot create a new session with an incoming connection, would leak");
		return;
		}

		@@ -314,7 +315,7 @@ void RpcServer::establishConnection(sp<RpcServer>&& server, base::unique_fd clie
		session = it->second;
		}

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

libs/binder/RpcSession.cpp

+63 −36

Original line number	Diff line number	Diff line
		@@ -175,9 +175,11 @@ bool RpcSession::FdTrigger::isTriggered() {
		return mWrite == -1;
		}

		status_t RpcSession::FdTrigger::triggerablePollRead(base::borrowed_fd fd) {
		status_t RpcSession::FdTrigger::triggerablePoll(base::borrowed_fd fd, int16_t event) {
		while (true) {
		pollfd pfd[]{{.fd = fd.get(), .events = POLLIN \| POLLHUP, .revents = 0},
		pollfd pfd[]{{.fd = fd.get(),
		.events = static_cast<int16_t>(event \| POLLHUP),
		.revents = 0},
		{.fd = mRead.get(), .events = POLLHUP, .revents = 0}};
		int ret = TEMP_FAILURE_RETRY(poll(pfd, arraysize(pfd), -1));
		if (ret < 0) {
		@@ -189,10 +191,31 @@ status_t RpcSession::FdTrigger::triggerablePollRead(base::borrowed_fd fd) {
		if (pfd[1].revents & POLLHUP) {
		return -ECANCELED;
		}
		return pfd[0].revents & POLLIN ? OK : DEAD_OBJECT;
		return pfd[0].revents & event ? OK : DEAD_OBJECT;
		}
		}

		status_t RpcSession::FdTrigger::interruptableWriteFully(base::borrowed_fd fd, const void* data,
		size_t size) {
		const uint8_t* buffer = reinterpret_cast<const uint8_t*>(data);
		const uint8_t* end = buffer + size;

		MAYBE_WAIT_IN_FLAKE_MODE;

		status_t status;
		while ((status = triggerablePoll(fd, POLLOUT)) == OK) {
		ssize_t writeSize = TEMP_FAILURE_RETRY(send(fd.get(), buffer, end - buffer, MSG_NOSIGNAL));
		if (writeSize == 0) return DEAD_OBJECT;

		if (writeSize < 0) {
		return -errno;
		}
		buffer += writeSize;
		if (buffer == end) return OK;
		}
		return status;
		}

		status_t RpcSession::FdTrigger::interruptableReadFully(base::borrowed_fd fd, void* data,
		size_t size) {
		uint8_t* buffer = reinterpret_cast<uint8_t*>(data);
		@@ -201,7 +224,7 @@ status_t RpcSession::FdTrigger::interruptableReadFully(base::borrowed_fd fd, voi
		MAYBE_WAIT_IN_FLAKE_MODE;

		status_t status;
		while ((status = triggerablePollRead(fd)) == OK) {
		while ((status = triggerablePoll(fd, POLLIN)) == OK) {
		ssize_t readSize = TEMP_FAILURE_RETRY(recv(fd.get(), buffer, end - buffer, MSG_NOSIGNAL));
		if (readSize == 0) return DEAD_OBJECT; // EOF

		@@ -330,7 +353,7 @@ bool RpcSession::setupSocketClient(const RpcSocketAddress& addr) {
		mOutgoingConnections.size());
		}

		if (!setupOneSocketConnection(addr, RpcAddress::zero(), false /reverse/)) return false;
		if (!setupOneSocketConnection(addr, RpcAddress::zero(), false /incoming/)) return false;

		// TODO(b/189955605): we should add additional sessions dynamically
		// instead of all at once.
		@@ -351,7 +374,7 @@ bool RpcSession::setupSocketClient(const RpcSocketAddress& addr) {
		// we've already setup one client
		for (size_t i = 0; i + 1 < numThreadsAvailable; i++) {
		// TODO(b/189955605): shutdown existing connections?
		if (!setupOneSocketConnection(addr, mId.value(), false /reverse/)) return false;
		if (!setupOneSocketConnection(addr, mId.value(), false /incoming/)) return false;
		}

		// TODO(b/189955605): we should add additional sessions dynamically
		@@ -361,14 +384,14 @@ bool RpcSession::setupSocketClient(const RpcSocketAddress& addr) {
		// any requests at all.

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

		return true;
		}

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

		@@ -395,7 +418,7 @@ bool RpcSession::setupOneSocketConnection(const RpcSocketAddress& addr, const Rp
		RpcConnectionHeader header{.options = 0};
		memcpy(&header.sessionId, &id.viewRawEmbedded(), sizeof(RpcWireAddress));

		if (reverse) header.options \|= RPC_CONNECTION_OPTION_REVERSE;
		if (incoming) header.options \|= RPC_CONNECTION_OPTION_INCOMING;

		if (sizeof(header) != TEMP_FAILURE_RETRY(write(serverFd.get(), &header, sizeof(header)))) {
		int savedErrno = errno;
		@@ -406,7 +429,18 @@ bool RpcSession::setupOneSocketConnection(const RpcSocketAddress& addr, const Rp

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

		if (reverse) {
		if (incoming) {
		return addIncomingConnection(std::move(serverFd));
		} else {
		return addOutgoingConnection(std::move(serverFd), true);
		}
		}

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

		bool RpcSession::addIncomingConnection(unique_fd fd) {
		std::mutex mutex;
		std::condition_variable joinCv;
		std::unique_lock<std::mutex> lock(mutex);
		@@ -415,13 +449,13 @@ bool RpcSession::setupOneSocketConnection(const RpcSocketAddress& addr, const Rp
		bool ownershipTransferred = false;
		thread = std::thread([&]() {
		std::unique_lock<std::mutex> threadLock(mutex);
		unique_fd fd = std::move(serverFd);
		unique_fd movedFd = std::move(fd);
		// NOLINTNEXTLINE(performance-unnecessary-copy-initialization)
		sp<RpcSession> session = thiz;
		session->preJoinThreadOwnership(std::move(thread));

		// only continue once we have a response or the connection fails
		auto setupResult = session->preJoinSetup(std::move(fd));
		auto setupResult = session->preJoinSetup(std::move(movedFd));

		ownershipTransferred = true;
		threadLock.unlock();
		@@ -433,13 +467,6 @@ bool RpcSession::setupOneSocketConnection(const RpcSocketAddress& addr, const Rp
		joinCv.wait(lock, [&] { return ownershipTransferred; });
		LOG_ALWAYS_FATAL_IF(!ownershipTransferred);
		return true;
		} else {
		return addOutgoingConnection(std::move(serverFd), true);
		}
		}

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

		bool RpcSession::addOutgoingConnection(unique_fd fd, bool init) {

libs/binder/RpcState.cpp

+6 −10

Original line number	Diff line number	Diff line
		@@ -275,23 +275,19 @@ status_t RpcState::rpcSend(const sp<RpcSession::RpcConnection>& connection,
		LOG_RPC_DETAIL("Sending %s on fd %d: %s", what, connection->fd.get(),
		hexString(data, size).c_str());

		MAYBE_WAIT_IN_FLAKE_MODE;

		if (size > std::numeric_limits<ssize_t>::max()) {
		ALOGE("Cannot send %s at size %zu (too big)", what, size);
		(void)session->shutdownAndWait(false);
		return BAD_VALUE;
		}

		ssize_t sent = TEMP_FAILURE_RETRY(send(connection->fd.get(), data, size, MSG_NOSIGNAL));

		if (sent < 0 \|\| sent != static_cast<ssize_t>(size)) {
		int savedErrno = errno;
		LOG_RPC_DETAIL("Failed to send %s (sent %zd of %zu bytes) on fd %d, error: %s", what, sent,
		size, connection->fd.get(), strerror(savedErrno));

		if (status_t status = session->mShutdownTrigger->interruptableWriteFully(connection->fd.get(),
		data, size);
		status != OK) {
		LOG_RPC_DETAIL("Failed to write %s (%zu bytes) on fd %d, error: %s", what, size,
		connection->fd.get(), statusToString(status).c_str());
		(void)session->shutdownAndWait(false);
		return -savedErrno;
		return status;
		}

		return OK;

libs/binder/RpcWireFormat.h

+2 −2

Original line number	Diff line number	Diff line
		@@ -21,7 +21,7 @@ namespace android {
		#pragma clang diagnostic error "-Wpadded"

		enum : uint8_t {
		RPC_CONNECTION_OPTION_REVERSE = 0x1,
		RPC_CONNECTION_OPTION_INCOMING = 0x1, // default is outgoing
		};

		constexpr uint64_t RPC_WIRE_ADDRESS_OPTION_CREATED = 1 << 0; // distinguish from '0' address
		@@ -47,7 +47,7 @@ struct RpcConnectionHeader {
		/**
		* Whenever a client connection is setup, this is sent as the initial
		* transaction. The main use of this is in order to control the timing for when
		* a reverse connection is setup.
		* an incoming connection is setup.
		*/
		struct RpcOutgoingConnectionInit {
		char msg[4];

libs/binder/include/binder/RpcSession.h

+7 −3

Original line number	Diff line number	Diff line
		@@ -152,20 +152,23 @@ private:
		/**
		* Poll for a read event.
		*
		* event - for pollfd
		*
		* Return:
		* true - time to read!
		* false - trigger happened
		*/
		status_t triggerablePollRead(base::borrowed_fd fd);
		status_t triggerablePoll(base::borrowed_fd fd, int16_t event);

		/**
		* Read, but allow the read to be interrupted by this trigger.
		* Read (or write), but allow to be interrupted by this trigger.
		*
		* Return:
		* true - read succeeded at 'size'
		* true - succeeded in completely processing 'size'
		* false - interrupted (failure or trigger)
		*/
		status_t interruptableReadFully(base::borrowed_fd fd, void* data, size_t size);
		status_t interruptableWriteFully(base::borrowed_fd fd, const void* data, size_t size);

		private:
		base::unique_fd mWrite;
		@@ -223,6 +226,7 @@ private:
		[[nodiscard]] bool setupSocketClient(const RpcSocketAddress& address);
		[[nodiscard]] bool setupOneSocketConnection(const RpcSocketAddress& address,
		const RpcAddress& sessionId, bool server);
		[[nodiscard]] bool addIncomingConnection(base::unique_fd fd);
		[[nodiscard]] bool addOutgoingConnection(base::unique_fd fd, bool init);
		[[nodiscard]] bool setForServer(const wp<RpcServer>& server,
		const wp<RpcSession::EventListener>& eventListener,