virtio前后端配合限速分析 ——VIRTIO_RING_F_EVENT_IDX

virtio前后端配合限速分析

——lvyilong316

在VIRTIO中，有个一个设备的特性叫做VIRTIO_RING_F_EVENT_IDX，这个特性是用来对前后端速率进行匹配限速的。我们知道avail ring，这个ring有两个用途，一是发送侧(send queue)前端驱动发送报文的时，将待发送报文加入avail ring等待后端的处理，后端处理完后，会将其放入used ring，并由前端将其释放desc中(free_old_xmit_skbs, detach_buf)，最后通过try_fill_recv重新装入availring中; 二是接收侧(receive qeueu)，前端将空白物理块加入avail ring中，提供给后端用来接收报文，后端接收完报文会放入used ring。可以看出：都是后端用完前端的avail ring的东西放入used ring，也就是前端消耗uesd，后端消耗avail。所以本特性中后端用了used ring的最后一个元素，告诉前端驱动后端处理到哪个avail ring上的元素了，同时前端使用avail ring的最后一个元素告诉后端，处理到那个used ring了。

发送方向限速

我们看发送方向的限速。首先看前端guest的发送逻辑(kernel 3.10: virtio-net)，在guest发送时会调用virtqueue_add函数：

    virtqueue_add函数将要发送的skb转换的sg再转换为desc chain。具体转换流程不是这里分析的重点，我们只看virtqueue_add函数的最后一部分。

l  virtqueue_add

static inline int virtqueue_add(struct virtqueue *_vq, …)
{
/*省略sg到desc chain的具体转换逻辑*/
add_head:
         /* Set token.记录数组记录本次发送的skb */
         vq->data[head] = data; /*head 为记录次skb的首个desc的下标，data为本skb的地址*/

          /*更新avail*/
         avail = (vq->vring.avail->idx & (vq->vring.num-1));
         vq->vring.avail->ring[avail] = head; /*将本次要发送的首个desc下标记录在avail->ring[vq->vring.avail->idx]，vq->vring.avail->idx 是avail ring下一个可用的index*/

         virtio_wmb(vq->weak_barriers);
         vq->vring.avail->idx++;
         vq->num_added++; /*更新num_added ，num_added 记录从上一次kick后端后，前端新增加的desc数量*/

         /* This is very unlikely, but theoretically possible. Kick
          * just in case. */
          /*如果avail的数量太多，则kick后端收包，这种情况是你很难发生的*/
         if (unlikely(vq->num_added == (1 << 16) - 1))
                   virtqueue_kick(_vq);

         pr_debug("Added buffer head %i to %p\n", head, vq);
         END_USE(vq);

         return 0;
}

这里一个关键点就是vq->num_added，这个变量在这里记录从上一次kick后端后，前端新增加的desc数量。我们看到在kick后端前（virtqueue_kick），有一个判断：vq->num_added == (1 << 16) - 1)，也就是当前端上层发送kick后，这段期间如果累计填充的desc不足65535个时，就先不去kick后端处理，这样不用每次发送skb都kick后端，提高后端的处理效率。

下面看如果达到了65535调用virtqueue_kick的逻辑。

l  virtqueue_kick

void virtqueue_kick(struct virtqueue *vq)
{
         if (virtqueue_kick_prepare(vq))
                   virtqueue_notify(vq);
}

在真正调用virtqueue_notify kick后端前，会调用virtqueue_kick_prepare来再次判断是否需要kick，这也是我们要分析的重点。

l  virtqueue_kick_prepare

