TCP的发送系列 — tcp_sendmsg()的实现（二）

主要内容：Socket发送函数在TCP层的实现

内核版本：3.15.2

我的博客：http://blog.csdn.net/zhangskd

在上篇blog中分析了tcp_sendmsg()这个主要函数的实现，现在来看下之前略过的一些细节，

包括等待连接的建立、tcp_push()的实现、tcp_autocorking和数据的复制。

等待连接建立

在tcp_sendmsg()中如果发现连接尚未建立，会调用sk_stream_wait_connect()来等待连接的建立，

连接成功建立时返回0，之后才能发送数据。

/* Wait for a socket to get into the connected state
 * @sk: sock to wait on
 * @timeo_p: for how long to wait
 * Must be called with the socket locked.
 */

int sk_stream_wait_connect(struc sock *sk, long *timeo_p)
{
    struct task_struct *tsk = current;
    DEFINE_WAIT(wait); /* 初始化等待任务 */
    int done;

    do {
        int err = sock_error(sk);

        /* 连接发生错误 */
        if (err) 
            return err;

        /* 此时连接必须处于SYN_SENT或SYN_RECV的状态 */
        if (1 << sk->sk_state) & ~(TCPF_SYN_SENT | TCPF_SYN_RECV))
            return -EPIPE; /* Broken pipe */

        /* 如果是非阻塞的，或者等待时间耗尽了，直接返回 */
        if (! *timeo_p)
            return -EAGAIN; /* Try again */

        /* 如果进程有待处理的信号，返回 */
        if (signal_pending(tsk))
            return sock_intr_errno(*timeo_p);

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

        /* 进入睡眠，返回值为真的条件：
         * 连接没有发生错误，且状态为ESTABLISHED或CLOSE_WAIT。
         */
        done = sk_wait_event(sk, timeo_p, ! sk->sk_err &&
                     ! ((1 << sk->sk_state) & ~(TCPF_ESTABLISHED | TCPF_CLOSE_WAIT))); 
 
        /* 把等待任务从等待队列中删除，把当前进程的状态设为TASK_RUNNING */
        finish_wait(sk_sleep(sk), &wait);
        sk->sk_write_pending--; /* 更新写等待计数 */
    } while (! done)

    return 0;
}

#define sk_wait_event(__sk, __timeo, __condition)    
    ({    int __rc;    
           release_sock(__sk);    
           __rc = __condition;    

          if (! __rc) {    
              *(__timeo) = schedule_timeout(*(__timeo));    
          }    

          lock_sock(__sk);    
          __rc = __condition;    
          __rc;    
    }) 

static inline long sock_sndtimeo(const struct sock *sk, bool noblock)
{
    return noblock ? 0 : sk->sk_sndtimeo;
}

tcp_push

tcp_sendmsg()中，在sock发送缓存不足、系统内存不足或应用层的数据都拷贝完毕等情况下，

都会调用tcp_push()来把已经拷贝到发送队列中的数据给发送出去。

tcp_push()主要做了以下事情：

1. 检查是否有未发送过的数据。

2. 检查是否需要设置PSH标志。

3. 检查是否使用了紧急模式。

4. 检查是否需要使用自动阻塞。

5. 尽可能地把发送队列中的skb给发送出去。

static void tcp_push(struct sock *sk, int flags, int mss_now, int nonagle, int size_goal)
{
    struct tcp_sock *tp = tcp_sk(sk);
    struct sk_buff *skb;

    /* 如果没有未发送过的数据 */
    if (! tcp_send_head(sk))
        return;
 
    /* 发送队列的最后一个skb */
    skb = tcp_write_queue_tail(sk);
 
    /* 如果接下来没有更多的数据需要发送，或者距离上次PUSH后又有比较多的数据，
     * 那么就需要设置PSH标志，让接收端马上把接收缓存中的数据提交给应用程序。
     */
    if (! (flags & MSG_MORE) || forced_push(tp))
        tcp_mark_push(tp, skb);
 
    /* 如果设置了MSG_OOB标志，就记录紧急指针 */
    tcp_mark_urg(tp, flags);

    /* 如果需要自动阻塞小包 */
    if (tcp_should_autocork(sk, skb, size_goal)) {
        /* avoid atomic op if TSQ_THROTTED bit is already set, 设置阻塞标志位 */
        if (! test_bit(TSQ_THROTTLED, &tp->tsq_flags)) {
            NET_INC_STATS(sock_net(sk), LINUX_MIB_TCPAUTOCORKING);
            set_bit(TSQ_THROTTLED, &tp->tsq_flags);
        }
       
        /* It is possible TX completion already happened before we set TSQ_THROTTED.
         * 我的理解是，当提交给IP层的数据包都发送出去后，sk_wmem_alloc的值就会变小，
         * 此时这个条件就为假，之后可以发送被阻塞的数据包了。
         */
        if (atomic_read(&sk->sk_wmem_alloc) > skb->truesize)
            return;
    }

    /* 如果之后还有更多的数据，那么使用TCP CORK，显式地阻塞发送 */
    if (flags & MSG_MORE)
        nonagle = TCP_NAGLE_CORK;

    /* 尽可能地把发送队列中的skb发送出去。
     * 如果发送失败，检查是否需要启动零窗口探测定时器。
     */
    __tcp_push_pending_frames(sk, mss_now, nonagle);
}

