TCP层sendmsg系统调用的实现分析

概述

sendmsg系统调用在tcp层的实现是tcp_sendmsg函数，该函数完成以下任务：从用户空间读取数据，拷贝到内核skb，将skb加入到发送队列的任务，调用发送函数；函数在执行过程中会锁定控制块，避免软中断在tcp层的影响；函数核心流程为，在发送数据时，查看是否能够将数据合并到发送队列中最后一个skb中，如果不能合并，则新申请一个skb；拷贝过程中，如果skb的线性区域有空间，则优先使用线性区域，线性区域空间不足，则使用分页区域；拷贝完成后，调用发送函数发送数据；

代码分析

int tcp_sendmsg(struct sock *sk, struct msghdr *msg, size_t size)
{
    struct tcp_sock *tp = tcp_sk(sk);
    struct sk_buff *skb;
    struct sockcm_cookie sockc;
    int flags, err, copied = 0;
    int mss_now = 0, size_goal, copied_syn = 0;
    bool process_backlog = false;
    bool sg;
    long timeo;

    /* 加锁，避免与软中断的冲突 */
    lock_sock(sk);

    /* 获取标记 */
    flags = msg->msg_flags;

    /* fastopen和defer */
    if (unlikely(flags & MSG_FASTOPEN || inet_sk(sk)->defer_connect)) {
        err = tcp_sendmsg_fastopen(sk, msg, &copied_syn, size);
        if (err == -EINPROGRESS && copied_syn > 0)
            goto out;
        else if (err)
            goto out_err;
    }

    /* 获取阻塞时间，非阻塞为0 */
    timeo = sock_sndtimeo(sk, flags & MSG_DONTWAIT);

    /* 限速检查 */
    tcp_rate_check_app_limited(sk);  /* is sending application-limited? */

    /* Wait for a connection to finish. One exception is TCP Fast Open
     * (passive side) where data is allowed to be sent before a connection
     * is fully established.
     */
    /* 等待连接完成状态，fastopen的被动打开方例外 */
    if (((1 << sk->sk_state) & ~(TCPF_ESTABLISHED | TCPF_CLOSE_WAIT)) &&
        !tcp_passive_fastopen(sk)) {
        /* 等待连接完成 */
        err = sk_stream_wait_connect(sk, &timeo);
        if (err != 0)
            goto do_error;
    }

    if (unlikely(tp->repair)) {
        if (tp->repair_queue == TCP_RECV_QUEUE) {
            copied = tcp_send_rcvq(sk, msg, size);
            goto out_nopush;
        }

        err = -EINVAL;
        if (tp->repair_queue == TCP_NO_QUEUE)
            goto out_err;

        /* 'common' sending to sendq */
    }

    sockc.tsflags = sk->sk_tsflags;
    if (msg->msg_controllen) {
        err = sock_cmsg_send(sk, msg, &sockc);
        if (unlikely(err)) {
            err = -EINVAL;
            goto out_err;
        }
    }

    /* This should be in poll */
    /* 清除异步队列已满标记 */
    sk_clear_bit(SOCKWQ_ASYNC_NOSPACE, sk);

    /* Ok commence sending. */
    copied = 0;

restart:
    /* 获取mss,gso情况下size_goal记录总mss=页数*mss */
    mss_now = tcp_send_mss(sk, &size_goal, flags);

    err = -EPIPE;
    /* 已经关闭了发送端 */
    if (sk->sk_err || (sk->sk_shutdown & SEND_SHUTDOWN))
        goto do_error;

    /* 网卡分散聚合标记 */
    sg = !!(sk->sk_route_caps & NETIF_F_SG);

    /* 遍历发送缓存 */
    while (msg_data_left(msg)) {
        int copy = 0;
        int max = size_goal;

        /* 拿到发送队列的尾部skb */
        skb = tcp_write_queue_tail(sk);
        /* 有skb未发送 */
        if (tcp_send_head(sk)) {
            /* 网卡不支持校验和，调整最大长度 */
            if (skb->ip_summed == CHECKSUM_NONE)
                max = mss_now;
            /* 该skb还能够容纳的数据长度 */
            copy = max - skb->len;
        }

        /* 剩余空间为0，或者不能合并，分配一个新的skb */
        if (copy <= 0 || !tcp_skb_can_collapse_to(skb)) {
            bool first_skb;

new_segment:
            /* Allocate new segment. If the interface is SG,
             * allocate skb fitting to single page.
             */
            /* 空闲内存不足，进入等待 */
            if (!sk_stream_memory_free(sk))
                goto wait_for_sndbuf;

            if (process_backlog && sk_flush_backlog(sk)) {
                process_backlog = false;
                goto restart;
            }
            first_skb = skb_queue_empty(&sk->sk_write_queue);

            /* 分配skb */
            skb = sk_stream_alloc_skb(sk,
                          select_size(sk, sg, first_skb),
                          sk->sk_allocation,
                          first_skb);
            /* 分配失败，等待 */
            if (!skb)
                goto wait_for_memory;

            process_backlog = true;
            /*
             * Check whether we can use HW checksum.
             */
            /* 网卡允许计算校验和 */
            if (sk_check_csum_caps(sk))
                skb->ip_summed = CHECKSUM_PARTIAL;

            /* 添加到发送队列 */
            skb_entail(sk, skb);
            copy = size_goal;
            max = size_goal;

            /* All packets are restored as if they have
             * already been sent. skb_mstamp isn't set to
             * avoid wrong rtt estimation.
             */
            if (tp->repair)
                TCP_SKB_CB(skb)->sacked |= TCPCB_REPAIRED;
        }

        /* Try to append data to the end of skb. */
        /* 拷贝的数据不能超过实际数据块的长度 */
        if (copy > msg_data_left(msg))
            copy = msg_data_left(msg);

        /* Where to copy to? */
        /* 线性区域还有空间 */
        if (skb_availroom(skb) > 0) {
            /* We have some space in skb head. Superb! */
            /* 取要拷贝的数量和线性区域的较小值 */
            copy = min_t(int, copy, skb_availroom(skb));
            /* 从用户空间拷贝到内核 */
            err = skb_add_data_nocache(sk, skb, &msg->msg_iter, copy);
            if (err)
                goto do_fault;
        }
        /* 线性区域没有空间，则使用分页区 */
        else {
            bool merge = true;
            /* 获取页数量 */
            int i = skb_shinfo(skb)->nr_frags;
            /* 获取缓存的页 */
            struct page_frag *pfrag = sk_page_frag(sk);

            /* 检查是否有足够的空间，空间不足则申请新页,失败则等待 */
            if (!sk_page_frag_refill(sk, pfrag))
                goto wait_for_memory;

            /* 不能合并 */
            if (!skb_can_coalesce(skb, i, pfrag->page,
                          pfrag->offset)) {
                /* 页数量超过限制 || 网卡不支持分散聚合*/
                if (i >= sysctl_max_skb_frags || !sg) {
                    /* 增加push标记，尽快处理数据 */
                    tcp_mark_push(tp, skb);
                    /* 新申请一个skb */
                    goto new_segment;
                }
                /* 不合并 */
                merge = false;
            }

            /* 获取能够合并的空间 */
            copy = min_t(int, copy, pfrag->size - pfrag->offset);

            /* 发送合法性检查 */
            if (!sk_wmem_schedule(sk, copy))
                goto wait_for_memory;

            /* 拷贝数据到分页 */
            err = skb_copy_to_page_nocache(sk, &msg->msg_iter, skb,
                               pfrag->page,
                               pfrag->offset,
                               copy);
            if (err)
                goto do_error;

            /* Update the skb. */
            if (merge) {
                /* 合并的，需要增加数据量 */
                skb_frag_size_add(&skb_shinfo(skb)->frags[i - 1], copy);
            } else {
                /* 非合并的，对新页进行初始化 */
                skb_fill_page_desc(skb, i, pfrag->page,
                           pfrag->offset, copy);
                page_ref_inc(pfrag->page);
            }

            /* 记录新的偏移 */
            pfrag->offset += copy;
        }

        /* 第一次拷贝，清除push标记 */
        if (!copied)
            TCP_SKB_CB(skb)->tcp_flags &= ~TCPHDR_PSH;

        /* 更新发送队列的最后一个序号 */
        tp->write_seq += copy;
        /* 更新skb结束序号 */
        TCP_SKB_CB(skb)->end_seq += copy;
        tcp_skb_pcount_set(skb, 0);

        /* 更新拷贝数量 */
        copied += copy;

        /* 数据都拷贝完成 */
        if (!msg_data_left(msg)) {
            if (unlikely(flags & MSG_EOR))
                TCP_SKB_CB(skb)->eor = 1;
            goto out;
        }

        /* 还能继续拷贝数据 带外数据 修复模式，继续拷贝 */
        if (skb->len < max || (flags & MSG_OOB) || unlikely(tp->repair))
            continue;

        /* 需要使用push标记 */
        if (forced_push(tp)) {
            /* 打psh标记 */
            tcp_mark_push(tp, skb);
            /* 发送队列中的多个skb */
            __tcp_push_pending_frames(sk, mss_now, TCP_NAGLE_PUSH);
        } 
        /* 否则，有数据要发送，则发送一个skb */
        else if (skb == tcp_send_head(sk))
            tcp_push_one(sk, mss_now);
        continue;

wait_for_sndbuf:
        /* 设置空间不足标记 */
        set_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
wait_for_memory:
        /* 已经有拷贝数据到发送队列，则发送之 */
        if (copied)
            tcp_push(sk, flags & ~MSG_MORE, mss_now,
                 TCP_NAGLE_PUSH, size_goal);

        /* 等待内存足够 */
        err = sk_stream_wait_memory(sk, &timeo);
        if (err != 0)
            goto do_error;

        /* 重新计算mss */
        mss_now = tcp_send_mss(sk, &size_goal, flags);
    }

out:
    /* 已经拷贝数据到发送队列，则发送之 */
    if (copied) {
        tcp_tx_timestamp(sk, sockc.tsflags, tcp_write_queue_tail(sk));
        tcp_push(sk, flags, mss_now, tp->nonagle, size_goal);
    }
out_nopush:
    release_sock(sk);
    return copied + copied_syn;

do_fault:
    /* skb中没有数据，从发送队列中删除skb */
    if (!skb->len) {
        tcp_unlink_write_queue(skb, sk);
        /* It is the one place in all of TCP, except connection
         * reset, where we can be unlinking the send_head.
         */
        tcp_check_send_head(sk, skb);
        sk_wmem_free_skb(sk, skb);
    }

do_error:
    if (copied + copied_syn)
        goto out;
out_err:
    err = sk_stream_error(sk, flags, err);
    /* make sure we wake any epoll edge trigger waiter */
    if (unlikely(skb_queue_len(&sk->sk_write_queue) == 0 &&
             err == -EAGAIN)) {
        sk->sk_write_space(sk);
        tcp_chrono_stop(sk, TCP_CHRONO_SNDBUF_LIMITED);
    }
    release_sock(sk);
    return err;
}