bool virtqueue_kick_prepare(struct virtqueue *_vq)
{
         struct vring_virtqueue *vq = to_vvq(_vq);
         u16 new, old;
         bool needs_kick;

         START_USE(vq);
         virtio_mb(vq->weak_barriers);
    /*old上次kick后的avail.idx */
         old = vq->vring.avail->idx - vq->num_added;
         /*new是当前的avail.idx*/
         new = vq->vring.avail->idx;
         vq->num_added = 0;
    /*当VIRTIO_RING_F_EVENT_IDX 被设置的时候vq->event 为1*/
         if (vq->event) {
                   needs_kick = vring_need_event(vring_avail_event(&vq->vring),
                                                     new, old);
         } else {
                   needs_kick = !(vq->vring.used->flags & VRING_USED_F_NO_NOTIFY);
         }
         END_USE(vq);
         return needs_kick;
}

    这里我们注意两个变量，old和new，old表示上次kick后的avail.idx，new是当前的avail.idx，两者的差值就是vq->num_added，也就是自上次kick后端后前端又积累的desc chain数量。另外vq->event是在vq初始化的时候设置的，在vring_new_virtqueue有如下代码：

vq->event = virtio_has_feature(vdev, VIRTIO_RING_F_EVENT_IDX);

所以当设置了VIRTIO_RING_F_EVENT_IDX后，vq->event就会被置位，这里就会调用needs_kick = vring_need_event(vring_avail_event(&vq->vring), new, old);

注意vring_avail_event(&vq->vring) 为：(vr)->used->ring[(vr)->num]，即uesd ring的最后一个元素，后端用used ring的最后一个元素告诉前端后端处理的位置。

l  vring_need_event

static inline int vring_need_event(__u16 event_idx, __u16 new_idx, __u16 old)
{
         return (__u16)(new_idx - event_idx - 1) < (__u16)(new_idx - old);
}

这个公式决定了是否想后端QEMU发送通知：

当满足公式的时候，后端处理的位置event_idx超过了old，表示后端QEMU处理的速度够快，索引返回true，通知(kick)后端，通知后端有新的avail 逻辑buf，请你继续处理

如下下面情况：

后端处理的位置event_idx落后于上次添加avail ring的位置，说明后端处理较慢，返回false，那么前端就先不通知(kick)，积攒一下，反正后端正处理不过来，下次退出的时候，让后端一起尽情处理。

     然后我们看下后端是如何处理的，看guest发方向，为了简单起见我们选择dpdk 18.02中的vhost_user来分析（vhost_net相对逻辑比较绕，另外较新版本的dpdk才支持VIRTIO_RING_F_EVENT_IDX）。guest的发送，对应后端的dequeue操作。

l  rte_vhost_dequeue_burst

uint16_t
rte_vhost_dequeue_burst(int vid, uint16_t queue_id,
         struct rte_mempool *mbuf_pool, struct rte_mbuf **pkts, uint16_t count)
{
/*省略前面dequeue操作*/
          if (likely(dev->dequeue_zero_copy == 0)) {
                   do_data_copy_dequeue(vq);
                   vq->last_used_idx += i; /*更新last_used_idx ，i为本次dequeue的mbuf数量，也就是清空的desc数量*/
                   update_used_idx(dev, vq, i);
          }
/*……*/
}

更新vq->last_used_idx后，调用update_used_idx。

l  update_used_idx

static __rte_always_inline void
update_used_idx(struct virtio_net *dev, struct vhost_virtqueue *vq,
                   uint32_t count)
{
         if (unlikely(count == 0))
                   return;

         rte_smp_wmb();
         rte_smp_rmb();

         vq->used->idx += count; /*更新vq->used->idx */
         vhost_log_used_vring(dev, vq, offsetof(struct vring_used, idx),
                            sizeof(vq->used->idx));
         vhost_vring_call(dev, vq); /*调用vhost_vring_call  call前端*/
}

vhost_vring_call主要作用是Call前端，告诉前端desc中的数据已经取出，前端可以回收了。

l  vhost_vring_call

