kk Blog —— 通用基础


date [-d @int|str] [+%s|"+%F %T"]
netstat -ltunp
sar -n DEV 1

Socket层实现系列 — I/O事件及其处理函数

http://blog.csdn.net/zhangskd/article/details/45787989

主要内容：Socket I/O事件的定义、I/O处理函数的实现。

内核版本：3.15.2

I/O事件定义

sock中定义了几个I/O事件，当协议栈遇到这些事件时，会调用它们的处理函数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

struct sock {
	...
	struct socket_wq __rcu *sk_wq; /* socket的等待队列和异步通知队列 */
	...
	/* callback to indicate change in the state of the sock.
	 * sock状态改变时调用，比如从TCP_SYN_SENT或TCP_SYN_RECV变为TCP_ESTABLISHED，
	 * 导致connect()的唤醒。比如从TCP_ESTABLISHED变为TCP_CLOSE_WAIT。
	 */
	void (*sk_state_change) (struct sock *sk);

	/* callback to indicate there is data to be processed.
	 * sock上有数据可读时调用，比如服务器端收到第三次握手的ACK时会调用，导致accept()的唤醒。
	 */
	void (*sk_data_ready) (struct sock *sk);

	/* callback to indicate there is buffer sending space available.
	 * sock上有发送空间可写时调用，比如发送缓存变得足够大了。
	 */
	void (*sk_write_space) (struct sock *sk);

	/* callback to indicate errors (e.g. %MSG_ERRQUEUE)
	 * sock上有错误发生时调用，比如收到RST包。
	 */
	void (*sk_error_report) (struct sock *sk);
	...
};

Socket I/O事件的默认处理函数在sock初始化时赋值。

对于SOCK_STREAM类型的Socket，sock有发送缓存可写事件会被更新为sk_stream_write_space。

1
2
3
4
5
6
7
8
9

void sock_init_data(struct socket *sock, struct sock *sk)
{
	...
	sk->sk_state_change = sock_def_wakeup; /* sock状态改变事件 */
	sk->sk_data_ready = sock_def_readable; /* sock有数据可读事件 */
	sk->sk_write_space = sock_def_write_space; /* sock有发送缓存可写事件 */
	sk->sk_error_report = sock_def_error_report; /* sock有IO错误事件 */
	...
}

判断socket的等待队列上是否有进程。

1
2
3
4
5

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

状态改变事件

sk->sk_state_change的实例为sock_def_wakeup()，当sock的状态发生改变时，会调用此函数来进行处理。

1
2
3
4
5
6
7
8
9
10

static void sock_def_wakeup(struct sock *sk)
{
	struct socket_wq *wq; /* socket的等待队列和异步通知队列 */

	rcu_read_lock();
	wq = rcu_dereference(sk->sk_wq);
	if (wq_has_sleeper(wq)) /* 有进程阻塞在此socket上 */
		wake_up_interruptible_all(&wq->wait); /* 唤醒此socket上的所有睡眠进程 */
	rcu_read_unlock();
}
1
2
3
4
5
6
7
8
9

#define wake_up_interruptible_all(x) __wake_up(x, TASK_INTERRUPTIBLE, 0, NULL)

void __wake_up(wait_queue_head_t *q, unsigned int mode, int nr_exclusive, void *key)
{
	unsigned long flags;
	spin_lock_irqsave(&q->lock, flags);
	__wake_up_common(q, mode, nr_exclusive, 0, key);
	spin_unlock_irqrestore(&q->lock, flags);
}

初始化等待任务时，如果flags设置了WQ_FLAG_EXCLUSIVE，那么传入的nr_exclusive为1，

表示只允许唤醒一个等待任务，这是为了避免惊群现象。否则会把t等待队列上的所有睡眠进程都唤醒。

1
2
3
4
5
6
7
8
9
10
11
12
13

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;
	}
}

最终调用的是等待任务中的处理函数，默认为autoremove_wake_function()。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

#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;
}

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);
}

try_to_wake_up()通过把进程的状态设置为TASK_RUNNING，并把进程插入CPU运行队列，来唤醒睡眠的进程。

有数据可读事件

sk->sk_data_ready的实例为sock_def_readable()，当sock有输入数据可读时，会调用此函数来处理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

