VNFs/DPPD-PROX/handle_aggregator.c

   1 /*
   2 // Copyright (c) 2010-2017 Intel Corporation
   3 //
   4 // Licensed under the Apache License, Version 2.0 (the "License");
   5 // you may not use this file except in compliance with the License.
   6 // You may obtain a copy of the License at
   7 //
   8 //     http://www.apache.org/licenses/LICENSE-2.0
   9 //
  10 // Unless required by applicable law or agreed to in writing, software
  11 // distributed under the License is distributed on an "AS IS" BASIS,
  12 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 // See the License for the specific language governing permissions and
  14 // limitations under the License.
  15 */
  16
  17 #include <rte_ip.h>
  18 #include <stdio.h>
  19 #include <string.h>
  20 #include <rte_version.h>
  21
  22 #include "prox_lua.h"
  23 #include "prox_lua_types.h"
  24
  25 #include "lconf.h"
  26 #include "task_base.h"
  27 #include "task_init.h"
  28 #include "defines.h"
  29 #include "prefetch.h"
  30 #include "qinq.h"
  31 #include "prox_cfg.h"
  32 #include "log.h"
  33 #include "quit.h"
  34 #include "prox_shared.h"
  35 #include "mbuf_utils.h"
  36 #include "handle_aggregator.h"
  37
  38 #define PRIORITY_DHCP   (HIGH_PRIORITY)
  39
  40 #define TASK_STATS_ADD_DROP_TX_FAIL_PRIO(stats, ntx, prio) do {    \
  41         (stats)->drop_tx_fail_prio[prio] += ntx;           \
  42         } while(0)
  43 #define TASK_STATS_ADD_TX_PRIO(stats, ntx, prio) do {    \
  44                 (stats)->rx_prio[prio] += ntx;           \
  45         } while(0)                                      \
  46
  47 static inline uint8_t detect_l4_priority(uint8_t l3_priority, const struct ipv4_hdr *ipv4_hdr)
  48 {
  49         if (ipv4_hdr->next_proto_id == IPPROTO_UDP) {
  50                 const struct udp_hdr *udp = (const struct udp_hdr *)((const uint8_t *)ipv4_hdr + sizeof(struct ipv4_hdr));
  51                 if (((udp->src_port == 0x67) && (udp->dst_port == 0x68)) || ((udp->src_port == 0x68) && (udp->dst_port == 0x67))) {
  52                         return PRIORITY_DHCP;
  53                 }
  54         }
  55         return l3_priority;
  56 }
  57
  58 static inline uint8_t detect_l3_priority(uint8_t l2_priority, const struct ipv4_hdr *ipv4_hdr)
  59 {
  60         uint8_t dscp;
  61         if ((ipv4_hdr->version_ihl >> 4) == 4) {
  62         } else if ((ipv4_hdr->version_ihl >> 4) == 6) {
  63                 plog_warn("IPv6 Not implemented\n");
  64                 return OUT_DISCARD;
  65         } else {
  66                 plog_warn("Unexpected IP version\n");
  67                 return OUT_DISCARD;
  68         }
  69         dscp = ipv4_hdr->type_of_service >> 2;
  70         if (dscp)
  71                 return MAX_PRIORITIES - dscp - 1;
  72         else
  73                 return l2_priority;
  74 }
  75
  76 static inline uint8_t detect_l2_priority(const struct qinq_hdr *pqinq)
  77 {
  78         if (pqinq->cvlan.eth_proto != ETYPE_VLAN) {
  79                 plog_warn("Unexpected proto in QinQ = %#04x\n", pqinq->cvlan.eth_proto);
  80                 return OUT_DISCARD;
  81         }
  82         uint16_t svlan_priority = ntohs(pqinq->svlan.vlan_tci >> 13);
  83         uint16_t cvlan_priority = ntohs(pqinq->cvlan.vlan_tci >> 13);
  84         if (svlan_priority)
  85                 return svlan_priority;
  86         else
  87                 return cvlan_priority;
  88 }
  89
  90 static inline void buffer_packet(struct task_aggregator *task, struct rte_mbuf *mbuf, uint8_t priority)
  91 {
  92         struct task_base *tbase = (struct task_base *)task;
  93
  94         struct task_buffer *prio = &task->priority[priority];
  95         if (prio->pkt_nb < BUFFER_LENGTH) {
  96                 prio->buffer[prio->pkt_pos] = mbuf;
  97                 prio->pkt_pos++;
  98                 if (prio->pkt_pos == BUFFER_LENGTH)
  99                         prio->pkt_pos = 0;
 100                 prio->pkt_nb++;
 101         } else {
 102                 task->drop.buffer[task->drop.pkt_nb] = mbuf;
 103                 task->drop.pkt_nb++;
 104                 TASK_STATS_ADD_DROP_TX_FAIL_PRIO(&task->stats, 1, priority);
 105         }
 106 }
 107
 108 static inline void handle_aggregator(struct task_aggregator *task, struct rte_mbuf *mbuf)
 109 {
 110         struct ether_hdr *peth = rte_pktmbuf_mtod(mbuf, struct ether_hdr *);
 111         uint8_t priority = 0;
 112         const struct qinq_hdr *pqinq;
 113         const struct ipv4_hdr *ipv4_hdr;
 114
 115         const uint16_t eth_type = peth->ether_type;
 116         switch (eth_type) {
 117         case ETYPE_MPLSU:
 118         case ETYPE_MPLSM:
 119                 break;
 120         case ETYPE_8021ad:
 121                 pqinq = rte_pktmbuf_mtod(mbuf, const struct qinq_hdr *);
 122                 if ((priority = detect_l2_priority(pqinq)) == OUT_DISCARD)
 123                         break;
 124                 ipv4_hdr = (const struct ipv4_hdr *)(pqinq + 1);
 125                 if ((priority = detect_l3_priority(priority, ipv4_hdr)) == OUT_DISCARD)
 126                         break;
 127                 if ((priority = detect_l4_priority(priority, ipv4_hdr)) == OUT_DISCARD)
 128                         break;
 129                 break;
 130         case ETYPE_VLAN:
 131                 break;
 132         case ETYPE_IPv4:
 133                 ipv4_hdr = (const struct ipv4_hdr *)(peth+1);
 134                 if ((priority = detect_l3_priority(LOW_PRIORITY, ipv4_hdr)) == OUT_DISCARD)
 135                         break;
 136                 if ((priority = detect_l4_priority(priority, ipv4_hdr)) == OUT_DISCARD)
 137                         break;
 138                 break;
 139         case ETYPE_IPv6:
 140                 break;
 141         case ETYPE_ARP:
 142                 break;
 143         default:
 144                 break;
 145         }
 146         if (priority == OUT_DISCARD) {
 147                 task->drop.buffer[task->drop.pkt_nb] = mbuf;
 148                 task->drop.pkt_nb++;
 149                 return;
 150         }
 151         buffer_packet(task, mbuf, priority);
 152 }
 153
 154 static int handle_aggregator_bulk(struct task_base *tbase, struct rte_mbuf **mbufs, uint16_t n_pkts)
 155 {
 156         struct task_aggregator *task = (struct task_aggregator *)tbase;
 157
 158         uint16_t j;
 159         uint32_t drop_bytes = 0;
 160 #ifdef PROX_PREFETCH_OFFSET
 161         for (j = 0; j < PROX_PREFETCH_OFFSET && j < n_pkts; ++j) {
 162                 prefetch_nta(mbufs[j]);
 163         }
 164         for (j = 1; j < PROX_PREFETCH_OFFSET && j < n_pkts; ++j) {
 165                 prefetch_nta(rte_pktmbuf_mtod(mbufs[j - 1], void *));
 166         }
 167 #endif
 168         for (j = 0; j + PREFETCH_OFFSET < n_pkts; ++j) {
 169 #ifdef PROX_PREFETCH_OFFSET
 170                 prefetch_nta(mbufs[j + PREFETCH_OFFSET]);
 171                 prefetch_nta(rte_pktmbuf_mtod(mbufs[j + PREFETCH_OFFSET - 1], void *));
 172 #endif
 173                 handle_aggregator(task, mbufs[j]);
 174         }
 175 #ifdef PROX_PREFETCH_OFFSET
 176         prefetch_nta(rte_pktmbuf_mtod(mbufs[n_pkts - 1], void *));
 177         for (; j < n_pkts; ++j) {
 178                 handle_aggregator(task, mbufs[j]);
 179         }
 180 #endif
 181
 182         for (int i = 0 ; i < task->drop.pkt_nb; i++) {
 183                 drop_bytes += mbuf_wire_size(task->drop.buffer[i]);
 184                 rte_pktmbuf_free(task->drop.buffer[i]);
 185         }
 186         TASK_STATS_ADD_DROP_TX_FAIL(&tbase->aux->stats, task->drop.pkt_nb);
 187         TASK_STATS_ADD_DROP_BYTES(&tbase->aux->stats, drop_bytes);
 188         task->drop.pkt_nb = 0;
 189
 190         for (int priority = 0; priority < MAX_PRIORITIES; priority++) {
 191                 struct task_buffer *prio = &task->priority[priority];
 192                 if (prio->pkt_nb) {
 193                         uint8_t n = 0;
 194                         if (prio->pkt_pos > prio->pkt_nb) {
 195                                 struct rte_mbuf **buf = prio->buffer + prio->pkt_pos - prio->pkt_nb;
 196                                 n = tbase->aux->tx_pkt_try(&task->base, buf, prio->pkt_nb);
 197                         } else {
 198                                 struct rte_mbuf **buf = prio->buffer + BUFFER_LENGTH + prio->pkt_pos - prio->pkt_nb;
 199                                 n = tbase->aux->tx_pkt_try(&task->base, buf, prio->pkt_nb - prio->pkt_pos);
 200                                 if (n == (prio->pkt_nb - prio->pkt_pos))
 201                                         n += tbase->aux->tx_pkt_try(&task->base, prio->buffer, prio->pkt_pos);
 202                         }
 203                         prio->pkt_nb -=n;
 204                         TASK_STATS_ADD_TX_PRIO(&task->stats, n, priority);
 205                         if (prio->pkt_nb)
 206                                 break;
 207                 }
 208         }
 209         return 0;
 210 }
 211
 212 static void init_task_aggregator(struct task_base *tbase, struct task_args *targ)
 213 {
 214         struct task_aggregator *task = (struct task_aggregator *)tbase;
 215         const int socket_id = rte_lcore_to_socket_id(targ->lconf->id);
 216 }
 217
 218 static struct task_init task_init_aggregator = {
 219         .mode_str = "aggreg",
 220         .init = init_task_aggregator,
 221         .handle = handle_aggregator_bulk,
 222         .flag_features = TASK_FEATURE_NEVER_DISCARDS,
 223         .size = sizeof(struct task_aggregator)
 224 };
 225
 226 __attribute__((constructor)) static void reg_task_aggregator(void)
 227 {
 228         reg_task(&task_init_aggregator);
 229 }