Line data    Source code

       1             : /*
       2             :  * Copyright (C) 2014 Semihalf.
       3             :  *
       4             :  * This program is free software; you can redistribute it and/or
       5             :  * modify it under the terms of the GNU General Public License as
       6             :  * published by the Free Software Foundation version 2.
       7             :  *
       8             :  * This program is distributed "as is" WITHOUT ANY WARRANTY of any
       9             :  * kind, whether express or implied; without even the implied warranty
      10             :  * of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
      11             :  * GNU General Public License for more details.
      12             :  *
      13             :  * vr_dpdk_ringdev.c -- DPDK ring device
      14             :  *
      15             :  */
      16             : 
      17             : #include "vr_dpdk.h"
      18             : 
      19             : #include <rte_malloc.h>
      20             : #include <rte_port_ring.h>
      21             : 
      22             : /* Allocates a new ring */
      23             : struct rte_ring *
      24         106 : vr_dpdk_ring_allocate(unsigned host_lcore_id, char *ring_name,
      25             :     unsigned vr_dpdk_tx_ring_sz, unsigned flags)
      26             : {
      27             :     int ret;
      28             :     ssize_t ring_size;
      29             :     struct rte_ring *ring;
      30             : 
      31         106 :     ring_size = rte_ring_get_memsize(vr_dpdk_tx_ring_sz);
      32         106 :     if (ring_size == -EINVAL)
      33           0 :         return NULL;
      34             : 
      35         106 :     ring = (struct rte_ring *)rte_malloc_socket(ring_name, ring_size,
      36         106 :         RTE_CACHE_LINE_SIZE,  rte_lcore_to_socket_id(host_lcore_id));
      37         106 :     if (ring == NULL)
      38           0 :         return NULL;
      39             : 
      40         106 :     ret = rte_ring_init(ring, ring_name, vr_dpdk_tx_ring_sz, flags);
      41         106 :     if (ret < 0) {
      42           0 :         rte_free(ring);
      43           0 :         return NULL;
      44             :     }
      45             : 
      46         106 :     return ring;
      47             : }
      48             : 
      49             : /* Add the ring to the list of rings to push */
      50             : void
      51           0 : dpdk_ring_to_push_add(unsigned lcore_id, struct rte_ring *tx_ring,
      52             :     struct vr_dpdk_queue *tx_queue)
      53             : {
      54           0 :     struct vr_dpdk_lcore *lcore = vr_dpdk.lcores[lcore_id];
      55           0 :     struct vr_dpdk_ring_to_push *rtp = &lcore->lcore_rings_to_push[0];
      56             : 
      57             :     /* find an empty ring to push */
      58           0 :     while (rtp->rtp_tx_ring) rtp++;
      59             : 
      60           0 :     rtp->rtp_tx_ring = tx_ring;
      61           0 :     rtp->rtp_tx_queue = tx_queue;
      62             :     rte_wmb();
      63           0 :     lcore->lcore_nb_rings_to_push++;
      64           0 :     RTE_LOG(INFO, VROUTER, "    lcore %u now has %" PRIu16
      65             :         " ring(s) to push\n",
      66             :         lcore_id, lcore->lcore_nb_rings_to_push);
      67           0 :     RTE_VERIFY(lcore->lcore_nb_rings_to_push <= VR_DPDK_MAX_RINGS);
      68           0 : }
      69             : 
      70             : /* Remove the ring from the list of rings to push
      71             :  * The function is called by the NetLink lcore only.
      72             :  */
      73             : void
      74           0 : dpdk_ring_to_push_remove(unsigned lcore_id, struct rte_ring *tx_ring)
      75             : {
      76           0 :     struct vr_dpdk_lcore *lcore = vr_dpdk.lcores[lcore_id];
      77           0 :     struct vr_dpdk_ring_to_push *rtp = &lcore->lcore_rings_to_push[0];
      78             :     struct vr_dpdk_ring_to_push *last_rtp;
      79             : 
      80             :     /* find the ring to push */
      81           0 :     while (rtp->rtp_tx_ring != tx_ring) rtp++;
      82             : 
      83           0 :     rtp->rtp_tx_ring = NULL;
      84           0 :     lcore->lcore_nb_rings_to_push--;
      85           0 :     RTE_VERIFY(lcore->lcore_nb_rings_to_push <= VR_DPDK_MAX_RINGS);
      86             :     rte_wmb();
      87             :     /* copy the last element to the empty spot */
      88           0 :     last_rtp = &lcore->lcore_rings_to_push[lcore->lcore_nb_rings_to_push];
      89           0 :     rtp->rtp_tx_queue = last_rtp->rtp_tx_queue;
      90             :     rte_wmb();
      91           0 :     rtp->rtp_tx_ring = last_rtp->rtp_tx_ring;
      92           0 :     last_rtp->rtp_tx_ring = NULL;
      93           0 :     last_rtp->rtp_tx_queue = NULL;
      94           0 : }
      95             : 
      96             : /* Release ring TX queue
      97             :  * The function is called by the NetLink lcore only.
      98             :  */
      99             : static void
     100           0 : dpdk_ring_tx_queue_release(unsigned lcore_id, unsigned queue_index,
     101             :         struct vr_interface *vif)
     102             : {
     103           0 :     struct vr_dpdk_lcore *lcore = vr_dpdk.lcores[lcore_id];
     104           0 :     struct vr_dpdk_queue *tx_queue =
     105           0 :         &lcore->lcore_tx_queues[vif->vif_idx][queue_index];
     106           0 :     struct vr_dpdk_queue_params *tx_queue_params
     107           0 :         = &lcore->lcore_tx_queue_params[vif->vif_idx][queue_index];
     108             : 
     109           0 :     tx_queue->txq_ops.f_tx = NULL;
     110             :     rte_wmb();
     111             : 
     112             :     /* remove the ring from the list of rings to push */
     113           0 :     dpdk_ring_to_push_remove(tx_queue_params->qp_ring.host_lcore_id,
     114             :             tx_queue_params->qp_ring.ring_p);
     115             : 
     116           0 :     rte_free(tx_queue_params->qp_ring.ring_p);
     117             : 
     118             :     /* flush and free the queue */
     119           0 :     if (tx_queue->txq_ops.f_free(tx_queue->q_queue_h)) {
     120           0 :         RTE_LOG(ERR, VROUTER, "    error freeing lcore %u ring\n", lcore_id);
     121             :     }
     122             : 
     123             :     /* reset the queue */
     124           0 :     vrouter_put_interface(tx_queue->q_vif);
     125           0 :     memset(tx_queue, 0, sizeof(*tx_queue));
     126           0 :     memset(tx_queue_params, 0, sizeof(*tx_queue_params));
     127           0 : }
     128             : 
     129             : /* Init ring TX queue */
     130             : struct vr_dpdk_queue *
     131           0 : vr_dpdk_ring_tx_queue_init(unsigned lcore_id, struct vr_interface *vif,
     132             :         unsigned int queue_id, unsigned host_lcore_id)
     133             : {
     134             :     int ret;
     135             :     uint8_t port_id;
     136           0 :     unsigned int vif_idx = vif->vif_idx, dpdk_queue_index;
     137           0 :     const unsigned int socket_id = rte_lcore_to_socket_id(lcore_id);
     138             : 
     139             :     char ring_name[RTE_RING_NAMESIZE];
     140             :     struct rte_ring *tx_ring;
     141           0 :     struct vr_dpdk_lcore *lcore = vr_dpdk.lcores[lcore_id];
     142           0 :     struct vr_dpdk_lcore *host_lcore = vr_dpdk.lcores[host_lcore_id];
     143             :     struct vr_dpdk_queue *tx_queue;
     144             :     struct vr_dpdk_queue_params *tx_queue_params;
     145             :     struct vr_dpdk_queue *host_tx_queue;
     146             : 
     147             : 
     148           0 :     if (lcore->lcore_hw_queue_to_dpdk_index[vif->vif_idx]) {
     149           0 :         dpdk_queue_index =
     150           0 :             lcore->lcore_hw_queue_to_dpdk_index[vif->vif_idx][queue_id];
     151             :     } else {
     152           0 :         dpdk_queue_index = 0;
     153             :     }
     154             : 
     155           0 :     tx_queue = &lcore->lcore_tx_queues[vif_idx][dpdk_queue_index];
     156           0 :     tx_queue_params = &lcore->lcore_tx_queue_params[vif_idx][dpdk_queue_index];
     157           0 :     host_tx_queue = &host_lcore->lcore_tx_queues[vif_idx][dpdk_queue_index];
     158             : 
     159           0 :     if (vif->vif_type == VIF_TYPE_PHYSICAL) {
     160           0 :         port_id = (((struct vr_dpdk_ethdev *)(vif->vif_os))->ethdev_port_id);
     161             :     } else {
     162           0 :         port_id = vif->vif_os_idx;
     163             :     }
     164             : 
     165             :     /* init queue */
     166           0 :     tx_queue->txq_ops = rte_port_ring_writer_ops;
     167           0 :     tx_queue->q_queue_h = NULL;
     168           0 :     tx_queue->q_vif = vrouter_get_interface(vif->vif_rid, vif_idx);
     169             : 
     170           0 :     ret = snprintf(ring_name, sizeof(ring_name), "vr_dpdk_ring_%u_%u_%u",
     171             :         host_lcore_id, vif_idx, lcore_id);
     172           0 :     if (ret >= sizeof(ring_name))
     173           0 :         goto error;
     174             : 
     175             :     /* allocate a ring on the host lcore (single-producer single-consumer) */
     176           0 :     tx_ring = vr_dpdk_ring_allocate(host_lcore_id, ring_name,
     177             :             vr_dpdk_tx_ring_sz, RING_F_SP_ENQ | RING_F_SC_DEQ);
     178           0 :     if (tx_ring == NULL)
     179           0 :         goto error;
     180             : 
     181             :     /* add the ring to the list of rings to push */
     182           0 :     dpdk_ring_to_push_add(host_lcore_id, tx_ring, host_tx_queue);
     183             : 
     184             :     /* create the queue */
     185           0 :     struct rte_port_ring_writer_params writer_params = {
     186             :         .ring = tx_ring,
     187             :         .tx_burst_sz = VR_DPDK_TX_BURST_SZ,
     188             :     };
     189           0 :     tx_queue->q_queue_h = tx_queue->txq_ops.f_create(&writer_params,
     190             :                                                         socket_id);
     191           0 :     if (tx_queue->q_queue_h == NULL)
     192           0 :         goto error;
     193             : 
     194             :     /* store queue params */
     195           0 :     tx_queue_params->qp_release_op = &dpdk_ring_tx_queue_release;
     196           0 :     tx_queue_params->qp_ring.ring_p = tx_ring;
     197           0 :     tx_queue_params->qp_ring.host_lcore_id = host_lcore_id;
     198             : 
     199           0 :     return tx_queue;
     200             : 
     201           0 : error:
     202           0 :     RTE_LOG(ERR, VROUTER, "    error initializing ring TX queue for device %"
     203             :         PRIu8 "\n", port_id);
     204           0 :     return NULL;
     205             : }
     206             : 
     207             : /* Init ring RX queue */
     208             : struct vr_dpdk_queue *
     209           0 : vr_dpdk_ring_rx_queue_init(unsigned lcore_id, struct vr_interface *vif,
     210             :     unsigned host_lcore_id)
     211             : {
     212           0 :     RTE_LOG(ERR, VROUTER, "%s: not implemented\n", __func__);
     213             : 
     214           0 :     return NULL;
     215             : }