tcp_recvmsg 函数具体解释

看了非常多网上关于tcp_recvmsg的文章，感觉解释的不太到位，或者非常多都是空口说白话，昨天分析了一下午tcp_recvmsg。感觉了解了十之八九，如今贴出来和大家分享一下。

须要背景：了解tcp三个接收队列 prequeue,backlog,receive的各自用处。

/*
*	This routine copies from a sock struct into the user buffer.
*
*	Technical note: in 2.3 we work on _locked_ socket, so that
*	tricks with *seq access order and skb->users are not required.
*	Probably, code can be easily improved even more.
*/

int tcp_recvmsg(struct kiocb *iocb, struct sock *sk, struct msghdr *msg,
size_t len, int nonblock, int flags, int *addr_len)
{
struct tcp_sock *tp = tcp_sk(sk);
int copied = 0;
u32 peek_seq;
u32 *seq;
unsigned long used;
int err;
int target;	 /* Read at least this many bytes */
long timeo;
struct task_struct *user_recv = NULL;
int copied_early = 0;
struct sk_buff *skb;
u32 urg_hole = 0;

//功能：“锁住sk”，并不是真正的加锁，而是运行sk->sk_lock.owned = 1
//目的：这样软中断上下文可以通过owned 。推断该sk是否处于进程上下文。
//提供一种同步机制。
lock_sock(sk);

TCP_CHECK_TIMER(sk);

err = -ENOTCONN;
if (sk->sk_state == TCP_LISTEN)
goto out;

//获取延迟，假设用户设置为非堵塞，那么timeo ==0000 0000 0000 0000
//假设用户使用默认recv系统调用
//则为堵塞，此时timeo ==0111 1111 1111 1111
//timeo 就2个值
timeo = sock_rcvtimeo(sk, nonblock);

/* Urgent data needs to be handled specially. */
if (flags & MSG_OOB)
goto recv_urg;

//待拷贝的下一个序列号
seq = &tp->copied_seq;

//设置了MSG_PEEK，表示不让数据从缓冲区移除。目的是下一次调用recv函数
//仍然可以读到同样数据
if (flags & MSG_PEEK) {
peek_seq = tp->copied_seq;
seq = &peek_seq;
}

//假设设置了MSG_WAITALL。则target  ==len，即recv函数中的參数len
//假设没设置MSG_WAITALL。则target  == 1
target = sock_rcvlowat(sk, flags & MSG_WAITALL, len);

//大循环
do {
u32 offset;

/* Are we at urgent data? Stop if we have read anything or have SIGURG pending. */
if (tp->urg_data && tp->urg_seq == *seq) {
if (copied)
break;
if (signal_pending(current)) {
copied = timeo ? sock_intr_errno(timeo) : -EAGAIN;
break;
}
}

/* Next get a buffer. */

//小循环
skb_queue_walk(&sk->sk_receive_queue, skb) {
/* Now that we have two receive queues this
* shouldn't happen.
*/
if (WARN(before(*seq, TCP_SKB_CB(skb)->seq),
KERN_INFO "recvmsg bug: copied %X "
"seq %X rcvnxt %X fl %X\n", *seq,
TCP_SKB_CB(skb)->seq, tp->rcv_nxt,
flags))
break;

//假设用户的缓冲区(即用户malloc的buf)长度够大。offset通常是0。

//即 “下次准备拷贝数据的序列号”==此时获取报文的起始序列号
//什么情况下offset >0呢？非常简答。假设用户缓冲区12字节，而这个skb有120字节
//那么一次recv系统调用，仅仅能获取skb中的前12个字节，下一次运行recv系统调用
//offset就是12了。<span style="font-family: Arial, Helvetica, sans-serif;">offset</span><span style="font-family: Arial, Helvetica, sans-serif;">表示从第12个字节開始读取数据，前12个字节已经读取了。</span>
//那这个"已经读取12字节"这个消息。存在哪呢？
//在*seq = &tp->copied_seq;中
offset = *seq - TCP_SKB_CB(skb)->seq;
if (tcp_hdr(skb)->syn)
offset--;
if (offset < skb->len)
goto found_ok_skb;
if (tcp_hdr(skb)->fin)
goto found_fin_ok;
WARN(!(flags & MSG_PEEK), KERN_INFO "recvmsg bug 2: "
"copied %X seq %X rcvnxt %X fl %X\n",
*seq, TCP_SKB_CB(skb)->seq,
tp->rcv_nxt, flags);
}

//运行到了这里。表明小循环中break了，既然break了，说明sk_receive_queue中
//已经没有skb能够读取了
//假设没有运行到这里说明前面的小循环中运行了goto，读到实用的skb，或者读到fin都会goto。

//没有skb能够读取，说明什么？
//可能性1：当用户第一次调用recv时，压根没有数据到来
//可能性2：skb->len一共20字节，假设用户调用一次 recv。读取12字节，再调用recv，
//读取12字节，此时skb由于上次已经被读取了12字节，仅仅剩下8字节。
//于是代码的逻辑上。再会要求获取skb，来读取剩下的8字节。

//可能性1的情况下，copied == 0。肯定不会进这个if。兴许将运行休眠
//可能性2的情况下，情况比較复杂。可能性2表明数据没有读够用户想要的len长度
//尽管进程上下文中，没有读够数据。可是可能我们在读数据的时候
//软中断把数据放到backlog队列中了，而backlog对队列中的数据也许恰好让我们读够数
//据。

//copied了数据的，copied肯定>=1，而target 是1或者len
//copied仅仅能取0(可能性1)，或者0~len(可能性2)
//copied >= target 表示我们取得我们想要的数据了，何必进行休眠。直接return
//假设copied 没有达到我们想要的数据。则看看sk_backlog是否为空
//空的话，尽力了。仅仅能尝试休眠
//非空的话，另一线希望。我们去sk_backlog找找数据，看看能否够达到我们想要的
//数据大小

//我认为copied == target是会出现的，可是出现的话，也不会进如今这个流程
//，如有不正确，请各位大神指正，告诉我
//说明情况下copied == target

/* Well, if we have backlog, try to process it now yet. */
if (copied >= target && !sk->sk_backlog.tail)
break;

if (copied) {
//可能性2，拷贝了数据。可是没有复制到指定大小
if (sk->sk_err ||
sk->sk_state == TCP_CLOSE ||
(sk->sk_shutdown & RCV_SHUTDOWN) ||
!timeo ||
signal_pending(current))
break;
} else {
//可能性1
if (sock_flag(sk, SOCK_DONE))
break;

if (sk->sk_err) {
copied = sock_error(sk);
break;
}

if (sk->sk_shutdown & RCV_SHUTDOWN)
break;

if (sk->sk_state == TCP_CLOSE) {
if (!sock_flag(sk, SOCK_DONE)) {
/* This occurs when user tries to read
* from never connected socket.
*/
copied = -ENOTCONN;
break;
}
break;
}

//是否是堵塞的。不是。就return了。
if (!timeo) {
copied = -EAGAIN;
break;
}

if (signal_pending(current)) {
copied = sock_intr_errno(timeo);
break;
}
}

tcp_cleanup_rbuf(sk, copied);

//sysctl_tcp_low_latency 默认0tp->ucopy.task == user_recv肯定也成立

if (!sysctl_tcp_low_latency && tp->ucopy.task == user_recv) {
/* Install new reader */
if (!user_recv && !(flags & (MSG_TRUNC | MSG_PEEK))) {
user_recv = current;
tp->ucopy.task = user_recv;
tp->ucopy.iov = msg->msg_iov;
}

tp->ucopy.len = len;

WARN_ON(tp->copied_seq != tp->rcv_nxt &&
!(flags & (MSG_PEEK | MSG_TRUNC)));

/* Ugly... If prequeue is not empty, we have to
* process it before releasing socket, otherwise
* order will be broken at second iteration.
* More elegant solution is required!!!
*
* Look: we have the following (pseudo)queues:
*
* 1. packets in flight
* 2. backlog
* 3. prequeue
* 4. receive_queue
*
* Each queue can be processed only if the next ones
* are empty. At this point we have empty receive_queue.
* But prequeue _can_ be not empty after 2nd iteration,
* when we jumped to start of loop because backlog
* processing added something to receive_queue.
* We cannot release_sock(), because backlog contains
* packets arrived _after_ prequeued ones.
*
* Shortly, algorithm is clear --- to process all
* the queues in order. We could make it more directly,
* requeueing packets from backlog to prequeue, if
* is not empty. It is more elegant, but eats cycles,
* unfortunately.
*/
//
if (!skb_queue_empty(&tp->ucopy.prequeue))
goto do_prequeue;

/* __ Set realtime policy in scheduler __ */
}

if (copied >= target) {
/* Do not sleep, just process backlog. */
release_sock(sk);
lock_sock(sk);
} else
sk_wait_data(sk, &timeo);
//在此处睡眠了，将在tcp_prequeue函数中调用wake_up_interruptible_poll唤醒

//软中断会推断用户是正在读取检查而且睡眠了，假设是的话，就直接把数据拷贝
//到prequeue队列，然后唤醒睡眠的进程。由于进程睡眠，表示没有读到想要的字节数
//此时，软中断有数据到来，直接给进程。这样进程就能以最快的速度被唤醒。

if (user_recv) {
int chunk;

/* __ Restore normal policy in scheduler __ */

if ((chunk = len - tp->ucopy.len) != 0) {
NET_ADD_STATS_USER(sock_net(sk), LINUX_MIB_TCPDIRECTCOPYFROMBACKLOG, chunk);
len -= chunk;
copied += chunk;
}

if (tp->rcv_nxt == tp->copied_seq &&
!skb_queue_empty(&tp->ucopy.prequeue)) {
do_prequeue:
tcp_prequeue_process(sk);

if ((chunk = len - tp->ucopy.len) != 0) {
NET_ADD_STATS_USER(sock_net(sk), LINUX_MIB_TCPDIRECTCOPYFROMPREQUEUE, chunk);
len -= chunk;
copied += chunk;
}
}
}
if ((flags & MSG_PEEK) &&
(peek_seq - copied - urg_hole != tp->copied_seq)) {
if (net_ratelimit())
printk(KERN_DEBUG "TCP(%s:%d): Application bug, race in MSG_PEEK.\n",
current->comm, task_pid_nr(current));
peek_seq = tp->copied_seq;
}
continue;

found_ok_skb:
/* Ok so how much can we use?

*/
//skb中还有多少聚聚没有拷贝。

//正如前面所说的，offset是上次已经拷贝了的，这次从offset開始接下去拷贝
used = skb->len - offset;
//非常有可能used的大小，即skb剩余长度。依旧大于用户的缓冲区大小(len)。所以依旧
//仅仅能拷贝len长度。一般来说，用户还得运行一次recv系统调用。直到skb中的数据读完
if (len < used)
used = len;

/* Do we have urgent data here?

*/
if (tp->urg_data) {
u32 urg_offset = tp->urg_seq - *seq;
if (urg_offset < used) {
if (!urg_offset) {
if (!sock_flag(sk, SOCK_URGINLINE)) {
++*seq;
urg_hole++;
offset++;
used--;
if (!used)
goto skip_copy;
}
} else
used = urg_offset;
}
}

if (!(flags & MSG_TRUNC)) {
{
//一般都会进这个if。进行数据的拷贝，把能够读到的数据，放到用户的缓冲区
err = skb_copy_datagram_iovec(skb, offset,
msg->msg_iov, used);
if (err) {
/* Exception. Bailout! */
if (!copied)
copied = -EFAULT;
break;
}
}
}

//更新标志位，seq 是指针，指向了tp->copied_seq
//used是我们有能力拷贝的数据大小，即已经复制到用户缓冲区的大小
//正如前面所说。假设用户的缓冲区非常小。一次recv拷贝不玩skb中的数据，
//我们须要保存已经拷贝了的大小，下次recv时。从这个大小处继续拷贝。

//所以须要更新copied_seq。
*seq += used;
copied += used;
len -= used;

tcp_rcv_space_adjust(sk);

skip_copy:
if (tp->urg_data && after(tp->copied_seq, tp->urg_seq)) {
tp->urg_data = 0;
tcp_fast_path_check(sk);
}

//这个就是推断我们是否拷贝完了skb中的数据，假设没有continue
//这样的情况下，len经过 len -= used; 。已经变成0，所以continue的效果相当于
//退出了这个大循环。

能够理解，你仅仅能拷贝len长度。拷贝完之后。那就return了。

//另一种情况used + offset ==  skb->len，表示skb拷贝完了。这时我们仅仅须要释放skb
//以下会讲到
if (used + offset < skb->len)
continue;

//看看这个数据报文是否含有fin，含有fin，则goto到found_fin_ok
if (tcp_hdr(skb)->fin)
goto found_fin_ok;

//运行到这里，标明used + offset ==  skb->len，报文也拷贝完了，那就把skb摘链释放
if (!(flags & MSG_PEEK)) {
sk_eat_skb(sk, skb, copied_early);
copied_early = 0;
}
//这个cintinue不一定是退出大循环，可能还会运行循环。

//假设用户设置缓冲区12字节。你skb->len长度20字节。
//第一次recv读取了12字节，skb剩下8。下一次调用recv再想读取12，
//可是仅仅能读取到这8字节了。

//此时len 变量长度为4，那么这个continue依旧在这个循环中，
//函数还是再次从do開始。使用skb_queue_walk，找skb
//假设sk_receive_queue中skb仍旧有，那么继续读，直到len == 0
//假设没有skb了，我们怎么办？我们的len还有4字节怎么办？
//这得看用户设置的recv函数堵塞与否。即和timeo变量相关了。
continue;

found_fin_ok:
/* Process the FIN. */
++*seq;
if (!(flags & MSG_PEEK)) {
//把skb从sk_receive_queue中摘链
sk_eat_skb(sk, skb, copied_early);
copied_early = 0;
}
break;
} while (len > 0);

//到这里是大循环退出
//休眠过的进程，然后退出大循环 。才满足 if (user_recv) 条件
if (user_recv) {
if (!skb_queue_empty(&tp->ucopy.prequeue)) {
int chunk;

tp->ucopy.len = copied > 0 ? len : 0;

tcp_prequeue_process(sk);

if (copied > 0 && (chunk = len - tp->ucopy.len) != 0) {
NET_ADD_STATS_USER(sock_net(sk), LINUX_MIB_TCPDIRECTCOPYFROMPREQUEUE, chunk);
len -= chunk;
copied += chunk;
}
}

//数据读取完成。清零
tp->ucopy.task = NULL;
tp->ucopy.len = 0;
}

/* According to UNIX98, msg_name/msg_namelen are ignored
* on connected socket. I was just happy when found this 8) --ANK
*/

/* Clean up data we have read: This will do ACK frames. */
//非常重要。将更新缓存，而且适当的时候发送ack
tcp_cleanup_rbuf(sk, copied);

TCP_CHECK_TIMER(sk);
release_sock(sk);
return copied;

out:
TCP_CHECK_TIMER(sk);
release_sock(sk);
return err;

recv_urg:
err = tcp_recv_urg(sk, msg, len, flags);
goto out;
}