NIO-EPollSelectorIpml源碼分析

本文轉載自查看原文 2020-03-02 13:52 1978 java/ nio/ linux/ epoll

NIO-EPollSelectorIpml源碼分析

NIO-EPollSelectorIpml源碼分析

前言

本來是想學習Netty的，但是Netty是一個NIO框架，因此在學習netty之前，還是先梳理一下NIO的知識。通過剖析源碼理解NIO的設計原理。

本系列文章針對的是JDK1.8.0.161的源碼。

NIO-Selector源碼分析對Selector的功能和創建過程進行了分析，本篇對Linux環境下JDK實現的EPollSelectorImpl源碼進行詳細講解。

本篇文章不會對EPoll算法進行詳細介紹，對epoll算法感興趣或還不了解的同學可以看epoll原理詳解及epoll反應堆模型先進行學習。

在詳細介紹EpollSelectorProvider之前我們先了解一下EPoll主要的三個步驟：

調用epoll_create建立一個epoll 對象(在epoll文件系統中給這個句柄分配資源)；
調用epoll_ctl向epoll對象中添加或刪除文件句柄及監控事件。
調用epoll_wait收集發生事件的文件描述符。

初始化EPollSelectorProvider

NIO-Selector源碼分析提到，若沒有進行配置時，默認通過sun.nio.ch.DefaultSelectorProvider.create()創建SelectorProvider。Linux下的代碼路徑在jdk\src\solaris\classes\sun\nio\ch\DefaultSelectorProvider.java。在其內部通過實際是創建了一個EPollSelectorProvider。

創建EPollSelectorImpl

EPollSelectorProvider是用於創建EPollSelectorImpl的。

Selector.Open()->
SelectorProvider.provider()->
sun.nio.ch.DefaultSelectorProvider.create()->
EPollSelectorProvider.openSelector()->
new EPollSelectorImpl(this)
public class EPollSelectorProvider extends SelectorProviderImpl
{
    public AbstractSelector openSelector() throws IOException {
        return new EPollSelectorImpl(this);
    }

    public Channel inheritedChannel() throws IOException {
        return InheritedChannel.getChannel();
    }
}

inheritedChannel()可以返回系統默認SelectorProvider創建的通道，主要有些操作系統底層需要調用默認的通道。

EPollSelectorImpl結構

在詳細講解EPollSelectorImpl源碼之前，先了解EPollSelectorImpl的主要的數據結構和屬性。

名稱	作用
Map<Integer,SelectionKeyImpl> fdToKey	保存文件描述符句柄和的SelectionKey的映射關系
int fd0	管道的讀端文件描述符
int fd1	管道的寫端文件描述符
EPollArrayWrapper pollWrapper	調用底層Epoll算法的包裝類


 EPollSelectorImpl(SelectorProvider sp) throws IOException {
    super(sp);
    long pipeFds = IOUtil.makePipe(false);
    fd0 = (int) (pipeFds >>> 32); //無符號移位
    fd1 = (int) pipeFds;
    pollWrapper = new EPollArrayWrapper();
    pollWrapper.initInterrupt(fd0, fd1);
    fdToKey = new HashMap<>();
}
void initInterrupt(int fd0, int fd1) {
    outgoingInterruptFD = fd1;
    incomingInterruptFD = fd0;
    //將管道的讀取端注冊
    epollCtl(epfd, EPOLL_CTL_ADD, fd0, EPOLLIN);
}

pipeFds高32位存放的是通道read端的文件描述符FD，低32位存放的是write端的文件描述符。這里做移位處理。

通過調用JNI的makePipe方法創建單向管道。

JNIEXPORT jlong JNICALL
Java_sun_nio_ch_IOUtil_makePipe(JNIEnv *env, jobject this, jboolean blocking)
{
    int fd[2];

    if (pipe(fd) < 0) {
        JNU_ThrowIOExceptionWithLastError(env, "Pipe failed");
        return 0;
    }
    if (blocking == JNI_FALSE) {
        //配置阻塞
        if ((configureBlocking(fd[0], JNI_FALSE) < 0)
            || (configureBlocking(fd[1], JNI_FALSE) < 0)) {
            JNU_ThrowIOExceptionWithLastError(env, "Configure blocking failed");
            close(fd[0]);
            close(fd[1]);
            return 0;
        }
    }
    //高32位存讀端，低32位存寫端
    return ((jlong) fd[0] << 32) | (jlong) fd[1];
}

