Handle message mechanism parsing
An overview of the
Handler messaging (by the Handler/stars/MessageQueue, etc), the Android there is a lot of the message-driven method to interact, For example, the four components of Android (Activity, Service, Broadcast, ContentProvider) can not be separated from the Handler message mechanism, so the Android system is also a message-driven system in a sense.
Architecture diagram
Classes (Android 8.0 source code)
Handler
Handler overview
- A Handler is a program that sends and processes messages and Runnable objects by associating them with a message queue. Each Handler is associated with a separate thread and message queue. When you create a new Handler, it binds to a thread or message queue on a thread. From there, this Handler will supply the message queue with messages or Runnable objects and process the messages or Runnable objects that the message queue distributes.
What is the use of handler
-
Send a message to MesageQueue
-
Process messages distributed by Looper
The constructor
public Handler(Callback callback, boolean async) {
// If it is an anonymous class, an inner class, or a local class (inside a method), and is not declared STATIC, then there is a risk of memory leakage, so print a log to alert the developer
if (FIND_POTENTIAL_LEAKS) {
final Class<? extends Handler> klass = getClass();
if((klass.isAnonymousClass() || klass.isMemberClass() || klass.isLocalClass()) && (klass.getModifiers() & Modifier.STATIC) = =0) {
Log.w(TAG, "The following Handler class should be static or leaks might occur: "+ klass.getCanonicalName()); }}// Get the current thread's Looper from ThreadLocal
mLooper = Looper.myLooper();
// If the current thread does not have Looper, looper.prepare () is not called and an exception is thrown
if (mLooper == null) {
throw new RuntimeException(
"Can't create handler inside thread that has not called Looper.prepare()");
}
//关联MessageQueue
mQueue = mLooper.mQueue;
//mCallback assigns a value. If not null, use this callback to process the message
mCallback = callback;
// Indicates whether this Handler is asynchronous. If true, messages sent by this Handler are asynchronous messages. Asynchronous messaging is from by MessageQueue. EnqueueSyncBarrier (long) (after change method signature to postSyncBarrier (long)) send synchronization barrier impact.
mAsynchronous = async;
}
Copy the codeLooper
Looper’s constructor is private and passes
public static void prepare(a) {
prepare(true);
}
//quiteAllowed - whether to exit the MessageQueue (MessageQueue)
private static void prepare(boolean quitAllowed) {
if(sThreadLocal.get() ! =null) {
throw new RuntimeException("Only one Looper may be created per thread");
}
sThreadLocal.set(new Looper(quitAllowed));
}
Copy the codeTo create a Looper object and bind it to the thread using ThreadLoacl.
There is another important method
public static void prepareMainLooper(a) {
/ / can't quit
prepare(false);
synchronized (Looper.class) {
// This method cannot be called manually because it was already called when the application process started (in activityThread.main ())
if(sMainLooper ! =null) {
throw new IllegalStateException("The main Looper has already been prepared."); } sMainLooper = myLooper(); }}Copy the codeIn the main function of ActivityThread, the entry to the Android VIRTUAL machine
public static void main(String[] args) {
//prepare mainLooper
Looper.prepareMainLooper();
ActivityThread thread = new ActivityThread();
thread.attach(false);
if (sMainThreadHandler == null) {
sMainThreadHandler = thread.getHandler();
}
if (false) {
Looper.myLooper().setMessageLogging(new
LogPrinter(Log.DEBUG, "ActivityThread"));
}
// End of event ActivityThreadMain.
Trace.traceEnd(Trace.TRACE_TAG_ACTIVITY_MANAGER);
Looper.loop();
throw new RuntimeException("Main thread loop unexpectedly exited");
}
Copy the codeInitialize the main thread Looper, which indirectly indicates that Android is an event-driven system to some extent.
ThreadLocal
An overview of the
The ThreadLocal class is used to provide local variables within a thread. It is suitable for scenarios where each thread needs its own independent instance and that instance needs to be used in multiple methods, that is, variables are isolated between threads and shared between methods or classes. ThreadLocal instances are typically of the private static type, used to associate threads.
Core method
1. Set ()- Gets the ThreadLocalMap object instance threadLocals from the current Thread and saves the required values to threadLocals
// Returns the value in the current thread copy of this thread-local variable. If the variable has no value for the current thread, it is first initialized to the value returned by calling the initialValue method.
public void set(T value) {
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
if(map ! =null)
map.set(this, value);
else
createMap(t, value);
}
// Returns threadLocals from the current thread object
ThreadLocalMap getMap(Thread t) {
return t.threadLocals;
}
Copy the code2. Get ()- Gets the ThreadLocalMap from the current Thread’s Thread object and reads the value of the ThreadLocal object from it
// Sets the current thread copy of this thread-local variable to the specified value. Most subclasses will not need to override this method and will rely only on the initialValue method to set the value of the thread initialValue.
public T get(a) {
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
if(map ! =null) {
ThreadLocalMap.Entry e = map.getEntry(this);
if(e ! =null) {
@SuppressWarnings("unchecked")
T result = (T)e.value;
returnresult; }}return setInitialValue();
}
Copy the codeMap is the static internal class of ThreadLocalMap. The operations such as storing and fetching are basically performed on this map.
ThreadLocalMap
Similar to HashMap, the bottom layer is an Entry<K,V> array, which inherits WeakReference. Key is a WeakReference type, and value is a strong reference type, so there will be old entries in the array (key == null). In get(),set() operations, Stale entries are removed via expungeStaleEntry ().
static class Entry extends WeakReference<ThreadLocal<? >>{
/** The value associated with this ThreadLocal. */
Object value;
// Key (threadLocal object) is a weak reference typeEntry(ThreadLocal<? > k, Object v) {super(k); value = v; }}// Store Entry objects. When a hash conflict occurs, the open address method of linear probe resolves the hash conflict.
private void set(ThreadLocal
key, Object value) {
// We don't use a fast path as with get() because it is at
// least as common to use set() to create new entries as
// it is to replace existing ones, in which case, a fast
// path would fail more often than not.
Entry[] tab = table;
int len = tab.length;
// Compute array subscripts, similar to HashMap
int i = key.threadLocalHashCode & (len-1);
for(Entry e = tab[i]; e ! =null; e = tab[i = nextIndex(i, len)]) { ThreadLocal<? > k = e.get();// There is a key
if (k == key) {
e.value = value;
return;
}
// Replace old entries (key == null)
if (k == null) {
replaceStaleEntry(key, value, i);
return; }}// Create an entry and insert it
tab[i] = new Entry(key, value);
int sz = ++size;
// Determine if any old entries have been removed && size >= threshold
if(! cleanSomeSlots(i, sz) && sz >= threshold) rehash(); }// First scan the entire table to remove stale entries. If the size is greater than 3/4 of the threshold, double the capacity
private void rehash(a) {
expungeStaleEntries();
// Use lower threshold for doubling to avoid hysteresis
if (size >= threshold - threshold / 4)
resize();
}
// Get the Entry object.
private Entry getEntry(ThreadLocal
key) {
// Compute the array subscript
int i = key.threadLocalHashCode & (table.length - 1);
Entry e = table[i];
// Do not return null
if(e ! =null && e.get() == key)
return e;
else
//1. It is possible that the current entry is stale
//2. A hash conflict exists
return getEntryAfterMiss(key, i, e);
}
// Iterate through the number group to find the corresponding Entry object
private Entry getEntryAfterMiss(ThreadLocal<? > key,int i, Entry e) {
Entry[] tab = table;
int len = tab.length;
while(e ! =null) { ThreadLocal<? > k = e.get();if (k == key)
return e;
if (k == null)
expungeStaleEntry(i);
else
i = nextIndex(i, len);
e = tab[i];
}
return null;
}
Copy the codeThreadLocal global model
ThreadLocal and memory leaks
The act of causing a leak
-
Static ThreadLocal extends the lifetime of ThreadLocal
-
When using thread pools, where the current thread does not necessarily exit (such as a fixed-size thread pool), placing large objects in a ThreadLocal can cause memory leaks (when objects are no longer in use, they cannot be reclaimed because references exist).
The root cause of the memory leak
- ThreadLocal itself is not designed to cause memory leaks. The ThreadLocalMap object is held by the current Thread object and is null when the Thread exits.
private void exit(a) {
if(group ! =null) {
group.threadTerminated(this);
group = null;
}
/* Aggressively null out all reference fields: see bug 4006245 */
target = null;
/* Speed the release of some of these resources */
threadLocals = null;
inheritableThreadLocals = null;
inheritedAccessControlContext = null;
blocker = null;
uncaughtExceptionHandler = null;
}
Copy the code-
Cause: Because the key of an Entry is a weak reference, when no strong reference exists, the Entry object and its corresponding value will not be reclaimed when the JVM GC. When an Entry and value are not reclaimed, strong references to the Entry will persist unless the current thread dies, resulting in memory leaks.
-
Suggestion: Call the remove method when you want to reclaim an object.
/** * Remove the entry for key. */
private void remove(ThreadLocal
key) {
Entry[] tab = table;
int len = tab.length;
int i = key.threadLocalHashCode & (len-1);
for(Entry e = tab[i]; e ! =null;
e = tab[i = nextIndex(i, len)]) {
if (e.get() == key) {
e.clear();
expungeStaleEntry(i);
return; }}}Copy the codeThreadLocal summary
-
ThreadLocal does not solve the problem of sharing data between threads
-
ThreadLocal avoids instance thread-safety problems by implicitly creating separate instance copies in different threads
-
Each thread holds a Map and maintains the mapping of ThreadLocal objects to specific instances. Since the Map is only accessed by the thread holding it, there are no thread-safety or locking issues
-
The Entry of a ThreadLocalMap refers to a ThreadLocal as a weak reference, avoiding the problem that ThreadLocal objects cannot be reclaimed
-
ThreadLocalMap’s set method prevents memory leaks by calling the replaceStaleEntry method to reclaim the value of an Entry object with a null key (that is, a concrete instance) as well as the Entry object itself
-
ThreadLocal is suitable for scenarios where variables are isolated between threads and shared between methods
MessageQueue
An overview of the
MessageQueue internally maintains a one-way linked list of messages. The Handler sends the Message to the Message queue, which picks up the Message to execute according to certain rules.
When Looper is created, the associated MessageQueu object is created
Message
Message is a linked list, and all messages in a MessageQueue form a one-way linked list based on time. A pool of recycled objects is maintained internally for reuse.
Get the message
public static Message obtain(a) {
// Ensure thread safety
synchronized (sPoolSync) {
// Get messages from the recycle pool first
if(sPool ! =null) {
Message m = sPool;
sPool = m.next;
m.next = null;
// Clear the usage flag
m.flags = 0; // clear in-use flag
sPoolSize--;
returnm; }}return new Message();
}
Copy the codeRecycling MSG
/** * Recycles a Message that may be in-use. * Used internally by the MessageQueue and Looper when disposing of queued Messages. */
void recycleUnchecked(a) {
// Mark the message as in use while it remains in the recycled object pool.
// Clear out all other details.
flags = FLAG_IN_USE;
what = 0;
arg1 = 0;
arg2 = 0;
obj = null;
replyTo = null;
sendingUid = -1;
when = 0;
target = null;
callback = null;
data = null;
synchronized (sPoolSync) {
// Add to the recycle pool
if (sPoolSize < MAX_POOL_SIZE) {
next = sPool;
sPool = this; sPoolSize++; }}}Copy the codeHow to use Handler
The next step is to thread through the process by sending a message using a Handler
How to send messages
Let’s start with synchronous message barriers.
The synchronization message barrier passes
// Publish a synchronization barrier to Looper's message queue. Message processing continues as usual until the message queue hits a published synchronization barrier. When a barrier is encountered, synchronization messages later in the queue are paused (preventing execution) until the barrier is released by calling removeSyncBarrier and specifying the token that identifies the synchronization barrier. This method is used to immediately delay the execution of all subsequent published synchronous messages until the conditions for releasing the barrier are met. Asynchronous messages (see message. isAsynchronous waivers barriers and continues processing as usual. This call must always match the removeSyncBarrier call with the same token to ensure that the message queue resumes normal operation. Otherwise the application may hang
public int postSyncBarrier(a) {
return postSyncBarrier(SystemClock.uptimeMillis());
}
Copy the code private int postSyncBarrier(long when) {
// Enqueue a new sync barrier token.
// We don't need to wake the queue because the purpose of a barrier is to stall it.
synchronized (this) {
final int token = mNextBarrierToken++;
final Message msg = Message.obtain();
msg.markInUse();
msg.when = when;
msg.arg1 = token;
Message prev = null;
Message p = mMessages;
if(when ! =0) {
// find the insertion position
while(p ! =null&& p.when <= when) { prev = p; p = p.next; }}// Insert message queue
if(prev ! =null) { // invariant: p == prev.next
msg.next = p;
prev.next = msg;
} else {
msg.next = p;
mMessages = msg;
}
returntoken; }}Copy the codeAfter inserting the message queue, you can see that the synchronization barrier target is NULL. For example, in view rule PL, before you go through the drawing process
void scheduleTraversals(a) {
if(! mTraversalScheduled) { mTraversalScheduled =true;
mTraversalBarrier = mHandler.getLooper().getQueue().postSyncBarrier();
mChoreographer.postCallback(
Choreographer.CALLBACK_TRAVERSAL, mTraversalRunnable, null);
if (!mUnbufferedInputDispatch) {
scheduleConsumeBatchedInput();
}
notifyRendererOfFramePending();
pokeDrawLockIfNeeded();
}
}
Copy the codeA message barrier is put in first, so that the message drawn on the interface will be executed before other messages, and the drawing message will be blocked due to too many messages in MessageQueue and the picture will be stuck. When the drawing is completed, the message barrier will be removed
A synchronous message
A synchronous message using send (sendMessageAtFrontOfQueue, sendMessageAtTime, etc.) or post (postAtFrontOfQueue, postAtTime, etc.), will call
sendMessageAtTime
public boolean sendMessageAtTime(Message msg, long uptimeMillis) {
MessageQueue queue = mQueue;
if (queue == null) {
RuntimeException e = new RuntimeException(
this + " sendMessageAtTime() called with no mQueue");
Log.w("Looper", e.getMessage(), e);
return false;
}
return enqueueMessage(queue, msg, uptimeMillis);
}
Copy the codeHandler.enqueueMessage
private boolean enqueueMessage(MessageQueue queue, Message msg, long uptimeMillis) {
// MSG is bound to handler
msg.target = this;
Asynchronous messages are not affected by the synchronization barrier
if (mAsynchronous) {
msg.setAsynchronous(true);
}
return queue.enqueueMessage(msg, uptimeMillis);
}
Copy the codeMessageQueue.enqueueMessage
boolean enqueueMessage(Message msg, long when) {
MSG target = postSyncBarrier(); MSG target = postSyncBarrier()
if (msg.target == null) {
throw new IllegalArgumentException("Message must have a target.");
}
// Check the MSG flag bit, because the MSG should be joining the queue, meaning that the MSG flag bit is not used. If the display has been used, obviously there is a problem, throw the exception directly
if (msg.isInUse()) {
throw new IllegalStateException(msg + " This message is already in use.");
}
/ / synchronization locks
synchronized (this) {
// If the message queue is exiting, reclaim the message
if (mQuitting) {
IllegalStateException e = new IllegalStateException(
msg.target + " sending message to a Handler on a dead thread");
Log.w(TAG, e.getMessage(), e);
msg.recycle();
return false;
}
// Set MSG when and change the flag bit of MSG
msg.markInUse();
msg.when = when;
Message p = mMessages;
// Whether to wake up
boolean needWake;
//p==null indicates that there is no message in the message queue. When == 0 inserts to the head of the queue; When < p.hen indicates that MSG is executed earlier than the header element in the linked list
if (p == null || when == 0 || when < p.when) {
// New head, wake up the event queue if blocked.
msg.next = p;
mMessages = msg;
If the message queue is blocked, it needs to wake up
needWake = mBlocked;
} else {
// Find the insertion position in the message queue
// Inserted within the middle of the queue. Usually we don't have to wake
// up the event queue unless there is a barrier at the head of the queue
// and the message is the earliest asynchronous message in the queue.
// If the queue is currently blocked and the head of the queue is a synchronization barrier and the current message is asynchronous, wake up is required
needWake = mBlocked && p.target == null && msg.isAsynchronous();
Message prev;
for (;;) {
prev = p;
p = p.next;
// Get to the end of the queue or find the insertion position according to when and break the loop
if (p == null || when < p.when) {
break;
}
if (needWake && p.isAsynchronous()) {
needWake = false; }}// Insert message queue
msg.next = p; // invariant: p == prev.next
prev.next = msg;
}
// We can assume mPtr ! = 0 because mQuitting is false.
// If you need to wake up, you need to wake up
if(needWake) { nativeWake(mPtr); }}// Join the team successfully
return true;
}
Copy the codeMessage extraction
Messages are fetched mainly through looper.loop (), which is an endless loop that iterates through the message queue, dispatches execution messages after retrieving them, and finally reclaims them.
public static void loop(a) {
/ / for stars
final Looper me = myLooper();
if (me == null) {
throw new RuntimeException("No Looper; Looper.prepare() wasn't called on this thread.");
}
final MessageQueue queue = me.mQueue;
// Make sure the identity of this thread is that of the local process,
// and keep track of what that identity token actually is.
Binder.clearCallingIdentity();
final long ident = Binder.clearCallingIdentity();
for (;;) {
Get the MSG object from the message queue
Message msg = queue.next(); // might block
if (msg == null) {
// No message indicates that the message queue is quitting.
return;
}
// Distribute the message
msg.target.dispatchMessage(msg);
// Message collectionmsg.recycleUnchecked(); }}Copy the codeLook at the next() method, which iterates through the message queue, fetching available messages from the message queue. The nativePollOnce method called at 1 inside the for loop, which is related to the Handler’s blocking wake mechanism, which we’ll talk about later.
The second is by retrieving subsequent asynchronous messages through the synchronization barrier. At 3 points, determine whether the message has reached the time to be sent. If it has reached the time to be sent, directly send MSG return; otherwise, just set nextPollTimeoutMillis to the remaining time. Execute idleHanlder at 4 if the message queue is not null or has not reached the time to send, and after execution, avoid multiple executions by setting pendingIdleHandlerCount to 0.
Message next(a) {
// 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
// which is not supported.
final long ptr = mPtr;
if (ptr == 0) {
return null;
}
// The number of idlerHandlers to process when idle
int pendingIdleHandlerCount = -1; // -1 only during first iteration
// The native layer uses variables related to the blocking welcome mechanism, which will be discussed later
int nextPollTimeoutMillis = 0;
for (;;) {
if(nextPollTimeoutMillis ! =0) {
Binder.flushPendingCommands();
}
//1. Call native layer for message marking,nextPollTimeoutMillis 0 immediately return, -1 block wait.
nativePollOnce(ptr, nextPollTimeoutMillis); //android_os_MessageQueue.cpp L188
/ / synchronization locks
synchronized (this) {
// Try to retrieve the next message. Return if found.
final long now = SystemClock.uptimeMillis();
Message prevMsg = null;
Message msg = mMessages;
//2. Determine if Message is a synchronization barrier, if so, execute a loop to intercept all synchronous messages until the first asynchronous Message is retrieved
if(msg ! =null && msg.target == null) {
// Stalled by a barrier. Find the next asynchronous message in the queue.
do {
prevMsg = msg;
msg = msg.next;
} while(msg ! =null && !msg.isAsynchronous());
}
// There are executable messages
if(msg ! =null) {
/ / 3
if (now < msg.when) {
// Next message is not ready. Set a timeout to wake up when it is ready.
nextPollTimeoutMillis = (int) Math.min(msg.when - now, Integer.MAX_VALUE);
} else {
// Fetch the Message from the queue
// Got a message.
mBlocked = false;
if(prevMsg ! =null) {
prevMsg.next = msg.next;
} else {
mMessages = msg.next;
}
MSG next cannot point to any element in the list, so next must be null
msg.next = null;
if (DEBUG) Log.v(TAG, "Returning message: " + msg);
// mark the MSG state
msg.markInUse();
returnmsg; }}else {
// No more messages.
nextPollTimeoutMillis = -1;
}
// Process the quit message now that all pending messages have been handled.
if (mQuitting) {
dispose();
return null;
}
// If first time idle, then get the number of idlers to run.
// Idle handles only run if the queue is empty or if the first message
// in the queue (possibly a barrier) is due to be handled in the future.
/ / 4
if (pendingIdleHandlerCount < 0
&& (mMessages == null || now < mMessages.when)) {
pendingIdleHandlerCount = mIdleHandlers.size();
}
// There is no idleHandler to execute
if (pendingIdleHandlerCount <= 0) {
// No idle handlers to run. Loop and wait some more.
mBlocked = true;
continue;
}
if (mPendingIdleHandlers == null) {
mPendingIdleHandlers = new IdleHandler[Math.max(pendingIdleHandlerCount, 4)];
}
mPendingIdleHandlers = mIdleHandlers.toArray(mPendingIdleHandlers);
}
// Run the idle handlers.
// We only ever reach this code block during the first iteration.
for (int i = 0; i < pendingIdleHandlerCount; i++) {
final IdleHandler idler = mPendingIdleHandlers[i];
mPendingIdleHandlers[i] = null; // release the reference to the handler
boolean keep = false;
try {
keep = idler.queueIdle();
} catch (Throwable t) {
Log.wtf(TAG, "IdleHandler threw exception", t);
}
if(! keep) {synchronized (this) { mIdleHandlers.remove(idler); }}}// idleHandler only executes once until the next call to the messagequyue.next () method.
pendingIdleHandlerCount = 0;
// While calling an idle handler, a new message could have been delivered
// so go back and look again for a pending message without waiting.
nextPollTimeoutMillis = 0; }}Copy the codeDistribution of the message
public void dispatchMessage(Message msg) {
if(msg.callback ! =null) {
// If Message has a callback method, call msg.callback.run();
handleCallback(msg);
} else {
if(mCallback ! =null) {
// Call handleMessage() when the Handler has a Callback member variable;
if (mCallback.handleMessage(msg)) {
return; }}//Handler's own callback method handleMessage()handleMessage(msg); }}Copy the codeMessage removal
Remove a single message
void removeMessages(Handler h, Runnable r, Object object) {
if (h == null || r == null) {
return;
}
synchronized (this) {
Message p = mMessages;
// Remove all messages at front.
while(p ! =null && p.target == h && p.callback == r
&& (object == null || p.obj == object)) {
Message n = p.next;
mMessages = n;
p.recycleUnchecked();
p = n;
}
// Remove all messages after front.
while(p ! =null) {
Message n = p.next;
if(n ! =null) {
if (n.target == h && n.callback == r
&& (object == null || n.obj == object)) {
Message nn = n.next;
n.recycleUnchecked();
p.next = nn;
continue; } } p = n; }}}Copy the codeRemove all messages
private void removeAllFutureMessagesLocked(a) {
final long now = SystemClock.uptimeMillis();
Message p = mMessages;
if(p ! =null) {
if (p.when > now) {
removeAllMessagesLocked();
} else {
Message n;
for (;;) {
n = p.next;
if (n == null) {
return;
}
if (n.when > now) {
break;
}
p = n;
}
p.next = null;
do {
p = n;
n = p.next;
p.recycleUnchecked();
} while(n ! =null); }}}Copy the codeBlocking wake up mechanism
Handler blocks the wake up mechanism to avoid looping when there is no message to execute. When there is no message to execute, the loop process of Looper is blocked by the blocking wake up mechanism until the specified time is reached or woken up to avoid the waste of resources.
File descriptor
The file descriptor is formally a non-negative integer. In fact, it is an index value that points to the record table of open files that the kernel maintains for each process. When a program opens an existing file or creates a new file, the kernel returns a file descriptor to the process. In programming, some low-level programming tends to revolve around file descriptors. However, the concept of file descriptors is usually only applicable to operating systems such as UNIX and Linux.
epoll
Epoll is an IO multiplexing mechanism designed by the Linux kernel to handle large numbers of file descriptors. It can significantly improve the CPU utilization of a program with a large number of concurrent connections and only a small number of active connections. Epoll is so efficient because when it retrieves ready events, it does not go through the entire set of file descriptors that are listened on, but only those that have been asynchronously awakened by device IO events (the CPU interrupt mechanism) and added to the ready list.
Epoll has three main methods: epoll_create, epoll_ctl, and epoll_wait.
epoll_create
int epoll_create(int size);
Copy the codeCreate a handle to epoll. The size argument is the maximum number of file descriptors that the kernel is guaranteed to handle correctly. Note that when the epoll handle is created, it will occupy a fd value.
epoll_ctl
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
Copy the codePerform operations on the created epoll FD. For example, add the newly created socket FD to epoll for epoll to monitor, or remove a socket FD that epoll is monitoring from epoll. This function is used to register an event with epoll and specify the type of event to listen for.
The first argument is an epoll handle.
The second parameter indicates that the FD listens for the event operation.
-
EPOLL_CTL_ADD: Registers fd to EPFD
-
EPOLL_CTL_DEL: Deletes fd from EPFD
-
EPOLL_CTL_MOD: Modifies the listening event of a registered FD
The third parameter is the FD to listen on.
The fourth parameter tells the kernel what to listen for. Epoll_event is a structure in which events represent the actions of the corresponding file operator.
-
EPOLLIN: indicates that the corresponding file descriptor can be read (including that the peer SOCKET is normally closed).
-
EPOLLOUT: indicates that the corresponding file descriptor can be written.
-
EPOLLPRI: indicates that the corresponding file descriptor has urgent data to read (it should indicate that out-of-band data arrives).
-
EPOLLERR: indicates that an error occurs in the corresponding file descriptor.
-
EPOLLHUP: the corresponding file descriptor is hung up.
-
EPOLLET: Set EPOLL to Edge Triggered mode, as opposed to Level Triggered.
-
EPOLLONESHOT: monitors only one event. If you want to continue monitoring the socket after this event, you need to add the socket to the EPOLL queue again
epoll_wait
int epoll_wait(int epfd, struct epoll_event *events,int maxevents, int timeout);
Copy the codeThe first argument is an epoll handle.
The second parameter is the set of events obtained from the kernel.
The third parameter is the number of events, which cannot exceed the size of create.
The fourth parameter is timeout.
When this method is called, it blocks and waits for IO events on the EPFD. If one of the file descriptors that the EPFD listens for an event specified above occurs, a callback is performed, causing epoll to wake up and return the number of events that need to be processed. If the timeout is exceeded, it will also wake up and return 0 to avoid constant blocking.
nativeInit
When MessageQueue is created, the nativeInit() method is called to create the NativeMessageQueue in the native layer and return its address to the Java layer mPtr, and then communicate with the NativeMessageQueue through this address. When NativeMessageQueue is created, Looper under Native layer will be created.
android_os_MessageQueue.cpp
NativeMessageQueue::NativeMessageQueue() :
mPollEnv(NULL), mPollObj(NULL), mExceptionObj(NULL) {
// Similar to the Java layer, looper is first fetched from the current thread
mLooper = Looper::getForThread(a);if (mLooper == NULL) {
/ / to create stars
mLooper = new Looper(false); //Looper.cpp L63
// Store it in the current thread
Looper::setForThread(mLooper); }}Copy the codeLooper.cpp
Looper::Looper(bool allowNonCallbacks) :
mAllowNonCallbacks(allowNonCallbacks), mSendingMessage(false),
mPolling(false), mEpollFd(- 1), mEpollRebuildRequired(false),
mNextRequestSeq(0), mResponseIndex(0), mNextMessageUptime(LLONG_MAX) {
// Construct the fd of the wake up event
mWakeEventFd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
LOG_ALWAYS_FATAL_IF(mWakeEventFd < 0."Could not make wake event fd: %s".strerror(errno));
AutoMutex _l(mLock);
rebuildEpollLocked(a);//L134
}
Copy the codeThe wake up event FD is constructed first, and epoll is initialized in rebuildEpollLocked(). The wake up event FD is constructed first, and epoll is initialized in rebuildEpollLocked().
void Looper::rebuildEpollLocked(a) {
// Close old epoll instance if we have one.
if (mEpollFd >= 0) {
#if DEBUG_CALLBACKS
ALOGD("%p ~ rebuildEpollLocked - rebuilding epoll set".this);
#endif
// Close the old epoll instance
close(mEpollFd);
}
// Allocate the new epoll instance and register the wake pipe.
// Create an epoll instance
mEpollFd = epoll_create(EPOLL_SIZE_HINT);
LOG_ALWAYS_FATAL_IF(mEpollFd < 0."Could not create epoll instance: %s".strerror(errno));
struct epoll_event eventItem;
// Set the unused data area to 0
memset(& eventItem, 0.sizeof(epoll_event)); // zero out unused members of data field union
// Register EPOLLIN events
eventItem.events = EPOLLIN;
// Set fd to wake up event fd
eventItem.data.fd = mWakeEventFd;
// Add the wake event (mWakeEventFd) to the epoll instance (mEpollFd)
int result = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, mWakeEventFd, & eventItem);
LOG_ALWAYS_FATAL_IF(result ! =0."Could not add wake event fd to epoll instance: %s".strerror(errno));
// The main additions here are Input events, such as keyboard and sensor Input, which are basically the responsibility of the system.
for (size_t i = 0; i < mRequests.size(a); i++) {const Request& request = mRequests.valueAt(i);
struct epoll_event eventItem;
request.initEventItem(&eventItem);
int epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, request.fd, & eventItem);
if (epollResult < 0) {
ALOGE("Error adding epoll events for fd %d while rebuilding epoll set: %s",
request.fd, strerror(errno)); }}}Copy the codeThe old epoll descriptor was closed, the new epoll descriptor was created by calling epoll_CREATE, and after some initialization, fd in mWakeEventFd and mRequests were registered in the epoll descriptor. The registered events are EPOLLIN.
This means that when an IO occurs on one of these file descriptors, epoll_wait is notified to wake it up, and Handler blocking is done with epoll_wait.
nativePollOnce
In MessageQueue’s next method, the nativePollOnce method is called to block the returned procedure until the specified time. This is a native method that ends up calling looper. CPP’s pollOnce(timeoutMillis) method.
Looper.cpp
int Looper::pollOnce(int timeoutMillis, int* outFd, int* outEvents, void** outData) {
int result = 0;
for (;;) {
// Handle the Response event with no Callback first
while (mResponseIndex < mResponses.size()) {
const Response& response = mResponses.itemAt(mResponseIndex++);
int ident = response.request.ident;
if (ident >= 0) {
int fd = response.request.fd;
int events = response.events;
void* data = response.request.data;
#if DEBUG_POLL_AND_WAKE
ALOGD("%p ~ pollOnce - returning signalled identifier %d: "
"fd=%d, events=0x%x, data=%p".this, ident, fd, events, data);
#endif
if(outFd ! =NULL) *outFd = fd;
if(outEvents ! =NULL) *outEvents = events;
if(outData ! =NULL) *outData = data;
returnident; }}if(result ! =0) {
#if DEBUG_POLL_AND_WAKE
ALOGD("%p ~ pollOnce - returning result %d".this, result);
#endif
if(outFd ! =NULL) *outFd = 0;
if(outEvents ! =NULL) *outEvents = 0;
if(outData ! =NULL) *outData = NULL;
return result;
}
result = pollInner(timeoutMillis); }}int Looper::pollInner(int timeoutMillis) {
// ...
int result = POLL_WAKE;
mResponses.clear(a); mResponseIndex =0;
mPolling = true;
struct epoll_event eventItems[EPOLL_MAX_EVENTS];
/ / 1
int eventCount = epoll_wait(mEpollFd, eventItems, EPOLL_MAX_EVENTS, timeoutMillis);
// ...
return result;
}
Copy the codeWe end up calling the epoll_wait method, passing in the timeoutMillis we passed in the natviePollOnce method earlier. This is the core implementation of our blocking function. After calling this method, it will block until the time we set or several FDS registered in epoll have IO.
nativeWake
android_os_messageQueue.cpp
static void android_os_MessageQueue_nativeWake(JNIEnv* env, jclass clazz, jlong ptr) {
NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);
nativeMessageQueue->wake(a); }void NativeMessageQueue::wake(a) {
mLooper->wake(a);//Looper.cpp
}
Copy the codeThe Wake () method of the Native layer Looper is called.
Looper.cpp
void Looper::wake(a) {
#if DEBUG_POLL_AND_WAKE
ALOGD("%p ~ wake".this);
#endif
// Write character 1 to mWakeEventFd
uint64_t inc = 1;
ssize_t nWrite = TEMP_FAILURE_RETRY(write(mWakeEventFd, &inc, sizeof(uint64_t)));
if(nWrite ! =sizeof(uint64_t)) {
if(errno ! = EAGAIN) {LOG_ALWAYS_FATAL("Could not write wake signal to fd %d: %s",
mWakeEventFd, strerror(errno)); }}}Copy the codeThe write method is called and 1 is written to mWakeEventFd, causing the pollOnce method listening on the FD to wake up and the Next method in Java to continue execution.
Time to wake up
-
When the quit method of MessageQueue is called to exit, it is woken up
-
When a message is queued, it is inserted at the top of the list (earliest to be executed) or at the top of the asynchronous message (earliest to be executed) with a synchronization barrier
-
When a synchronous barrier is removed, if the message list is empty or there are no asynchronous messages behind the synchronous barrier
Dequeue message
void quit(boolean safe) {
if(! mQuitAllowed) {throw new IllegalStateException("Main thread not allowed to quit.");
}
synchronized (this) {
if (mQuitting) {
return;
}
mQuitting = true;
// If safe is true, all future messages are deleted based on the current time. If safe is false, all messages in the current message queue are deleted
if (safe) {
removeAllFutureMessagesLocked();
} else {
removeAllMessagesLocked();
}
// We can assume mPtr ! = 0 because mQuitting was previously false.nativeWake(mPtr); }}Copy the codeRefer to the link
Android Handler mechanism 2 ThreadLocal
ThreadLocal principle and Magic 0x61C88647
Epoll,
Synchronization barrier? Blocking wake up? Reread the Handler source with me
File descriptor