1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
|
#include <stdio.h>
#include "bp_sim.h"
#include "flow_stat_parser.h"
#include "latency.h"
#include "trex_stateless_messaging.h"
#include "trex_stateless_rx_core.h"
void CRxCoreStateless::create(const CRxSlCfg &cfg) {
m_max_ports = cfg.m_max_ports;
CMessagingManager * cp_rx = CMsgIns::Ins()->getCpRx();
m_ring_from_cp = cp_rx->getRingCpToDp(0);
m_ring_to_cp = cp_rx->getRingDpToCp(0);
m_state = STATE_IDLE;
for (int i = 0; i < m_max_ports; i++) {
CLatencyManagerPerPort * lp = &m_ports[i];
lp->m_io = cfg.m_ports[i];
}
m_cpu_cp_u.Create(&m_cpu_dp_u);
}
void CRxCoreStateless::handle_cp_msg(TrexStatelessCpToRxMsgBase *msg) {
msg->handle(this);
delete msg;
}
bool CRxCoreStateless::periodic_check_for_cp_messages() {
/* fast path */
if ( likely ( m_ring_from_cp->isEmpty() ) ) {
return false;
}
while ( true ) {
CGenNode * node = NULL;
if (m_ring_from_cp->Dequeue(node) != 0) {
break;
}
assert(node);
TrexStatelessCpToRxMsgBase * msg = (TrexStatelessCpToRxMsgBase *)node;
handle_cp_msg(msg);
}
return true;
}
void CRxCoreStateless::idle_state_loop() {
const int SHORT_DELAY_MS = 2;
const int LONG_DELAY_MS = 50;
const int DEEP_SLEEP_LIMIT = 2000;
int counter = 0;
while (m_state == STATE_IDLE) {
bool had_msg = periodic_check_for_cp_messages();
if (had_msg) {
counter = 0;
continue;
}
/* enter deep sleep only if enough time had passed */
if (counter < DEEP_SLEEP_LIMIT) {
delay(SHORT_DELAY_MS);
counter++;
} else {
delay(LONG_DELAY_MS);
}
}
}
void CRxCoreStateless::start() {
static int count = 0;
static int i = 0;
bool do_try_rx_queue =CGlobalInfo::m_options.preview.get_vm_one_queue_enable() ? true : false;
while (true) {
if (m_state == STATE_WORKING) {
i++;
if (i == 100) {
i = 0;
// if no packets in 100 cycles, sleep for a while to spare the cpu
if (count == 0) {
delay(1);
}
count = 0;
periodic_check_for_cp_messages(); // m_state might change in here
}
} else {
if (m_state == STATE_QUIT)
break;
idle_state_loop();
}
if (do_try_rx_queue) {
try_rx_queues();
}
count += try_rx();
}
}
void CRxCoreStateless::handle_rx_pkt(CLatencyManagerPerPort *lp, rte_mbuf_t *m) {
Cxl710Parser parser;
if (parser.parse(rte_pktmbuf_mtod(m, uint8_t *), m->pkt_len) == 0) {
uint16_t ip_id;
if (parser.get_ip_id(ip_id) == 0) {
if (is_flow_stat_id(ip_id)) {
uint16_t hw_id = get_hw_id(ip_id);
lp->m_port.m_rx_pg_stat[hw_id].add_pkts(1);
lp->m_port.m_rx_pg_stat[hw_id].add_bytes(m->pkt_len);
}
}
}
}
// In VM setup, handle packets coming as messages from DP cores.
void CRxCoreStateless::handle_rx_queue_msgs(uint8_t thread_id, CNodeRing * r) {
while ( true ) {
CGenNode * node;
if ( r->Dequeue(node) != 0 ) {
break;
}
assert(node);
CGenNodeMsgBase * msg = (CGenNodeMsgBase *)node;
CGenNodeLatencyPktInfo * l_msg;
uint8_t msg_type = msg->m_msg_type;
uint8_t rx_port_index;
CLatencyManagerPerPort * lp;
switch (msg_type) {
case CGenNodeMsgBase::LATENCY_PKT:
l_msg = (CGenNodeLatencyPktInfo *)msg;
assert(l_msg->m_latency_offset == 0xdead);
rx_port_index = (thread_id << 1) + (l_msg->m_dir & 1);
assert( rx_port_index < m_max_ports );
lp = &m_ports[rx_port_index];
handle_rx_pkt(lp, (rte_mbuf_t *)l_msg->m_pkt);
break;
default:
printf("ERROR latency-thread message type is not valid %d \n", msg_type);
assert(0);
}
CGlobalInfo::free_node(node);
}
}
// VM mode function. Handle messages from DP
void CRxCoreStateless::try_rx_queues() {
CMessagingManager * rx_dp = CMsgIns::Ins()->getRxDp();
uint8_t threads=CMsgIns::Ins()->get_num_threads();
int ti;
for (ti = 0; ti < (int)threads; ti++) {
CNodeRing * r = rx_dp->getRingDpToCp(ti);
if ( ! r->isEmpty() ) {
handle_rx_queue_msgs((uint8_t)ti, r);
}
}
}
int CRxCoreStateless::try_rx() {
rte_mbuf_t * rx_pkts[64];
int i, total_pkts = 0;
for (i = 0; i < m_max_ports; i++) {
CLatencyManagerPerPort * lp = &m_ports[i];
rte_mbuf_t * m;
m_cpu_dp_u.start_work();
/* try to read 64 packets clean up the queue */
uint16_t cnt_p = lp->m_io->rx_burst(rx_pkts, 64);
total_pkts += cnt_p;
if (cnt_p) {
int j;
for (j = 0; j < cnt_p; j++) {
m = rx_pkts[j];
handle_rx_pkt(lp, m);
rte_pktmbuf_free(m);
}
/* commit only if there was work to do ! */
m_cpu_dp_u.commit();
}/* if work */
}// all ports
return total_pkts;
}
bool CRxCoreStateless::is_flow_stat_id(uint16_t id) {
if ((id & 0xff00) == IP_ID_RESERVE_BASE) return true;
return false;
}
uint16_t CRxCoreStateless::get_hw_id(uint16_t id) {
return (0x00ff & id);
}
void CRxCoreStateless::reset_rx_stats(uint8_t port_id) {
for (int hw_id = 0; hw_id < MAX_FLOW_STATS; hw_id++) {
m_ports[port_id].m_port.m_rx_pg_stat[hw_id].clear();
}
}
int CRxCoreStateless::get_rx_stats(uint8_t port_id, rx_per_flow_t *rx_stats, int min, int max, bool reset) {
for (int hw_id = min; hw_id <= max; hw_id++) {
rx_stats[hw_id - min] = m_ports[port_id].m_port.m_rx_pg_stat[hw_id];
if (reset) {
m_ports[port_id].m_port.m_rx_pg_stat[hw_id].clear();
}
}
return 0;
}
double CRxCoreStateless::get_cpu_util() {
m_cpu_cp_u.Update();
return m_cpu_cp_u.GetVal();
}
|
