Detailed explanation of Looper mechanism

preface

Previously, there was an article on JAVA layer Looper Introduction and MessageQueue Introduction of
There is also a very detailed article on the Internet introduce

Use a diagram to illustrate the mechanism of looper operation

As the name suggests, looper is polling. The purpose of polling is to find messages in the Message queue and process messages. MessageQueue and looper are in one-to-one correspondence. There is only one looper object and MessageQueue in a prepare d thread, and the thread is unique. The Message can be from the thread where looper is located or from other sub threads. Post a piece of code to explain

        HandlerThread handlerThread = new HandlerThread("test");
        handlerThread.start();
        Handler mainHandler = new Handler() {
            @Override
            public void handleMessage(Message msg) {
                super.handleMessage(msg);
                //Here is what runs in the main thread
                switch (msg.what) {
                    case 0:
                        //todo
                        break;
                    case 2:
                        //todo
                        break;
                }
            }
        };
        Handler workHandler = new Handler(handlerThread.getLooper()) {
            @Override
            public void handleMessage(Message msg) {
                super.handleMessage(msg);
                switch (msg.what) {
                    case 1:
                        //It runs in the sub thread. If it needs to be executed in the main thread, it needs to send a message message to MQ in the loop of the main thread
                        mainHandler.sendEmptyMessage(2);
                        break;
                }
            }
        };
        mainHandler.sendEmptyMessage(0);//The main thread sends a message message to MQ in the loop of the main thread
        workHandler.sendEmptyMessage(1);//The main thread sends a message message to MQ in the child thread looper

Polling underlying mechanism

In the previous article, I wrote that after MQ blocking, I didn't go deep into the cause of block. Here to solve my doubts. Before the process
Loop will be in queue Next() blocked

 public static void loop() {
        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;

        for (;;) {
            Message msg = queue.next(); // Light block blocking
            if (msg == null) {
                // No message indicates that the message queue is quitting.
                return;
            }
			msg.target.dispatchMessage(msg);
		}
}

nativePollOnce() in MessageQueue

Message next() {
    final long ptr = mPtr;
    ...

    int pendingIdleHandlerCount = -1; // -1 only during first iteration
    int nextPollTimeoutMillis = 0;
    for (;;) {
  		...
        //Blocking, unless timeout or wake-up occurs
        nativePollOnce(ptr, nextPollTimeoutMillis);
    }
    ...}

nativePollOnce blocking method: the blocking state will be released when the timeout (nextPollTimeoutMillis) or wake-up (nativeWake) is reached

  1. If nextPollTimeoutMillis is greater than or equal to zero, sleep will be specified during this period, and then wake up
  2. When the message queue is empty, nextPollTimeoutMillis is - 1, entering blocking; When a message enters the queue again, the nativeWake wake-up method will be triggered when it is inserted into the header node

In order to understand the principle of nativePollOnce blocking, let's first understand the Linux epoll mechanism

epoll mechanism

Here's an article article Not bad.

concept

epoll is an I/O event notification mechanism and an implementation of IO multiplexing in linux kernel.
IO multiplexing refers to monitoring multiple I / O sources at the same time in one operation, returning when one or more I / O sources are available, and then reading and writing their.
epoll's popular explanation is a mechanism that sends a readable signal to notify when the kernel buffer of the file descriptor is not empty, and sends a writable signal to notify when the write buffer is not full.

I/O

The objects of input / output can be files, sockets and pipes between processes. In linux system, it is represented by file descriptor (fd).

event

Readable event: when the kernel read buffer associated with the file descriptor is readable, a readable event is triggered.
(readable: the kernel buffer is not empty and data can be read)
Writable event: when the kernel write buffer associated with the file descriptor is writable, a writable event is triggered.
(writable: the kernel buffer is insufficient and there is free space to write)

graphic


epoll_create(): create and initialize the eventpoll structure ep, put ep into file - > private, and return fd. A red black tree and a ready list (which stores ready file descriptors) will be established in the high-speed cache area of the kernel
epoll_ctl(): register fd to epoll instance. Executing epoll_ During the add operation of CTL, not only the file descriptor is placed on the red black tree, but also the callback function is registered. When the kernel detects that a file descriptor is readable / writable, it will call the callback function, which puts the file descriptor in the ready list.
epoll_wait(): the application process is blocked on epoll. Setting the timeout to - 1 means that it will not return until there is a target event. epoll_wait only needs to observe whether there is data in the ready linked list. Finally, it returns the data of the linked list to the array and returns the ready quantity