static __rte_always_inline void
vhost_vring_call(struct virtio_net *dev, struct vhost_virtqueue *vq)
{
         /* Flush used->idx update before we read avail->flags. */
         rte_mb();

         /* Don't kick guest if we don't reach index specified by guest. */
         if (dev->features & (1ULL << VIRTIO_RING_F_EVENT_IDX)) {
                   uint16_t old = vq->signalled_used;
                   uint16_t new = vq->last_used_idx;

                   LOG_DEBUG(VHOST_DATA, "%s: used_event_idx=%d, old=%d, new=%d\n",
                            __func__,
                            vhost_used_event(vq),
                            old, new);
                   if (vhost_need_event(vhost_used_event(vq), new, old)
                            && (vq->callfd >= 0)) {
                            vq->signalled_used = vq->last_used_idx; /*更新vq->signalled_used为本次Call前端后的used idx*/
                            eventfd_write(vq->callfd, (eventfd_t) 1);
                   }
         } else {
                   /* Kick the guest if necessary. */
                   if (!(vq->avail->flags & VRING_AVAIL_F_NO_INTERRUPT)
                                     && (vq->callfd >= 0))
                            eventfd_write(vq->callfd, (eventfd_t)1);
         }
}

当设置VIRTIO_RING_F_EVENT_IDX后，会调用vhost_need_event判断是否需要Call前端，否则直接调用eventfd_write Call前端。在看vhost_need_event实现之前，先看其后的一行代码：

vq->signalled_used = vq->last_used_idx;

这里将vq->signalled_used更新为本次Call前端后的used idx，那么对于调用vhost_need_event时，这里存放的应该就是上次Call前端的used idx，弄清楚这个看vhost_need_event的实现就很容易了。

l  vhost_need_event

static __rte_always_inline int
vhost_need_event(uint16_t event_idx, uint16_t new_idx, uint16_t old)
{
         return (uint16_t)(new_idx - event_idx - 1) < (uint16_t)(new_idx - old);
}

其中第一个参数为(vq)->avail->ring[(vq)->size]，前端使用avail ring最后一个元素通知后端。

new_idx – old大于new_idx - event_idx – 1，说明前端也更新了used idx，则后端可以进行Call通知前端，否则则暂时不通知。

这里还有一点要注意，我们看到后端使用avail->ring的最后一个元素来判断前端的使用情况，那么前端是什么时候更新这个值呢？答案是在virtqueue_get_buf中（kernel 3.10 virtio-net），这个函数在前端发送和接收时都会被调用，根据used ring从desc中得到skbbuf（发送时得到的是待填入数据的skbbuf，接收时得到的是带有有效数据的skbbuf）。在函数的末尾有如下逻辑：

if (!(vq->vring.avail->flags & VRING_AVAIL_F_NO_INTERRUPT)) {
                   vring_used_event(&vq->vring) = vq->last_used_idx;
                   virtio_mb(vq->weak_barriers);
}

其中vring_used_event定义如下：

#define vring_used_event(vr) ((vr)->avail->ring[(vr)->num])

   这样前端就更新了avail ring最后一个元素。

   好像还是缺少点什么？我们开始看到前端virtio-net是根据used ring的最后一个元素来判断是否要kick后端，按照这个逻辑后端应该有地方来更新used ring的最后一个元素才对。可惜我们再vhost-user的逻辑中并没有发现相关操作。这是为什么呢？我们回头想想整个逻辑的目的，是降低前端kick，后端Call的频率，是只在不影响对端使用的情况，尽可能的多积攒一些再通知。而我们知道vhost-user采用的是pmd，根本不去管这个通知，所以也就没有必要去配合前端了，前端想kick就使劲kick，反正也不受影响。

接收方向限速

收方向类似，这里不再展开，只给出调用路径。先从后端vhost-user开始分析。

后端：

virtio_dev_rxàvhost_vring_callàvhost_need_eventà通过avail->ring的最后一个元素判断是否Call 前端；

前端：

try_fill_recvàvirtqueue_kickàvirtqueue_kick_prepareàvring_need_eventà通过used->ring的最后一个元素判断是否kick后端。