JNI內部則通過pipe創建管道。

對於管道的詳細邏輯可以看《Linux管道 - 系統調用pipe()函數實現》

fdToKey

在注冊時會將文件描述符的句柄和對應的SelectionKey保存到Map<Integer,SelectionKeyImpl> fdToKey中

管道文件描述符

在EPollSelectorImpl創建的時候會使用IOUtil.makePipe(false)調用創建一個管道，用於喚醒線程用。當線程中斷時通過向寫管道寫入一個字節來喚醒線程，具體可以看doSelect邏輯。

EPollArrayWrapper

PollArrayWrapper用於存放linux的epoll_event結構。

 typedef union epoll_data {
     void *ptr;
     int fd;
     __uint32_t u32;
     __uint64_t u64;
  } epoll_data_t;

 struct epoll_event {
     __uint32_t events;
     epoll_data_t data;
 };

創建EPoll文件描述符

在EPollArrayWrapper創建時候會創建epoll文件描述符和epoll_event數組結構

EPollArrayWrapper() throws IOException {
    // creates the epoll file descriptor
    epfd = epollCreate();

    // the epoll_event array passed to epoll_wait
    int allocationSize = NUM_EPOLLEVENTS * SIZE_EPOLLEVENT;
    pollArray = new AllocatedNativeObject(allocationSize, true);
    pollArrayAddress = pollArray.address();

    // eventHigh needed when using file descriptors > 64k
    if (OPEN_MAX > MAX_UPDATE_ARRAY_SIZE)
        eventsHigh = new HashMap<>();
}
//最大不超過64K
private static final int MAX_UPDATE_ARRAY_SIZE = AccessController.doPrivileged(
        new GetIntegerAction("sun.nio.ch.maxUpdateArraySize", Math.min(OPEN_MAX, 64*1024)));

EPollArrayWrapper內部會為維護兩個結構，當句柄值小於MAX_UPDATE_ARRAY_SIZE時會保存到數組結構中。否則會存儲到Map中。主要是優化效率。

private static final int MAX_UPDATE_ARRAY_SIZE = AccessController.doPrivileged(
        new GetIntegerAction("sun.nio.ch.maxUpdateArraySize", Math.min(OPEN_MAX, 64*1024)));

通過epollCreate方法創建epoll文件描述符,JNI調用底層的epoll_create方法。傳入的參數位最大注冊的socket fd數量。

JNIEXPORT jint JNICALL
Java_sun_nio_ch_EPoll_epollCreate(JNIEnv *env, jclass c) {
    /*
     * epoll_create expects a size as a hint to the kernel about how to
     * dimension internal structures. We can't predict the size in advance.
     */
    int epfd = epoll_create(256);
    if (epfd < 0) {
       JNU_ThrowIOExceptionWithLastError(env, "epoll_create failed");
    }
    return epfd;
}

epoll_create用於創建EPoll事件所需的內存空間，默認為256，在Linux 2.6.8以后，傳入的size就沒用了，底層會動態調整所需數據結構的大小。詳情可以看下epoll_create的方法描述

初始化epoll_event數組

epfd創建完后，創建epoll_event的數組，首先查詢epoll_event結構的大小

private static final int SIZE_EPOLLEVENT  = sizeofEPollEvent();
Java_sun_nio_ch_EPollArrayWrapper_sizeofEPollEvent(JNIEnv* env, jclass this)
{
    return sizeof(struct epoll_event);
}

查詢配置的文件描述符最大數量

private static final int OPEN_MAX         = IOUtil.fdLimit();
private static final int NUM_EPOLLEVENTS  = Math.min(OPEN_MAX, 8192);
Java_sun_nio_ch_IOUtil_fdLimit(JNIEnv *env, jclass this)
{
    struct rlimit rlp;
    if (getrlimit(RLIMIT_NOFILE, &rlp) < 0) {
        JNU_ThrowIOExceptionWithLastError(env, "getrlimit failed");
        return -1;
    }
    if (rlp.rlim_max < 0 || rlp.rlim_max > java_lang_Integer_MAX_VALUE) {
        return java_lang_Integer_MAX_VALUE;
    } else {
        return (jint)rlp.rlim_max;
    }
}