static void sock_def_readable(struct sock *sk)
{
	struct socket_wq *wq; /* socket的等待队列和异步通知队列 */

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

	if (wq_has_sleeper(wq)) /* 有进程在此socket的等待队列 */
		wake_up_interruptible_sync_poll(&wq->wait, POLLIN | POLLPRI |
			POLLRDNORM | POLLRDBAND); /* 唤醒等待进程 */

	/* 异步通知队列的处理。
	 * 检查应用程序是否通过recv()类调用来等待接收数据，如果没有就发送SIGIO信号，
	 * 告知它有数据可读。
	 * how为函数的处理方式，band为用来告知的IO类型。
	 */
	sk_wake_async(sk, SOCK_WAKE_WAITD, POLL_IN);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

#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)
{
	unsigned long flags;
	int wake_flags = 1; /* XXX WF_SYNC */

	if (unlikely(!q))
		return;
	if (unlikely(nr_exclusive != 1))
		wake_flags = 0;

	spin_lock_irqsave(&q->lock, flags);
	__wake_up_common(q, mode, nr_exclusive, wake_flags, key);
	spin_unlock_irqrestore(&q->lock, flags);
}

最终也是调用__wake_up_common()。初始化等待任务时，flags |= WQ_FLAG_EXCLUSIVE。

传入的nr_exclusive为1，表示只允许唤醒一个等待任务。所以这里只会唤醒一个等待的进程。

有缓存可写事件

sk->sk_write_space的实例为sock_def_write_space()。

如果socket是SOCK_STREAM类型的，那么函数指针的值会更新为sk_stream_write_space()。

sk_stream_write_space()在TCP中的调用路径为：

1
2
3
4

tcp_rcv_established / tcp_rcv_state_process
	tcp_data_snd_check
		tcp_check_space
			tcp_new_space
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

/* When incoming ACK allowed to free some skb from write_queue,
 * we remember this event in flag SOCK_QUEUE_SHRUNK and wake up socket
 * on the exit from tcp input handler.
 */
static void tcp_new_space(struct sock *sk)
{
	struct tcp_sock *tp = tcp_sk(sk);

	if (tcp_should_expand_sndbuf(sk)) {
		tcp_sndbuf_expand(sk);
		tp->snd_cwnd_stamp = tcp_time_stamp;
	}

	/* 检查是否需要触发有缓存可写事件 */
	sk->sk_write_space(sk);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

void sk_stream_write_space(struct sock *sk)
{
	struct socket *sock = sk->sk_socket;
	struct socket_wq *wq; /* 等待队列和异步通知队列 */

	/* 如果剩余的发送缓存不低于发送缓存上限的1/3，且尚未发送的数据不高于一定值时 */
	if (sk_stream_is_writeable(sk) && sock) {
		clear_bit(SOCK_NOSPACE, &sock->flags); /* 清除发送缓存不够的标志 */

		rcu_read_lock();
		wq = rcu_dereference(sk->sk_wq); /* socket的等待队列和异步通知队列 */
		if (wq_has_sleeper(wq)) /* 如果等待队列不为空，则唤醒一个睡眠进程 */
			wake_up_interruptible_poll(&wq->wait, POLLOUT | POLLWRNORM | POLLWRBAND);

		/* 异步通知队列不为空，且允许发送数据时。
		 * 检测sock的发送队列是否曾经到达上限，如果有的话发送SIGIO信号，告知异步通知队列上
		 * 的进程有发送缓存可写。
		 */
		if (wq && wq->fasync_list && !(sk->sk_shutdown & SEND_SHUTDOWN))
			sock_wake_async(sock, SOCK_WAKE_SPACE, POLL_OUT);

		rcu_read_unlock();
	}
}

#define wake_up_interruptible_poll(x, m) \
	__wake_up(x, TASK_INTERRUPTIBLE, 1, (void *) (m))

最终也是调用__wake_up_common()。初始化等待任务时，flags |= WQ_FLAG_EXCLUSIVE。

传入的nr_exclusive为1，表示只允许唤醒一个等待进程。

1
2
3
4
5
6
7
8
9
10
11
12
13
14

struct sock {
	...
	/* 发送队列中，skb数据区的总大小 */
	atomic_t sk_wmem_alloc;
	...
	int sk_sndbuf; /* 发送缓冲区大小的上限 */
	struct sk_buff_head sk_write_queue; /* 发送队列 */
	...
	/* 发送队列的总大小，包含发送队列中skb数据区的总大小，
	 * 以及sk_buff、sk_shared_info结构体、协议头的额外开销。
	 */
	int sk_wmem_queued;
	...
};

如果剩余的发送缓存大于发送缓存上限的1/3，且尚未发送的数据少于一定值时，才会触发有发送

缓存可写的事件。

1
2
3
4
5
6
7
8
9
10
11
12
13
14

static inline bool sk_stream_is_writeable(const struct sock *sk)
{
	return sk_stream_wspace(sk) >= sk_stream_min_wspace(sk) &&
}

static inline int sk_stream_wspace(const struct sock *sk)
{
	return sk->sk_sndbuf - sk->sk_wmem_queued;
}

static inline int sk_stream_min_wspace(const struct sock *sk)
{
	return sk->sk_wmem_queued >> 1;
}

检查尚未发送的数据是否已经够多了，如果超过了用户设置的值，就不用触发有发送缓存可写事件，

以免使用过多的内存。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

static inline bool sk_stream_memory_free(const struct sock *sk)
{
	if (sk->sk_wmem_queued >= sk->sk_sndbuf)
		return false;

	return sk->sk_prot->stream_memory_free ? sk->sk_prot->stream_memory_free(sk) : true;
}

struct proto tcp_prot = {
	...
	.stream_memory_free = tcp_stream_memory_free,
	...
};

static inline bool tcp_stream_memory_free(const struct sock *sk)
{
	const struct tcp_sock *tp = tcp_sk(sk);
	u32 notsent_bytes = tp->write_seq - tp->snd_nxt; /* 尚未发送的数据大小 */

	/* 当尚未发送的数据，少于配置的值时，才触发有发送缓存可写的事件。
	 * 这是为了避免发送缓存占用过多的内存。
	 */
	return notsent_bytes < tcp_notsent_lowat(tp);
}

如果有使用TCP_NOTSENT_LOWAT选项，则使用用户设置的值。

否则使用sysctl_tcp_notsent_lowat，默认为无穷大。

1
2
3
4

static inline u32 tcp_notsent_lowat(const struct tcp_sock *tp)
{
	return tp->notsent_lowat ?: sysctl_tcp_notsent_lowat;
}

有I/O错误事件

sk->sk_error_report的实例为sock_def_error_report()。

在以下函数中会调用I/O错误事件处理函数：

1
2
3
4

tcp_disconnect
tcp_reset
tcp_v4_err
tcp_write_err
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

static void sock_def_error_report(struct sock *sk)
{
	struct socket_wq *wq; /* 等待队列和异步通知队列 */

	rcu_read_lock();
	wq = rcu_dereference(sk->sk_wq);
	if (wq_has_sleeper(wq)) /* 有进程阻塞在此socket上 */
		wake_up_interruptible_poll(&wq->wait, POLLERR);

	/* 如果使用了异步通知，则发送SIGIO信号通知进程有错误 */
	sk_wake_async(sk, SOCK_WAKE_IO, POLL_ERR);
}

#define wake_up_interruptible_poll(x, m) \
	__wake_up(x, TASK_INTERRUPTIBLE, 1, (void *) (m))

最终也是调用__wake_up_common()，由于nr_exclusive为1，只会唤醒socket上的一个等待进程。

Socket层实现系列 — 睡眠驱动的同步等待

http://blog.csdn.net/zhangskd/article/details/45770323

主要内容：Socket的同步等待机制，connect和accept等待的实现。

内核版本：3.15.2

概述

socket上定义了几个IO事件：状态改变事件、有数据可读事件、有发送缓存可写事件、有IO错误事件。对于这些事件，socket中分别定义了相应的事件处理函数，也称回调函数。

Socket I/O事件的处理过程中，要使用到sock上的两个队列：等待队列和异步通知队列，这两个队列中都保存着等待该Socket I/O事件的进程。

Q：为什么要使用两个队列，等待队列和异步通知队列有什么区别呢？
A：等待队列上的进程会睡眠，直到Socket I/O事件的发生，然后在事件处理函数中被唤醒。异步通知队列上的进程则不需要睡眠，Socket I/O事件发时，事件处理函数会给它们发送到信号，这些进程事先注册的信号处理函数就能够被执行。

等待队列

Socket层使用等待队列来进行阻塞等待，在等待期间，阻塞在此socket上的进程会睡眠。

1
2
3
4
5
6
7
8
9
10
11
12

struct sock {
	...
	struct socket_wq __rcu *sk_wq; /* socket的等待队列和异步通知队列 */
	...
}

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;
};
(1) socket的等待队列头
1
2
3
4
5

struct __wait_queue_head {
	spinlock_t lock;
	struct list_head task_list;
};
typedef struct __wait_queue_head wait_queue_head_t;
(2) 进程的等待任务
1
2
3
4
5
6
7
8
9
10

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);
(3) 初始化等待任务
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

