1. Basic Concepts

epoll is an efficient I/O event notification mechanism provided by the Linux kernel. It is used to monitor multiple file descriptors and get notified immediately when I/O events occur on them. epoll is more efficient than traditional select and poll, especially suitable for handling large numbers of concurrent connections.

An epoll instance is created by epoll_create or epoll_create1. Its essentially a kernel object that tracks and manages events on a set of file descriptors. When an epoll instance is created, a file descriptor is returned, which is used for subsequent epoll operations.

2. File Descriptors

In epoll, file descriptors can be any file descriptors that support non-blocking I/O, such as sockets, pipes, files, etc.

These file descriptors can be added to an epoll instance using the epoll_ctl function to monitor events on them.

3. Events

epoll supports monitoring various events, such as read events (EPOLLIN), write events (EPOLLOUT), error events, etc.

Events are described by the struct epoll_event structure.

4. Basic Usage Flow of epoll

4.1 Creating an epoll Instance

Use epoll_create or epoll_create1 to create an epoll instance. The returned file descriptor identifies this instance.

int epfd = epoll_create1(EPOLL_CLOEXEC);
if (epfd < 0) {
    perror("epoll_create1");
    return -1;
}

4.2 Adding, Modifying, or Removing File Descriptors

Use the epoll_ctl function to add (EPOLL_CTL_ADD), modify (EPOLL_CTL_MOD), or remove (EPOLL_CTL_DEL) file descriptors to be monitored.

struct epoll_event ev;
ev.events = EPOLLIN;  // Monitor read events
ev.data.fd = fd;      // File descriptor to monitor

if (epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ev) < 0) {
    perror("epoll_ctl");
    return -1;
}

4.3 Waiting for Events

Use epoll_wait to wait for events on file descriptors. This function blocks until one or more events occur, or a timeout occurs.

struct epoll_event events[MAX_EVENTS];
int nfds = epoll_wait(epfd, events, MAX_EVENTS, -1); 
// Returns 0 if timeout, without any event on monitored fds
if (nfds < 0) {
    perror("epoll_wait");
    return -1;
}

for (int i = 0; i < nfds; ++i) {
    if (events[i].events & EPOLLIN) {
        // Handle read event
    }
    if (events[i].events & EPOLLOUT) {
        // Handle write event
    }
    // Handle other events, possibly errors
}

5. Code Analysis

The initialization function is defined as follows:

int nni_init(void)
{
    int rv;
    if ((rv = nni_plat_init(nni_init_helper)) != 0) {
        nng_log_err("NNG-INIT", "NNG library initialization failed: %s", nng_strerror(rv));
    }
    return rv;
}

The function discussed in this article is called within nni_plat_init:

// Line 396
if ((rv = nni_posix_pollq_sysinit()) != 0) {
    // ...
}

5.1 Functon Description

static int nni_posix_pollq_create(nni_posix_pollq *pq)

This function creates an event file descriptor and adds it to an epoll instance so that the program can use this file descriptor to trigger an epoll event when self-wakeup is needed. The main steps include creating the event file descriptor, setting file descriptor attributes, configuring the epoll event, and adding it to the epoll instance.

if ((pq->epfd = epoll_create1(EPOLL_CLOEXEC)) < 0) {
    return (nni_plat_errno(errno));
}

This code creates an epoll instance and stores the file descriptor in pq->epfd. The EPOLL_CLOEXEC flag indicates that the file descriptor will be automatically closed when an exec family function is called.

5.2 nni_posix_pollq_add_eventfd(pq)

This function is called within nni_posix_pollq_create. It performs the following operations:

1. Define an epoll_event variable ev.
2. Create an event file descriptor using eventfd with an initial value of 0 and the flag EFD_NONBLOCK (non-blocking).

if ((fd = eventfd(0, EFD_NONBLOCK)) < 0) {
    return (nni_plat_errno(errno));
}

3. Set file descriptor attributes:

(void) fcntl(fd, F_SETFD, FD_CLOEXEC);
(void) fcntl(fd, F_SETFL, O_NONBLOCK);

fcntl is used to set the FD_CLOEXEC flag, ensuring the file descriptor is closed after an exec family function is called. It is also used to set the file descriptor to non-blocking mode.

4. Configure the epoll event:

ev.events = EPOLLIN;
ev.data.ptr = 0;

Set the event type to EPOLLIN (read event). ev.data.ptr is set to 0 and can be used to store user data.

5. Add the event to the epoll instance:

if (epoll_ctl(pq->epfd, EPOLL_CTL_ADD, fd, &ev) != 0) {
    (void) close(fd);
    return (nni_plat_errno(errno));
}

6. Save the file descriptor and return success:

pq->evfd = fd;
return (0);

In summary, this function creates an event file descriptor and adds it to the epoll instance for self-wakeup purposes.

6. nni_thr_init(&pq->thr, nni_posix_poll_thr, pq) Initialization

First, let's look at the structure definitions:

struct nni_posix_pollq {
    nni_mtx  mtx;
    int      epfd;
    int      evfd;
    bool     close;
    nni_thr  thr;
    nni_list reqpq;
};

struct nni_thr {
    nni_plat_thr   thr;
    nni_plat_mtx   mtx;
    nni_plat_cv    cv;
    nni_thr_func   fn;
    void *         arg;
    int            start;
    int            stop;
    int            done;
    int            init;
};

