Merge "Fix latency accuracy and dumping latencies to file"
[samplevnf.git] / 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 }