#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;
}

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);
}

try_to_wake_up()通过把进程的状态设置为TASK_RUNNING，并把进程插入CPU运行队列，来唤醒睡眠的进程。

(4) 把等待任务插入到等待队列中

获取sock的等待队列。

1
2
3
4
5

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;
}

把等待任务加入到等待队列中，同时设置当前进程的状态，TASK_INTERRUPTIBLE或TASK_UNINTERRUPTIBLE。

1
2
3
4
5
6
7
8
9
10
11
12
13
14

void prepare_to_wait(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(q, wait); /* 把等待任务加入到等待队列的头部，会最先被唤醒 */

	set_current_state(state); /* 设置进程的状态 */

	spin_unlock_irqrestore(&q->lock, flags);
}

prepare_to_wait()和prepare_to_wait_exclusive()都是用来把等待任务加入到等待队列中，不同之处在于使用prepare_to_wait_exclusive()时，会在等待任务中添加WQ_FLAG_EXCLUSIVE标志，表示一次只能唤醒一个等待任务，目的是为了避免惊群现象。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

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))
(5) 删除等待任务

从等待队列中删除等待任务，同时把等待进程的状态置为可运行状态，即TASK_RUNNING。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

/**
 * 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);
	}
}

connect等待

(1) 睡眠

connect()的超时时间为sk->sk_sndtimeo，在sock_init_data()中初始化为MAX_SCHEDULE_TIMEOUT，表示无限等待，可以通过SO_SNDTIMEO选项来修改。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

static long inet_wait_for_connect(struct sock *sk, long timeo, int writebias)
{
	DEFINE_WAIT(wait);  /* 初始化等待任务 */

	/* 把等待任务加入到socket的等待队列头部，把进程的状态设为TASK_INTERRUPTIBLE */
	prepare_to_wait(sk_sleep(sk), &wait, TASK_INTERRUPTIBLE);
	sk->sk_write_pending += writebias;

	/* Basic assumption: if someone sets sk->sk_err, he _must_ change state of the socket
	 * from TCP_SYN_*. Connect() does not allow to get error notifications without closing
	 * the socket.
	 */

	/* 完成三次握手后，状态就会变为TCP_ESTABLISHED，从而退出循环 */
	while ((1 << sk->sk_state) & (TCPF_SYN_SENT | TCPF_SYN_RECV)) {
		release_sock(sk); /* 等下要睡觉了，先释放锁 */

		/* 进入睡眠，直到超时或收到信号，或者被I/O事件处理函数唤醒。
		 * 1. 如果是收到信号退出的，timeo为剩余的jiffies。
		 * 2. 如果使用了SO_SNDTIMEO选项，超时退出后，timeo为0。
		 * 3. 如果没有使用SO_SNDTIMEO选项，timeo为无穷大，即MAX_SCHEDULE_TIMEOUT，
		 *      那么返回值也是这个，而超时时间不定。为了无限阻塞，需要上面的while循环。
		 */
		timeo = schedule_timeout(timeo);

		lock_sock(sk); /* 被唤醒后重新上锁 */

		/* 如果进程有待处理的信号，或者睡眠超时了，退出循环，之后会返回错误码 */
		if (signal_pending(current) || !timeo)
			break;

		/* 继续睡眠吧 */
		prepare_to_wait(sk_sleep(sk), &wait, TASK_INTERRUPTIBLE);
	}

	/* 等待结束时，把等待进程从等待队列中删除，把当前进程的状态设为TASK_RUNNING */
	finish_wait(sk_sleep(sk), &wait);
	sk->sk_write_pending -= writebias;
	return timeo;
}
(2) 唤醒

