libbinder: RPC more symmetrical max threads

Now, RpcServer and RpcSession both keep track of their max threads using
the same variable, and the server can therefore request the number of
reverse connections possible.

Bug: 185167543
Test: N/A
Change-Id: Ieaff69c8c2da2faf7598aed7e862601a1fcd7a00

5 files changed, 47 insertions, 42 deletions

diff --git a/libs/binder/RpcServer.cpp b/libs/binder/RpcServer.cpp
index b146bb063e..5ffee0600f 100644
--- a/libs/binder/RpcServer.cpp
+++ b/libs/binder/RpcServer.cpp
@@ -280,6 +280,7 @@ void RpcServer::establishConnection(sp<RpcServer>&& server, base::unique_fd clie
             server->mSessionIdCounter++;
 
             session = RpcSession::make();
+            session->setMaxThreads(server->mMaxThreads);
             session->setForServer(server,
                                   sp<RpcServer::EventListener>::fromExisting(
                                           static_cast<RpcServer::EventListener*>(server.get())),
diff --git a/libs/binder/RpcSession.cpp b/libs/binder/RpcSession.cpp
index a3efa56e3b..a27dff5c8f 100644
--- a/libs/binder/RpcSession.cpp
+++ b/libs/binder/RpcSession.cpp
@@ -59,15 +59,18 @@ 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;
+void RpcSession::setMaxThreads(size_t threads) {
+    std::lock_guard<std::mutex> _l(mMutex);
+    LOG_ALWAYS_FATAL_IF(!mClientConnections.empty() || !mServerConnections.empty(),
+                        "Must set max threads before setting up connections, but has %zu client(s) "
+                        "and %zu server(s)",
+                        mClientConnections.size(), mServerConnections.size());
+    mMaxThreads = threads;
+}
+
+size_t RpcSession::getMaxThreads() {
+    std::lock_guard<std::mutex> _l(mMutex);
+    return mMaxThreads;
 }
 
 bool RpcSession::setupUnixDomainClient(const char* path) {
@@ -309,7 +312,7 @@ bool RpcSession::setupSocketClient(const RpcSocketAddress& addr) {
     // 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++) {
+    for (size_t i = 0; i < mMaxThreads; i++) {
         if (!setupOneSocketConnection(addr, mId.value(), true /*reverse*/)) return false;
     }
 
diff --git a/libs/binder/RpcState.cpp b/libs/binder/RpcState.cpp
index e18179e4a8..76df97069c 100644
--- a/libs/binder/RpcState.cpp
+++ b/libs/binder/RpcState.cpp
@@ -625,34 +625,34 @@ status_t RpcState::processTransactInternal(const base::unique_fd& fd, const sp<R
         } else {
             LOG_RPC_DETAIL("Got special transaction %u", transaction->code);
 
-            sp<RpcServer> server = session->server().promote();
-            if (server) {
-                // special case for 'zero' address (special server commands)
-                switch (transaction->code) {
-                    case RPC_SPECIAL_TRANSACT_GET_ROOT: {
-                        replyStatus = reply.writeStrongBinder(server->getRootObject());
-                        break;
-                    }
-                    case RPC_SPECIAL_TRANSACT_GET_MAX_THREADS: {
-                        replyStatus = reply.writeInt32(server->getMaxThreads());
-                        break;
-                    }
-                    case RPC_SPECIAL_TRANSACT_GET_SESSION_ID: {
-                        // only sessions w/ services can be the source of a
-                        // session ID (so still guarded by non-null server)
-                        //
-                        // sessions associated with servers must have an ID
-                        // (hence abort)
-                        int32_t id = session->mId.value();
-                        replyStatus = reply.writeInt32(id);
-                        break;
-                    }
-                    default: {
-                        replyStatus = UNKNOWN_TRANSACTION;
+            switch (transaction->code) {
+                case RPC_SPECIAL_TRANSACT_GET_MAX_THREADS: {
+                    replyStatus = reply.writeInt32(session->getMaxThreads());
+                    break;
+                }
+                case RPC_SPECIAL_TRANSACT_GET_SESSION_ID: {
+                    // for client connections, this should always report the value
+                    // originally returned from the server
+                    int32_t id = session->mId.value();
+                    replyStatus = reply.writeInt32(id);
+                    break;
+                }
+                default: {
+                    sp<RpcServer> server = session->server().promote();
+                    if (server) {
+                        switch (transaction->code) {
+                            case RPC_SPECIAL_TRANSACT_GET_ROOT: {
+                                replyStatus = reply.writeStrongBinder(server->getRootObject());
+                                break;
+                            }
+                            default: {
+                                replyStatus = UNKNOWN_TRANSACTION;
+                            }
+                        }
+                    } else {
+                        ALOGE("Special command sent, but no server object attached.");
                     }
                 }
-            } else {
-                ALOGE("Special command sent, but no server object attached.");
             }
         }
     }
diff --git a/libs/binder/include/binder/RpcSession.h b/libs/binder/include/binder/RpcSession.h
index 9d314e4ac9..a6bc1a9cc7 100644
--- a/libs/binder/include/binder/RpcSession.h
+++ b/libs/binder/include/binder/RpcSession.h
@@ -47,16 +47,17 @@ public:
     static sp<RpcSession> make();
 
     /**
-     * Set the maximum number of reverse connections allowed to be made (for
-     * things like callbacks). By default, this is 0. This must be called before
-     * setting up this connection as a client.
+     * Set the maximum number of threads allowed to be made (for things like callbacks).
+     * By default, this is 0. This must be called before setting up this connection as a client.
+     * Server sessions will inherits this value from RpcServer.
      *
      * If this is called, 'shutdown' on this session must also be called.
      * Otherwise, a threadpool will leak.
      *
      * TODO(b/185167543): start these dynamically
      */
-    void setMaxReverseConnections(size_t connections);
+    void setMaxThreads(size_t threads);
+    size_t getMaxThreads();
 
     /**
      * This should be called once per thread, matching 'join' in the remote
@@ -257,7 +258,7 @@ private:
 
     std::mutex mMutex; // for all below
 
-    size_t mMaxReverseConnections = 0;
+    size_t mMaxThreads = 0;
 
     std::condition_variable mAvailableConnectionCv; // for mWaitingThreads
     size_t mWaitingThreads = 0;
diff --git a/libs/binder/tests/binderRpcTest.cpp b/libs/binder/tests/binderRpcTest.cpp
index 80708df1df..e48b7c54a5 100644
--- a/libs/binder/tests/binderRpcTest.cpp
+++ b/libs/binder/tests/binderRpcTest.cpp
@@ -446,7 +446,7 @@ public:
 
         for (size_t i = 0; i < numSessions; i++) {
             sp<RpcSession> session = RpcSession::make();
-            session->setMaxReverseConnections(numReverseConnections);
+            session->setMaxThreads(numReverseConnections);
 
             switch (socketType) {
                 case SocketType::UNIX: