Add support for non-blocking I/O with Looper.
Bug: 10349083
Change-Id: I4a94b1eac53df57c05103913bd593d92b1e062d7
diff --git a/core/java/android/os/Looper.java b/core/java/android/os/Looper.java
index 9b3880b..34c880f 100644
--- a/core/java/android/os/Looper.java
+++ b/core/java/android/os/Looper.java
@@ -273,6 +273,7 @@
mQueue.dump(pw, prefix + " ");
}
+ @Override
public String toString() {
return "Looper (" + mThread.getName() + ", tid " + mThread.getId()
+ ") {" + Integer.toHexString(System.identityHashCode(this)) + "}";
diff --git a/core/java/android/os/MessageQueue.java b/core/java/android/os/MessageQueue.java
index d672f9bb..7dd4f94 100644
--- a/core/java/android/os/MessageQueue.java
+++ b/core/java/android/os/MessageQueue.java
@@ -16,9 +16,15 @@
package android.os;
+import android.annotation.IntDef;
+import android.annotation.NonNull;
import android.util.Log;
import android.util.Printer;
+import android.util.SparseArray;
+import java.io.FileDescriptor;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
import java.util.ArrayList;
/**
@@ -30,6 +36,9 @@
* {@link Looper#myQueue() Looper.myQueue()}.
*/
public final class MessageQueue {
+ private static final String TAG = "MessageQueue";
+ private static final boolean DEBUG = false;
+
// True if the message queue can be quit.
private final boolean mQuitAllowed;
@@ -38,6 +47,7 @@
Message mMessages;
private final ArrayList<IdleHandler> mIdleHandlers = new ArrayList<IdleHandler>();
+ private SparseArray<FileDescriptorRecord> mFileDescriptorRecords;
private IdleHandler[] mPendingIdleHandlers;
private boolean mQuitting;
@@ -50,9 +60,10 @@
private native static long nativeInit();
private native static void nativeDestroy(long ptr);
- private native static void nativePollOnce(long ptr, int timeoutMillis);
+ private native void nativePollOnce(long ptr, int timeoutMillis); /*non-static for callbacks*/
private native static void nativeWake(long ptr);
private native static boolean nativeIsPolling(long ptr);
+ private native static void nativeSetFileDescriptorEvents(long ptr, int fd, int events);
MessageQueue(boolean quitAllowed) {
mQuitAllowed = quitAllowed;
@@ -101,7 +112,7 @@
*
* @param handler The IdleHandler to be added.
*/
- public void addIdleHandler(IdleHandler handler) {
+ public void addIdleHandler(@NonNull IdleHandler handler) {
if (handler == null) {
throw new NullPointerException("Can't add a null IdleHandler");
}
@@ -119,7 +130,7 @@
*
* @param handler The IdleHandler to be removed.
*/
- public void removeIdleHandler(IdleHandler handler) {
+ public void removeIdleHandler(@NonNull IdleHandler handler) {
synchronized (this) {
mIdleHandlers.remove(handler);
}
@@ -148,6 +159,151 @@
return !mQuitting && nativeIsPolling(mPtr);
}
+ /**
+ * Registers a file descriptor callback to receive notification when file descriptor
+ * related events occur.
+ * <p>
+ * If the file descriptor has already been registered, the specified events
+ * and callback will replace any that were previously associated with it.
+ * It is not possible to set more than one callback per file descriptor.
+ * </p><p>
+ * It is important to always unregister the callback when the file descriptor
+ * is no longer of use.
+ * </p>
+ *
+ * @param fd The file descriptor for which a callback will be registered.
+ * @param events The set of events to receive: a combination of the
+ * {@link FileDescriptorCallback#EVENT_INPUT},
+ * {@link FileDescriptorCallback#EVENT_OUTPUT}, and
+ * {@link FileDescriptorCallback#EVENT_ERROR} event masks. If the requested
+ * set of events is zero, then the callback is unregistered.
+ * @param callback The callback to invoke when file descriptor events occur.
+ *
+ * @see FileDescriptorCallback
+ * @see #unregisterFileDescriptorCallback
+ */
+ public void registerFileDescriptorCallback(@NonNull FileDescriptor fd,
+ @FileDescriptorCallback.Events int events,
+ @NonNull FileDescriptorCallback callback) {
+ if (fd == null) {
+ throw new IllegalArgumentException("fd must not be null");
+ }
+ if (callback == null) {
+ throw new IllegalArgumentException("callback must not be null");
+ }
+
+ synchronized (this) {
+ setFileDescriptorCallbackLocked(fd, events, callback);
+ }
+ }
+
+ /**
+ * Unregisters a file descriptor callback.
+ * <p>
+ * This method does nothing if no callback has been registered for the
+ * specified file descriptor.
+ * </p>
+ *
+ * @param fd The file descriptor whose callback will be unregistered.
+ *
+ * @see FileDescriptorCallback
+ * @see #registerFileDescriptorCallback
+ */
+ public void unregisterFileDescriptorCallback(@NonNull FileDescriptor fd) {
+ if (fd == null) {
+ throw new IllegalArgumentException("fd must not be null");
+ }
+
+ synchronized (this) {
+ setFileDescriptorCallbackLocked(fd, 0, null);
+ }
+ }
+
+ private void setFileDescriptorCallbackLocked(FileDescriptor fd, int events,
+ FileDescriptorCallback callback) {
+ final int fdNum = fd.getInt$();
+
+ int index = -1;
+ FileDescriptorRecord record = null;
+ if (mFileDescriptorRecords != null) {
+ index = mFileDescriptorRecords.indexOfKey(fdNum);
+ if (index >= 0) {
+ record = mFileDescriptorRecords.valueAt(index);
+ if (record != null && record.mEvents == events) {
+ return;
+ }
+ }
+ }
+
+ if (events != 0) {
+ events |= FileDescriptorCallback.EVENT_ERROR;
+ if (record == null) {
+ if (mFileDescriptorRecords == null) {
+ mFileDescriptorRecords = new SparseArray<FileDescriptorRecord>();
+ }
+ record = new FileDescriptorRecord(fd, events, callback);
+ mFileDescriptorRecords.put(fdNum, record);
+ } else {
+ record.mCallback = callback;
+ record.mEvents = events;
+ record.mSeq += 1;
+ }
+ nativeSetFileDescriptorEvents(mPtr, fdNum, events);
+ } else if (record != null) {
+ record.mEvents = 0;
+ mFileDescriptorRecords.removeAt(index);
+ }
+ }
+
+ // Called from native code.
+ private int dispatchEvents(int fd, int events) {
+ // Get the file descriptor record and any state that might change.
+ final FileDescriptorRecord record;
+ final int oldWatchedEvents;
+ final FileDescriptorCallback callback;
+ final int seq;
+ synchronized (this) {
+ record = mFileDescriptorRecords.get(fd);
+ if (record == null) {
+ return 0; // spurious, no callback registered
+ }
+
+ oldWatchedEvents = record.mEvents;
+ events &= oldWatchedEvents; // filter events based on current watched set
+ if (events == 0) {
+ return oldWatchedEvents; // spurious, watched events changed
+ }
+
+ callback = record.mCallback;
+ seq = record.mSeq;
+ }
+
+ // Invoke the callback outside of the lock.
+ int newWatchedEvents = callback.onFileDescriptorEvents(
+ record.mDescriptor, events);
+ if (newWatchedEvents != 0) {
+ newWatchedEvents |= FileDescriptorCallback.EVENT_ERROR;
+ }
+
+ // Update the file descriptor record if the callback changed the set of
+ // events to watch and the callback itself hasn't been updated since.
+ if (newWatchedEvents != oldWatchedEvents) {
+ synchronized (this) {
+ int index = mFileDescriptorRecords.indexOfKey(fd);
+ if (index >= 0 && mFileDescriptorRecords.valueAt(index) == record
+ && record.mSeq == seq) {
+ record.mEvents = newWatchedEvents;
+ if (newWatchedEvents == 0) {
+ mFileDescriptorRecords.removeAt(index);
+ }
+ }
+ }
+ }
+
+ // Return the new set of events to watch for native code to take care of.
+ return newWatchedEvents;
+ }
+
Message next() {
// Return here if the message loop has already quit and been disposed.
// This can happen if the application tries to restart a looper after quit
@@ -191,7 +347,8 @@
mMessages = msg.next;
}
msg.next = null;
- if (false) Log.v("MessageQueue", "Returning message: " + msg);
+ if (DEBUG) Log.v(TAG, "Returning message: " + msg);
+ msg.markInUse();
return msg;
}
} else {
@@ -234,7 +391,7 @@
try {
keep = idler.queueIdle();
} catch (Throwable t) {
- Log.wtf("MessageQueue", "IdleHandler threw exception", t);
+ Log.wtf(TAG, "IdleHandler threw exception", t);
}
if (!keep) {
@@ -385,7 +542,7 @@
if (mQuitting) {
IllegalStateException e = new IllegalStateException(
msg.target + " sending message to a Handler on a dead thread");
- Log.w("MessageQueue", e.getMessage(), e);
+ Log.w(TAG, e.getMessage(), e);
msg.recycle();
return false;
}
@@ -627,4 +784,94 @@
*/
boolean queueIdle();
}
+
+ /**
+ * A callback which is invoked when file descriptor related events occur.
+ */
+ public static abstract class FileDescriptorCallback {
+ /**
+ * File descriptor event: Indicates that the file descriptor is ready for input
+ * operations, such as reading.
+ * <p>
+ * The callback should read all available data from the file descriptor
+ * then return <code>true</code> to keep the callback active or <code>false</code>
+ * to remove the callback.
+ * </p><p>
+ * In the case of a socket, this event may be generated to indicate
+ * that there is at least one incoming connection that the callback
+ * should accept.
+ * </p><p>
+ * This event will only be generated if the {@link #EVENT_INPUT} event mask was
+ * specified when the callback was added.
+ * </p>
+ */
+ public static final int EVENT_INPUT = 1 << 0;
+
+ /**
+ * File descriptor event: Indicates that the file descriptor is ready for output
+ * operations, such as writing.
+ * <p>
+ * The callback should write as much data as it needs. If it could not
+ * write everything at once, then it should return <code>true</code> to
+ * keep the callback active. Otherwise, it should return <code>false</code>
+ * to remove the callback then re-register it later when it needs to write
+ * something else.
+ * </p><p>
+ * This event will only be generated if the {@link #EVENT_OUTPUT} event mask was
+ * specified when the callback was added.
+ * </p>
+ */
+ public static final int EVENT_OUTPUT = 1 << 1;
+
+ /**
+ * File descriptor event: Indicates that the file descriptor encountered a
+ * fatal error.
+ * <p>
+ * File descriptor errors can occur for various reasons. One common error
+ * is when the remote peer of a socket or pipe closes its end of the connection.
+ * </p><p>
+ * This event may be generated at any time regardless of whether the
+ * {@link #EVENT_ERROR} event mask was specified when the callback was added.
+ * </p>
+ */
+ public static final int EVENT_ERROR = 1 << 2;
+
+ /** @hide */
+ @Retention(RetentionPolicy.SOURCE)
+ @IntDef(flag=true, value={EVENT_INPUT, EVENT_OUTPUT, EVENT_ERROR})
+ public @interface Events {}
+
+ /**
+ * Called when a file descriptor receives events.
+ * <p>
+ * The default implementation does nothing and returns 0 to unregister the callback.
+ * </p>
+ *
+ * @param fd The file descriptor.
+ * @param events The set of events that occurred: a combination of the
+ * {@link #EVENT_INPUT}, {@link #EVENT_OUTPUT}, and {@link #EVENT_ERROR} event masks.
+ * @return The new set of events to watch, or 0 to unregister the callback.
+ *
+ * @see #EVENT_INPUT
+ * @see #EVENT_OUTPUT
+ * @see #EVENT_ERROR
+ */
+ public @Events int onFileDescriptorEvents(@NonNull FileDescriptor fd, @Events int events) {
+ return 0;
+ }
+ }
+
+ private static final class FileDescriptorRecord {
+ public final FileDescriptor mDescriptor;
+ public int mEvents;
+ public FileDescriptorCallback mCallback;
+ public int mSeq;
+
+ public FileDescriptorRecord(FileDescriptor descriptor,
+ int events, FileDescriptorCallback callback) {
+ mDescriptor = descriptor;
+ mEvents = events;
+ mCallback = callback;
+ }
+ }
}
diff --git a/core/jni/android_os_MessageQueue.cpp b/core/jni/android_os_MessageQueue.cpp
index ee64044..854e6b7 100644
--- a/core/jni/android_os_MessageQueue.cpp
+++ b/core/jni/android_os_MessageQueue.cpp
@@ -29,22 +29,31 @@
static struct {
jfieldID mPtr; // native object attached to the DVM MessageQueue
+ jmethodID dispatchEvents;
} gMessageQueueClassInfo;
+// Must be kept in sync with the constants in Looper.FileDescriptorCallback
+static const int CALLBACK_EVENT_INPUT = 1 << 0;
+static const int CALLBACK_EVENT_OUTPUT = 1 << 1;
+static const int CALLBACK_EVENT_ERROR = 1 << 2;
-class NativeMessageQueue : public MessageQueue {
+
+class NativeMessageQueue : public MessageQueue, public LooperCallback {
public:
NativeMessageQueue();
virtual ~NativeMessageQueue();
virtual void raiseException(JNIEnv* env, const char* msg, jthrowable exceptionObj);
- void pollOnce(JNIEnv* env, int timeoutMillis);
-
+ void pollOnce(JNIEnv* env, jobject obj, int timeoutMillis);
void wake();
+ void setFileDescriptorEvents(int fd, int events);
+
+ virtual int handleEvent(int fd, int events, void* data);
private:
- bool mInCallback;
+ JNIEnv* mPollEnv;
+ jobject mPollObj;
jthrowable mExceptionObj;
};
@@ -66,10 +75,11 @@
return false;
}
-NativeMessageQueue::NativeMessageQueue() : mInCallback(false), mExceptionObj(NULL) {
+NativeMessageQueue::NativeMessageQueue() :
+ mPollEnv(NULL), mPollObj(NULL), mExceptionObj(NULL) {
mLooper = Looper::getForThread();
if (mLooper == NULL) {
- mLooper = new Looper(false);
+ mLooper = new Looper(true);
Looper::setForThread(mLooper);
}
}
@@ -79,7 +89,7 @@
void NativeMessageQueue::raiseException(JNIEnv* env, const char* msg, jthrowable exceptionObj) {
if (exceptionObj) {
- if (mInCallback) {
+ if (mPollEnv == env) {
if (mExceptionObj) {
env->DeleteLocalRef(mExceptionObj);
}
@@ -94,10 +104,13 @@
}
}
-void NativeMessageQueue::pollOnce(JNIEnv* env, int timeoutMillis) {
- mInCallback = true;
+void NativeMessageQueue::pollOnce(JNIEnv* env, jobject pollObj, int timeoutMillis) {
+ mPollEnv = env;
+ mPollObj = pollObj;
mLooper->pollOnce(timeoutMillis);
- mInCallback = false;
+ mPollObj = NULL;
+ mPollEnv = NULL;
+
if (mExceptionObj) {
env->Throw(mExceptionObj);
env->DeleteLocalRef(mExceptionObj);
@@ -109,6 +122,46 @@
mLooper->wake();
}
+void NativeMessageQueue::setFileDescriptorEvents(int fd, int events) {
+ if (events) {
+ int looperEvents = 0;
+ if (events & CALLBACK_EVENT_INPUT) {
+ looperEvents |= Looper::EVENT_INPUT;
+ }
+ if (events & CALLBACK_EVENT_OUTPUT) {
+ looperEvents |= Looper::EVENT_OUTPUT;
+ }
+ mLooper->addFd(fd, Looper::POLL_CALLBACK, looperEvents, this,
+ reinterpret_cast<void*>(events));
+ } else {
+ mLooper->removeFd(fd);
+ }
+}
+
+int NativeMessageQueue::handleEvent(int fd, int looperEvents, void* data) {
+ int events = 0;
+ if (looperEvents & Looper::EVENT_INPUT) {
+ events |= CALLBACK_EVENT_INPUT;
+ }
+ if (looperEvents & Looper::EVENT_OUTPUT) {
+ events |= CALLBACK_EVENT_OUTPUT;
+ }
+ if (looperEvents & (Looper::EVENT_ERROR | Looper::EVENT_HANGUP | Looper::EVENT_INVALID)) {
+ events |= CALLBACK_EVENT_ERROR;
+ }
+ int oldWatchedEvents = reinterpret_cast<int>(data);
+ int newWatchedEvents = mPollEnv->CallIntMethod(mPollObj,
+ gMessageQueueClassInfo.dispatchEvents, fd, events);
+ if (!newWatchedEvents) {
+ return 0; // unregister the fd
+ }
+ if (newWatchedEvents != oldWatchedEvents) {
+ setFileDescriptorEvents(fd, newWatchedEvents);
+ }
+ return 1;
+}
+
+
// ----------------------------------------------------------------------------
sp<MessageQueue> android_os_MessageQueue_getMessageQueue(JNIEnv* env, jobject messageQueueObj) {
@@ -132,15 +185,15 @@
nativeMessageQueue->decStrong(env);
}
-static void android_os_MessageQueue_nativePollOnce(JNIEnv* env, jclass clazz,
+static void android_os_MessageQueue_nativePollOnce(JNIEnv* env, jobject obj,
jlong ptr, jint timeoutMillis) {
NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);
- nativeMessageQueue->pollOnce(env, timeoutMillis);
+ nativeMessageQueue->pollOnce(env, obj, timeoutMillis);
}
static void android_os_MessageQueue_nativeWake(JNIEnv* env, jclass clazz, jlong ptr) {
NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);
- return nativeMessageQueue->wake();
+ nativeMessageQueue->wake();
}
static jboolean android_os_MessageQueue_nativeIsPolling(JNIEnv* env, jclass clazz, jlong ptr) {
@@ -148,6 +201,12 @@
return nativeMessageQueue->getLooper()->isPolling();
}
+static void android_os_MessageQueue_nativeSetFileDescriptorEvents(JNIEnv* env, jclass clazz,
+ jlong ptr, jint fd, jint events) {
+ NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);
+ nativeMessageQueue->setFileDescriptorEvents(fd, events);
+}
+
// ----------------------------------------------------------------------------
static JNINativeMethod gMessageQueueMethods[] = {
@@ -156,7 +215,9 @@
{ "nativeDestroy", "(J)V", (void*)android_os_MessageQueue_nativeDestroy },
{ "nativePollOnce", "(JI)V", (void*)android_os_MessageQueue_nativePollOnce },
{ "nativeWake", "(J)V", (void*)android_os_MessageQueue_nativeWake },
- { "nativeIsPolling", "(J)Z", (void*)android_os_MessageQueue_nativeIsPolling }
+ { "nativeIsPolling", "(J)Z", (void*)android_os_MessageQueue_nativeIsPolling },
+ { "nativeSetFileDescriptorEvents", "(JII)V",
+ (void*)android_os_MessageQueue_nativeSetFileDescriptorEvents },
};
int register_android_os_MessageQueue(JNIEnv* env) {
@@ -164,8 +225,9 @@
NELEM(gMessageQueueMethods));
jclass clazz = FindClassOrDie(env, "android/os/MessageQueue");
-
gMessageQueueClassInfo.mPtr = GetFieldIDOrDie(env, clazz, "mPtr", "J");
+ gMessageQueueClassInfo.dispatchEvents = GetMethodIDOrDie(env, clazz,
+ "dispatchEvents", "(II)I");
return res;
}
diff --git a/core/jni/android_os_MessageQueue.h b/core/jni/android_os_MessageQueue.h
index 49d2aa0..1e49b5f 100644
--- a/core/jni/android_os_MessageQueue.h
+++ b/core/jni/android_os_MessageQueue.h
@@ -22,7 +22,7 @@
namespace android {
-class MessageQueue : public RefBase {
+class MessageQueue : public virtual RefBase {
public:
/* Gets the message queue's looper. */
inline sp<Looper> getLooper() const {