Linux 内核之旅（十三）：epoll

void epoll_server_run(){   
    //....
	char buf[BUF_SIZE];
	struct sockaddr_in srv_addr;
	struct sockaddr_in cli_addr;
	struct epoll_event events[MAX_EVENTS];

	listen_sock = socket(AF_INET, SOCK_STREAM, 0);

	set_sockaddr(&srv_addr);
	bind(listen_sock, (struct sockaddr *)&srv_addr, sizeof(srv_addr));

	//注意需要设置为NON-BLOCK
	setnonblocking(listen_sock);
	listen(listen_sock, MAX_CONN);
    
    //1.   创建epoll fd 并对listener fd加入监听
	epfd = epoll_create(1);
	epoll_ctl_add(epfd, listen_sock, EPOLLIN | EPOLLOUT | EPOLLET);

	socklen = sizeof(cli_addr);
	for (;;) {
        //2.   调用epoll_wait并等待返回就绪事件
		nfds = epoll_wait(epfd, events, MAX_EVENTS, -1);
        //3.   处理就绪事件（读、写、退出等），要区分是否为listener fd或者accept fd
		for (i = 0; i < nfds; i++) {
			if (events[i].data.fd == listen_sock) {
				/* handle new connection */
				// 若为listenfd，在ET模式下
				// 用 while 循环包裹住 accept 调用，处理完 TCP 就绪队列中的所有连接后再退出循环
				// 如何知道是否处理完就绪队列中的所有连接呢？ 
				// accept  返回 -1 并且 errno 设置为 EAGAIN 就表示所有连接都处理完
				// 当然读/写在ET模式下也是一样
				//读：只要可读, 就一直读, 直到返回 0, 或者 errno = EAGAIN 
				//写：只要可写, 就一直写, 直到数据发送完, 或者 errno = EAGAIN 
				while(1){
					conn_sock =
						accept(listen_sock,
						(struct sockaddr *)&cli_addr,
						&socklen);
					if (conn_sock == -1) {
						if ((errno == EAGAIN) || (errno == EWOULDBLOCK)) {
							break;  // We have processed all incoming connections.
						} else {
							perror("accept");
							break;
						}
					}

					inet_ntop(AF_INET, (char *)&(cli_addr.sin_addr),
						buf, sizeof(cli_addr));
					printf("[+] connected with %s:%d\n", buf,
						ntohs(cli_addr.sin_port));

					setnonblocking(conn_sock);
					epoll_ctl_add(epfd, conn_sock,
							EPOLLIN | EPOLLET | EPOLLRDHUP |
							EPOLLHUP);
					//check epoll_ctl_add return value
				}
			} else if (events[i].events & EPOLLIN) {
				/* handle EPOLLIN event */
				for (;;) {
					bzero(buf, sizeof(buf));
					n = read(events[i].data.fd, buf,
						 sizeof(buf));
					if (n <= 0 /* || errno == EAGAIN */ ) {
						break;
					} else {
						printf("[+] data: %s\n", buf);
						write(events[i].data.fd, buf,
						      strlen(buf));
					}
				}
			} else {
				printf("[+] unexpected\n");
			}
			/* check if the connection is closing */
			if (events[i].events & (EPOLLRDHUP | EPOLLHUP)) {
				printf("[+] connection closed\n");
                // 将fd移除epoll监听
				epoll_ctl(epfd, EPOLL_CTL_DEL,
					  events[i].data.fd, NULL);
				close(events[i].data.fd);
				continue;
			}
		}
	}
}

struct socket_wq {
	/* Note: wait MUST be first field of socket_wq */
	wait_queue_head_t	wait;
	......
};

// https://elixir.bootlin.com/linux/v4.11.6/source/include/net/sock.h#L1822
static inline wait_queue_head_t *sk_sleep(struct sock *sk)
{
    BUILD_BUG_ON(offsetof(struct socket_wq, wait) != 0);
    return &rcu_dereference_raw(sk->sk_wq)->wait;
}

//https://elixir.bootlin.com/linux/v4.11.6/source/fs/eventpoll.c#L1793
SYSCALL_DEFINE1(epoll_create1, int, flags)
{
	int error, fd;
	struct eventpoll *ep = NULL;
	struct file *file;
    // ...

    //1. 创建一个 eventpoll 对象
	error = ep_alloc(&ep);
	if (error < 0)
		return error;

	fd = get_unused_fd_flags(O_RDWR | (flags & O_CLOEXEC));
	if (fd < 0) {
		error = fd;
		goto out_free_ep;
	}
	file = anon_inode_getfile("[eventpoll]", &eventpoll_fops, ep,
				 O_RDWR | (flags & O_CLOEXEC));
	if (IS_ERR(file)) {
		error = PTR_ERR(file);
		goto out_free_fd;
	}
	ep->file = file;
    //安装fd
	fd_install(fd, file);
	return fd;
    //....
}

struct eventpoll {
	/* Protect the access to this structure */
	spinlock_t lock;

	/*
	 * This mutex is used to ensure that files are not removed
	 * while epoll is using them. This is held during the event
	 * collection loop, the file cleanup path, the epoll file exit
	 * code and the ctl operations.
	 */
	struct mutex mtx;

	/* Wait queue used by sys_epoll_wait() */
	wait_queue_head_t wq;

	/* Wait queue used by file->poll() */
	wait_queue_head_t poll_wait;

	/* List of ready file descriptors */
	struct list_head rdllist;

	/* RB tree root used to store monitored fd structs */
	struct rb_root rbr;

	/*
	 * This is a single linked list that chains all the "struct epitem" that
	 * happened while transferring ready events to userspace w/out
	 * holding ->lock.
	 */
	struct epitem *ovflist;

	/* wakeup_source used when ep_scan_ready_list is running */
	struct wakeup_source *ws;

	/* The user that created the eventpoll descriptor */
	struct user_struct *user;

	struct file *file;

	/* used to optimize loop detection check */
	int visited;
	struct list_head visited_list_link;
};

static int ep_alloc(struct eventpoll **pep)
{
	int error;
	struct user_struct *user;
	struct eventpoll *ep;

	user = get_current_user();
	error = -ENOMEM;

    //1.  申请 epollevent 内存
	ep = kzalloc(sizeof(*ep), GFP_KERNEL);
	if (unlikely(!ep))
		goto free_uid;

	spin_lock_init(&ep->lock);
	mutex_init(&ep->mtx);

    //2. 初始化等待队列头
	init_waitqueue_head(&ep->wq);
	init_waitqueue_head(&ep->poll_wait);

    //3. 初始化就绪列表
	INIT_LIST_HEAD(&ep->rdllist);

    //4. 初始化红黑树指针
	ep->rbr = RB_ROOT;
	ep->ovflist = EP_UNACTIVE_PTR;
	ep->user = user;

	*pep = ep;

	return 0;

free_uid:
	free_uid(user);
	return error;
}