JAVA layer Looper

Waiting for message

Message next() {
    final long ptr = mPtr;
    ...

    int pendingIdleHandlerCount = -1; // -1 only during first iteration
    int nextPollTimeoutMillis = 0;
    for (;;) {
  		...
        //Blocking, unless the timeout or wake-up occurs, will eventually call epoll_wait() method, waiting for the message
        nativePollOnce(ptr, nextPollTimeoutMillis);
    }
    ...}

Production message

 boolean enqueueMessage(Message msg, long when) {
        if (msg.target == null) {
            throw new IllegalArgumentException("Message must have a target.");
        }

        synchronized (this) {
			//balabala
            // We can assume mPtr != 0 because mQuitting is false.
            if (needWake) {
            	//nativewake will execute the write() operation to write messages
                nativeWake(mPtr);
            }
        }
        return true;
    }

C + + layer Looper

Looper and messagequeue objects are created when the thread prepare s
Looper is an STL object, unique to the thread, and MessageQueue is a member variable of looper

	//Create Looper
    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));
    }
    //Create MQ
    private Looper(boolean quitAllowed) {
        mQueue = new MessageQueue(quitAllowed);
        mThread = Thread.currentThread();
    }
    MessageQueue(boolean quitAllowed) {
      mQuitAllowed = quitAllowed;
      mPtr = nativeInit();    
    }

android_os_MessageQueue.cpp

static jlong android_os_MessageQueue_nativeInit(JNIEnv* env, jclass clazz) {
	//New NativeMQ
    NativeMessageQueue* nativeMessageQueue = new NativeMessageQueue();
    if (!nativeMessageQueue) {
        jniThrowRuntimeException(env, "Unable to allocate native queue");
        return 0;
    }

    nativeMessageQueue->incStrong(env);
    return reinterpret_cast<jlong>(nativeMessageQueue);
}

NativeMessageQueue::NativeMessageQueue() :
        mPollEnv(NULL), mPollObj(NULL), mExceptionObj(NULL) {
    mLooper = Looper::getForThread();
    if (mLooper == NULL) {
    	//New Looper
        mLooper = new Looper(false);
        Looper::setForThread(mLooper);
    }
}

Looper::Looper(bool allowNonCallbacks)
    : mAllowNonCallbacks(allowNonCallbacks),
      mSendingMessage(false),
      mPolling(false),
      mEpollRebuildRequired(false),
      mNextRequestSeq(WAKE_EVENT_FD_SEQ + 1),
      mResponseIndex(0),
      mNextMessageUptime(LLONG_MAX) {
    mWakeEventFd.reset(eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC));
    LOG_ALWAYS_FATAL_IF(mWakeEventFd.get() < 0, "Could not make wake event fd: %s", strerror(errno));

    AutoMutex _l(mLock);
    //epoll operation
    rebuildEpollLocked();
}

void Looper::rebuildEpollLocked() {
    // Close old epoll instance if we have one.
    if (mEpollFd >= 0) {
#if DEBUG_CALLBACKS
        ALOGD("%p ~ rebuildEpollLocked - rebuilding epoll set", this);
#endif
        mEpollFd.reset();
    }

    // Allocate the new epoll instance and register the WakeEventFd.
    //Create epoll instance
    mEpollFd.reset(epoll_create1(EPOLL_CLOEXEC));
    LOG_ALWAYS_FATAL_IF(mEpollFd < 0, "Could not create epoll instance: %s", strerror(errno));

    epoll_event wakeEvent = createEpollEvent(EPOLLIN, WAKE_EVENT_FD_SEQ);
    //Monitor fd
    int result = epoll_ctl(mEpollFd.get(), EPOLL_CTL_ADD, mWakeEventFd.get(), &wakeEvent);
    LOG_ALWAYS_FATAL_IF(result != 0, "Could not add wake event fd to epoll instance: %s",
                        strerror(errno));

    for (const auto& [seq, request] : mRequests) {
        epoll_event eventItem = createEpollEvent(request.getEpollEvents(), seq);

        int epollResult = epoll_ctl(mEpollFd.get(), 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));
        }
    }
}

Keywords: Android andorid

Added by VladSun on Wed, 23 Feb 2022 15:56:15 +0200