eventfd(2) Combined with seelect(2) Source Code Analysis

The code examples are from Linux kernel 4.17

eventfd(2) - Creates a file descriptor for event notification.

• Usage
• Source Code Analysis
• References

#include <sys/eventfd.h>

int eventfd(unsigned int initval, int flags);

int eventfd2(unsigned int initval, int flags);

Parameters
- \initval  Initial value (associated with the internal count)
- \flags    This parameter was invalid before kernel 2.6.26 and must be specified as 0

Meaningful flags include:
- EFD_CLOEXEC, equivalent to O_CLOEXEC
- EFD_NONBLOCK, equivalent to O_NONBLOCK
- EFD_SEMAPHORE, semaphore option affecting read(2) value


Return
- Returns a new file descriptor on success, -1 on failure with errno set

eventfd acts as a simple abstract file, where each file descriptor corresponds to a __u64 count maintained in kernel space. All file operations related to eventfd are tied to this counter.

Supported file operations:

• read(2), reads a value that reduces the count. If EFD_SEMAPHORE is set in flags, count -= 1; otherwise, count -= count. Returns 8 on success.
• write(2), writes a cnt, count += cnt, returns 8 on success.
• poll(2), poll operation, core of event notification, detailed below.
• close(2), decrements the reference count of the eventfd structure. If it reaches 0, releases the allocated memory.

Usage

The core of eventfd(2) is its poll operation. The most common usage is combining it with select(2)/poll(2)/epoll(2) for inter-thread communication.

#include <poll.h>
#include <unistd.h>
#include <stdio.h>
#include <sys/eventfd.h>
#include <pthread.h>

int event_fd;

void *writer_thread(void *arg) {
    uint64_t value = 1;
    while (1) {
        printf("Writing value: %zu\n", value);
        write(event_fd, &value, sizeof(value));
        value++;
        sleep(2);
    }
}

int main() {
    struct pollfd poll_fds;
    pthread_t thread_id;

    unsigned int initial_value = 1000;  // Observe the output order when changing to 0
    int flags = 0;
    int timeout_ms = 1000;

//     flags |= EFD_SEMAPHORE; // Uncomment to observe different output
    event_fd = eventfd(initial_value, flags);
    poll_fds.fd = event_fd;
    poll_fds.events |= POLLIN;

    pthread_create(&thread_id, NULL, writer_thread, NULL);

    while (1) {
        int result = poll(&poll_fds, 1, timeout_ms);
        if (result > 0) {
            uint64_t read_value;
            read(event_fd, &read_value, sizeof(read_value));
            printf("Read value: %zu\n", read_value);
        }
    }
}

Read value: 1000
Writing value: 1
Read value: 1
Writing value: 2
Read value: 2
Writing value: 3
Read value: 3
Writing value: 4
Read value: 4

This example demonstrates simple inter-thread communication. The child thread writes incrementing unsigned long integers to eventfd every second. The main thread uses poll(2) to detect readiness and reads the reduced counter value from kernel space.

The parameters for read/write system calls follow:

int read(int, void *, size_t);

Since eventfd maintains an internal counter, the second and third parameters should be uint64_t * and sizeof(uint64_t) respectively.

Implementation

eventfd(2) implementation is located in fs/eventfd.c

The directory structure reveals eventfd is implemented as a file. The code is concise (under 500 lines) and easy to understand. Studying eventfd provides insight into kernel driver logic.

struct eventfd_ctx

struct eventfd_ctx is the structure maintained in kernel space for eventfd, lightweight and simple.

struct eventfd_ctx {
        struct kref kref;  // Reference count, memory freed when reaches 0
        wait_queue_head_t wqh;  // Wait queue head
        /*
         * Every write(2) operation adds the written __u64 value to "count" and wakes up "wqh".
         * A read(2) returns the "count" value to userspace and resets "count" to zero.
         * The kernel-side eventfd_signal() also adds to the "count" counter and issues a wakeup.
         */
        __u64 count;  // Counter closely related to file operations
        unsigned int flags;  // Various flags
};

