Linux eventfd分析

2017-07-20

eventfd在用户层下有函数

#include <sys/eventfd.h>

 int eventfd(unsigned int initval, int flags);

该函数返回一个文件描述符，类似于其他的文件描述符操作，可以对该描述符进行一系列的操作，如读、写、poll、select等，当然这里我们仅仅考虑read、write。看下该函数的内核实现

SYSCALL_DEFINE2(eventfd2, unsigned int, count, int, flags)
{
    int fd, error;
    struct file *file;
    error = get_unused_fd_flags(flags & EFD_SHARED_FCNTL_FLAGS);
    if (error < 0)
        return error;
    fd = error;
    file = eventfd_file_create(count, flags);
    if (IS_ERR(file)) {
        error = PTR_ERR(file);
        goto err_put_unused_fd;
    }
    fd_install(fd, file);
    return fd;
err_put_unused_fd:
    put_unused_fd(fd);
    return error;
}

代码本身很是简单，首先 获取一个空闲的文件描述符，这个和普通的文件描述符没有区别。然后调用eventfd_file_create创建了一个file结构。该函数中有针对eventfd的一系列操作，看下该函数

struct file *eventfd_file_create(unsigned int count, int flags)
{
    struct file *file;
    struct eventfd_ctx *ctx;

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

    if (flags & ~EFD_FLAGS_SET)
        return ERR_PTR(-EINVAL);

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

    kref_init(&ctx->kref);
    init_waitqueue_head(&ctx->wqh);
    ctx->count = count;
    ctx->flags = flags;

    file = anon_inode_getfile("[eventfd]", &eventfd_fops, ctx,
                  O_RDWR | (flags & EFD_SHARED_FCNTL_FLAGS));
    if (IS_ERR(file))
        eventfd_free_ctx(ctx);

    return file;
}

这里说明下，每个eventfd在内核对应一个eventfd_ctx结构，该结构后面咱们再细讲，函数中首先给该结构分配 了内存然后做初始化，注意有个等待队列和count，等待队列就是当进程需要阻塞的时候挂在对应evnetfd的等待队列上，而count就是read、write操作的值。接着就调用anon_inode_getfile获取一个file对象，具体也没什么好说的，只是注意这里把刚才分配好的eventfd_ctx作为file结构的私有成员即private_data，并且关联了eventfd自身的操作函数表eventfd_fops， 里面实现的函数不多，如下

static const struct file_operations eventfd_fops = {
#ifdef CONFIG_PROC_FS
    .show_fdinfo    = eventfd_show_fdinfo,
#endif
    .release    = eventfd_release,
    .poll        = eventfd_poll,
    .read        = eventfd_read,
    .write        = eventfd_write,
    .llseek        = noop_llseek,
};

我们重点看read和write函数。当用户空间对eventfd文件描述符发起read操作时，最终要调用到上面函数表中的eventfd_read函数，

static ssize_t eventfd_read(struct file *file, char __user *buf, size_t count,
                loff_t *ppos)
{
    struct eventfd_ctx *ctx = file->private_data;
    ssize_t res;
    __u64 cnt;
    if (count < sizeof(cnt))
        return -EINVAL;
    res = eventfd_ctx_read(ctx, file->f_flags & O_NONBLOCK, &cnt);
    if (res < 0)
        return res;
    return put_user(cnt, (__u64 __user *) buf) ? -EFAULT : sizeof(cnt);
}

首先从private_data获取eventfd_ctx，然后判断请求读取的大小是否满足条件，这里count是64位即8个字节，所以最小读取8个字节，如果不足则错误。没问题就调用eventfd_ctx_read，该函数实际返回eventfd_ctx中的count计数，并清零，如果读取有问题则返回，否则把值写入到用户空间。前面eventfd_ctx_read是读取的核心，什么时候会返回小于0的值呢，我们看下该函数的实现