struct nni_plat_thr {
    pthread_t  tid;
    void(*func)(void *);
    void *arg;
};

nni_thr_init initializes the thr member of the global nni_posix_global_pollq.

First-level initialization: Assign basic members of pq->thr.

• First parameter: pq->thr
• Second parameter: function pointer nni_posix_poll_thr
• Third parameter: pq

thr->done  = 0;
thr->start = 0;
thr->stop  = 0;
thr->fn    = fn;   // nni_posix_poll_thr
thr->arg   = arg;  // pq

After the first-level initialization, the values are:

pq->thr->done  = 0;
pq->thr->start = 0;
pq->thr->stop  = 0;
pq->thr->fn    = nni_posix_poll_thr;
pq->thr->arg   = pq;

Second-level initialization: Initialize the thr member of nni_thr, specifically pq->thr->thr, using nni_plat_thr_init:

Parameters:

1. thr->thr (i.e., pq->thr->thr) - the object to initialize
2. Function pointer nni_thr_wrap
3. thr (i.e., pq->thr)

// nni_plat_thr_init(&thr->thr, nni_thr_wrap, thr)
// Actual assignment:
thr->func = fn;  // pq->thr->thr->func = nni_thr_wrap
thr->arg  = arg; // pq->thr->thr->arg = pq->thr

After nni_plat_thr_init, if function pointers and arguments are non-null, the final values for pq->thr members are:

pq->thr->done     = 0;
pq->thr->start    = 0;
pq->thr->stop     = 0;
pq->thr->fn       = nni_posix_poll_thr;
pq->thr->arg      = pq;
pq->thr->thr->func = nni_thr_wrap;
pq->thr->thr->arg  = pq->thr;

Next, a thread is created using thread attributes set earlier in nni_init. The thread ID is pq->thr->thr->tid, the function to execute is nni_plat_thr_main, and the argument is pq->thr->thr.

In nni_plat_thr_main, the member function of the nni_plat_thr structure is executed:

thr->func(thr->arg)

This actually executes pq->thr->thr->fn, which is nni_thr_wrap, with the argument pq->thr.

Here is the nni_thr_wrap function:

static void
nni_thr_wrap(void *arg)
{
    nni_thr *thr = arg;
    int      start;

    nni_plat_mtx_lock(&thr->mtx);
    while (((start = thr->start) == 0) && (thr->stop == 0)) {
        nni_plat_cv_wait(&thr->cv);
    }
    nni_plat_mtx_unlock(&thr->mtx);
    if ((start) && (thr->fn != NULL)) {
        thr->fn(thr->arg);
    }
    nni_plat_mtx_lock(&thr->mtx);
    thr->done = 1;
    nni_plat_cv_wake(&thr->cv);
    nni_plat_mtx_unlock(&thr->mtx);
}

Since thr->start == 0 and thr->stop == 0 after initialization, the thread will wait on the condition variable thr->cv until the while condition becomes false.

while (((start = thr->start) == 0) && (thr->stop == 0)) {
    nni_plat_cv_wait(&thr->cv);
}

7. Setting the Thread Name

This step is omitted.

8. Executing nni_thr_run

In nni_thr_run, thr->start is set to 1, and the thread waiting on pq->thr->cv is woken up. The condition is then satisfied, and pq->thr->fn(), which is nni_posix_poll_thr, is executed with pq as the argument. The system then starts polling for readable events on the pq->epfd instance. Initialization is complete!

Here is the nni_posix_poll_thr function:

static void
nni_posix_poll_thr(void *arg)
{
    nni_posix_pollq *  pq = arg;
    struct epoll_event events[NNI_MAX_EPOLL_EVENTS];

    for (;;) {
        int  n;
        bool reap = false;

        n = epoll_wait(pq->epfd, events, NNI_MAX_EPOLL_EVENTS, -1);
        if ((n < 0) && (errno == EBADF)) {
            // Epoll fd closed, bail.
            return;
        }

        // dispatch events
        for (int i = 0; i < n; ++i) {
            const struct epoll_event *ev;

            ev = &events[i];
            // If the waker pipe was signaled, read from it.
            if ((ev->data.ptr == NULL) &&
                (ev->events & (unsigned) POLLIN)) {
                uint64_t clear;
                if (read(pq->evfd, &clear, sizeof(clear)) !=
                    sizeof(clear)) {
                    nni_panic("read from evfd incorrect!");
                }
                reap = true;
            } else {
                nni_posix_pfd *  pfd = ev->data.ptr;
                nni_posix_pfd_cb cb;
                void *           cbarg;
                unsigned         mask;

                mask = ev->events &
                    ((unsigned) EPOLLIN | (unsigned) EPOLLOUT |
                     (unsigned) EPOLLERR);

                nni_mtx_lock(&pfd->mtx);
                pfd->events &= ~mask;
                cb    = pfd->cb;
                cbarg = pfd->arg;
                nni_mtx_unlock(&pfd->mtx);

                // Execute the callback with lock released
                if (cb != NULL) {
                    cb(pfd, mask, cbarg);
                }
            }
        }

        if (reap) {
            nni_posix_pollq_reap(pq);
            nni_mtx_lock(&pq->mtx);
            if (pq->close) {
                nni_mtx_unlock(&pq->mtx);
                return;
            }
            nni_mtx_unlock(&pq->mtx);
        }
    }
}

Some implementation details, such as thread attributes, signal handling, and locking/unlocking operations, are omitted here.