eventfd(2)

This system call creates a new file descriptor, initializes the kernel-space counter, and initializes the wait queue head. Subsequent read/write operations will add themselves to this wait queue.

static int do_eventfd(unsigned int count, int flags)
{
        struct eventfd_ctx *ctx;
        int fd;

        /* Check EFD_* constants for consistency */
        BUILD_BUG_ON(EFD_CLOEXEC != O_CLOEXEC);
        BUILD_BUG_ON(EFD_NONBLOCK != O_NONBLOCK);

        // Flags must be among EFD_CLOEXEC, EFD_NONBLOCK, EFD_SEMAPHORE
        if (flags & ~EFD_FLAGS_SET)
                return -EINVAL;

        ctx = kmalloc(sizeof(*ctx), GFP_KERNEL);
        if (!ctx)
                return -ENOMEM;

        kref_init(&ctx->kref);  // Initialize reference count to 1
        init_waitqueue_head(&ctx->wqh);  // Initialize wait queue head
        ctx->count = count;  // Initialize counter
        ctx->flags = flags;  // Set flags

        // Create new file descriptor and set eventfd file operations
        fd = anon_inode_getfd("[eventfd]", &eventfd_fops, ctx,
                              O_RDWR | (flags & EFD_SHARED_FCNTL_FLAGS));
        if (fd < 0)
                eventfd_free_ctx(ctx);

        return fd;
}

eventfd_fops is the file operations structure for eventfd, registered in the file's f_op structure.

static const struct file_operations eventfd_fops = {
        .release        = eventfd_release,  // File close operation
        .poll           = eventfd_poll,    // File poll operation
        .read           = eventfd_read,    // Read
        .write          = eventfd_write,   // Write
        .llseek         = noop_llseek,
};

eventfd_read(2), read(2), eventfd_write(2), write(2)

static ssize_t eventfd_read(struct file *file, char __user *buf, size_t count,
                            loff_t *ppos)
{
        struct eventfd_ctx *ctx = file->private_data;  // Retrieve eventfd structure from file private data
        ssize_t res;
        __u64 ucnt = 0;
        DECLARE_WAITQUEUE(wait, current);  // Declare a wait queue entry

        if (count < sizeof(ucnt))  // Must be able to hold a u64
                return -EINVAL;

        spin_lock_irq(&ctx->wqh.lock);
        res = -EAGAIN;  // Initial EAGAIN for non-blocking mode when unreadable
        if (ctx->count > 0)  // Counter > 0 means readable, return 8
                res = sizeof(ucnt);
        else if (!(file->f_flags & O_NONBLOCK)) {  // count = 0 and not non-blocking
                __add_wait_queue(&ctx->wqh, &wait);  // Add wait entry to queue
                for (;;) {
                        set_current_state(TASK_INTERRUPTIBLE);  // Set task state to interruptible
                        if (ctx->count > 0) {  // Counter > 0, exit loop
                                res = sizeof(ucnt);
                                break;
                        }
                        if (signal_pending(current)) {  // Signal pending, exit loop
                                res = -ERESTARTSYS;
                                break;
                        }
                        spin_unlock_irq(&ctx->wqh.lock);
                        schedule();  // Schedule
                        spin_lock_irq(&ctx->wqh.lock);
                }
                __remove_wait_queue(&ctx->wqh, &wait);  // Remove wait entry
                __set_current_state(TASK_RUNNING);  // Set task to running
        }
        if (likely(res > 0)) {
                eventfd_ctx_do_read(ctx, &ucnt);  // Read based on flags
                if (waitqueue_active(&ctx->wqh))
                        wake_up_locked_poll(&ctx->wqh, EPOLLOUT);  // Wake up threads
        }
        spin_unlock_irq(&ctx->wqh.lock);

        if (res > 0 && put_user(ucnt, (__u64 __user *)buf))  // Copy reduced count to userspace
                return -EFAULT;

        return res;
}


