libbinder: finalize connect/server APIs

Before, you needed to manually setup the required number of sockets on
the client and server sides of a connection and manually setup threads.
Now, you configure the thread count on RpcServer and call join once, and
on the client side, you connect once, and the connection figured out how
many connections it will make.

Now, we will be able to manage how these sockets/threads get setup
without affecting any client code in various tests.

So, a server looks like this:

    sp<RpcServer> server = RpcServer::make();
    // still until we are ready to open this up
    server->iUnderstandThisCodeIsExperimentalAndIWillNotUseItInProduction();
    server->setMaxThreads(3 /* for example */);
    // call this for each client (currently this must be setup in
    // advance)
    sp<RpcConnection> connection = server->addClientConnection();
    // other server types are supported
    if (!connection->setupInetServer(1234 /*some port*/)) .. error ..
    // process requests for each client
    server->join();

And a client looks like this:

    sp<RpcConnection> connection = RpcConnection::make();
    if (!connection->setupInetClient(/*some IP address*/, 1234 /*some port*/))
        .. error ..

The above code will create 3 threads on the server serving 3 separate
socket connections that the client can use to make up to 3 simultaneous
sets of syncrhonous calls (this can't be shared because the sockets may
be needed for binder socket calls).

This means that each address (ip + port) in this case can server a single process.

Future considerations:
- if we wanted, we could dynamically setup this connection, so that
  extra threads and sockets are only created as needed. This would be at
  parity with binder, but also it opens up the possibility for later
  errors. TODOs are added in the code for this.
- a single server should be able to share a threadpool between multiple
  clients. Currently a new threadpool is created for each client.
- new client connections should be able to be setup dynamically.
  Currently, once the threadpool is started, we don't support making
  more connections, but we should.

Bug: 185167543
Test: binderRpcTest
Change-Id: I4c11ab64bf7c1c19ca67f6a1c4be21de52358a5c

1 files changed, 44 insertions, 7 deletions

diff --git a/libs/binder/RpcConnection.cpp b/libs/binder/RpcConnection.cpp
index 2502d1bf94..ee5f50868a 100644
--- a/libs/binder/RpcConnection.cpp
+++ b/libs/binder/RpcConnection.cpp
@@ -139,8 +139,8 @@ bool RpcConnection::setupUnixDomainServer(const char* path) {
     return setupSocketServer(UnixSocketAddress(path));
 }
 
-bool RpcConnection::addUnixDomainClient(const char* path) {
-    return addSocketClient(UnixSocketAddress(path));
+bool RpcConnection::setupUnixDomainClient(const char* path) {
+    return setupSocketClient(UnixSocketAddress(path));
 }
 
 #ifdef __BIONIC__
@@ -171,8 +171,8 @@ bool RpcConnection::setupVsockServer(unsigned int port) {
     return setupSocketServer(VsockSocketAddress(kAnyCid, port));
 }
 
-bool RpcConnection::addVsockClient(unsigned int cid, unsigned int port) {
-    return addSocketClient(VsockSocketAddress(cid, port));
+bool RpcConnection::setupVsockClient(unsigned int cid, unsigned int port) {
+    return setupSocketClient(VsockSocketAddress(cid, port));
 }
 
 #endif // __BIONIC__
@@ -240,12 +240,12 @@ bool RpcConnection::setupInetServer(unsigned int port, unsigned int* assignedPor
     return false;
 }
 
-bool RpcConnection::addInetClient(const char* addr, unsigned int port) {
+bool RpcConnection::setupInetClient(const char* addr, unsigned int port) {
     auto aiStart = 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 (addSocketClient(socketAddress)) return true;
+        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;
@@ -268,6 +268,11 @@ sp<IBinder> RpcConnection::getRootObject() {
     return state()->getRootObject(socket.fd(), sp<RpcConnection>::fromExisting(this));
 }
 
+status_t RpcConnection::getMaxThreads(size_t* maxThreads) {
+    ExclusiveSocket socket(sp<RpcConnection>::fromExisting(this), SocketUse::CLIENT);
+    return state()->getMaxThreads(socket.fd(), sp<RpcConnection>::fromExisting(this), maxThreads);
+}
+
 status_t RpcConnection::transact(const RpcAddress& address, uint32_t code, const Parcel& data,
                                  Parcel* reply, uint32_t flags) {
     ExclusiveSocket socket(sp<RpcConnection>::fromExisting(this),
@@ -348,7 +353,39 @@ bool RpcConnection::setupSocketServer(const SocketAddress& addr) {
     return true;
 }
 
-bool RpcConnection::addSocketClient(const SocketAddress& addr) {
+bool RpcConnection::setupSocketClient(const SocketAddress& addr) {
+    {
+        std::lock_guard<std::mutex> _l(mSocketMutex);
+        LOG_ALWAYS_FATAL_IF(mClients.size() != 0,
+                            "Must only setup connection once, but already has %zu clients",
+                            mClients.size());
+    }
+
+    if (!setupOneSocketClient(addr)) return false;
+
+    // TODO(b/185167543): we should add additional connections dynamically
+    // instead of all at once.
+    // TODO(b/186470974): first risk of blocking
+    size_t numThreadsAvailable;
+    if (status_t status = getMaxThreads(&numThreadsAvailable); status != OK) {
+        ALOGE("Could not get max threads after initial connection 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): avoid race w/ accept4 not being called on server
+        for (size_t tries = 0; tries < 5; tries++) {
+            if (setupOneSocketClient(addr)) break;
+            usleep(10000);
+        }
+    }
+
+    return true;
+}
+
+bool RpcConnection::setupOneSocketClient(const SocketAddress& addr) {
     unique_fd serverFd(
             TEMP_FAILURE_RETRY(socket(addr.addr()->sa_family, SOCK_STREAM | SOCK_CLOEXEC, 0)));
     if (serverFd == -1) {