Switch Looper back to using poll() instead of epoll().
Added a couple of micro-optimizations to avoid calling wake() unnecessarily
and reduce JNI overhead slightly.
Fixed a minor issue where we were not clearing the "next" field of Messages
returned by the MessageQueue so the Message would hold on to its successor
and potentially prevent the GC from collecting it if the message were leaked
somehow.
Change-Id: I488d29417ce0cdd7d0e447cda76ec978ef7f811c
diff --git a/include/utils/Looper.h b/include/utils/Looper.h
index 3f00b78..cc51490 100644
--- a/include/utils/Looper.h
+++ b/include/utils/Looper.h
@@ -20,9 +20,22 @@
#include <utils/threads.h>
#include <utils/RefBase.h>
#include <utils/KeyedVector.h>
+#include <utils/Timers.h>
#include <android/looper.h>
+// Currently using poll() instead of epoll_wait() since it does a better job of meeting a
+// timeout deadline. epoll_wait() typically causes additional delays of up to 10ms
+// beyond the requested timeout.
+//#define LOOPER_USES_EPOLL
+//#define LOOPER_STATISTICS
+
+#ifdef LOOPER_USES_EPOLL
+#include <sys/epoll.h>
+#else
+#include <sys/poll.h>
+#endif
+
/*
* Declare a concrete type for the NDK's looper forward declaration.
*/
@@ -190,13 +203,54 @@
const bool mAllowNonCallbacks; // immutable
- int mEpollFd; // immutable
int mWakeReadPipeFd; // immutable
int mWakeWritePipeFd; // immutable
+ Mutex mLock;
+
+#ifdef LOOPER_USES_EPOLL
+ int mEpollFd; // immutable
// Locked list of file descriptor monitoring requests.
- Mutex mLock;
- KeyedVector<int, Request> mRequests;
+ KeyedVector<int, Request> mRequests; // guarded by mLock
+#else
+ // The lock guards state used to track whether there is a poll() in progress and whether
+ // there are any other threads waiting in wakeAndLock(). The condition variables
+ // are used to transfer control among these threads such that all waiters are
+ // serviced before a new poll can begin.
+ // The wakeAndLock() method increments mWaiters, wakes the poll, blocks on mAwake
+ // until mPolling becomes false, then decrements mWaiters again.
+ // The poll() method blocks on mResume until mWaiters becomes 0, then sets
+ // mPolling to true, blocks until the poll completes, then resets mPolling to false
+ // and signals mResume if there are waiters.
+ bool mPolling; // guarded by mLock
+ uint32_t mWaiters; // guarded by mLock
+ Condition mAwake; // guarded by mLock
+ Condition mResume; // guarded by mLock
+
+ Vector<struct pollfd> mRequestedFds; // must hold mLock and mPolling must be false to modify
+ Vector<Request> mRequests; // must hold mLock and mPolling must be false to modify
+
+ ssize_t getRequestIndexLocked(int fd);
+ void wakeAndLock();
+#endif
+
+#ifdef LOOPER_STATISTICS
+ static const int SAMPLED_WAKE_CYCLES_TO_AGGREGATE = 100;
+ static const int SAMPLED_POLLS_TO_AGGREGATE = 1000;
+
+ nsecs_t mPendingWakeTime;
+ int mPendingWakeCount;
+
+ int mSampledWakeCycles;
+ int mSampledWakeCountSum;
+ nsecs_t mSampledWakeLatencySum;
+
+ int mSampledPolls;
+ int mSampledZeroPollCount;
+ int mSampledZeroPollLatencySum;
+ int mSampledTimeoutPollCount;
+ int mSampledTimeoutPollLatencySum;
+#endif
// This state is only used privately by pollOnce and does not require a lock since
// it runs on a single thread.
@@ -204,6 +258,8 @@
size_t mResponseIndex;
int pollInner(int timeoutMillis);
+ void awoken();
+ void pushResponse(int events, const Request& request);
static void initTLSKey();
static void threadDestructor(void *st);
diff --git a/libs/utils/Looper.cpp b/libs/utils/Looper.cpp
index d2dd6eb..a5363d6 100644
--- a/libs/utils/Looper.cpp
+++ b/libs/utils/Looper.cpp
@@ -19,16 +19,17 @@
#include <unistd.h>
#include <fcntl.h>
-#include <sys/epoll.h>
namespace android {
+#ifdef LOOPER_USES_EPOLL
// Hint for number of file descriptors to be associated with the epoll instance.
static const int EPOLL_SIZE_HINT = 8;
// Maximum number of file descriptors for which to retrieve poll events each iteration.
static const int EPOLL_MAX_EVENTS = 16;
+#endif
static pthread_once_t gTLSOnce = PTHREAD_ONCE_INIT;
static pthread_key_t gTLSKey = 0;
@@ -36,9 +37,6 @@
Looper::Looper(bool allowNonCallbacks) :
mAllowNonCallbacks(allowNonCallbacks),
mResponseIndex(0) {
- mEpollFd = epoll_create(EPOLL_SIZE_HINT);
- LOG_ALWAYS_FATAL_IF(mEpollFd < 0, "Could not create epoll instance. errno=%d", errno);
-
int wakeFds[2];
int result = pipe(wakeFds);
LOG_ALWAYS_FATAL_IF(result != 0, "Could not create wake pipe. errno=%d", errno);
@@ -54,6 +52,11 @@
LOG_ALWAYS_FATAL_IF(result != 0, "Could not make wake write pipe non-blocking. errno=%d",
errno);
+#ifdef LOOPER_USES_EPOLL
+ // Allocate the epoll instance and register the wake pipe.
+ mEpollFd = epoll_create(EPOLL_SIZE_HINT);
+ LOG_ALWAYS_FATAL_IF(mEpollFd < 0, "Could not create epoll instance. errno=%d", errno);
+
struct epoll_event eventItem;
memset(& eventItem, 0, sizeof(epoll_event)); // zero out unused members of data field union
eventItem.events = EPOLLIN;
@@ -61,12 +64,45 @@
result = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, mWakeReadPipeFd, & eventItem);
LOG_ALWAYS_FATAL_IF(result != 0, "Could not add wake read pipe to epoll instance. errno=%d",
errno);
+#else
+ // 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.push(requestedFd);
+
+ Request request;
+ request.fd = mWakeReadPipeFd;
+ request.callback = NULL;
+ request.ident = 0;
+ request.data = NULL;
+ mRequests.push(request);
+
+ mPolling = false;
+ mWaiters = 0;
+#endif
+
+#ifdef LOOPER_STATISTICS
+ mPendingWakeTime = -1;
+ mPendingWakeCount = 0;
+ mSampledWakeCycles = 0;
+ mSampledWakeCountSum = 0;
+ mSampledWakeLatencySum = 0;
+
+ mSampledPolls = 0;
+ mSampledZeroPollCount = 0;
+ mSampledZeroPollLatencySum = 0;
+ mSampledTimeoutPollCount = 0;
+ mSampledTimeoutPollLatencySum = 0;
+#endif
}
Looper::~Looper() {
close(mWakeReadPipeFd);
close(mWakeWritePipeFd);
+#ifdef LOOPER_USES_EPOLL
close(mEpollFd);
+#endif
}
void Looper::initTLSKey() {
@@ -157,45 +193,61 @@
#if DEBUG_POLL_AND_WAKE
LOGD("%p ~ pollOnce - waiting: timeoutMillis=%d", this, timeoutMillis);
#endif
+
+ int result = ALOOPER_POLL_WAKE;
+ mResponses.clear();
+ mResponseIndex = 0;
+
+#ifdef LOOPER_STATISTICS
+ nsecs_t pollStartTime = systemTime(SYSTEM_TIME_MONOTONIC);
+#endif
+
+#ifdef LOOPER_USES_EPOLL
struct epoll_event eventItems[EPOLL_MAX_EVENTS];
int eventCount = epoll_wait(mEpollFd, eventItems, EPOLL_MAX_EVENTS, timeoutMillis);
+ bool acquiredLock = false;
+#else
+ // Wait for wakeAndLock() waiters to run then set mPolling to true.
+ mLock.lock();
+ while (mWaiters != 0) {
+ mResume.wait(mLock);
+ }
+ mPolling = true;
+ mLock.unlock();
+
+ size_t requestedCount = mRequestedFds.size();
+ int eventCount = poll(mRequestedFds.editArray(), requestedCount, timeoutMillis);
+#endif
+
if (eventCount < 0) {
if (errno == EINTR) {
- return ALOOPER_POLL_WAKE;
+ goto Done;
}
LOGW("Poll failed with an unexpected error, errno=%d", errno);
- return ALOOPER_POLL_ERROR;
+ result = ALOOPER_POLL_ERROR;
+ goto Done;
}
if (eventCount == 0) {
#if DEBUG_POLL_AND_WAKE
LOGD("%p ~ pollOnce - timeout", this);
#endif
- return ALOOPER_POLL_TIMEOUT;
+ result = ALOOPER_POLL_TIMEOUT;
+ goto Done;
}
- int result = ALOOPER_POLL_WAKE;
- mResponses.clear();
- mResponseIndex = 0;
-
#if DEBUG_POLL_AND_WAKE
LOGD("%p ~ pollOnce - handling events from %d fds", this, eventCount);
#endif
- bool acquiredLock = false;
+
+#ifdef LOOPER_USES_EPOLL
for (int i = 0; i < eventCount; i++) {
int fd = eventItems[i].data.fd;
uint32_t epollEvents = eventItems[i].events;
if (fd == mWakeReadPipeFd) {
if (epollEvents & EPOLLIN) {
-#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 == -1 && errno == EINTR) || nRead == sizeof(buffer));
+ awoken();
} else {
LOGW("Ignoring unexpected epoll events 0x%x on wake read pipe.", epollEvents);
}
@@ -212,11 +264,7 @@
if (epollEvents & EPOLLOUT) events |= ALOOPER_EVENT_OUTPUT;
if (epollEvents & EPOLLERR) events |= ALOOPER_EVENT_ERROR;
if (epollEvents & EPOLLHUP) events |= ALOOPER_EVENT_HANGUP;
-
- Response response;
- response.events = events;
- response.request = mRequests.valueAt(requestIndex);
- mResponses.push(response);
+ pushResponse(events, mRequests.valueAt(requestIndex));
} else {
LOGW("Ignoring unexpected epoll events 0x%x on fd %d that is "
"no longer registered.", epollEvents, fd);
@@ -226,6 +274,66 @@
if (acquiredLock) {
mLock.unlock();
}
+Done: ;
+#else
+ for (size_t i = 0; i < requestedCount; i++) {
+ const struct pollfd& requestedFd = mRequestedFds.itemAt(i);
+
+ short pollEvents = requestedFd.revents;
+ if (pollEvents) {
+ if (requestedFd.fd == mWakeReadPipeFd) {
+ if (pollEvents & POLLIN) {
+ awoken();
+ } else {
+ LOGW("Ignoring unexpected poll events 0x%x on wake read pipe.", pollEvents);
+ }
+ } else {
+ int events = 0;
+ if (pollEvents & POLLIN) events |= ALOOPER_EVENT_INPUT;
+ if (pollEvents & POLLOUT) events |= ALOOPER_EVENT_OUTPUT;
+ if (pollEvents & POLLERR) events |= ALOOPER_EVENT_ERROR;
+ if (pollEvents & POLLHUP) events |= ALOOPER_EVENT_HANGUP;
+ if (pollEvents & POLLNVAL) events |= ALOOPER_EVENT_INVALID;
+ pushResponse(events, mRequests.itemAt(i));
+ }
+ if (--eventCount == 0) {
+ break;
+ }
+ }
+ }
+
+Done:
+ // Set mPolling to false and wake up the wakeAndLock() waiters.
+ mLock.lock();
+ mPolling = false;
+ if (mWaiters != 0) {
+ mAwake.broadcast();
+ }
+ mLock.unlock();
+#endif
+
+#ifdef LOOPER_STATISTICS
+ nsecs_t pollEndTime = systemTime(SYSTEM_TIME_MONOTONIC);
+ mSampledPolls += 1;
+ if (timeoutMillis == 0) {
+ mSampledZeroPollCount += 1;
+ mSampledZeroPollLatencySum += pollEndTime - pollStartTime;
+ } else if (timeoutMillis > 0 && result == ALOOPER_POLL_TIMEOUT) {
+ mSampledTimeoutPollCount += 1;
+ mSampledTimeoutPollLatencySum += pollEndTime - pollStartTime
+ - milliseconds_to_nanoseconds(timeoutMillis);
+ }
+ if (mSampledPolls == SAMPLED_POLLS_TO_AGGREGATE) {
+ LOGD("%p ~ poll latency statistics: %0.3fms zero timeout, %0.3fms non-zero timeout", this,
+ 0.000001f * float(mSampledZeroPollLatencySum) / mSampledZeroPollCount,
+ 0.000001f * float(mSampledTimeoutPollLatencySum) / mSampledTimeoutPollCount);
+ mSampledPolls = 0;
+ mSampledZeroPollCount = 0;
+ mSampledZeroPollLatencySum = 0;
+ mSampledTimeoutPollCount = 0;
+ mSampledTimeoutPollLatencySum = 0;
+ }
+#endif
for (size_t i = 0; i < mResponses.size(); i++) {
const Response& response = mResponses.itemAt(i);
@@ -278,6 +386,13 @@
LOGD("%p ~ wake", this);
#endif
+#ifdef LOOPER_STATISTICS
+ // FIXME: Possible race with awoken() but this code is for testing only and is rarely enabled.
+ if (mPendingWakeCount++ == 0) {
+ mPendingWakeTime = systemTime(SYSTEM_TIME_MONOTONIC);
+ }
+#endif
+
ssize_t nWrite;
do {
nWrite = write(mWakeWritePipeFd, "W", 1);
@@ -290,23 +405,51 @@
}
}
+void Looper::awoken() {
+#if DEBUG_POLL_AND_WAKE
+ LOGD("%p ~ awoken", this);
+#endif
+
+#ifdef LOOPER_STATISTICS
+ if (mPendingWakeCount == 0) {
+ LOGD("%p ~ awoken: spurious!", this);
+ } else {
+ mSampledWakeCycles += 1;
+ mSampledWakeCountSum += mPendingWakeCount;
+ mSampledWakeLatencySum += systemTime(SYSTEM_TIME_MONOTONIC) - mPendingWakeTime;
+ mPendingWakeCount = 0;
+ mPendingWakeTime = -1;
+ if (mSampledWakeCycles == SAMPLED_WAKE_CYCLES_TO_AGGREGATE) {
+ LOGD("%p ~ wake statistics: %0.3fms wake latency, %0.3f wakes per cycle", this,
+ 0.000001f * float(mSampledWakeLatencySum) / mSampledWakeCycles,
+ float(mSampledWakeCountSum) / mSampledWakeCycles);
+ mSampledWakeCycles = 0;
+ mSampledWakeCountSum = 0;
+ mSampledWakeLatencySum = 0;
+ }
+ }
+#endif
+
+ char buffer[16];
+ ssize_t nRead;
+ do {
+ nRead = read(mWakeReadPipeFd, buffer, sizeof(buffer));
+ } while ((nRead == -1 && errno == EINTR) || nRead == sizeof(buffer));
+}
+
+void Looper::pushResponse(int events, const Request& request) {
+ Response response;
+ response.events = events;
+ response.request = request;
+ mResponses.push(response);
+}
+
int Looper::addFd(int fd, int ident, int events, ALooper_callbackFunc callback, void* data) {
#if DEBUG_CALLBACKS
LOGD("%p ~ addFd - fd=%d, ident=%d, events=0x%x, callback=%p, data=%p", this, fd, ident,
events, callback, data);
#endif
- int epollEvents = 0;
- if (events & ALOOPER_EVENT_INPUT) epollEvents |= EPOLLIN;
- if (events & ALOOPER_EVENT_OUTPUT) epollEvents |= EPOLLOUT;
- if (events & ALOOPER_EVENT_ERROR) epollEvents |= EPOLLERR;
- if (events & ALOOPER_EVENT_HANGUP) epollEvents |= EPOLLHUP;
-
- if (epollEvents == 0) {
- LOGE("Invalid attempt to set a callback with no selected poll events.");
- return -1;
- }
-
if (! callback) {
if (! mAllowNonCallbacks) {
LOGE("Invalid attempt to set NULL callback but not allowed for this looper.");
@@ -319,6 +462,11 @@
}
}
+#ifdef LOOPER_USES_EPOLL
+ int epollEvents = 0;
+ if (events & ALOOPER_EVENT_INPUT) epollEvents |= EPOLLIN;
+ if (events & ALOOPER_EVENT_OUTPUT) epollEvents |= EPOLLOUT;
+
{ // acquire lock
AutoMutex _l(mLock);
@@ -350,6 +498,33 @@
mRequests.replaceValueAt(requestIndex, request);
}
} // release lock
+#else
+ int pollEvents = 0;
+ if (events & ALOOPER_EVENT_INPUT) pollEvents |= POLLIN;
+ if (events & ALOOPER_EVENT_OUTPUT) pollEvents |= POLLOUT;
+
+ wakeAndLock(); // acquire lock
+
+ struct pollfd requestedFd;
+ requestedFd.fd = fd;
+ requestedFd.events = pollEvents;
+
+ Request request;
+ request.fd = fd;
+ request.ident = ident;
+ request.callback = callback;
+ request.data = data;
+ ssize_t index = getRequestIndexLocked(fd);
+ if (index < 0) {
+ mRequestedFds.push(requestedFd);
+ mRequests.push(request);
+ } else {
+ mRequestedFds.replaceAt(requestedFd, size_t(index));
+ mRequests.replaceAt(request, size_t(index));
+ }
+
+ mLock.unlock(); // release lock
+#endif
return 1;
}
@@ -358,6 +533,7 @@
LOGD("%p ~ removeFd - fd=%d", this, fd);
#endif
+#ifdef LOOPER_USES_EPOLL
{ // acquire lock
AutoMutex _l(mLock);
ssize_t requestIndex = mRequests.indexOfKey(fd);
@@ -372,8 +548,49 @@
}
mRequests.removeItemsAt(requestIndex);
- } // request lock
+ } // release lock
return 1;
+#else
+ wakeAndLock(); // acquire lock
+
+ ssize_t index = getRequestIndexLocked(fd);
+ if (index >= 0) {
+ mRequestedFds.removeAt(size_t(index));
+ mRequests.removeAt(size_t(index));
+ }
+
+ mLock.unlock(); // release lock
+ return index >= 0;
+#endif
}
+#ifndef LOOPER_USES_EPOLL
+ssize_t Looper::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 Looper::wakeAndLock() {
+ mLock.lock();
+
+ mWaiters += 1;
+ while (mPolling) {
+ wake();
+ mAwake.wait(mLock);
+ }
+
+ mWaiters -= 1;
+ if (mWaiters == 0) {
+ mResume.signal();
+ }
+}
+#endif
+
} // namespace android
diff --git a/libs/utils/tests/Looper_test.cpp b/libs/utils/tests/Looper_test.cpp
index afc92f8..cea1313 100644
--- a/libs/utils/tests/Looper_test.cpp
+++ b/libs/utils/tests/Looper_test.cpp
@@ -354,14 +354,6 @@
<< "addFd should return 1 because FD was added";
}
-TEST_F(LooperTest, AddFd_WhenEventsIsZero_ReturnsError) {
- Pipe pipe;
- int result = mLooper->addFd(pipe.receiveFd, 0, 0, NULL, NULL);
-
- EXPECT_EQ(-1, result)
- << "addFd should return -1 because arguments were invalid";
-}
-
TEST_F(LooperTest, AddFd_WhenIdentIsNegativeAndCallbackIsNull_ReturnsError) {
Pipe pipe;
int result = mLooper->addFd(pipe.receiveFd, -1, ALOOPER_EVENT_INPUT, NULL, NULL);