libs/utils/PollLoop.cpp

//
// Copyright 2010 The Android Open Source Project
//
// A select loop implementation.
//
#define LOG_TAG "PollLoop"

//#define LOG_NDEBUG 0

// Debugs poll and wake interactions.
#define DEBUG_POLL_AND_WAKE 0

// Debugs callback registration and invocation.
#define DEBUG_CALLBACKS 0

#include <cutils/log.h>
#include <utils/PollLoop.h>

#include <unistd.h>
#include <fcntl.h>

namespace android {

static pthread_mutex_t gTLSMutex = PTHREAD_MUTEX_INITIALIZER;
static bool gHaveTLS = false;
static pthread_key_t gTLS = 0;

PollLoop::PollLoop(bool allowNonCallbacks) :
        mAllowNonCallbacks(allowNonCallbacks), mPolling(false),
        mWaiters(0), mPendingFdsPos(0) {
    openWakePipe();
}

PollLoop::~PollLoop() {
    closeWakePipe();
}

void PollLoop::threadDestructor(void *st) {
    PollLoop* const self = static_cast<PollLoop*>(st);
    if (self != NULL) {
        self->decStrong((void*)threadDestructor);
    }
}

void PollLoop::setForThread(const sp<PollLoop>& pollLoop) {
    sp<PollLoop> old = getForThread();
    
    if (pollLoop != NULL) {
        pollLoop->incStrong((void*)threadDestructor);
    }
    
    pthread_setspecific(gTLS, pollLoop.get());
    
    if (old != NULL) {
        old->decStrong((void*)threadDestructor);
    }
}
    
sp<PollLoop> PollLoop::getForThread() {
    if (!gHaveTLS) {
        pthread_mutex_lock(&gTLSMutex);
        if (pthread_key_create(&gTLS, threadDestructor) != 0) {
            pthread_mutex_unlock(&gTLSMutex);
            return NULL;
        }
        gHaveTLS = true;
        pthread_mutex_unlock(&gTLSMutex);
    }
    
    return (PollLoop*)pthread_getspecific(gTLS);
}

void PollLoop::openWakePipe() {
    int wakeFds[2];
    int result = pipe(wakeFds);
    LOG_ALWAYS_FATAL_IF(result != 0, "Could not create wake pipe.  errno=%d", errno);

    mWakeReadPipeFd = wakeFds[0];
    mWakeWritePipeFd = wakeFds[1];

    result = fcntl(mWakeReadPipeFd, F_SETFL, O_NONBLOCK);
    LOG_ALWAYS_FATAL_IF(result != 0, "Could not make wake read pipe non-blocking.  errno=%d",
            errno);

    result = fcntl(mWakeWritePipeFd, F_SETFL, O_NONBLOCK);
    LOG_ALWAYS_FATAL_IF(result != 0, "Could not make wake write pipe non-blocking.  errno=%d",
            errno);

    // Add the wake pipe to the head of the request list with a null callback.
    struct pollfd requestedFd;
    requestedFd.fd = mWakeReadPipeFd;
    requestedFd.events = POLLIN;
    mRequestedFds.insertAt(requestedFd, 0);

    RequestedCallback requestedCallback;
    requestedCallback.callback = NULL;
    requestedCallback.looperCallback = NULL;
    requestedCallback.data = NULL;
    mRequestedCallbacks.insertAt(requestedCallback, 0);
}

void PollLoop::closeWakePipe() {
    close(mWakeReadPipeFd);
    close(mWakeWritePipeFd);

    // Note: We don't need to remove the poll structure or callback entry because this
    //       method is currently only called by the destructor.
}

int32_t PollLoop::pollOnce(int timeoutMillis, int* outEvents, void** outData) {
    // If there are still pending fds from the last call, dispatch those
    // first, to avoid an earlier fd from starving later ones.
    const size_t pendingFdsCount = mPendingFds.size();
    if (mPendingFdsPos < pendingFdsCount) {
        const PendingCallback& pending = mPendingFds.itemAt(mPendingFdsPos);
        mPendingFdsPos++;
        if (outEvents != NULL) *outEvents = pending.events;
        if (outData != NULL) *outData = pending.data;
        return pending.fd;
    }
    
    mLock.lock();
    while (mWaiters != 0) {
        mResume.wait(mLock);
    }
    mPolling = true;
    mLock.unlock();

    int32_t result;
    size_t requestedCount = mRequestedFds.size();

#if DEBUG_POLL_AND_WAKE
    LOGD("%p ~ pollOnce - waiting on %d fds", this, requestedCount);
    for (size_t i = 0; i < requestedCount; i++) {
        LOGD("  fd %d - events %d", mRequestedFds[i].fd, mRequestedFds[i].events);
    }
#endif

    int respondedCount = poll(mRequestedFds.editArray(), requestedCount, timeoutMillis);

    if (respondedCount == 0) {
        // Timeout
#if DEBUG_POLL_AND_WAKE
        LOGD("%p ~ pollOnce - timeout", this);
#endif
        result = POLL_TIMEOUT;
        goto Done;
    }

    if (respondedCount < 0) {
        // Error
#if DEBUG_POLL_AND_WAKE
        LOGD("%p ~ pollOnce - error, errno=%d", this, errno);
#endif
        if (errno != EINTR) {
            LOGW("Poll failed with an unexpected error, errno=%d", errno);
        }
        result = POLL_ERROR;
        goto Done;
    }

#if DEBUG_POLL_AND_WAKE
    LOGD("%p ~ pollOnce - handling responses from %d fds", this, respondedCount);
    for (size_t i = 0; i < requestedCount; i++) {
        LOGD("  fd %d - events %d, revents %d", mRequestedFds[i].fd, mRequestedFds[i].events,
                mRequestedFds[i].revents);
    }
#endif

    mPendingCallbacks.clear();
    mPendingFds.clear();
    mPendingFdsPos = 0;
    if (outEvents != NULL) *outEvents = 0;
    if (outData != NULL) *outData = NULL;
    
    result = POLL_CALLBACK;
    for (size_t i = 0; i < requestedCount; i++) {
        const struct pollfd& requestedFd = mRequestedFds.itemAt(i);

        short revents = requestedFd.revents;
        if (revents) {
            const RequestedCallback& requestedCallback = mRequestedCallbacks.itemAt(i);
            PendingCallback pending;
            pending.fd = requestedFd.fd;
            pending.events = revents;
            pending.callback = requestedCallback.callback;
            pending.looperCallback = requestedCallback.looperCallback;
            pending.data = requestedCallback.data;

            if (pending.callback || pending.looperCallback) {
                mPendingCallbacks.push(pending);
            } else if (pending.fd != mWakeReadPipeFd) {
                if (result == POLL_CALLBACK) {
                    result = pending.fd;
                    if (outEvents != NULL) *outEvents = pending.events;
                    if (outData != NULL) *outData = pending.data;
                } else {
                    mPendingFds.push(pending);
                }
            } else {
#if DEBUG_POLL_AND_WAKE
                LOGD("%p ~ pollOnce - awoken", this);
#endif
                char buffer[16];
                ssize_t nRead;
                do {
                    nRead = read(mWakeReadPipeFd, buffer, sizeof(buffer));
                } while (nRead == sizeof(buffer));
            }

            respondedCount -= 1;
            if (respondedCount == 0) {
                break;
            }
        }
    }

Done:
    mLock.lock();
    mPolling = false;
    if (mWaiters != 0) {
        mAwake.broadcast();
    }
    mLock.unlock();

    if (result == POLL_CALLBACK || result >= 0) {
        size_t pendingCount = mPendingCallbacks.size();
        for (size_t i = 0; i < pendingCount; i++) {
            const PendingCallback& pendingCallback = mPendingCallbacks.itemAt(i);
#if DEBUG_POLL_AND_WAKE || DEBUG_CALLBACKS
            LOGD("%p ~ pollOnce - invoking callback for fd %d", this, pendingCallback.fd);
#endif

            bool keep = true;
            if (pendingCallback.callback != NULL) {
                keep = pendingCallback.callback(pendingCallback.fd, pendingCallback.events,
                        pendingCallback.data);
            } else {
                keep = pendingCallback.looperCallback(pendingCallback.fd, pendingCallback.events,
                        pendingCallback.data) != 0;
            }
            if (! keep) {
                removeCallback(pendingCallback.fd);
            }
        }
    }

#if DEBUG_POLL_AND_WAKE
    LOGD("%p ~ pollOnce - done", this);
#endif
    return result;
}

void PollLoop::wake() {
#if DEBUG_POLL_AND_WAKE
    LOGD("%p ~ wake", this);
#endif

    ssize_t nWrite = write(mWakeWritePipeFd, "W", 1);
    if (nWrite != 1) {
        if (errno != EAGAIN) {
            LOGW("Could not write wake signal, errno=%d", errno);
        }
    }
}

bool PollLoop::getAllowNonCallbacks() const {
    return mAllowNonCallbacks;
}

void PollLoop::setCallback(int fd, int events, Callback callback, void* data) {
    setCallbackCommon(fd, events, callback, NULL, data);
}

void PollLoop::setLooperCallback(int fd, int events, ALooper_callbackFunc* callback,
        void* data) {
    setCallbackCommon(fd, events, NULL, callback, data);
}

void PollLoop::setCallbackCommon(int fd, int events, Callback callback,
        ALooper_callbackFunc* looperCallback, void* data) {

#if DEBUG_CALLBACKS
    LOGD("%p ~ setCallback - fd=%d, events=%d", this, fd, events);
#endif

    if (! events) {
        LOGE("Invalid attempt to set a callback with no selected poll events.");
        removeCallback(fd);
        return;
    }

    if (! callback && ! looperCallback && ! mAllowNonCallbacks) {
        LOGE("Invalid attempt to set NULL callback but not allowed.");
        removeCallback(fd);
        return;
    }
    
    wakeAndLock();

    struct pollfd requestedFd;
    requestedFd.fd = fd;
    requestedFd.events = events;

    RequestedCallback requestedCallback;
    requestedCallback.callback = callback;
    requestedCallback.looperCallback = looperCallback;
    requestedCallback.data = data;

    ssize_t index = getRequestIndexLocked(fd);
    if (index < 0) {
        mRequestedFds.push(requestedFd);
        mRequestedCallbacks.push(requestedCallback);
    } else {
        mRequestedFds.replaceAt(requestedFd, size_t(index));
        mRequestedCallbacks.replaceAt(requestedCallback, size_t(index));
    }

    mLock.unlock();
}

bool PollLoop::removeCallback(int fd) {
#if DEBUG_CALLBACKS
    LOGD("%p ~ removeCallback - fd=%d", this, fd);
#endif

    wakeAndLock();

    ssize_t index = getRequestIndexLocked(fd);
    if (index >= 0) {
        mRequestedFds.removeAt(size_t(index));
        mRequestedCallbacks.removeAt(size_t(index));
    }

    mLock.unlock();
    return index >= 0;
}

ssize_t PollLoop::getRequestIndexLocked(int fd) {
    size_t requestCount = mRequestedFds.size();

    for (size_t i = 0; i < requestCount; i++) {
        if (mRequestedFds.itemAt(i).fd == fd) {
            return i;
        }
    }

    return -1;
}

void PollLoop::wakeAndLock() {
    mLock.lock();
    mWaiters += 1;
    while (mPolling) {
        wake();
        mAwake.wait(mLock);
    }
    mWaiters -= 1;
    if (mWaiters == 0) {
        mResume.signal();
    }
}

} // namespace android