2 // Copyright (c) 2010-2017 Intel Corporation
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
8 // http://www.apache.org/licenses/LICENSE-2.0
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.
20 #include <rte_version.h>
23 #include "prox_lua_types.h"
26 #include "task_base.h"
27 #include "task_init.h"
34 #include "prox_shared.h"
35 #include "mbuf_utils.h"
36 #include "handle_aggregator.h"
38 #define PRIORITY_DHCP (HIGH_PRIORITY)
40 #define TASK_STATS_ADD_DROP_TX_FAIL_PRIO(stats, ntx, prio) do { \
41 (stats)->drop_tx_fail_prio[prio] += ntx; \
43 #define TASK_STATS_ADD_TX_PRIO(stats, ntx, prio) do { \
44 (stats)->rx_prio[prio] += ntx; \
47 static inline uint8_t detect_l4_priority(uint8_t l3_priority, const struct ipv4_hdr *ipv4_hdr)
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))) {
58 static inline uint8_t detect_l3_priority(uint8_t l2_priority, const struct ipv4_hdr *ipv4_hdr)
61 if ((ipv4_hdr->version_ihl >> 4) == 4) {
62 } else if ((ipv4_hdr->version_ihl >> 4) == 6) {
63 plog_warn("IPv6 Not implemented\n");
66 plog_warn("Unexpected IP version\n");
69 dscp = ipv4_hdr->type_of_service >> 2;
71 return MAX_PRIORITIES - dscp - 1;
76 static inline uint8_t detect_l2_priority(const struct qinq_hdr *pqinq)
78 if (pqinq->cvlan.eth_proto != ETYPE_VLAN) {
79 plog_warn("Unexpected proto in QinQ = %#04x\n", pqinq->cvlan.eth_proto);
82 uint16_t svlan_priority = ntohs(pqinq->svlan.vlan_tci >> 13);
83 uint16_t cvlan_priority = ntohs(pqinq->cvlan.vlan_tci >> 13);
85 return svlan_priority;
87 return cvlan_priority;
90 static inline void buffer_packet(struct task_aggregator *task, struct rte_mbuf *mbuf, uint8_t priority)
92 struct task_base *tbase = (struct task_base *)task;
94 struct task_buffer *prio = &task->priority[priority];
95 if (prio->pkt_nb < BUFFER_LENGTH) {
96 prio->buffer[prio->pkt_pos] = mbuf;
98 if (prio->pkt_pos == BUFFER_LENGTH)
102 task->drop.buffer[task->drop.pkt_nb] = mbuf;
104 TASK_STATS_ADD_DROP_TX_FAIL_PRIO(&task->stats, 1, priority);
108 static inline void handle_aggregator(struct task_aggregator *task, struct rte_mbuf *mbuf)
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;
115 const uint16_t eth_type = peth->ether_type;
121 pqinq = rte_pktmbuf_mtod(mbuf, const struct qinq_hdr *);
122 if ((priority = detect_l2_priority(pqinq)) == OUT_DISCARD)
124 ipv4_hdr = (const struct ipv4_hdr *)(pqinq + 1);
125 if ((priority = detect_l3_priority(priority, ipv4_hdr)) == OUT_DISCARD)
127 if ((priority = detect_l4_priority(priority, ipv4_hdr)) == OUT_DISCARD)
133 ipv4_hdr = (const struct ipv4_hdr *)(peth+1);
134 if ((priority = detect_l3_priority(LOW_PRIORITY, ipv4_hdr)) == OUT_DISCARD)
136 if ((priority = detect_l4_priority(priority, ipv4_hdr)) == OUT_DISCARD)
146 if (priority == OUT_DISCARD) {
147 task->drop.buffer[task->drop.pkt_nb] = mbuf;
151 buffer_packet(task, mbuf, priority);
154 static int handle_aggregator_bulk(struct task_base *tbase, struct rte_mbuf **mbufs, uint16_t n_pkts)
156 struct task_aggregator *task = (struct task_aggregator *)tbase;
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]);
164 for (j = 1; j < PROX_PREFETCH_OFFSET && j < n_pkts; ++j) {
165 prefetch_nta(rte_pktmbuf_mtod(mbufs[j - 1], void *));
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 *));
173 handle_aggregator(task, mbufs[j]);
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]);
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]);
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;
190 for (int priority = 0; priority < MAX_PRIORITIES; priority++) {
191 struct task_buffer *prio = &task->priority[priority];
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);
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);
204 TASK_STATS_ADD_TX_PRIO(&task->stats, n, priority);
212 static void init_task_aggregator(struct task_base *tbase, struct task_args *targ)
214 struct task_aggregator *task = (struct task_aggregator *)tbase;
215 const int socket_id = rte_lcore_to_socket_id(targ->lconf->id);
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)
226 __attribute__((constructor)) static void reg_task_aggregator(void)
228 reg_task(&task_init_aggregator);