三次握手中，当客户端收到SYNACK、发出ACK后，连接就成功建立了。此时连接的状态从TCP_SYN_SENT或TCP_SYN_RECV变为TCP_ESTABLISHED，sock的状态发生变化，会调用sock_def_wakeup()来处理连接状态变化事件，唤醒进程，connect()就能成功返回了。

sock_def_wakeup()的函数调用路径如下：

1
2
3
4
5
6
7
8
9

tcp_v4_rcv
	tcp_v4_do_rcv
		tcp_rcv_state_process
			tcp_rcv_synsent_state_process
				tcp_finish_connect
					sock_def_wakeup
						wake_up_interruptible_all
							__wake_up
								__wake_up_common
1
2
3
4
5
6
7
8
9
10

void tcp_finish_connect(struct sock *sk, struct sk_buff *skb)
{
	...
	tcp_set_state(sk, TCP_ESTABLISHED); /* 在这里设置为连接已建立的状态 */
	...
	if (! sock_flag(sk, SOCK_DEAD)) {
		sk->sk_state_change(sk); /* 指向sock_def_wakeup，会唤醒调用connect()的进程，完成连接的建立 */
		sk_wake_async(sk, SOCK_WAKE_IO, POLL_OUT); /* 如果使用了异步通知，则发送SIGIO通知进程可写 */
	}
}