判断是否要设置PSH标志：

如果此时发送队列的最后一个字节序号，和上次PSH的最后一个字节序号，

它们的间隔超过了对端通告过的最大接收窗口的一半，就需要设置。

static inline bool forced_push(const struct tcp_sock *tp)
{
    /* write_seq：发送队列最后一个字节的序号+1
    * pushed_seq：上次PUSH的最后一个字节
    * max_window：对端层通告过的最大接收窗口
    */
    return after(tp->write_seq, tp->pushed_seq + (tp->max_window >> 1));
} 

static inline void tcp_mark_push(struct tcp_sock *tp, struct sk_buff *skb)
{
    TCP_SKB_CB(skb)->tcp_flags |= TCPHDR_PSH; /* 设置PSH标志 */
    tp->pushed_seq = tp->write_seq; /* 记录本次PUSH的最后一个字节序号 */
}

static inline void tcp_mark_urg(struct tcp_sock *tp, int flags)
{
    if (flags & MSG_OOB)
        tp->snd_up = tp->write_seq; 
}

tcp_push_pending_frames()和__tcp_push_pending_frames()简单的封装了下tcp_write_xmit()。

从tcp_write_xmit()开始，TCP层才真正开始发送数据。 

/* Push out any pending frames which were held back due to TCP_CORK
 * or attempt at coalescing tiny packets.
 * The socket must be locked by the caller.
 */
void __tcp_push_pending_frames(struct sock *sk, unsigned int cur_mss, int nonagle)
{
    /* If we are closed, the bytes will have to remain here.
     * In time closedown will finish, we empty the write queue and
     * all will be happy.
     */
    if (unlikely(sk->sk_state == TCP_CLOSE))
        return;

    /* 如果发送失败 */
    if (tcp_write_xmit(sk, cur_mss, nonagle, 0, sk_gfp_atomic(sk, GFP_ATOMIC)))
        tcp_check_probe_timer(sk); /* 检查是否需要启用0窗口探测定时器*/
} 

tcp_autocorking

当应用程序连续地发送小包时，如果能够把这些小包合成一个全尺寸的包再发送，无疑可以减少

总的发包个数。tcp_autocorking的思路是当规则队列Qdisc、或网卡的发送队列中有尚未发出的

数据包时，那么就延迟小包的发送，等待应用层的后续数据，直到Qdisc或网卡发送队列的数据

包成功发送出去为止。

sysctl_tcp_autocorking - BOOLEAN

Enable TCP auto corking:

When  applications do consecutive small write()/sendmsg() system calls,

we try to coalesce these small writes as much as possible, to lower total

amount of sent packets. This is done if at least one prior packet for the flow

is waiting in Qdisc queues or device transmit queue. Applications can still

use TCP_CORK for optimal behavior when they know how/when to uncork

their sockets.

Default: 1

Patch: http://lwn.net/Articles/576263/

同时满足以下条件时，tcp_push()才会自动阻塞：

1. 数据包为小包，即数据长度小于最大值。

2. 使用了tcp_autocorking，这个值默认为1。

3. 此数据包不是发送队列的第一个包，即前面有数据包被发送了。

4. Qdisc或Nic queues必须有数据包，而不能只是纯ACK包。

/* If a not yet filled skb is pushed, do not send it if we have data packets
 * in Qdisc or NIC queues: Because TX completion will happen shortly,
 * it gives a chance to coalesce future sendmsg() payload into this skb,
 * without need for a timer, and with no latency trade off.
 * As packts containing data payload have a bigger truesize than pure
 * acks (dataless) packets, the last checks prevent autocorking if we only
 * have an ACK in Qdisc/NIC queues, or if TX completion was delayed
 * after we processed ACK packet.
 */

