2017-07-20
eventfd在linux中是一個較新的進程通信方式，和信號量等不同的是event不僅可以用於進程間的通信，還可以用戶內核發信號給用戶層的進程。eventfd在virtIO后端驅動vHost的實現中作為vhost和KVM交互的媒介，起到了重大作用。本節結合linux源碼就eventfd的具體實現坐下簡要分析。
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源碼