static void eventfd_ctx_do_read(struct eventfd_ctx *ctx, __u64 *cnt)
{
        *cnt = (ctx->flags & EFD_SEMAPHORE) ? 1 : ctx->count;  // EFD_SEMAPHORE reads 1
        ctx->count -= *cnt;
}

static ssize_t eventfd_write(struct file *file, const char __user *buf, size_t count,
                             loff_t *ppos)
{
        struct eventfd_ctx *ctx = file->private_data;
        ssize_t res;
        __u64 ucnt;
        DECLARE_WAITQUEUE(wait, current);

        if (count < sizeof(ucnt))
                return -EINVAL;
        if (copy_from_user(&ucnt, buf, sizeof(ucnt)))  // Copy 8 bytes from userspace
                return -EFAULT;
        if (ucnt == ULLONG_MAX)  // Max count value
                return -EINVAL;
        spin_lock_irq(&ctx->wqh.lock);
        res = -EAGAIN;  // Initial EAGAIN for non-blocking mode when unwritable
        if (ULLONG_MAX - ctx->count > ucnt)  // Can write
                res = sizeof(ucnt);
        else if (!(file->f_flags & O_NONBLOCK)) {  // Cannot write and not non-blocking
                __add_wait_queue(&ctx->wqh, &wait);  // Add to wait queue
                for (res = 0;;) {  // Clear EAGAIN
                        set_current_state(TASK_INTERRUPTIBLE);  // Set task to interruptible
                        if (ULLONG_MAX - ctx->count > ucnt) {  // Can write, set return value
                                res = sizeof(ucnt);
                                break;
                        }
                        if (signal_pending(current)) {  // Signal pending
                                res = -ERESTARTSYS;
                                break;
                        }
                        spin_unlock_irq(&ctx->wqh.lock);
                        schedule();  // Schedule
                        spin_lock_irq(&ctx->wqh.lock);
                }
                __remove_wait_queue(&ctx->wqh, &wait);  // Remove from wait queue
                __set_current_state(TASK_RUNNING);  // Set task to running
        }
        if (likely(res > 0)) {
                ctx->count += ucnt;  // Increment counter
                if (waitqueue_active(&ctx->wqh))
                        wake_up_locked_poll(&ctx->wqh, EPOLLIN);  // Wake up threads
        }
        spin_unlock_irq(&ctx->wqh.lock);

        return res;
}

Besides parameter validation, the handling of blocking mode differs slightly before the loop: write(2) sets res from -EAGAIN to 0, while read(2) doesn't modify it. However, both achieve the same effect as res will be set before returning from blocking mode.

During each blocking operation, read(2)/write(2) add themselves to the internal wait queue via __add_wait_queue(). After the counter becomes available, a wakeup occurs by traversing the wait queue and waking threads.

poll

static __poll_t eventfd_poll(struct file *file, poll_table *wait)
{
        struct eventfd_ctx *ctx = file->private_data;
        __poll_t events = 0;
        u64 count;

        poll_wait(file, &ctx->wqh, wait);  // Key function when combined with select

        // Critical section access comments

        count = READ_ONCE(ctx->count);

        if (count > 0)  // Readable when count > 0
                events |= EPOLLIN;
        if (count == ULLONG_MAX)  // Counter at max, error
                events |= EPOLLERR;
        if (ULLONG_MAX - 1 > count)  // Writable
                events |= EPOLLOUT;

        return events;
}

The poll implementation is straightforward, returning events based on the counter value.

In select(2)/poll(2), the file's f_op->poll() corresponds to eventfd_poll(). The select(2) implementation (do_select() → poll_initwait()) sets pt->_qproc to __pollwait(). During the select loop, each file descriptor's poll method is called, which for eventfd is eventfd_poll().

eventfd_poll() calls poll_wait() → __pollwait(), which sets the queue entry's callback to pollwake() and adds it to the file's wait queue, returning the readiness mask.

When read(2)/write(2) occurs, before returning, wake_up_locked_poll() traverses the file's wait queue, executes the queue entry's callback (pollwake() for select(2)), and wakes the thread.