getrlimit用於獲取資源使用限制，RLIMIT_NOFILE獲取最大文件打開數量。對於getrlimit詳細介紹可以看一下
Linux系統調用--getrlimit()與setrlimit()函數詳解

根據查詢到的epoll_event結構大小和數量初始化數組大小。

int allocationSize = NUM_EPOLLEVENTS * SIZE_EPOLLEVENT;
pollArray = new AllocatedNativeObject(allocationSize, true);

在 EPollArrayWrapper 內部使用 AllocatedNativeObject對象創建的堆外(native)內存對象。
將數組的首地址保存到pollArrayAddress中，在調用epollWait的時候需要傳遞該參數給JNI。

和Windows的PollArrayWrapper一樣，EPollArrayWrapper也暴露了讀寫FD和Event的方法供EPollSelectorImpl使用。

void putEventOps(int i, int event) {
    int offset = SIZE_EPOLLEVENT * i + EVENT_OFFSET;
    pollArray.putInt(offset, event);
}

void putDescriptor(int i, int fd) {
    int offset = SIZE_EPOLLEVENT * i + FD_OFFSET;
    pollArray.putInt(offset, fd);
}

int getEventOps(int i) {
    int offset = SIZE_EPOLLEVENT * i + EVENT_OFFSET;
    return pollArray.getInt(offset);
}

int getDescriptor(int i) {
    int offset = SIZE_EPOLLEVENT * i + FD_OFFSET;
    return pollArray.getInt(offset);
}

注冊

protected void implRegister(SelectionKeyImpl ski) {
    if (closed)
        throw new ClosedSelectorException();
    SelChImpl ch = ski.channel;
    //獲取通道的句柄
    int fd = Integer.valueOf(ch.getFDVal());
    //加入到緩存中
    fdToKey.put(fd, ski);
    //加入到數組緩存
    pollWrapper.add(fd);
    keys.add(ski);
}

在注冊的時候會將SelectionKey加入到fdToKey和keys，同時會將文件描述符加入到pollWrapper

pollWrapper.add(fd);
void add(int fd) {
    // force the initial update events to 0 as it may be KILLED by a
    // previous registration.
    synchronized (updateLock) {
        assert !registered.get(fd);
        //初始化事件掩碼為0
        setUpdateEvents(fd, (byte)0, true);
    }
}
private void setUpdateEvents(int fd, byte events, boolean force) {
    //小於MAX_UPDATE_ARRAY_SIZE存到數組中
    if (fd < MAX_UPDATE_ARRAY_SIZE) {
        if ((eventsLow[fd] != KILLED) || force) {
            eventsLow[fd] = events;
        }
    } else {
        //大於MAX_UPDATE_ARRAY_SIZE存到map中
        Integer key = Integer.valueOf(fd);
        if (!isEventsHighKilled(key) || force) {
            eventsHigh.put(key, Byte.valueOf(events));
        }
    }
}
private boolean isEventsHighKilled(Integer key) {
    assert key >= MAX_UPDATE_ARRAY_SIZE;
    Byte value = eventsHigh.get(key);
    return (value != null && value == KILLED);
}

若文件描述符的值為KILLED(-1)時，該管道被釋放。不再加入。如上面所述，這里會根據key的大小存放到mapeventsHigh或字節數組eventsLow中。

在調用poll的時候才會調用epollCtl進行注冊。

int poll(long timeout) throws IOException {
    //更新epoll事件，實際調用`epollCtl`加入到epollfd中
    updateRegistrations();
    ...
}
private void updateRegistrations() {
    synchronized (updateLock) {
        int j = 0;
        while (j < updateCount) {
            int fd = updateDescriptors[j];
            short events = getUpdateEvents(fd);
            boolean isRegistered = registered.get(fd);
            int opcode = 0;

            if (events != KILLED) {
                //已經注冊過
                if (isRegistered) {
                    //修改或刪除
                    opcode = (events != 0) ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
                } else {
                    //新增
                    opcode = (events != 0) ? EPOLL_CTL_ADD : 0;
                }
                if (opcode != 0) {
                    epollCtl(epfd, opcode, fd, events);
                    if (opcode == EPOLL_CTL_ADD) {
                        //增加到registered緩存是否已注冊
                        registered.set(fd);
                    } else if (opcode == EPOLL_CTL_DEL) {
                        registered.clear(fd);
                    }
                }
            }
            j++;
        }
        updateCount = 0;
    }
}
private byte getUpdateEvents(int fd) {
    if (fd < MAX_UPDATE_ARRAY_SIZE) {
        return eventsLow[fd];
    } else {
        Byte result = eventsHigh.get(Integer.valueOf(fd));
        // result should never be null
        return result.byteValue();
    }
}