accept等待

(1) 睡眠

accept()超时时间为sk->sk_rcvtimeo，在sock_init_data()中初始化为MAX_SCHEDULE_TIMEOUT，表示无限等待。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

/* 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); /* 进入睡眠直到超时或收到信号，或被IO事件处理函数唤醒 */

		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;
	}

	/* 从等待队列中删除等待任务，把等待进程的状态设为TASK_RUNNING */
	finish_wait(sk_sleep(sk), &wait);
	return err;
}
(2) 唤醒

三次握手中，当服务器端接收到ACK完成连接建立的时候，会把新的连接链入全连接队列中，然后唤醒监听socket上的等待进程，accept()就能成功返回了。

三次握手时，当收到客户端的ACK后，经过如下调用：

1
2
3
4
5
6
7

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()是如何避免惊群现象的。

1
2
3
4
5
6
7
8
9
10
11
12
13

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，表示只允许唤醒一个等待任务。

所以这里只会唤醒一个等待的进程，不会导致惊群现象。

Socket层实现系列 — 信号驱动的异步等待

http://blog.csdn.net/zhangskd/article/details/45932775

主要内容：Socket的异步通知机制。

内核版本：3.15.2

概述

socket上定义了几个IO事件：状态改变事件、有数据可读事件、有发送缓存可写事件、有IO错误事件。对于这些事件，socket中分别定义了相应的事件处理函数，也称回调函数。

Socket I/O事件的处理过程中，要使用到sock上的两个队列：等待队列和异步通知队列，这两个队列中都保存着等待该Socket I/O事件的进程。

Q：为什么要使用两个队列，等待队列和异步通知队列有什么区别呢？
A：等待队列上的进程会睡眠，直到Socket I/O事件的发生，然后在事件处理函数中被唤醒。异步通知队列上的进程则不需要睡眠，Socket I/O事件发时，事件处理函数会给它们发送到信号，这些进程事先注册的信号处理函数就能够被执行。

异步通知队列

Socket层使用异步通知队列来实现异步等待，此时等待Socket I/O事件的进程不用睡眠。

1
2
3
4
5
6
7
8
9
10
11
12

struct sock {
	...
	struct socket_wq __rcu *sk_wq; /* socket的等待队列和异步通知队列 */
	...
}

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;
};
1
2
3
4
5
6
7
8

struct fasync_struct {
	spinlock_t fa_lock;
	int magic;
	int fa_fd; /* 文件描述符 */
	struct fasync_struct *fa_next; /* 用于链入单向链表 */
	struct file *fa_file; /* fa_file->f_owner记录接收信号的进程 */
	struct rcu_head fa_rcu;
};

通过之前的blog《linux的异步通知机制》，我们知道为了能处理协议栈发出的SIGIO信号，