SYSCALL_DEFINE4(accept4, int, fd, struct sockaddr __user *, upeer_sockaddr,
        int __user *, upeer_addrlen, int, flags)
{
    struct socket *sock, *newsock;

    //根据 fd 查找到监听的 socket
    sock = sockfd_lookup_light(fd, &err, &fput_needed);

    //1、申请并初始化新的 socket
    newsock = sock_alloc();

	//2、把 listen 状态的 socket 对象（关联AF_INET）上的协议操作函数集合 ops等赋值给新的 socket
    newsock->type = sock->type;
    newsock->ops = sock->ops;

    //3、 申请新的 file 对象，并设置到新 socket 上
	// sock_alloc_file最终会完成 sock->file = file
	// 
    newfile = sock_alloc_file(newsock, flags, sock->sk->sk_prot_creator->name);
    ......

    //4、 接收连接
    err = sock->ops->accept(sock, newsock, sock->file->f_flags);

    //5、 添加新文件到当前进程的打开文件列表
    fd_install(newfd, newfile);

static const struct file_operations socket_file_ops = {
	.llseek =	no_llseek,
	.read_iter =	sock_read_iter,
	.write_iter =	sock_write_iter,
	.poll =		sock_poll,		//记住这个sock_poll函数
	.unlocked_ioctl = sock_ioctl,
	.mmap =		sock_mmap,
	.release =	sock_close,
	.fasync =	sock_fasync,
	.sendpage =	sock_sendpage,
	.splice_write = generic_splice_sendpage,
	.splice_read =	sock_splice_read,
};

struct sock *inet_csk_accept(struct sock *sk, int flags, int *err, bool kern)
{
    struct inet_connection_sock *icsk = inet_csk(sk);
    struct request_sock_queue *queue = &icsk->icsk_accept_queue;
    struct request_sock *req;
    struct sock *newsk;

	// ......

    /* 从已完成连接队列中移除 */
    req = reqsk_queue_remove(queue, sk);

    /* 设置新控制块指针 */
	/* sk为 struct sock *sk 类型*/
    newsk = req->sk;

	// ......

    /* 返回找到的连接控制块 */
    return newsk;
}

// https://elixir.bootlin.com/linux/v4.11.6/source/net/core/sock.c#L2460
void sock_init_data(struct socket *sock, struct sock *sk)
{
    // 关键：把 sock->wq 挂到 sk->sk_wq，从此 sk_sleep(sk) 与 socket 等待队列一一对应
    sk->sk_wq           =   sock ? &sock->wq : NULL;

    // 注册四类默认回调（v4.11.6 中均为单参签名）
    sk->sk_state_change =   sock_def_wakeup;        // 状态变化（如 tcp_fin、tcp_done）
    sk->sk_data_ready   =   sock_def_readable;      // 数据可读（receive_queue 有新数据）
    sk->sk_write_space  =   sock_def_write_space;   // 发送空间可用
    sk->sk_error_report =   sock_def_error_report;  // 错误上报（如 tcp_reset）
}

//https://elixir.bootlin.com/linux/v4.11.6/source/fs/eventpoll.c#L141
struct epitem {
	union {
		/* RB tree node links this structure to the eventpoll RB tree */
		struct rb_node rbn;
		/* Used to free the struct epitem */
		struct rcu_head rcu;
	};

	/* List header used to link this structure to the eventpoll ready list */
	struct list_head rdllink;

	/*
	 * Works together "struct eventpoll"->ovflist in keeping the
	 * single linked chain of items.
	 */
	struct epitem *next;

	/* The file descriptor information this item refers to */
	struct epoll_filefd ffd;

	/* Number of active wait queue attached to poll operations */
	int nwait;

	/* List containing poll wait queues */
	struct list_head pwqlist;

	/* The "container" of this item */
	struct eventpoll *ep;

	/* List header used to link this item to the "struct file" items list */
	struct list_head fllink;

	/* wakeup_source used when EPOLLWAKEUP is set */
	struct wakeup_source __rcu *ws;

	/* The structure that describe the interested events and the source fd */
	struct epoll_event event;
};

//https://elixir.bootlin.com/linux/v4.11.6/source/fs/eventpoll.c#L230
struct eppoll_entry {
	/* List header used to link this structure to the "struct epitem" */
	struct list_head llink;

	/* The "base" pointer is set to the container "struct epitem" */
	struct epitem *base;

	/*
	 * Wait queue item that will be linked to the target file wait
	 * queue head.
	 */
	wait_queue_t wait;

	/* The wait queue head that linked the "wait" wait queue item */
	wait_queue_head_t *whead;
};

typedef struct poll_table_struct {
	poll_queue_proc _qproc;
	unsigned long _key;
} poll_table;

/* Wrapper struct used by poll queueing */
struct ep_pqueue {
	poll_table pt;
	struct epitem *epi;
};

// epoll_ctl的参数包括两个fd
SYSCALL_DEFINE4(epoll_ctl, int, epfd, int, op, int, fd,
        struct epoll_event __user *, event)
{
    struct eventpoll *ep;
    struct file *file, *tfile;

    //根据 epfd 找到 eventpoll 内核对象
    file = fget(epfd);
    ep = file->private_data;	// 又一次看到了private_data

    //根据 socket 句柄号， 找到其 file 内核对象
    tfile = fget(fd);

    switch (op) {
    case EPOLL_CTL_ADD:
        if (!epi) {
            epds.events |= POLLERR | POLLHUP;
			// for add
            error = ep_insert(ep, &epds, tfile, fd);
        } else
            error = -EEXIST;
        clear_tfile_check_list();
        break;

    //.....
}

static int ep_insert(struct eventpoll *ep, 
                struct epoll_event *event,
                struct file *tfile, int fd)
{
    //1、分配并初始化 epitem
    struct epitem *epi;	    //分配一个epi对象
    if (!(epi = kmem_cache_alloc(epi_cache, GFP_KERNEL)))
        return -ENOMEM;

    //对分配的epi进行初始化
    //epi->ffd中存了句柄号和struct file对象地址
    INIT_LIST_HEAD(&epi->pwqlist);

	// 设置epitem的ep成员指向eventpoll管理对象
    epi->ep = ep;

	// 设置epitem的ffd 指向要添加的fd 相关对象
    ep_set_ffd(&epi->ffd, tfile, fd);

    //2、设置 socket 等待队列
    //定义并初始化 ep_pqueue 对象
    struct ep_pqueue epq;
    epq.epi = epi;
    init_poll_funcptr(&epq.pt, ep_ptable_queue_proc);

    //调用 ep_ptable_queue_proc 注册回调函数 
    //实际注入的函数为 ep_poll_callback
    revents = ep_item_poll(epi, &epq.pt);

    ......
    //3、将epi插入到 eventpoll 对象中的红黑树中
    ep_rbtree_insert(ep, epi);
    ......
}

// 参数ffd为 新建的epitem成员
// 参数file/fd为 要添加监听的socket成员（epoll_ctl的操作对象）
static inline void ep_set_ffd(struct epoll_filefd *ffd,
                        struct file *file, int fd)
{
    ffd->file = file;
    ffd->fd = fd;
}

static int ep_insert(...)
{
    ...
	// 新建一个 ep_pqueue 对象
    struct ep_pqueue epq;
	// 将新建的epitem挂在 epq上
	// 记住ep_pqueue结构包含了两个成员：
	// 1、poll_table pt
	// 2、struct epitem *epi
    epq.epi = epi;
    init_poll_funcptr(&epq.pt, ep_ptable_queue_proc);
    ...
	//ep_item_poll：核心
	//作用：调用 ep_ptable_queue_proc 注册回调函数（实际注入的函数为 ep_poll_callback）
	revents = ep_item_poll(epi, &epq.pt);
	// ......
}

static inline void init_poll_funcptr(poll_table *pt, 
    poll_queue_proc qproc)
{
    pt->_qproc = qproc;	// 这里的qproc就是ep_ptable_queue_proc
    pt->_key   = ~0UL; /* all events enabled */
}

static inline unsigned int ep_item_poll(struct epitem *epi, poll_table *pt)
{
	//......
    pt->_key = epi->event.events;

	// 重要：epi->ffd.file->f_op 找到被操作的fd的f_op成员
	// sock_poll
    return epi->ffd.file->f_op->poll(epi->ffd.file, pt) & epi->event.events;
}

//https://elixir.bootlin.com/linux/v4.11.6/source/net/socket.c#L1041
static unsigned int sock_poll(struct file *file, poll_table *wait)
{
    struct socket *sock;
	sock = file->private_data;
	// ......
	//tcp_poll
    return sock->ops->poll(file, sock, wait);
}

unsigned int tcp_poll(struct file *file, struct socket *sock, poll_table *wait)
{
    struct sock *sk = sock->sk;
	// ......
    sock_poll_wait(file, sk_sleep(sk), wait);
}

static inline wait_queue_head_t *sk_sleep(struct sock *sk)
{
    BUILD_BUG_ON(offsetof(struct socket_wq, wait) != 0);
    return &rcu_dereference_raw(sk->sk_wq)->wait;
}

static inline void sock_poll_wait(struct file *filp,
        wait_queue_head_t *wait_address, poll_table *p)
{	
	//poll_wait
    poll_wait(filp, wait_address, p);
}

static inline void poll_wait(struct file * filp, wait_queue_head_t * wait_address, poll_table *p)
{
    if (p && p->_qproc && wait_address)
		// qproc 为函数指针，
		// 在前面的 init_poll_funcptr 调用时被设置成了 ep_ptable_queue_proc 函数
		// 调用其实是ep_ptable_queue_proc方法
        p->_qproc(filp, wait_address, p);
}

// https://elixir.bootlin.com/linux/v4.11.6/source/fs/eventpoll.c#L1090
static void ep_ptable_queue_proc(struct file *file, wait_queue_head_t *whead,
                 poll_table *pt)
{
    // 从 poll_table 反向拿到 epitem（依赖 ep_pqueue 的内存布局：epi 紧跟 pt）
    struct epitem *epi = ep_item_from_epqueue(pt);
    // 新建一个 eppoll_entry 对象（重要！这才是 socket 等待队列的真正驻留项）
    struct eppoll_entry *pwq;

    if (epi->nwait >= 0 && (pwq = kmem_cache_alloc(pwq_cache, GFP_KERNEL))) {
        // 初始化 pwq->wait（wait_queue_t）：把回调 func 设为 ep_poll_callback，private 留 NULL
        init_waitqueue_func_entry(&pwq->wait, ep_poll_callback);
        pwq->whead = whead;     // 记录所属 socket 等待队列头
        pwq->base  = epi;       // 反向指针：回调时通过 base 拿到 epitem
        // 关键：根据是否设置 EPOLLEXCLUSIVE，决定挂入方式（决定唤醒时是否被 break）
        if (epi->event.events & EPOLLEXCLUSIVE)
            add_wait_queue_exclusive(whead, &pwq->wait); // 设置 WQ_FLAG_EXCLUSIVE
        else
            add_wait_queue(whead, &pwq->wait);
        // 同时把 pwq 挂到 epitem.pwqlist，方便 ep_remove 时反向清理
        list_add_tail(&pwq->llink, &epi->pwqlist);
        epi->nwait++;
    } else {
        epi->nwait = -1;
    }
}

// https://elixir.bootlin.com/linux/v4.11.6/source/include/linux/wait.h#L72
static inline void init_waitqueue_func_entry(
    wait_queue_t *q, wait_queue_func_t func)
{
    q->flags   = 0;
    q->private = NULL;  // 不绑定进程：唤醒时不会走 try_to_wake_up（设置为NULL的原因）
    q->func    = func;  // ep_poll_callback：有数据到达时被调用
}

static void ep_rbtree_insert(struct eventpoll *ep, struct epitem *epi)
{
	int kcmp;
	struct rb_node **p = &ep->rbr.rb_node, *parent = NULL;
	struct epitem *epic;

	while (*p) {
		parent = *p;
		epic = rb_entry(parent, struct epitem, rbn);
		kcmp = ep_cmp_ffd(&epi->ffd, &epic->ffd);
		if (kcmp > 0)
			p = &parent->rb_right;
		else
			p = &parent->rb_left;
	}
	rb_link_node(&epi->rbn, parent, p);
	rb_insert_color(&epi->rbn, &ep->rbr);
}

SYSCALL_DEFINE4(epoll_wait, int, epfd, struct epoll_event __user *, events,
		int, maxevents, int, timeout)
{
	int error;
	struct fd f;
	struct eventpoll *ep;

	/* Get the "struct file *" for the eventpoll file */
	f = fdget(epfd);
	if (!f.file)
		return -EBADF;

	/*
	 * We have to check that the file structure underneath the fd
	 * the user passed to us _is_ an eventpoll file.
	 */
	error = -EINVAL;
	if (!is_file_epoll(f.file))
		goto error_fput;

	/*
	 * At this point it is safe to assume that the "private_data" contains
	 * our own data structure.
	 */
	ep = f.file->private_data;

	/* Time to fish for events ... */
	error = ep_poll(ep, events, maxevents, timeout);

	//.....
}

static int ep_poll(struct eventpoll *ep, struct epoll_event __user *events,
		   int maxevents, long timeout)
{
	int res = 0, eavail, timed_out = 0;
	unsigned long flags;
	u64 slack = 0;
	// 定义一个等待队列
	wait_queue_t wait;
	ktime_t expires, *to = NULL;

	//.......

fetch_events:
	spin_lock_irqsave(&ep->lock, flags);

	//1、判断就绪队列上有没有事件就绪
	if (!ep_events_available(ep)) {

		//2、定义等待事件并关联当前进程（current为当前进程）
		init_waitqueue_entry(&wait, current);

		//3、 把新 waitqueue 添加到 epoll->wq 链表（eventpoll的等待队列）里
		__add_wait_queue_exclusive(&ep->wq, &wait);
		
		//4、启动等待-唤醒 循环机制
		for (;;) {
			// 将当前进程状态设置为可打断状态
			set_current_state(TASK_INTERRUPTIBLE);

			//5、current进入CPU切换前，先检查唤醒事件是否满足
			// 满足则退出等待
			// 不满足继续执行，让出CPU（进程调度切换）
			if (ep_events_available(ep) || timed_out)
				break;
			if (signal_pending(current)) {
				res = -EINTR;
				break;
			}

			spin_unlock_irqrestore(&ep->lock, flags);
			//6、让出CPU 主动进入睡眠状态
			if (!schedule_hrtimeout_range(to, slack, HRTIMER_MODE_ABS))
				timed_out = 1;
			
			//7、说明被唤醒了，拿到CPU的控制权，继续循环（检测等待条件是否满足....）
			spin_lock_irqsave(&ep->lock, flags);
		}

		__remove_wait_queue(&ep->wq, &wait);
		__set_current_state(TASK_RUNNING);
	}
check_events:
	/* Is it worth to try to dig for events ? */
	eavail = ep_events_available(ep);

	spin_unlock_irqrestore(&ep->lock, flags);

	/*
	 * Try to transfer events to user space. In case we get 0 events and
	 * there's still timeout left over, we go trying again in search of
	 * more luck.
	 */
	if (!res && eavail &&
	    !(res = ep_send_events(ep, events, maxevents)) && !timed_out)
		goto fetch_events;

	return res;
}

//https://elixir.bootlin.com/linux/v4.11.6/source/fs/eventpoll.c#L374
static inline int ep_events_available(struct eventpoll *ep)
{
	return !list_empty(&ep->rdllist) || ep->ovflist != EP_UNACTIVE_PTR;
}

static inline void init_waitqueue_entry(wait_queue_t *q, struct task_struct *p)
{
    q->flags = 0;
    q->private = p;	//current
    q->func = default_wake_function;	//等待被唤醒时的调用函数为default_wake_function
}

//https://elixir.bootlin.com/linux/v4.11.6/source/kernel/sched/core.c#L3665
int default_wake_function(wait_queue_t *curr, unsigned mode, int wake_flags,
			  void *key)
{
	return try_to_wake_up(curr->private, mode, wake_flags);
}

static inline void __add_wait_queue_exclusive(wait_queue_head_t *q,
                                wait_queue_t *wait)
{
    wait->flags |= WQ_FLAG_EXCLUSIVE;	//注意这个WQ_FLAG_EXCLUSIVE
    __add_wait_queue(q, wait);
}

int __sched schedule_hrtimeout_range(ktime_t *expires, 
    unsigned long delta, const enum hrtimer_mode mode)
{
    return schedule_hrtimeout_range_clock(
            expires, delta, mode, CLOCK_MONOTONIC);
}

int __sched schedule_hrtimeout_range_clock(...)
{
	//在 schedule 中选择下一个进程调度
    schedule();
    ...
}

static void __sched __schedule(void)
{
    next = pick_next_task(rq);
    ...
    context_switch(rq, prev, next);
}

/* Get the "struct epitem" from a wait queue pointer */
static inline struct epitem *ep_item_from_wait(wait_queue_t *p)
{
	return container_of(p, struct eppoll_entry, wait)->base;
}

//https://elixir.bootlin.com/linux/v4.11.6/source/net/ipv4/tcp_ipv4.c#L1605
int tcp_v4_rcv(struct sk_buff *skb)
{
	struct sock *sk;
    ......
    th = tcp_hdr(skb); //获取tcp header
    iph = ip_hdr(skb); //获取ip header

    //1、根据数据包 header 中的 ip、端口信息查找到对应的sock结构
    sk = __inet_lookup_skb(&tcp_hashinfo, skb, __tcp_hdrlen(th), th->source,
			       th->dest, &refcounted);
    ......

    //sock 未被用户锁定
    if (!sock_owned_by_user(sk)) {
        {
            if (!tcp_prequeue(sk, skb))
				//2、go tcp_v4_do_rcv
                ret = tcp_v4_do_rcv(sk, skb);
        }
    }
	......
}

//tcp_v4_do_rcv
//https://elixir.bootlin.com/linux/v4.11.6/source/net/ipv4/tcp_ipv4.c#L1413
int tcp_v4_do_rcv(struct sock *sk, struct sk_buff *skb)
{
    if (sk->sk_state == TCP_ESTABLISHED) {
        // TCP_ESTABLISHED：执行连接状态下的数据处理（暂时只关注此状态）
        // v4.11.6 中 tcp_rcv_established 返回 void，无须 if 包裹
        tcp_rcv_established(sk, skb, tcp_hdr(skb), skb->len);
        return 0;
    }
    //其它非 ESTABLISH 状态的数据包处理
    ......
}

//tcp_rcv_established：非常复杂，基于TCP FSM处理
//https://elixir.bootlin.com/linux/v4.11.6/source/net/ipv4/tcp_input.c#L5388
void tcp_rcv_established(struct sock *sk, struct sk_buff *skb,
            const struct tcphdr *th, unsigned int len)
{
    ......

    // 1、调用 tcp_queue_rcv ，将接收数据放到 sock 的接收队列sk_receive_queue
    eaten = tcp_queue_rcv(sk, skb, tcp_header_len, &fragstolen);
	......
    // 2、数据 ready，唤醒 sock 等待队列上的等待者
    sk->sk_data_ready(sk);

	......
}

//tcp_queue_rcv
static int __must_check tcp_queue_rcv(struct sock *sk, struct sk_buff *skb, int hdrlen,
            bool *fragstolen)
{
    //1、把接收到的数据放到 sock 的接收队列的尾部
    if (!eaten) {
        __skb_queue_tail(&sk->sk_receive_queue, skb);
        skb_set_owner_r(skb, sk);
    }
    return eaten;
}

// 参数whead 来自于fd关联的sock对象的等待队列头： (sk->sk_wq)->wait
// 在ep_ptable_queue_proc会把新建的eppoll_entry对象挂到上面这个等待队列链表里面
static void ep_ptable_queue_proc(struct file *file, wait_queue_head_t *whead, poll_table *pt)
{
    struct epitem *epi = ep_item_from_epqueue(pt);
    struct eppoll_entry *pwq;  //新建一个eppoll_entry对象
    if (epi->nwait >= 0 && (pwq = kmem_cache_alloc(pwq_cache, GFP_KERNEL))) {
        //初始化pwq->wait（等待队列项）的回调方法
        init_waitqueue_func_entry(&pwq->wait, ep_poll_callback);
        pwq->whead = whead;
        pwq->base  = epi;
        //将 pwq->wait 放入socket的等待队列whead（注意不是epoll的等待队列）
        if (epi->event.events & EPOLLEXCLUSIVE)
            add_wait_queue_exclusive(whead, &pwq->wait);
        else
            add_wait_queue(whead, &pwq->wait);
        list_add_tail(&pwq->llink, &epi->pwqlist);
        epi->nwait++;
    }
    ...
}

// 初始化等待队列项wait_queue_t
static inline void
init_waitqueue_func_entry(wait_queue_t *q, wait_queue_func_t func)
{
	q->flags	= 0;
	q->private	= NULL;		//注意！这里是NULL
	q->func		= func;		// 这里是ep_poll_callback
}

// 追加到等待队列的成员task_list上
static inline void __add_wait_queue(wait_queue_head_t *head, wait_queue_t *new)
{
	list_add(&new->task_list, &head->task_list);
}

// https://elixir.bootlin.com/linux/v4.11.6/source/net/core/sock.c#L2397
static void sock_def_readable(struct sock *sk)
{
    struct socket_wq *wq;

    rcu_read_lock();
    wq = rcu_dereference(sk->sk_wq);

    //判断等待队列不为空
    if (skwq_has_sleeper(wq))
        //唤醒等待队列上的所有 EXCLUSIVE 项（受 nr_exclusive 控制）
        //会逐个调用 wait_queue_t.func，即 ep_poll_callback
        wake_up_interruptible_sync_poll(&wq->wait, POLLIN | POLLPRI |
                        POLLRDNORM | POLLRDBAND);
    //向通过 fcntl(F_SETOWN) 注册了 SIGIO 的进程发信号（与 epoll 路径无关）
    sk_wake_async(sk, SOCK_WAKE_WAITD, POLL_IN);
    rcu_read_unlock();
}

//https://elixir.bootlin.com/linux/v4.11.6/source/include/linux/wait.h#L106
static inline int waitqueue_active(wait_queue_head_t *q)
{
	return !list_empty(&q->task_list);
}

static inline bool wq_has_sleeper(wait_queue_head_t *wq)
{
	smp_mb();
	return waitqueue_active(wq);
}

static inline bool skwq_has_sleeper(struct socket_wq *wq)
{
	return wq && wq_has_sleeper(&wq->wait);
}

#define wake_up_interruptible_sync_poll(x, m)       \
    __wake_up_sync_key((x), TASK_INTERRUPTIBLE, 1, (void *) (m))


void __wake_up_sync_key(wait_queue_head_t *q, unsigned int mode,
            int nr_exclusive, void *key)
{
    ...
    __wake_up_common(q, mode, nr_exclusive, wake_flags, key);
}

static void __wake_up_common(wait_queue_head_t *q, unsigned int mode,
            int nr_exclusive, int wake_flags, void *key)
{
    wait_queue_t *curr, *next;

    list_for_each_entry_safe(curr, next, &q->task_list, task_list) {
        unsigned flags = curr->flags;

		//curr->func 为 ep_poll_callback
		// 注意ep_poll_callback的参数为
		// wait_queue_t *wait, unsigned mode, int sync, void *key
        if (curr->func(curr, mode, wake_flags, key) &&
                (flags & WQ_FLAG_EXCLUSIVE) && !--nr_exclusive)
            break;
    }
}

//https://elixir.bootlin.com/linux/v4.11.6/source/fs/eventpoll.c#L1004
static int ep_poll_callback(wait_queue_t *wait, unsigned mode, int sync, void *key)
{
	// 1、根据sk的等待队列中的等待项wait_queue_t
	// 获取 waitqueue 对应的 epitem（前文已述）
	struct epitem *epi = ep_item_from_wait(wait);

	// 2、获取 epitem 对应的 eventpoll 结构体
	struct eventpoll *ep = epi->ep;
	int ewake = 0;

	//......

	// 3、将当前 epitem 添加到 eventpoll 的就绪队列中（尾部）
	if (!ep_is_linked(&epi->rdllink)) {
		list_add_tail(&epi->rdllink, &ep->rdllist);
		ep_pm_stay_awake_rcu(epi);
	}

	// 4、查看 eventpoll 的等待队列上是否有在等待
	// waitqueue_active：ep->wq的等待队列不为空返回true
	if (waitqueue_active(&ep->wq)) {
		if ((epi->event.events & EPOLLEXCLUSIVE) &&
					!((unsigned long)key & POLLFREE)) {
			switch ((unsigned long)key & EPOLLINOUT_BITS) {
			case POLLIN:
				if (epi->event.events & POLLIN)
					ewake = 1;
				break;
			case POLLOUT:
				if (epi->event.events & POLLOUT)
					ewake = 1;
				break;
			case 0:
				ewake = 1;
				break;
			}
		}
		wake_up_locked(&ep->wq);
	}

	//.......
}

static void __wake_up_common(wait_queue_head_t *q, unsigned int mode,
            int nr_exclusive, int wake_flags, void *key)
{
    wait_queue_t *curr, *next;

    list_for_each_entry_safe(curr, next, &q->task_list, task_list) {
        unsigned flags = curr->flags;
		//curr->func：default_wake_function
        if (curr->func(curr, mode, wake_flags, key) &&
                (flags & WQ_FLAG_EXCLUSIVE) && !--nr_exclusive)
            break;
    }
}

int default_wake_function(wait_queue_t *curr, unsigned mode, int wake_flags, void *key)
{
    return try_to_wake_up(curr->private, mode, wake_flags);
}

static int ep_poll(struct eventpoll *ep, struct epoll_event __user *events,
             int maxevents, long timeout)
{
    ......
	for (;;) {
			set_current_state(TASK_INTERRUPTIBLE);

			if (ep_events_available(ep) || timed_out)
                // 就绪队列有事件了，所以condition成立，跳出for(;;)循环
				break;

			if (!schedule_hrtimeout_range(to, slack, HRTIMER_MODE_ABS))
				timed_out = 1;
			// 唤醒后，继续从这里开始执行
	}

	// 将先前epoll_wait阻塞的进程，移除epoll的等待队列
    __remove_wait_queue(&ep->wq, &wait);
    set_current_state(TASK_RUNNING);
check_events:
    //返回就绪事件给用户进程
    ep_send_events(ep, events, maxevents)
}

//https://elixir.bootlin.com/linux/v4.11.6/source/fs/eventpoll.c#L584
static int ep_send_events(struct eventpoll *ep,
			  struct epoll_event __user *events, int maxevents)
{
	struct ep_send_events_data esed;

	esed.maxevents = maxevents;
	esed.events = events;
    // 遍历事件就绪列表，发送就绪事件到用户空间
	return ep_scan_ready_list(ep, ep_send_events_proc/**/, &esed, 0, false);
}

static __poll_t ep_scan_ready_list(struct eventpoll *ep,
                  __poll_t (*sproc)(struct eventpoll *,
                       struct list_head *, void *),
                  void *priv, int depth, bool ep_locked) {
    __poll_t res;
    struct epitem *epi, *nepi;
    LIST_HEAD(txlist);
    ......
    // 将就绪队列分片链接到 txlist 链表中
	//分割就绪队列`rdllist`，将 `rdllist` 中的事件全部转移到临时链表 `txlist` 中，并清空 `rdllist`
    list_splice_init(&ep->rdllist, &txlist);

	//CALL ep_send_events_proc
    res = (*sproc)(ep, &txlist, priv);
    ......
    // 在处理 sproc 回调处理过程中，可能产生新的就绪事件被写入 ovflist，将 ovflist 回写 rdllist
    for (nepi = READ_ONCE(ep->ovflist); (epi = nepi) != NULL;
         nepi = epi->next, epi->next = EP_UNACTIVE_PTR) {
        if (!ep_is_linked(epi)) {
            list_add(&epi->rdllink, &ep->rdllist);
            ep_pm_stay_awake(epi);
        }
    }
    ......
    // txlist 在 epitem 回调中，可能没有完全处理完，那么重新放回到 rdllist，下次处理
    list_splice(&txlist, &ep->rdllist);
    ......
}

static int ep_send_events_proc(struct eventpoll *ep, struct list_head *head,
			       void *priv)
{
	struct ep_send_events_data *esed = priv;
	int eventcnt;
	unsigned int revents;
	struct epitem *epi;
	struct epoll_event __user *uevent;
	struct wakeup_source *ws;
	poll_table pt;

	......
	for (eventcnt = 0, uevent = esed->events;
	     !list_empty(head) && eventcnt < esed->maxevents;) {
		epi = list_first_entry(head, struct epitem, rdllink);
		......

		list_del_init(&epi->rdllink);

		revents = ep_item_poll(epi, &pt);

		if (revents) {
			// copy to 用户态
			if (__put_user(revents, &uevent->events) ||
			    __put_user(epi->event.data, &uevent->data)) {
				//copy error
				//把未处理的事件，再加回rdllink
				list_add(&epi->rdllink, head);
				ep_pm_stay_awake(epi);
				return eventcnt ? eventcnt : -EFAULT;
			}
			eventcnt++;
			uevent++;
			if (epi->event.events & EPOLLONESHOT)
				epi->event.events &= EP_PRIVATE_BITS;
			else if (!(epi->event.events & EPOLLET)) {
				// 非ET模式下，需要加回ep->rdllist
				list_add_tail(&epi->rdllink, &ep->rdllist);
				ep_pm_stay_awake(epi);
			}//ET模式，不加回
		}
	}

	return eventcnt;
}

//https://elixir.bootlin.com/linux/v4.11.6/source/net/ipv4/tcp_input.c#L5388
void tcp_rcv_established(struct sock *sk, struct sk_buff *skb,
             const struct tcphdr *th, unsigned int len)
{
    ......
    // 头预测命中三连：
    // 1) flag 字段与 pred_flags 完全一致
    // 2) seq == rcv_nxt（顺序段）
    // 3) ack_seq 未超 snd_nxt
    if ((tcp_flag_word(th) & TCP_HP_BITS) == tp->pred_flags &&
        TCP_SKB_CB(skb)->seq == tp->rcv_nxt &&
        !after(TCP_SKB_CB(skb)->ack_seq, tp->snd_nxt)) {
        ......
        // ── Fast Path：直接入 receive_queue ──
        eaten = tcp_queue_rcv(sk, skb, tcp_header_len, &fragstolen);
        ......
        sk->sk_data_ready(sk);   // ★ 触发 sock_def_readable → ep_poll_callback
        return;
    }
slow_path:
    ......
    tcp_data_queue(sk, skb);     // 走慢速路径
    ......
}

//https://elixir.bootlin.com/linux/v4.11.6/source/net/ipv4/tcp_input.c#L4561
static void tcp_data_queue(struct sock *sk, struct sk_buff *skb)
{
    ......
    if (TCP_SKB_CB(skb)->seq == tp->rcv_nxt) {
        //── 分支 A：有序段（in-sequence in-window）──
queue_and_out:
        eaten = tcp_queue_rcv(sk, skb, 0, &fragstolen);
        tcp_rcv_nxt_update(tp, TCP_SKB_CB(skb)->end_seq);
        ......
        if (!RB_EMPTY_ROOT(&tp->out_of_order_queue)) {
            //── 分支 B：本段填补了空洞，把 OFO 队列里能转正的段批量迁出 ──
            tcp_ofo_queue(sk);    // OFO -> receive_queue
            ......
        }
        if (!sock_flag(sk, SOCK_DEAD))
            sk->sk_data_ready(sk);     // ★ 此处触发 ep_poll_callback
        return;
    }
    ......
    //── 分支 C：纯乱序段（seq > rcv_nxt）──
    tcp_data_queue_ofo(sk, skb);   // 只挂 out_of_order_queue（rbtree）
    // ★ 注意：这里不调用 sk_data_ready，所以 epoll 不会收到通知！
}

// https://elixir.bootlin.com/linux/v4.11.6/source/fs/file_table.c#L187
static void __fput(struct file *file)
{
	struct dentry *dentry = file->f_path.dentry;
	struct vfsmount *mnt = file->f_path.mnt;
	struct inode *inode = file->f_inode;

	might_sleep();

	fsnotify_close(file);
	/*
	 * The function eventpoll_release() should be the first called
	 * in the file cleanup chain.
	 */
	eventpoll_release(file);
    ......
}

//https://elixir.bootlin.com/linux/v4.11.6/source/include/linux/eventpoll.h#L42
static inline void eventpoll_release(struct file *file)
{

	/*
	 * Fast check to avoid the get/release of the semaphore. Since
	 * we're doing this outside the semaphore lock, it might return
	 * false negatives, but we don't care. It'll help in 99.99% of cases
	 * to avoid the semaphore lock. False positives simply cannot happen
	 * because the file in on the way to be removed and nobody ( but
	 * eventpoll ) has still a reference to this file.
	 */
	if (likely(list_empty(&file->f_ep_links)))
		return;

	/*
	 * The file is being closed while it is still linked to an epoll
	 * descriptor. We need to handle this by correctly unlinking it
	 * from its containers.
	 */
	eventpoll_release_file(file);
}

// 当 file 引用计数归零时被 __fput 调用
void eventpoll_release_file(struct file *file)
{
    struct eventpoll *ep;
    struct epitem *epi, *next;

    mutex_lock(&epmutex);
    // 一个 file 可能同时被多个 epoll 监听，所以是 list 遍历
    list_for_each_entry_safe(epi, next, &file->f_ep_links, fllink) {
        ep = epi->ep;
        mutex_lock_nested(&ep->mtx, 0);
        ep_remove(ep, epi);   // 从对应 eventpoll 上彻底卸载
        mutex_unlock(&ep->mtx);
    }
    mutex_unlock(&epmutex);
}

// https://elixir.bootlin.com/linux/v4.11.6/source/net/ipv4/tcp_input.c#L4135
void tcp_fin(struct sock *sk)
{
    inet_csk_schedule_ack(sk);
    sk->sk_shutdown |= RCV_SHUTDOWN;
    sock_set_flag(sk, SOCK_DONE);
    // 状态机切换：ESTABLISHED -> CLOSE_WAIT，FIN_WAIT_2 -> TIME_WAIT 等
    switch (sk->sk_state) { case TCP_ESTABLISHED: tcp_set_state(sk, TCP_CLOSE_WAIT); break; ... }
    ......
    if (!sock_flag(sk, SOCK_DEAD)) {
        sk->sk_state_change(sk);    // = sock_def_wakeup -> 唤醒 sk_wq，上报 POLLHUP（仅 shutdown==SHUTDOWN_MASK 时）
        if (sk->sk_shutdown == SHUTDOWN_MASK || sk->sk_state == TCP_CLOSE)
            sk_wake_async(sk, SOCK_WAKE_WAITD, POLL_HUP);
        else
            sk_wake_async(sk, SOCK_WAKE_WAITD, POLL_IN);
    }
    ......
}

// https://elixir.bootlin.com/linux/v4.11.6/source/net/ipv4/tcp_input.c#L4097
void tcp_reset(struct sock *sk)
{
    switch (sk->sk_state) {
    case TCP_SYN_SENT:  sk->sk_err = ECONNREFUSED; break;
    case TCP_CLOSE_WAIT: sk->sk_err = EPIPE;        break;
    case TCP_CLOSE:     return;
    default:            sk->sk_err = ECONNRESET;
    }
    smp_wmb();
    if (!sock_flag(sk, SOCK_DEAD))
        sk->sk_error_report(sk);    // = sock_def_error_report
    tcp_done(sk);                   // 状态切到 TCP_CLOSE，触发 sk_state_change
}

static int ep_remove(struct eventpoll *ep, struct epitem *epi)
{
	unsigned long flags;
	struct file *file = epi->ffd.file;
	//1、解绑等待队列，移除事件回调
	//从 sock 等待队列中移除 eppoll_entry（通过 epi->pwqlist 链表定位）
	//解除事件回调函数 ep_poll_callback 的注册，确保后续事件不再触发通知
	ep_unregister_pollwait(ep, epi);

	/* Remove the current item from the list of epoll hooks */
	spin_lock(&file->f_lock);
	//将epitem从该fd关联的file结构的f_ep_links链表上移除
	// 从文件的 f_ep_links 链表中移除 epi（反向链接关系）
	// 解除文件与 epitem 的关联，确保文件关闭时不会误操作已移除的监听项
	// 添加代码位于：https://elixir.bootlin.com/linux/v4.11.6/source/fs/eventpoll.c#L1349
	list_del_rcu(&epi->fllink);
	spin_unlock(&file->f_lock);

	//2、从红黑树中删除epitem节点，解除全局监听
	//将 epi 从 epoll 实例的红黑树（rbr）中移除，此后内核不再监听该 fd 的事件
	rb_erase(&epi->rbn, &ep->rbr);

	//3、清理就绪队列，移除待处理事件（fd）
	//若 epi 已就绪但尚未被 epoll_wait 处理，需将其移出队列，避免无效事件上报
	spin_lock_irqsave(&ep->lock, flags);
	if (ep_is_linked(&epi->rdllink))
		list_del_init(&epi->rdllink);	// 从 rdllist 中移除epi
	spin_unlock_irqrestore(&ep->lock, flags);

	wakeup_source_unregister(ep_wakeup_source(epi));

	//5、释放资源：销毁 epitem，并减少文件引用计数
	// 若文件无其他引用，可触发关闭操作
	atomic_long_dec(&ep->user->epoll_watches);

	return 0;
}

// 检查就绪队列是否可用（不为空）
static inline int ep_events_available(struct eventpoll *ep)
{
	return !list_empty(&ep->rdllist) || ep->ovflist != EP_UNACTIVE_PTR;
}


//https://elixir.bootlin.com/linux/v4.11.6/source/fs/eventpoll.c#L1614
static int ep_poll(struct eventpoll *ep, struct epoll_event __user *events,
		   int maxevents, long timeout)
{
	int res = 0, eavail, timed_out = 0;
	unsigned long flags;
	u64 slack = 0;
	wait_queue_t wait;
	ktime_t expires, *to = NULL;

	if (timeout > 0) {
		struct timespec64 end_time = ep_set_mstimeout(timeout);

		slack = select_estimate_accuracy(&end_time);
		to = &expires;
		*to = timespec64_to_ktime(end_time);
	} else if (timeout == 0) {
		/*
		 * Avoid the unnecessary trip to the wait queue loop, if the
		 * caller specified a non blocking operation.
		 */
		timed_out = 1;
		spin_lock_irqsave(&ep->lock, flags);
		goto check_events;
	}

fetch_events:
	spin_lock_irqsave(&ep->lock, flags);

	if (!ep_events_available(ep)) {
		/*
		 * We don't have any available event to return to the caller.
		 * We need to sleep here, and we will be wake up by
		 * ep_poll_callback() when events will become available.
		 */
        // 1. 先初始化等待队列并移入
		init_waitqueue_entry(&wait, current);
		__add_wait_queue_exclusive(&ep->wq, &wait);

		for (;;) {
			/*
			 * We don't want to sleep if the ep_poll_callback() sends us
			 * a wakeup in between. That's why we set the task state
			 * to TASK_INTERRUPTIBLE before doing the checks.
			 */
			set_current_state(TASK_INTERRUPTIBLE);

            // 2. 检查condition条件是否成立
			if (ep_events_available(ep) || timed_out)
                // 3. 若condition成立，则跳出for(;;)循环
				break;
			if (signal_pending(current)) {
				res = -EINTR;
				break;
			}

			spin_unlock_irqrestore(&ep->lock, flags);
            // 4. 走到这里说明condition不成立，那么就先让出CPU
            // 进程第一调度定律
            // 那么内核执行进程切换，上下文保存到堆栈至此
			if (!schedule_hrtimeout_range(to, slack, HRTIMER_MODE_ABS))
				timed_out = 1;

            //5. 执行到这里大概率说明内核通过wakeup机制唤醒了当前的进程，继续for(;;)循环
            // 下一步是检查condition是否满足
			spin_lock_irqsave(&ep->lock, flags);
		}

        // 6. 已经被唤醒了，且condition也满足了，那么就移除等待队列
        // 且设置当前的进程的运行状态的TASK_RUNNING
		__remove_wait_queue(&ep->wq, &wait);
		__set_current_state(TASK_RUNNING);
	}
check_events:
	/* Is it worth to try to dig for events ? */
	eavail = ep_events_available(ep);

	spin_unlock_irqrestore(&ep->lock, flags);

	/*
	 * Try to transfer events to user space. In case we get 0 events and
	 * there's still timeout left over, we go trying again in search of
	 * more luck.
	 */
	if (!res && eavail &&
	    !(res = ep_send_events(ep, events, maxevents)) && !timed_out)
		goto fetch_events;

	return res;
}

// https://elixir.bootlin.com/linux/v4.11.6/source/fs/eventpoll.c#L1090
static void ep_ptable_queue_proc(struct file *file, wait_queue_head_t *whead,
                 poll_table *pt)
{
    struct epitem *epi = ep_item_from_epqueue(pt);
    struct eppoll_entry *pwq;
    if (epi->nwait >= 0 && (pwq = kmem_cache_alloc(pwq_cache, GFP_KERNEL))) {
        init_waitqueue_func_entry(&pwq->wait, ep_poll_callback);
        pwq->whead = whead;
        pwq->base  = epi;
        if (epi->event.events & EPOLLEXCLUSIVE)
            add_wait_queue_exclusive(whead, &pwq->wait); // ★ 设置 WQ_FLAG_EXCLUSIVE
        else
            add_wait_queue(whead, &pwq->wait);
        ......
    }
}

// https://elixir.bootlin.com/linux/v4.11.6/source/kernel/sched/wait.c#L74
static void __wake_up_common(wait_queue_head_t *q, unsigned int mode,
            int nr_exclusive, int wake_flags, void *key)
{
    wait_queue_t *curr, *next;
    list_for_each_entry_safe(curr, next, &q->task_list, task_list) {
        unsigned flags = curr->flags;
        // ★ 关键：func 返回非 0 表示"已消费"
        //   若该项是 EXCLUSIVE，且 nr_exclusive 减到 0，就 break
        if (curr->func(curr, mode, wake_flags, key) &&
                (flags & WQ_FLAG_EXCLUSIVE) && !--nr_exclusive)
            break;
    }
}

// https://elixir.bootlin.com/linux/v4.11.6/source/fs/eventpoll.c#L1057
static int ep_poll_callback(wait_queue_t *wait, unsigned mode, int sync, void *key)
{
    int pwake = 0;
    struct epitem *epi = ep_item_from_wait(wait);
    struct eventpoll *ep = epi->ep;
    int ewake = 0;
    ......
    if (waitqueue_active(&ep->wq)) {
        // ★ 只有 ep->wq 上确实有进程在等 + 事件类型匹配，才置 ewake=1
        if ((epi->event.events & EPOLLEXCLUSIVE) &&
                    !((unsigned long)key & POLLFREE)) {
            switch ((unsigned long)key & EPOLLINOUT_BITS) {
            case POLLIN:
                if (epi->event.events & POLLIN)  ewake = 1; break;
            case POLLOUT:
                if (epi->event.events & POLLOUT) ewake = 1; break;
            case 0:
                ewake = 1; break;
            }
        }
        wake_up_locked(&ep->wq);
    }
    ......
    // ★ EPOLLEXCLUSIVE 模式下，把"是否真消费"的信号返回给 __wake_up_common
    if (epi->event.events & EPOLLEXCLUSIVE)
        return ewake;
    return 1;
}

// https://elixir.bootlin.com/linux/v4.11.6/source/include/net/sock_reuseport.h#L13
struct sock_reuseport {
    struct rcu_head     rcu;
    u16                 max_socks;     // socks[] 容量
    u16                 num_socks;     // 当前数量
    struct bpf_prog __rcu *prog;       // 可选：用户态 BPF 选择程序
    struct sock        *socks[0];      // 所有 reuseport 成员 sock
};

// https://elixir.bootlin.com/linux/v4.11.6/source/net/core/sock_reuseport.c#L196
struct sock *reuseport_select_sock(struct sock *sk, u32 hash,
                                   struct sk_buff *skb, int hdr_len)
{
    struct sock_reuseport *reuse;
    struct bpf_prog *prog;
    struct sock *sk2 = NULL;
    u16 socks;

    rcu_read_lock();
    reuse = rcu_dereference(sk->sk_reuseport_cb);
    if (!reuse) goto out;
    prog  = rcu_dereference(reuse->prog);
    socks = READ_ONCE(reuse->num_socks);
    if (likely(socks)) {
        smp_rmb();   // 与 reuseport_add_sock 的 smp_wmb 配对
        if (prog && skb)
            // 路径1: 有 BPF 程序 → 跑 BPF 拿到 index
            sk2 = run_bpf(reuse, socks, prog, skb, hdr_len);
        else
            // 路径2: 默认按 hash 取模（基于源 IP/port + 目的 IP/port 计算）
            sk2 = reuse->socks[reciprocal_scale(hash, socks)];
    }
out:
    rcu_read_unlock();
    return sk2;
}

struct ep_pqueue {
    poll_table pt;          // 内含 _qproc 回调函数指针
    struct epitem *epi;     // 指向监听的 epitem
};
init_poll_funcptr(&epq.pt, ep_ptable_queue_proc); // 关键初始化

struct eppoll_entry {
    struct epitem *base;        // 指向 epitem（来自 ep_pqueue.epi）
    wait_queue_entry_t wait;     // 等待队列项，回调函数设为 ep_poll_callback
    wait_queue_head_t *whead;    // 指向文件的等待队列头
};

init_waitqueue_func_entry(&pwq->wait, ep_poll_callback); // 注册回调