doSelect


protected int doSelect(long timeout) throws IOException {
    if (closed)
        throw new ClosedSelectorException();
    //1. 刪除取消的key
    processDeregisterQueue();
    try {
        begin();
        //2. 獲取就緒文件描述符
        pollWrapper.poll(timeout);
    } finally {
        end();
    }
    //3. 再次刪除取消的key
    processDeregisterQueue();
    //4. 將就緒的key加入到selectedKeys中
    int numKeysUpdated = updateSelectedKeys();
    //5. 若管道被喚醒清理喚醒的數據
    if (pollWrapper.interrupted()) {
        // Clear the wakeup pipe
        pollWrapper.putEventOps(pollWrapper.interruptedIndex(), 0);
        synchronized (interruptLock) {
            pollWrapper.clearInterrupted();
            IOUtil.drain(fd0);
            interruptTriggered = false;
        }
    }
    return numKeysUpdated;
}

刪除取消的key,當channel關閉時，對應的Key會被取消,被取消的key會加入到cancelledKeys中。調用processDeregisterQueue遍歷所有的key進行卸載。


processDeregisterQueue();
//遍歷所有已取消的key，取消他們
void processDeregisterQueue() throws IOException {
    // Precondition: Synchronized on this, keys, and selectedKeys
    Set<SelectionKey> cks = cancelledKeys();
    //遍歷每個key調用卸載
    implDereg(ski);
}
protected void implDereg(SelectionKeyImpl ski) throws IOException {
    assert (ski.getIndex() >= 0);
    SelChImpl ch = ski.channel;
    int fd = ch.getFDVal();
    //根據文件句柄值移除
    fdToKey.remove(Integer.valueOf(fd));
    //從堆外內存溢出epoll_event結構
    pollWrapper.remove(fd);
    ski.setIndex(-1);
    keys.remove(ski);
    selectedKeys.remove(ski);
    //將key設置為無效
    deregister((AbstractSelectionKey)ski);
    SelectableChannel selch = ski.channel();
    if (!selch.isOpen() && !selch.isRegistered())
        ((SelChImpl)selch).kill();
}

從pollWrapper移除，會將句柄值設置為KILLED(-1)

pollWrapper.remove(fd);
void remove(int fd) {
        synchronized (updateLock) {
            //設置實現值為-1 取消
            setUpdateEvents(fd, KILLED, false);
            // remove from epoll
            if (registered.get(fd)) {
                //從epool對象中刪除
                epollCtl(epfd, EPOLL_CTL_DEL, fd, 0);
                registered.clear(fd);
            }
        }
    }

EPOLL_CTL_DEL操作符將文件描述符從epoll fd中移除。

獲取就緒文件描述符

通過調用epollWait方法，獲取到已就緒的文件描述符，存放在pollArrayAddress地址中。

pollWrapper.poll(timeout);
int poll(long timeout) throws IOException {
    //更新epoll事件，實際調用`epollCtl`加入到epollfd中
    updateRegistrations();
    //獲取已就緒的文件句柄
    updated = epollWait(pollArrayAddress, NUM_EPOLLEVENTS, timeout, epfd);
    //如是喚醒文件句柄，則跳過，設置interrupted=true
    for (int i=0; i<updated; i++) {
        if (getDescriptor(i) == incomingInterruptFD) {
            interruptedIndex = i;
            interrupted = true;
            break;
        }
    }
    return updated;
}

再次嘗試刪除取消的key。epollWait阻塞的時候可能會有channel被關閉，因此需要再次調用刪除取消key。
將就緒的key加入到selectedKeys中