用户程序需要做的事情有：
1. 通过signal()指定SIGIO的处理函数。
2. 设置sockfd的拥有者为本进程，如此一来本进程才能收到协议栈发出的SIGIO信号。
3. 设置sockfd支持异步通知，即设置O_ASYNC标志。

对应的用户程序函数调用大概如下：

1
2
3
4

signal(SIGIO, my_handler); /* set new SIGIO handler */
fcntl(sockfd, F_SETOWN, getpid()); /* set sockfd's owner process */
oflags = fcntl(sockfd, F_GETFL); /* get old sockfd flags */
fcntl(sockfd, F_SETFL, oflags | O_ASYNC); /* set new sockfd flags */

下文关注的是内核层面的一些工作：
1. 如何把进程加入Socket的异步通知队列，或者把进程从Socket的异步通知队列中删除。
2. 协议栈何时发送信号给Socket异步通知队列上的进程。

插入和删除

首先来看下fcntl()的系统调用。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

SYSCALL_DEFINE3(fcntl, unsigned int, fd, unsigned int, cmd, unsigned long, arg)
{
	struct fd f = fdget_raw(fd);
	long err = -EBADF; /* Bad file number */

	if (! f.file)
		goto out;

	/* File is opened with O_PATH, almost nothing can be done with it */
	if (unlikely(f.file->f_mode & FMODE_PATH)) {
		if (! check_fcntl_cmd(cmd))
			goto out1;
	}

	err = security_file_fcntl(f.file, cmd, arg);
	if (! err)
		err = do_fcntl(fd, cmd, arg, f.file); /* 实际的处理函数 */

out1:
	fdput(f);
out:
	return err;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

static long do_fcntl(int fd, unsigned int cmd, unsigned long arg, struct fil *filp)
{
	long err = -EINVAL;

	switch(cmd) {
	...
	case F_SETFL: /* 在这里设置O_ASYNC标志 */
		err = setfl(fd, filp, arg);
		break;
	...
	case F_SETOWN: /* 在这里设置所有者进程 */
		err = f_setown(filp, arg, 1);
		break;
	....
	}

	return err;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

static int setfl(int fd, struct file *filp, unsigned long arg)
{
	...
	/* ->fasync() is responsible for setting the FASYNC bit. */
	if (((arg ^ filp->f_flags) & FASYNC) && filp->f_op->fasync) {
		error = filp->f_op->fasync(fd, filp, (arg & FASYNC) != 0);

		if (error < 0)
			goto out;
		if (error > 0)
			error = 0;
	}
	...
}

Socket文件的操作函数集为socket_file_ops。

1
2
3
4
5

static const struct file_operations socket_file_ops = {
	...
	.fasync = sock_fasync,
	...
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

/* Update the socket async list. */
static int sock_fasync(int fd, struct file *filp, int on)
{
	struct socket *sock = filp->private_data;
	struct sock *sk = sock->sk;
	struct socket_wq *wq; /* Socket的等待队列和异步通知队列 */

	if (sk == NULL)
		return -EINVAL;

	lock_sock(sk);
	wq = rcu_dereference_protected(sock->wq, sock_owned_by_user(sk));

	fasync_helper(fd, filp, on, &wq->fasync_list); /* 使用此函数来插入或删除 */

	/* 设置或取消SOCK_FASYNC标志 */
	if (! wq->fasync_list)
		sock_reset_flag(sk, SOCK_FASYNC);
	else
		sock_set_flag(sk, SOCK_FASYNC);

	release_sock(sk);

	return 0;
}

和设备驱动一样，最终调用fasync_helper()来把进程插入异步通知队列，或者把进程从异步通知队列中删除。

1
2
3
4
5
6
7
8
9
10
11
12
13

/*
 * fasync_helper() is used by almost all character device drivers to set up the fasync
 * queue, and for regular files by the file lease code. It returns negative on error, 0 if
 * it did no changes and positive if it added / deleted the entry.
 */

int fasync_helper(int fd, struct file *filp, int on, struct fasync_struct **fapp)
{
	if (! on)
		return fasync_remove_entry(filp, fapp); /* 插入 */

	return fasync_add_entry(fd, filp, fapp); /* 删除 */
}

发送信号

当Socket I/O事件触发时，协议栈会调用sk_wake_async()来进行异步通知。

函数的处理方式：

1
2
3
4
5
6

enum {
	SOCK_WAKE_IO, /* 直接发送SIGIO信号 */
	SOCK_WAKE_WAITD, /* 检测应用程序是否通过recv()类调用来等待接收数据，如果没有才发送SIGIO信号 */
	SOCK_WAKE_SPACE, /* 检测sock的发送队列是否曾经到达上限，如果有的话发送SIGIO信号 */
	SOCK_WAKE_URG, /* 直接发送SIGURG信号 */
};

通告的IO类型，常用的有：

1
2
3
4
5
6
7

#define __SI_POLL 0
#define POLL_IN (__SI_POLL | 1) /* data input available, 有接收数据可读 */
#define POLL_OUT (__SI_POLL | 2) /* output buffers available, 有输出缓存可写 */
#define POLL_MSG (__SI_POLL | 3) /* input message available, 有输入消息可读 */
#define POLL_ERR (__SI_POLL | 4) /* i/0 error, I/O错误 */
#define POLL_PRI (__SI_POLL | 5) /* high priority input available, 有紧急数据可读 */
#define POLL_HUP (__SI_POLL | 6) /* device disconnected, 设备关闭或文件关闭，无法继续读写 */

how为函数的处理方式，band为通告的IO类型。

1
2
3
4
5

static inline void sk_wake_async(struct sock *sk, int how, int band)
{
	if (sock_flag(sk, SOCK_FASYNC)) /* sock需要支持异步通知 */
		sock_wake_async(sk->sk_socket, how, band);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

int sock_wake_async(struct socket *sock, int how, int band)
{
	struct socket_wq *wq;

	if (! sock)
		return -1;

	rcu_read_lock();
	wq = rcu_dereference(sock->wq); /* socket的等待队列和异步通知队列 */

	if (! wq || !wq->fasync_list) { /* 如果有队列没有实例 */
		rcu_read_unlock();
		return -1;
	}

	switch(how) {
	/* 检测应用程序是否通过recv()类调用来等待接收数据，如果没有才发送SIGIO信号 */
	case SOCK_WAKE_WAITD:
		if (test_bit(SOCK_ASYNC_WAITDATA, &sock->flags))
			break;
		goto call_kill;

	/* 检测sock的发送队列是否曾经到达上限，如果有的话发送SIGIO信号 */
	case SOCK_WAKE_SPACE:
		if (! test_and_clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags))
			break;
	/* fall_through */

	case SOCK_WAKE_IO: /* 直接发送SIGIO信号 */
call_kill:
			/* 发送SIGIO信号给异步通知队列上的进程，告知IO消息 */
			kill_fasync(&wq->fasync_list, SIGIO, band);
			break;

	case SOCK_WAKE_URG:
			/* 发送SIGURG信号给异步通知队列上的进程 */
			kill_fasync(&wq->fasync_list, SIGURG, band);
	}

	rcu_read_unlock();
	return 0;
}

和设备驱动一样，最终调用kill_fasync()来发送信号给用户进程。

1
2
3
4
5
6
7
8
9

void kill_fasync(struct fasync_struct **fp, int sig, int band)
{
	/* First a quick test without locking: usually the list is empty. */
	if (*f) {
		rcu_read_lock();
		kill_fasync_rcu(rcu_dereference(*fp), sig, band);
		rcu_read_unlock();
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

static void kill_fasync_rcu(struct fasync_struct *fa, int sig, int band)
{
	while (fa) {
		struct fown_struct *fown;
		unsigned long flags;

		if (fa->magic != FASYNC_MAGIC) {
			printk(KERN_ERR "kill_fasync: bad magic number in fasync_struct!\n");
			return;
		}

		spin_lock_irqsave(&fa->fa_lock, flags);
		if (fa->fa_file) {
			fown = &fa->file->f_owner; /* 持有文件的进程 */

			/* Don't send SIGURG to processes which have not set a queued signum:
			 * SIGURG has its own default signalling mechanism. */

			if (! (sig == SIGURG && fown->signum == 0))
				send_sigio(fown, fa->fa_fd, band); /* 发送信号给持有文件的进程 */
		}
		spin_unlock_irqrestore(&fa->fa_lock, flags);

		fa = rcu_dereference(fa->fa_next); /* 指向下一个异步通知结构体 */
	}
}