ssize_t eventfd_ctx_read(struct eventfd_ctx *ctx, int no_wait, __u64 *cnt)
{
    ssize_t res;
    DECLARE_WAITQUEUE(wait, current);

    spin_lock_irq(&ctx->wqh.lock);
    *cnt = 0;
    res = -EAGAIN;
    if (ctx->count > 0)
        res = 0;
    else if (!no_wait) {
        /*add to wait queue*/
        __add_wait_queue(&ctx->wqh, &wait);
        for (;;) {
            /*设置阻塞状态*/
            set_current_state(TASK_INTERRUPTIBLE);
            /*如果信号变为有状态。则break*/
            if (ctx->count > 0) {
                res = 0;
                break;
            }
            /*如果有未处理的信号，也break，进行处理*/
            if (signal_pending(current)) {
                res = -ERESTARTSYS;
                break;
            }
            /*否则触发调度器执行调度*/
            spin_unlock_irq(&ctx->wqh.lock);
            schedule();
            spin_lock_irq(&ctx->wqh.lock);
        }
        /*remove from the wait queue*/
        __remove_wait_queue(&ctx->wqh, &wait);
        /*set processs state*/
        __set_current_state(TASK_RUNNING);
    }
    if (likely(res == 0)) {
        /*read fdcount again*/
        eventfd_ctx_do_read(ctx, cnt);
        /**/
        if (waitqueue_active(&ctx->wqh))
            wake_up_locked_poll(&ctx->wqh, POLLOUT);
    }
    spin_unlock_irq(&ctx->wqh.lock);

    return res;
}

该函数比较长，我们慢慢分析，首先操作eventfd_ctx要加锁保证安全。起初res初始化为-EAGAIN，如果count计数大于0，那么对res置0，否则意味着count=0（count不会小于0），这种情况下看传递进来的参数标志，如果设置了O_NONBLOCK，则就不需等待，直接返回res.这正是前面说的返回值小于0的情况。如果没有指定O_NONBLOCK标志，此时由于读取不到count值（count值为0），就会在这里阻塞。具体把当前进程加入到eventfd_ctx的等待队列，这里有必要说下DECLARE_WAITQUEUE(wait, current)，该宏声明并初始化一个wait_queue_t对象，其关联的函数为default_wake_function，是作为唤醒函数存在。OK，接下上面，加入到队列后进入一个死循环，设置当前进程状态为TASK_INTERRUPTIBLE，并不断检查count值，如果count大于0了，意味着有信号了，就设置res=0,然后break，然后把进程从等待队列去掉，然后设置状态TASK_RUNNING。如果count值为0，则检查是否有挂起的信号，如果有信号，同样需要先对信号进行处理，不过这就以为这read失败了。都么有的话就正常阻塞，调用调度器进行调度。break之后，如果res==0，对count值进行读取，这里对应上面循环中判断count值大于0的情况。具体读取通过eventfd_ctx_do_read函数，该函数很简单

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

如果没有指定EFD_SEMAPHORE标志就返回count值，该标志是指定eventfd像信号量一样使用，不过在2.6之后的内核都设置为0了。然后对count做减法，实际上减去之后就为0了。在读取值之后count值就变小了，之前如果有在该eventfd上阻塞的write进程，现在就可以唤醒了，所以这里检查了下，如果等待队列还有进程，则调用wake_up_locked_poll对对应的进程进行唤醒。

用户空间的write操作最终要调用到eventfd_write，不过该函数的实现和上面read操作类似，这里就不重复，感兴趣可以自行分析源码。前面说内核也可以主动的对eventfd发送信号，这里就是通过eventfd_signal函数实现

__u64 eventfd_signal(struct eventfd_ctx *ctx, __u64 n)
{
    unsigned long flags;

    spin_lock_irqsave(&ctx->wqh.lock, flags);
    if (ULLONG_MAX - ctx->count < n)
        n = ULLONG_MAX - ctx->count;
    ctx->count += n;
    /*mainly judge if wait is empty*/
    if (waitqueue_active(&ctx->wqh))
        wake_up_locked_poll(&ctx->wqh, POLLIN);
    spin_unlock_irqrestore(&ctx->wqh.lock, flags);

    return n;
}

该函数和write函数类似，不过不会阻塞，如果指定的n太大导致count加上之后超过ULLONG_MAX，就去n为当前count和ULLONG_MAX的差值，即不会让count溢出。然后如果等待队列有等待的进程，则对其进程唤醒，当然唤醒的应该是需要读操作的进程。

到这里对于eventfd的介绍基本就完成了，总的来说很简单的一个东西，不过经过上面分析不难发现，eventfd应该归结于低级通信行列，即不适用于传递大量数据，仅仅用于通知或者同步操作。

关于eventfd的使用方法，参考手册：https://linux.die.net/man/2/eventfd

以马内利

参考资料：

linux内核3.10.1源码