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
diff --git a/libs/binder/RpcConnection.cpp b/libs/binder/RpcConnection.cpp
index 2502d1b..ee5f508 100644
--- a/libs/binder/RpcConnection.cpp
+++ b/libs/binder/RpcConnection.cpp
@@ -139,8 +139,8 @@
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 @@
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 @@
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 @@
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 @@
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) {