static bool tcp_should_autocork(struct sock *sk, struct sk_buff *skb, int size_goal)
{
    return skb->len < size_goal && sysctl_tcp_autocorking &&
        skb != tcp_write_queue_head(sk) &&
        atomic_read(&sk->sk_wmem_alloc) > skb->truesize;
}

Q：什么时候会取消自动阻塞呢？

A：在tcp_push()中会检查，if (atomic_read(&sk->sk_wmem_alloc) > skb->truesize)

当提交给IP层的数据包都发送出去后，sk_wmem_alloc的值就会变小，此时这个条件就为假，

之后可以发送被阻塞的数据包了。

数据的复制

tcp_sendmsg()的一项主要工作，就是把用户层的数据填充到发送队列的skb中。

skb_availroom返回skb的data room大小，如果有非线性数据区，就返回0。

/* 
 * bytes at buffer end.
 * Return the number of bytes of free space at the tail of an sk_buff
 * allocated by sk_stream_alloc()
 */
static inline int skb_availroom(const struct sk_buff *skb)
{
    /* skb->data_len不为零，表示有非线性的数据区 */
    if (skb_is_nonlinear(skb)) 
        return 0;
 
    /* data room的大小 */
    return skb->end - skb->tail - skb->reserved_tailroom;
}

验证用户空间的数据可读，拷贝用户空间的数据到内核空间。如果需要TCP自己计算校验和，

那么同时计算用户层数据的校验和。

static inline int skb_add_data_nocache(struct sock *sk, struct sk_buff *skb,
    char __user *from, int copy)
{
    int err, offset = skb->len;

    /* 拷贝用户空间的数据到内核空间，同时计算校验和 */
    err = skb_do_copy_data_nocache(sk, skb, from, skb_put(skb, copy), copy, offset);

    /* 如果拷贝失败，恢复skb->len和data room的大小 */
    if (err)
        __skb_trim(skb, offset);

    return err;
}

static inline int skb_do_copy_data_nocache(struct sock *sk, struct sk_buff *skb,
    char __user *from, char *to, int copy, int offset)
{
    /* 需要TCP自己计算校验和 */
    if (skb->ip_summed == CHECKSUM_NONE) {
        int err = 0;

        /* 拷贝用户空间的数据到内核空间，同时计算用户数据的校验和 */
        __wsum csum = csum_and_copy_from_user(from, to, copy, 0, &err);

        skb->csum = csum_block_add(skb->csum, csum, offset); /* 累加校验和 */

    } else if (sk->sk_route_caps & NETIF_F_NOCACHE_COPY) {
        if (! access_ok(VERIFY_READ, from, copy) || 
            __copy_from_user_nocache(to, from, copy))

        return -EFAULT;

    } else if (copy_from_user(to, from, copy))
        return -EFAULT;

    return 0;
}

向下扩大data room，返回扩大之前的tail指针。

/* add data to a buffer.
 * This function extends the used data area of the buffer. If this would
 * exceed the total buffer size the kernel will panic. A pointer to the first
 * byte of the extra data is returned.
 */
unsigned char *skb_put(struct sk_buff *skb, unsigned int len)
{
    unsigned char *tmp = skb_tail_pointer(skb);
    SKB_LINEAR_ASSERT(skb);

    skb->tail += len;
    skb->len += len;
    if (unlikely(skb->tail > skb->end))
        skb_over_panic(skb, len, __builtin_return_address(0));

    return tmp;
}

static inline void __skb_trim(struct sk_buff *skb, unsigned int len)
{
    if (unlikely(skb_is_nolinear(skb))) {
        WARN_ON(1);
        return;
    }

    skb->len = len; /* 恢复原来的长度 */
    skb_set_tail_pointer(skb, len); /* 恢复data room的大小 */
}

拷贝用户空间的数据到内核空间，同时计算校验和。同时更新skb->len、skb->data_len、

skb->truesize、sk->sk_wmem_queued和sk->sk_forward_alloc。

static inline int skb_copy_to_page_nocache(struct sock *sk, char __user *from,
    struct sk_buff *skb, struct page *page, int off, int copy)
{
    int err;

    /* 拷贝用户空间的数据到内核空间，同时计算校验和 */
    err = skb_do_copy_data_nocache(sk, skb, from, page_address(page) + off,
                  copy, skb->len);
    if (err)
        return err;

    skb->len += copy;
    skb->data_len += copy;
    skb->truesize += copy;

    sk->sk_wmem_queued += copy;
    sk_mem_charge(sk, copy);

    return 0;
}