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

Commit 798e0d1a authored by Steven Moreland's avatar Steven Moreland
Browse files

libbinder: send is non-blocking and interruptible 

We actually don't use MSG_DONTBLOCK now, but we only read after
receiving POLLOUT. However, by polling, send also now plugs into the
shutdown triggers. As a side effect, this causes send calls to fail
earlier.

Bug: 186470974
Test: binderRpcTest
Change-Id: Ia13d0dc1749fa7c10340e07e221b4b682cc59cc8
parent e6d7b589

libs/binder/RpcServer.cpp

+2 −1
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)));

libs/binder/RpcSession.cpp

+27 −4
Original line number Diff line number Diff line
@@ -178,9 +178,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) {
@@ -192,10 +194,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);
@@ -204,7 +227,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

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/include/binder/RpcSession.h

+6 −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;

libs/binder/tests/binderRpcTest.cpp

+15 −8
Original line number Diff line number Diff line
@@ -127,11 +127,6 @@ public: 
        out->clear();

        for (auto session : spServer->listSessions()) {

            size_t count = session->state()->countBinders();

            if (count != 1) {

                // this is called when there is only one binder held remaining,

                // so to aid debugging

                session->state()->dump();

            }

            out->push_back(count);

        }

        return Status::ok();
@@ -360,7 +355,11 @@ struct BinderRpcTestProcessSession { 
                EXPECT_EQ(remoteCount, 1);

            }



            EXPECT_OK(rootIface->scheduleShutdown());

            // even though it is on another thread, shutdown races with

            // the transaction reply being written

            if (auto status = rootIface->scheduleShutdown(); !status.isOk()) {

                EXPECT_EQ(DEAD_OBJECT, status.transactionError()) << status;

            }

        }



        rootIface = nullptr;
@@ -968,6 +967,12 @@ TEST_P(BinderRpc, OnewayCallExhaustion) { 
    Status status = iface->sleepMsAsync(kTooLongMs);

    EXPECT_EQ(DEAD_OBJECT, status.transactionError()) << status;



    // now that it has died, wait for the remote session to shutdown

    std::vector<int32_t> remoteCounts;

    do {

        EXPECT_OK(proc.rootIface->countBinders(&remoteCounts));

    } while (remoteCounts.size() == kNumClients);



    // the second session should be shutdown in the other process by the time we

    // are able to join above (it'll only be hung up once it finishes processing

    // any pending commands). We need to erase this session from the record
@@ -1007,7 +1012,9 @@ TEST_P(BinderRpc, Callbacks) { 


                // since we are severing the connection, we need to go ahead and

                // tell the server to shutdown and exit so that waitpid won't hang

                EXPECT_OK(proc.rootIface->scheduleShutdown());

                if (auto status = proc.rootIface->scheduleShutdown(); !status.isOk()) {

                    EXPECT_EQ(DEAD_OBJECT, status.transactionError()) << status;

                }



                // since this session has a reverse connection w/ a threadpool, we

                // need to manually shut it down
@@ -1112,7 +1119,7 @@ static bool testSupportVsockLoopback() { 


    sp<RpcSession> session = RpcSession::make();

    bool okay = session->setupVsockClient(VMADDR_CID_LOCAL, vsockPort);

    CHECK(server->shutdown());

    while (!server->shutdown()) usleep(10000);

    ALOGE("Detected vsock loopback supported: %d", okay);

    return okay;

}