应用层
int accept(int sockfd, struct sockaddr *addr, socklen_t *addrlen);
It extracts the first connection request on the queue of pending connections (backlog), creates a new connected socket, and returns a new file descriptor referring to that socket.
If no pending connections are present on the queue, and the socket is not marked as non-blocking, accept() blocks the caller until a connection is present. If the socket is marked non-blocking and no pending connections are present on the queue, accept() fails with the error EAGAIN.
在建立好接收队列以后，服务器就调用accept()，然后睡眠直到有客户端的连接请求到达。
addr用于保存客户端的地址。
系统调用
accept()是由glibc提供的，声明位于include/sys/socket.h中，实现位于sysdeps/mach/hurd/accept.c中，主要是用来从用户空间进入名为sys_socketcall的系统调用，并传递参数。sys_socketcall()实际上是所有socket函数进入内核空间的共同入口。
在sys_socketcall()中会调用sys_accept4()。
SYSCALL_DEFINE2(socketcall, int, call, unsigned long __user *, args)
{
...
switch(call) {
...
case SYS_ACCEPT:
err = sys_accpet4(a0, (struct sockaddr __user *)a1, (int __user *)a[2], 0);
break;
...
}
return err;
}
经过了socket层的总入口sys_socketcall()，现在进入sys_accpet4()。
SYSCALL_DEFINE4(accept4, int, fd, struct sockaddr __user *, upeer_sockaddr,
int __user *, upeer_addrlen, int flags)
{
struct socket *sock, *newsock;
struct file *newfile;
int err, len, newfd, fput_needed;
struct sockaddr_storage address;
/* 只允许使用这两个标志 */
if (flags & ~(SOCK_CLOSEXEC | SOCK_NONBLOCK))
return -EINVAL;
if (SOCK_NONBLOCK != O_NONBLOCK && (flags & SOCK_NONBLOCK))
flags = (flags & ~SOCK_NONBLOCK) | O_NONBLOCK;
/* 通过文件描述符fd，找到对应的socket。
* 以fd为索引从当前进程的文件描述符表files_struct中找到对应的file实例，
* 然后从file实例的private_data成员中获取socket实例。
*/
sock = sockfd_lookup_light(fd, &err, &fput_needed);
if (! sock)
goto out;
err = -ENFILE; /* File table overflow */
newsock = sock_alloc(); /* 创建一个新的inode和socket */
if (! newsock)
goto out_put;
newsock->type = sock->type; /* 新socket的类型 */
newsock->ops = sock->ops; /* 新socket的socket层操作 */
/* We don't need try_module_get here, as the listening socket (sock)
* has the protocol module (sock->ops->owner) held.
* Socekt层协议，对SOCK_STREAM来说是inet_stream_ops，它的引用计数加一。
*/
__module_get(newsock->ops->owner);
/* 为socket创建一个对应的file结构sock->file，返回fd */
newfd = sock_alloc_file(newsock, &newfile, flags);
if (unlikely(newfd < 0)) {
err = newfd;
sock_release(newsock);
goto out_put;
}
err = security_socket_accept(sock, newsock); /* SELinux相关 */
if (err)
goto out_fd;
/* SOCKET层的操作函数，如果是SOCK_STREAM，proto_ops为inet_stream_ops，
* 接下来调用inet_accept()。
*/
err = sock->ops->accept(sock, newsock, sock->file->f_flags);
if (err < 0)
goto out_fd;
if (upeer_sockaddr) { /* 如果要保存对端地址 */
/* 获取对端的地址，以及地址的长度 */
if (newsock->ops->getname(newsock, (struct sockaddr *)&address, &len, 2) < 0) {
err = -ECONNABORTED; /* Software caused connection abort */
goto out_fd;
}
/* 把内核空间的socket地址复制到用户空间 */
err = move_addr_to_user(&address, len, upeer_sockaddr, upeer_addrlen);
if (err < 0)
goto out_fd;
}
/* File flags are not inherited via accept() unlike another OSes.
* 以newfd为索引，把newfile加入当前进程的文件描述符表files_struct中。
*/
fd_install(newfd, newfile);
err = newfd;
out_put:
fput_light(sock->file, fput_needed);
out:
return err;
out_fd:
fput(newfile);
put_unused_fd(newfd);
goto out_put;
}
sys_accept4()主要做了：
创建了一个新的socket和inode，以及它所对应的fd、file。
调用Socket层操作函数inet_accept()。
保存对端地址到指定的用户空间地址。
Socket层
SOCK_STREAM套接口的Socket层操作函数集实例为inet_stream_ops，连接接收函数为inet_accept()。
const struct proto_ops inet_stream_ops = {
.family = PF_INET,
.owner = THIS_MODULE,
...
.accept = inet_accept,
...
};
inet_accept()主要做了：
调用TCP层的操作函数，获取已建立的连接sock。
把新socket和sock关联起来。
把新socket的状态设为SS_CONNECTED。
至此，新socket的各个字段都赋值完毕了。
/* Accept a pending connection.
* The TCP layer now gives BSD semantics.
*/
int inet_accept(struct socket *sock, struct socket *newsock, int flags)
{
struct sock *sk1 = sock->sk;
int err = -EINVAL;
/* 如果使用的是TCP，则sk_prot为tcp_prot，accept为inet_csk_accept()
* 获取新连接的sock。
*/
struct sock *sk2 = sk1->sk_prot->accept(sk1, flags, &err);
if (! sk2)
goto do_err;
lock_sock(sk2);
sock_rps_record_flow(sk2); /* RPS补丁 */
WARN_ON(! ((1 << sk2->sk_state) & (TCPF_ESTABLISHED | TCPF_CLOSE_WAIT | TCPF_CLOSE)));
sock_graft(sk2, newsock); /* 把sock和socket嫁接起来，让它们能相互索引 */
newsock->state = SS_CONNECTED; /* 把新socket的状态设为已连接 */
err = 0;
release_sock(sk2);
do_err:
return err;
}
/*
* struct callback_head - callback structure for use with RCU and task_work
* @next: next update requests in a list
* @func: actual update function to call after the grace period.
*/
struct callback_head {
struct callback_head *next;
void (*func) (struct callback_head *head);
};
#define rcu_head callback_head
/* 把sock和socket嫁接起来。*/
static inline void sock_graft(struct sock *sk, struct socket *parent)
{
write_lock_bh(&sk->sk_callback_lock);
sk->sk_wq = parent->wq; /* 等待队列 */
parent->sk = sk;
sk_set_socket(sk, parent);
security_sock_graft(sk, parent);
write_unlock_bh(&sk->sk_callback_lock);
}
static inline void sk_set_socket(struct sock *sk, struct socket *sock)
{
sk_tx_queue_clear(sk);
sk->sk_socket = sock;
}
TCP层实现
SOCK_STREAM套接口的TCP层操作函数集实例为tcp_prot，其中连接接收函数为inet_csk_accept()。
struct proto tcp_prot = {
.name = "TCP",
.owner = THIS_MODULE,
...
.accept = inet_csk_accept,
...
};
inet_csk_accept()用于从backlog队列（全连接队列）中取出一个ESTABLISHED状态的连接请求块，返回它所对应的连接sock。
- 非阻塞的，且当前没有已建立的连接，则直接退出，返回-EAGAIN。
- 阻塞的，且当前没有已建立的连接：
2.1 用户没有设置超时时间，则无限期阻塞。
2.2 用户设置了超时时间，超时后会退出。
/* This will accept the next outstanding connection. */
struct sock *inet_csk_accept(struct sock *sk, int flags, int *err)
{
struct inet_connection_sock *icsk = inet_csk(sk);
struct sock *newsk;
int error;
lock_sock(sk);
/* We need to make sure that this socket is listening,
* and that it has something pending.
*/
error = -EINVAL;
if (sk->sk_state != TCP_LISTEN) /* socket必须处于监听状态 */
goto out_err;
/* Find already established connection.
* 发没有现ESTABLISHED状态的连接请求块。
*/
if (reqsk_queue_empty(&icsk->icsk_accept_queue)) {
/* 等待超时时间，如果是非阻塞则为0 */
long timeo = sock_rcvtimeo(sk, flags & O_NONBLOCK);
/* If this is a non blocking socket don't sleep */
error = -EAGAIN; /* Try again */
if (! timeo) /* 如果是非阻塞的，则直接退出 */
goto out_err;
/* 阻塞等待，直到有全连接。如果用户有设置等待超时时间，超时后会退出 */
error = inet_csk_wait_for_connect(sk, timeo);
if (error)
goto out_err;
}
/* 获取新连接的sock，释放连接控制块 */
newsk = reqsk_queue_get_child(&icsk->icsk_accept_queue, sk);
WARN_ON(newsk->sk_state == TCP_SYN_RECV);
out:
release_sock(sk);
return newsk;
out err:
newsk = NULL;
*err = error;
goto out;
}
检查ESTABLISHED状态的连接请求块队列是否为空。
static inline int reqsk_queue_empty(struct request_sock_queue *queue)
{
return queue->rskq_accept_head == NULL;
}
static inline long sock_rcvtimeo(const struct sock *sk, bool noblock)
{
return noblock ? 0 : sk->sk_rcvtimeo; /* accept()超时时间 */
}
从backlog队列（全连接队列）中取出一个ESTABLISHED状态的连接请求块，返回它所对应的连接sock。同时更新backlog队列的全连接数，释放取出的连接控制块。
static inline struct sock *reqsk_queue_get_child(struct request_sock_queue *queue,
struct sock *parent)
{
/* 从全连接队列中，取出第一个ESTABLISHED状态的连接请求块 */
struct request_sock *req = reqsk_queue_remove(queue);
struct sock *child = req->sk; /* 一个已建立的连接 */
WARN_ON(child == NULL);
sk_acceptq_removed(parent); /* 当前backlog队列的全连接数减一 */
__reqsk_free(req); /* 释放取出的连接请求控制块 */
return child;
}
static inline struct request_sock *reqsk_queue_remove(struct request_sock_queue *queue)
{
struct request_sock *req = queue->rskq_accept_head; /* 第一个ESTABLISHED状态的连接请求块 */
WARN_ON(req == NULL);
queue->rskq_accept_head = req->dl_next;
if (queue->rskq_accept_head == NULL)
queue->rskq_accept_tail = NULL;
return req;
}
static inline void sk_acceptq_removed(struct sock *sk)
{
sk->sk_ack_backlog--; /* 全连接的数量减一 */
}
/* 释放连接请求控制块 */
static inline void __reqsk_free(struct request_sock *req)
{
kmem_cache_free(req->rsk_ops->slab, req);
}
等待队列
（1）socket的等待队列
/*
* @sk_wq: sock wait queue head and async head
*/
struct sock {
...
struct socket_wq __rcu *sk_wq; /* 套接字的等待队列 */
...
};
struct socket_wq {
/* Note: wait MUST be first field of socket_wq */
wait_queue_head_t wait; /* 等待队列 */
struct fasync_struct *fasync_list; /* 异步文件操作 */
struct rcu_head rcu; /* 更新时的回调函数 */
} __cacheline_aligned_in_smp;
struct __wait_queue_head {
spinlock_t lock;
struct list_head task_list;
};
typedef struct __wait_queue_head wait_queue_head_t; /* 等待队列头 */
（2）进程的等待任务
struct __wait_queue {
unsigned int flags;
#define WQ_FLAG_EXCLUSIVE 0x01
void *private; /* 指向当前的进程控制块 */
wait_queue_func_t func; /* 唤醒函数 */
struct list_head task_list; /* 用于链接入等待队列 */
};
typedef struct __wait_queue wait_queue_t;
typedef int (*wait_queue_func_t) (wait_queue_t *wait, unsigned mode, int flags, void *key);
int default_wake_function(wait_queue_t *wait, unsigned mode, int flags, void *key);
初始化等待任务。
#define DEFINE_WAIT(name) DEFINE_WAIT_FUNC(name, autoremove_wake_function)
#define DEFINE_WAIT_FUNC(name, function) \
wait_queue_t name = { \
.private = current, \
.func = function, \
.task_list = LIST_HEAD_INIT((name).task_list), \
}
int autoremove_wake_function(wait_queue_t *wait, unsigned mode, int sync, void *key)
{
int ret = default_wake_function(wait, mode, sync, key); /* 默认的唤醒函数 */
if (ret)
list_del_init(&wait->task_list); /* 从等待队列中删除，初始化此等待任务 */
return ret;
}
获取sock的等待队列。
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;
}
把等待任务加入到等待队列中，设置当前进程的状态。
void prepare_to_wait_exclusive(wait_queue_head_t *q, wait_queue_t *wait, int state)
{
unsigned long flags;
/* 这个标志表示一次只唤醒一个等待任务，避免惊群现象 */
wait->flags |= WQ_FLAG_EXCLUSIVE;
spin_lock_irqsave(&q->lock, flags);
if (list_empty(&wait->task_list))
__add_wait_queue_tail(q, wait); /* 把此等待任务加入到等待队列中 */
set_current_state(state); /* 设置当前进程的状态 */
spin_unlock_irqrestore(&q->lock, flags);
}
static inline void __add_wait_queue_tail(wait_queue_head_t *head, wait_queue_t *new)
{
list_add_tail(&new->task_list, &head->task_list);
}
#define set_current_state(state_value) \
set_mb(current->state, (state_value))
（3）accept()的阻塞等待
accept()超时时间为sk->sk_rcvtimeo，在sock_init_data()中初始化为MAX_SCHEDULE_TIMEOUT，表示无限等待。
/* Wait for an incoming connection, avoid race conditions.
* This must be called with the socket locked.
*/
static int inet_csk_wait_for_connect(struct sock *sk, long timeo)
{
struct inet_connection_sock *icsk = inet_csk(sk);
DEFINE_WAIT(wait); /* 初始化等待任务 */
int err;
for (; ;) {
/* 把等待任务加入到socket的等待队列中，把进程状态设置为TASK_INTERRUPTIBLE */
prepare_to_wait_exclusive(sk_sleep(sk), &wait, TASK_INTERRUPTIBLE);
release_sock(sk); /* 等下可能要睡觉了，先释放 */
if (reqsk_queue_empty(&icsk->icsk_accept_queue)) /* 如果全连接队列为空 */
timeo = schedule_timeout(timeo); /* 进入睡眠，直到超时或收到信号 */
lock_sock(sk); /* 醒来后重新上锁 */
err = 0;
if (! reqsk_queue_empty(&icsk->icsk_accept_queue)) /* 全连接队列不为空时，退出 */
break;
err = -EINVAL;
if (sk->sk_state != TCP_LISTEN) /* 如果sock不处于监听状态了，退出 */
break;
err = sock_intr_errno(timeo);
/* 如果进程有待处理的信号，退出。
* 因为timeo默认为MAX_SCHEDULE_TIMEOUT，所以err默认为-ERESTARTSYS。
* 接下来会重新调用此函数，所以accept()依然阻塞。
*/
if (signal_pending(current))
break;
err = -EAGAIN;
if (! timeo) /* 如果等待超时，即超过用户设置的sk->sk_rcvtimeo，退出 */
break;
}
finish_wait(sk_sleep(sk), &wait);
return err;
}
从等待队列中删除等待任务，把当前进程的状态置为可运行。
/**
* finish_wait - clean up after waiting in a queue
* @q: waitqueue waited on，等待队列头
* @wait: wait descriptor，等待任务
*
* Sets current thread back to running state and removes the wait
* descriptor from the given waitqueue if still queued.
*/
void finish_wait(wait_queue_head_t *q, wait_queue_t *wait)
{
unsigned long flags;
__set_current_state(TASK_RUNNING);
if (! list_empty_careful(&wait->task_list)) {
spin_lock_irqsave(&q->lock, flags);
list_del_init(&wait->task_list); /* 从等待队列中删除，初始化此等待任务 */
spin_unlock_irqrestore(&q->lock, flags);
}
}
（4）accept()的唤醒
当收到客户端的ACK后，经过如下调用：
tcp_v4_rcv
tcp_v4_do_rcv
tcp_child_process
sock_def_readable
wake_up_interruptible_sync_poll
__wake_up_sync_key
__wake_up_common
最终调用我们给等待任务注册的唤醒函数。
我们来看下accept()是如何避免惊群现象的。
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;
if (curr->func(curr, mode, wake_flags, key) && (flags & WQ_FLAG_EXCLUSIVE)
!--nr_exclusive)
break;
}
}
初始化等待任务时，flags |= WQ_FLAG_EXCLUSIVE。传入的nr_exclusive为1，表示只允许唤醒一个等待任务。
所以这里只会唤醒一个等待的进程，不会导致惊群现象。