private int updateSelectedKeys() {
        int entries = pollWrapper.updated;
        int numKeysUpdated = 0;
        for (int i=0; i<entries; i++) {
            int nextFD = pollWrapper.getDescriptor(i);
            SelectionKeyImpl ski = fdToKey.get(Integer.valueOf(nextFD));
            // ski is null in the case of an interrupt
            if (ski != null) {
                int rOps = pollWrapper.getEventOps(i);
                if (selectedKeys.contains(ski)) {
                    if (ski.channel.translateAndSetReadyOps(rOps, ski)) {
                        numKeysUpdated++;
                    }
                } else {
                    ski.channel.translateAndSetReadyOps(rOps, ski);
                    if ((ski.nioReadyOps() & ski.nioInterestOps()) != 0) {
                        //加入到selectedKeys中
                        selectedKeys.add(ski);
                        numKeysUpdated++;
                    }
                }
            }
        }
        return numKeysUpdated;
    }

當epollWait是被管道喚醒時，則將管道數據都讀取出來以清除管道數據

EPoll有水平喚醒觸發和邊緣觸發兩種觸發模式，水平觸發有數據可讀，若不讀取完，下次調用poll時會一致被喚醒。而邊緣觸發則觸發一次后不處理，下次除非有新的事件到來否則不會再喚醒。邊緣觸發性能更好。這里必須將管道數據全部讀取完才行，避免設置為水平觸發時管道一值喚醒。

當線程中斷時，會調用wakeup喚醒，向管道中寫入一個字節數據使其讀事件就緒被喚醒。在前面的文章提到過線程中斷接口。

public Selector wakeup() {
    synchronized (interruptLock) {
        if (!interruptTriggered) {
            pollWrapper.interrupt();
            interruptTriggered = true;
        }
    }
    return this;
}
JNIEXPORT void JNICALL
Java_sun_nio_ch_EPollArrayWrapper_interrupt(JNIEnv *env, jobject this, jint fd)
{
    int fakebuf[1];
    fakebuf[0] = 1;
    if (write(fd, fakebuf, 1) < 0) {
        JNU_ThrowIOExceptionWithLastError(env,"write to interrupt fd failed");
    }
}

清理喚醒管道數據，將數據讀出來。

IOUtil.drain(fd0);
JNIEXPORT jboolean JNICALL
Java_sun_nio_ch_IOUtil_drain(JNIEnv *env, jclass cl, jint fd)
{
    char buf[128];
    int tn = 0;

    for (;;) {
        int n = read(fd, buf, sizeof(buf));
        tn += n;
        if ((n < 0) && (errno != EAGAIN))
            JNU_ThrowIOExceptionWithLastError(env, "Drain");
        if (n == (int)sizeof(buf))
            continue;
        return (tn > 0) ? JNI_TRUE : JNI_FALSE;
    }
}

關閉EpollSelectorImpl

關閉EpollSelectorImpl時會將所有注冊的通道一同關閉


    protected void implClose() throws IOException {
        if (closed)
            return;
        closed = true;

        // prevent further wakeup
        synchronized (interruptLock) {
            interruptTriggered = true;
        }
        //關閉管道文件描述符
        FileDispatcherImpl.closeIntFD(fd0);
        FileDispatcherImpl.closeIntFD(fd1);
        //關閉epoll fd，並釋放堆外內存。
        pollWrapper.closeEPollFD();
        // it is possible
        selectedKeys = null;

        // 情理所有通道
        Iterator<SelectionKey> i = keys.iterator();
        while (i.hasNext()) {
            SelectionKeyImpl ski = (SelectionKeyImpl)i.next();
            SelectableChannel selch = ski.channel();
            if (!selch.isOpen() && !selch.isRegistered())
                ((SelChImpl)selch).kill();
            i.remove();
        }

        fd0 = -1;
        fd1 = -1;
    }
pollWrapper.closeEPollFD();
void closeEPollFD() throws IOException {
    //關閉epfd
    FileDispatcherImpl.closeIntFD(epfd);
    //釋放堆外內存
    pollArray.free();
}

總結

本文對EPollSelectorImpl的代碼實現進行詳細解析。相比WindowsSelectorImpl的select模型而言，因為沒有最大文件描述符的限制，因此也無需調用poll多次。通過簡單的調用JNI方法輕易的實現了高性能的I/O模型。

至此，本系列NIO源碼分析章節已經結束。通過9篇文章對NIO的各個塊的源碼進行分析，為后續對Netty的源碼分析打下基礎。