Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit 7c5e6c2d authored by Steven Moreland's avatar Steven Moreland
Browse files

libbinder: RpcConnection: add ID 

In preparation to have the server be able to distinguish clients and
clients to be able to dynamically create threads that are assigned to
them.

Future considerations:
- make ID impossible to guess (right now, one client might be able to
  get ahold of a thread from a server). We may implement something here
  or go for something existing like TLS.
- combine getting max threads and this? will wait until dynamic threads
  are actually implemented and we know we need this ID and we're looking
  at performance. For now this is a placeholder to enable dynamic client
  APIs.

Bug: 185167543
Test: binderRpcTest
Change-Id: If8563c69930c23b9ca91090b4f59ef1f51073f24
parent 611d15fe

libs/binder/RpcConnection.cpp

+29 −4
Original line number Diff line number Diff line
@@ -115,6 +115,24 @@ status_t RpcConnection::sendDecStrong(const RpcAddress& address) { 
    return state()->sendDecStrong(socket.fd(), address);

}



status_t RpcConnection::readId() {

    {

        std::lock_guard<std::mutex> _l(mSocketMutex);

        LOG_ALWAYS_FATAL_IF(mForServer != nullptr, "Can only update ID for client.");

    }



    int32_t id;



    ExclusiveSocket socket(sp<RpcConnection>::fromExisting(this), SocketUse::CLIENT);

    status_t status =

            state()->getConnectionId(socket.fd(), sp<RpcConnection>::fromExisting(this), &id);

    if (status != OK) return status;



    LOG_RPC_DETAIL("RpcConnection %p has id %d", this, id);

    mId = id;

    return OK;

}



void RpcConnection::join(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)
@@ -134,10 +152,6 @@ void RpcConnection::join(unique_fd client) { 
                        "bad state: socket object guaranteed to be in list");

}



void RpcConnection::setForServer(const wp<RpcServer>& server) {

    mForServer = server;

}



wp<RpcServer> RpcConnection::server() {

    return mForServer;

}
@@ -162,6 +176,12 @@ bool RpcConnection::setupSocketClient(const RpcSocketAddress& addr) { 
        return false;

    }



    if (status_t status = readId(); status != OK) {

        ALOGE("Could not get connection id 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
@@ -202,6 +222,11 @@ void RpcConnection::addClient(unique_fd fd) { 
    mClients.push_back(connection);

}



void RpcConnection::setForServer(const wp<RpcServer>& server, int32_t connectionId) {

    mId = connectionId;

    mForServer = server;

}



sp<RpcConnection::ConnectionSocket> RpcConnection::assignServerToThisThread(unique_fd fd) {

    std::lock_guard<std::mutex> _l(mSocketMutex);

    sp<ConnectionSocket> connection = sp<ConnectionSocket>::make();

libs/binder/RpcServer.cpp

+1 −1
Original line number Diff line number Diff line
@@ -128,7 +128,7 @@ void RpcServer::join() { 
        LOG_ALWAYS_FATAL_IF(mServer.get() == -1, "RpcServer must be setup to join.");

        // TODO(b/185167543): support more than one client at once

        mConnection = RpcConnection::make();

        mConnection->setForServer(sp<RpcServer>::fromExisting(this));

        mConnection->setForServer(sp<RpcServer>::fromExisting(this), 42 /*placeholder id*/);



        mStarted = true;

            for (size_t i = 0; i < mMaxThreads; i++) {

libs/binder/RpcState.cpp

+37 −6
Original line number Diff line number Diff line
@@ -249,7 +249,7 @@ sp<IBinder> RpcState::getRootObject(const base::unique_fd& fd, 
}



status_t RpcState::getMaxThreads(const base::unique_fd& fd, const sp<RpcConnection>& connection,

                                 size_t* maxThreads) {

                                 size_t* maxThreadsOut) {

    Parcel data;

    data.markForRpc(connection);

    Parcel reply;
@@ -261,15 +261,36 @@ status_t RpcState::getMaxThreads(const base::unique_fd& fd, const sp<RpcConnecti 
        return status;

    }



    int32_t threads;

    status = reply.readInt32(&threads);

    int32_t maxThreads;

    status = reply.readInt32(&maxThreads);

    if (status != OK) return status;

    if (threads <= 0) {

        ALOGE("Error invalid max threads: %d", threads);

    if (maxThreads <= 0) {

        ALOGE("Error invalid max maxThreads: %d", maxThreads);

        return BAD_VALUE;

    }



    *maxThreads = threads;

    *maxThreadsOut = maxThreads;

    return OK;

}



status_t RpcState::getConnectionId(const base::unique_fd& fd, const sp<RpcConnection>& connection,

                                   int32_t* connectionIdOut) {

    Parcel data;

    data.markForRpc(connection);

    Parcel reply;



    status_t status = transact(fd, RpcAddress::zero(), RPC_SPECIAL_TRANSACT_GET_CONNECTION_ID, data,

                               connection, &reply, 0);

    if (status != OK) {

        ALOGE("Error getting connection ID: %s", statusToString(status).c_str());

        return status;

    }



    int32_t connectionId;

    status = reply.readInt32(&connectionId);

    if (status != OK) return status;



    *connectionIdOut = connectionId;

    return OK;

}
@@ -554,6 +575,16 @@ status_t RpcState::processTransactInternal(const base::unique_fd& fd, 
                        replyStatus = reply.writeInt32(server->getMaxThreads());

                        break;

                    }

                    case RPC_SPECIAL_TRANSACT_GET_CONNECTION_ID: {

                        // only connections w/ services can be the source of a

                        // connection ID (so still guarded by non-null server)

                        //

                        // connections associated with servers must have an ID

                        // (hence abort)

                        int32_t id = connection->getPrivateAccessorForId().get().value();

                        replyStatus = reply.writeInt32(id);

                        break;

                    }

                    default: {

                        replyStatus = UNKNOWN_TRANSACTION;

                    }

libs/binder/RpcState.h

+3 −0
Original line number Diff line number Diff line
@@ -50,9 +50,12 @@ public: 
    RpcState();

    ~RpcState();



    // TODO(b/182940634): combine some special transactions into one "getServerInfo" call?

    sp<IBinder> getRootObject(const base::unique_fd& fd, const sp<RpcConnection>& connection);

    status_t getMaxThreads(const base::unique_fd& fd, const sp<RpcConnection>& connection,

                           size_t* maxThreadsOut);

    status_t getConnectionId(const base::unique_fd& fd, const sp<RpcConnection>& connection,

                             int32_t* connectionIdOut);



    [[nodiscard]] status_t transact(const base::unique_fd& fd, const RpcAddress& address,

                                    uint32_t code, const Parcel& data,

libs/binder/RpcWireFormat.h

+1 −0
Original line number Diff line number Diff line
@@ -48,6 +48,7 @@ enum : uint32_t { 
enum : uint32_t {

    RPC_SPECIAL_TRANSACT_GET_ROOT = 0,

    RPC_SPECIAL_TRANSACT_GET_MAX_THREADS = 1,

    RPC_SPECIAL_TRANSACT_GET_CONNECTION_ID = 2,

};



// serialization is like: