dpdk vhost_user desc chain的形成

vhost_user desc chain的形成

——lvyilong316

    在前面分析vhost_user mergeable特性的时候，我们提到了一个”desc chain”的概念。本节重点分析一下这个chain是如何建立的。

所谓desc chain即各个desc通过其next成员相互关联，但前提是desc->falg被设置了VRING_DESC_F_NEXT，只有这种情况next的值才有意义。这个desc chain的建立是由前端驱动建立并维护的，后端不能去更改。我们以kernel 3.10的virtio-net为例。当前端要接受数据包时，首先要准备好要接受数据包的内存，也就是avail ring中的desc。这是通过try_fill_recv来完成的。

l  try_fill_recv

static bool try_fill_recv(struct receive_queue *rq, gfp_t gfp)
{
         struct virtnet_info *vi = rq->vq->vdev->priv;
         int err;
         bool oom;

         do {
                   if (vi->mergeable_rx_bufs)
                            err = add_recvbuf_mergeable(rq, gfp); /*后端支持VIRTIO_NET_F_MRG_RXBUF*/
                   else if (vi->big_packets)
                            err = add_recvbuf_big(rq, gfp); /*后端支持GUEST_GSO/GUEST_TSO，相当于LRO*/
                   else
                            err = add_recvbuf_small(rq, gfp);

                   oom = err == -ENOMEM;
                   if (err)
                            break;
                   ++rq->num;
         } while (rq->vq->num_free);
         if (unlikely(rq->num > rq->max))
                   rq->max = rq->num;
         virtqueue_kick(rq->vq); /*通知后端avail ring更新*/
         return !oom;
}

     这里会根据是否开启guest tso，是否开启mergeable分为三种情况，分别会调用add_recvbuf_mergeable，add_recvbuf_big，add_recvbuf_small三个函数。这两个函数最终又都会调用virtqueue_add_inbuf函数。这个函数最终又会调用到virtqueue_add，后者在virtio-net发送逻辑中已经分析过了。其中主要是通过以下逻辑建立desc chain关系的。

for (sg = sgs[n]; sg; sg = next(sg, &total_out)) {
                            vq->vring.desc[i].flags = VRING_DESC_F_NEXT;
                            vq->vring.desc[i].addr = sg_phys(sg);
                            vq->vring.desc[i].len = sg->length;
                            prev = i;
                            i = vq->vring.desc[i].next;
                   }

这里的sgs指的是接受队列的rq->sg，也就是每个desc chain的长度由rq->sg的长度来决定。所以我们分别看下三种情况是如果初始化rq->sg的。

l  add_recvbuf_mergeable

static int add_recvbuf_mergeable(struct receive_queue *rq, gfp_t gfp)
{
         struct page *page;
         int err;

         page = get_a_page(rq, gfp);
         if (!page)
                   return -ENOMEM;

         sg_init_one(rq->sg, page_address(page), PAGE_SIZE);

         err = virtqueue_add_inbuf(rq->vq, rq->sg, 1, page, gfp);
         if (err < 0)
                   give_pages(rq, page);

         return err;
}

在add_recvbuf_mergeable中，首先调用get_a_page分配一个page，然后调用sg_init_one将这个page转换为对应的sg，设置在rq->sg。

l  sg_init_one

void sg_init_one(struct scatterlist *sg, const void *buf, unsigned int buflen)
{
         sg_init_table(sg, 1);
         sg_set_buf(sg, buf, buflen);
}

    首先sg_init_table初始化rq->sg的长度为一个元素，所以从然后调用sg_set_buf将之前的page信息存放在这一个sg中。

void sg_init_table(struct scatterlist *sgl, unsigned int nents)
{
         memset(sgl, 0, sizeof(*sgl) * nents);
         sg_mark_end(&sgl[nents - 1]);
}

从这个过程可以看出当开启mergeable，时每次rq->sg只有一个元素，所以对应的每个desc chain也只有一个，即每个desc 都是独立的。为什么会这样呢？因为开启mergeable后，后端可以不必只能用一个chain接收数据，既然这样要chain也就没有什么意义了。

下面看普通场景add_recvbuf_small的调用。

l  add_recvbuf_small

static int add_recvbuf_small(struct receive_queue *rq, gfp_t gfp)
{
         struct virtnet_info *vi = rq->vq->vdev->priv;
         struct sk_buff *skb;
         struct skb_vnet_hdr *hdr;
         int err;
    /*这里MAX_PACKET_LEN的值为1500+ETH_HLEN(14) + VLAN_HLEN(4)*/
         skb = __netdev_alloc_skb_ip_align(vi->dev, MAX_PACKET_LEN, gfp);
         if (unlikely(!skb))
                   return -ENOMEM;

         skb_put(skb, MAX_PACKET_LEN);

         hdr = skb_vnet_hdr(skb);
         sg_set_buf(rq->sg, &hdr->hdr, sizeof hdr->hdr);

         skb_to_sgvec(skb, rq->sg + 1, 0, skb->len);

         err = virtqueue_add_inbuf(rq->vq, rq->sg, 2, skb, gfp);
         if (err < 0)
                   dev_kfree_skb(skb);

         return err;
}

这个函数首先会按照1500的mtu分配一个skb，然后调用sg_set_buf将skb_vnet_hdr转换为一个sg放入rq->sg中，紧接着又调用skb_to_sgvec将skb的数据放入rq->sg的第二个sg中，所以这种情况每次rq->sg的长度就是2，即第一个用来存放skb_vnet_hdr，第二个用来存放数据。所以这种情况下每个desc chain的长度也是2。

下面看如果开启GUEST_GSO/GUEST_TSO的情况。

l  add_recvbuf_big

static int add_recvbuf_big(struct receive_queue *rq, gfp_t gfp)
{
         struct page *first, *list = NULL;
         char *p;
         int i, err, offset;

         /* page in rq->sg[MAX_SKB_FRAGS + 1] is list tail */
         /* MAX_SKB_FRAGS为16，即(65536/PAGE_SIZE + 1)， 因为当开启GUEST_GSO/GUEST_TSO
          * 时会使用一个长的chain接收大包， 这里使用17个page分别初始化rq->sg[0]~rq->[sg][17],
          * 其中rq->sg[0]和rq->sg[1]共享一个page*/
         for (i = MAX_SKB_FRAGS + 1; i > 1; --i) {
                   first = get_a_page(rq, gfp);
                   if (!first) {
                            if (list)
                                     give_pages(rq, list);
                            return -ENOMEM;
                   }
                   sg_set_buf(&rq->sg[i], page_address(first), PAGE_SIZE);

                   /* chain new page in list head to match sg */
                   first->private = (unsigned long)list;
                   list = first;
         }

         first = get_a_page(rq, gfp);
         if (!first) {
                   give_pages(rq, list);
                   return -ENOMEM;
         }
         p = page_address(first);

         /* rq->sg[0], rq->sg[1] share the same page */
         /* a separated rq->sg[0] for virtio_net_hdr only due to QEMU bug */
         /* rq->sg[0]用来存放virtio_net_hdr*/
         sg_set_buf(&rq->sg[0], p, sizeof(struct virtio_net_hdr));

         /* rq->sg[1] for data packet, from offset */
         /*rq->sg[1]~rq->sg[17]用来存放真正的数据*/
         offset = sizeof(struct padded_vnet_hdr);
         sg_set_buf(&rq->sg[1], p + offset, PAGE_SIZE - offset);

         /* chain first in list head */
         first->private = (unsigned long)list;
         err = virtqueue_add_inbuf(rq->vq, rq->sg, MAX_SKB_FRAGS + 2,
                                       first, gfp);
         if (err < 0)
                   give_pages(rq, first);

         return err;
}

    当开启GUEST_GSO/GUEST_TSO时，每个desc chain的长度为17，其中第一个用来触发virtio header，其他用来存放数据。也就是说如果希望guest能接受大包，可以有两种方式，一种是开启GUEST_GSO/GUEST_TSO，另一种是开启mergeable。对比之下，前者无论接受大包还是小包都会使用一个长chain来接收，对于小包会造成浪费，后者会使用多个chain来接受；此外，前者的chain长度为17，所以最大收包为65535，而后者没有这个限制。