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
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
|
/*
\brief 数据分发虚拟设备实现
数据分发虚设备用于MARSIO服务进程向各从应用分发数据，并将各从应用发送的数据转发到物理网卡。
\author Qiuwen Lu<[email protected]>
\date 2017-05-09
*/
#include <assert.h>
#include <rte_malloc.h>
#include <common.h>
#include <sc_common.h>
#include <sc_vdev.h>
static int vdev_data_dispatch(struct _vdev * _vdev, queue_id_t qid, struct rte_mbuf * pkts[], unsigned int nr_pkts,
int flags)
{
hash_t hash_value[MR_BURST_MAX];
assert(nr_pkts <= RTE_DIM(hash_value));
for (unsigned int i = 0; i < nr_pkts; i++)
{
hash_value[i] = pkts[i]->hash.usr;
}
for (unsigned int i = 0; i < nr_pkts; i++)
{
__rte_mbuf_sanity_check(pkts[i], 1);
}
return vnode_mirror_enqueue_bulk(_vdev->vnode_rx_prod, qid, pkts, hash_value, nr_pkts);
}
static int vdev_data_collect(struct _vdev * _vdev, queue_id_t qid, struct rte_mbuf * pkts[], unsigned int nr_pkts,
int flags)
{
int __nr_mbufs_out = 0;
int __nr_mbufs_left = nr_pkts;
int ret;
if (flags & VDEV_COLLECT_CTRL)
{
// 从CTRL通道取包
ret = vnode_mirror_dequeue_burst(_vdev->vnode_ctrl_cons, qid, &pkts[__nr_mbufs_out], __nr_mbufs_left);
__nr_mbufs_out += ret;
__nr_mbufs_left -= ret;
assert(__nr_mbufs_left >= 0);
}
if (flags & VDEV_COLLECT_DATA)
{
// 从FTX通道取包
ret = vnode_mirror_dequeue_burst(_vdev->vnode_ftx_cons, qid, &pkts[__nr_mbufs_out], __nr_mbufs_left);
__nr_mbufs_out += ret;
__nr_mbufs_left -= ret;
assert(__nr_mbufs_left >= 0);
// 剩余的容量，从TX通道取
ret = vnode_mirror_dequeue_burst(_vdev->vnode_tx_cons, qid, &pkts[__nr_mbufs_out], __nr_mbufs_left);
__nr_mbufs_out += ret;
__nr_mbufs_left -= ret;
assert(__nr_mbufs_left >= 0);
}
for (int i = 0; i < __nr_mbufs_out; i++)
{
__rte_mbuf_sanity_check(pkts[i], 1);
}
return __nr_mbufs_out;
}
static int vdev_data_idle_poll(struct _vdev * _vdev, queue_id_t qid)
{
vnode_mirror_flush(_vdev->vnode_rx_prod, qid);
return 0;
}
static int vdev_data_stats_get(struct _vdev * _vdev, struct vdev_stat_info * stat_info)
{
struct vnode_prod_stat * st_prod_rx = vnode_mirror_prod_stat_get(_vdev->vnode_rx_prod);
struct vnode_cons_stat * st_cons_tx = vnode_mirror_cons_stat_get(_vdev->vnode_tx_cons);
struct vnode_cons_stat * st_cons_ftx = vnode_mirror_cons_stat_get(_vdev->vnode_ftx_cons);
struct vnode_cons_stat * st_cons_ltx = vnode_mirror_cons_stat_get(_vdev->vnode_ctrl_cons);
// TODO: What about nr_rxstream, nr_txstream is different?
for (int i = 0; i < _vdev->nr_rxstream; i++)
{
stat_info->rx_on_line[i] = VNODE_STAT_READ(&st_prod_rx[i].on_line);
stat_info->rx_deliver[i] = VNODE_STAT_READ(&st_prod_rx[i].deliver);
stat_info->rx_missed[i] = VNODE_STAT_READ(&st_prod_rx[i].missed);
stat_info->rx_total_len[i] = VNODE_STAT_READ(&st_prod_rx[i].total_len);
#if 0
stat_info->notify_state_running[i] = rte_atomic64_read(&st_prod_rx[i].notify_state_running);
stat_info->notify_state_ready[i] = rte_atomic64_read(&st_prod_rx[i].notify_state_ready);
stat_info->notify_state_waiting[i] = rte_atomic64_read(&st_prod_rx[i].notify_state_waiting);
#endif
stat_info->batch_size_count[i] = VNODE_STAT_READ(&st_prod_rx[i].batch_size_count);
stat_info->batch_size_total[i] = VNODE_STAT_READ(&st_prod_rx[i].batch_size_total);
}
for (int i = 0; i < _vdev->nr_txstream; i++)
{
stat_info->tx_on_line[i] = VNODE_STAT_READ(&st_cons_tx[i].on_line);
stat_info->tx_deliver[i] = VNODE_STAT_READ(&st_cons_tx[i].deliver);
stat_info->tx_missed[i] = VNODE_STAT_READ(&st_cons_tx[i].missed);
stat_info->tx_total_len[i] = VNODE_STAT_READ(&st_cons_tx[i].total_len);
stat_info->ftx_on_line[i] = VNODE_STAT_READ(&st_cons_ftx[i].on_line);
stat_info->ftx_deliver[i] = VNODE_STAT_READ(&st_cons_ftx[i].deliver);
stat_info->ftx_missed[i] = VNODE_STAT_READ(&st_cons_ftx[i].missed);
stat_info->ftx_total_len[i] = VNODE_STAT_READ(&st_cons_ftx[i].total_len);
stat_info->ltx_on_line[i] = VNODE_STAT_READ(&st_cons_ltx[i].on_line);
stat_info->ltx_deliver[i] = VNODE_STAT_READ(&st_cons_ltx[i].deliver);
stat_info->ltx_missed[i] = VNODE_STAT_READ(&st_cons_ltx[i].missed);
stat_info->ltx_total_len[i] = VNODE_STAT_READ(&st_cons_ltx[i].total_len);
stat_info->ltx_on_line[i] = VNODE_STAT_READ(&st_cons_ltx[i].on_line);
stat_info->ltx_deliver[i] = VNODE_STAT_READ(&st_cons_ltx[i].deliver);
stat_info->ltx_missed[i] = VNODE_STAT_READ(&st_cons_ltx[i].missed);
stat_info->ltx_total_len[i] = VNODE_STAT_READ(&st_cons_ltx[i].total_len);
}
return 0;
}
static int vdev_data_instance_destory(struct vdev_instance * vdi)
{
if (vdi->vnode_rx_cons != NULL)
vnode_mirror_delete_cons(vdi->vnode_rx_cons);
if (vdi->vnode_tx_prod != NULL)
vnode_mirror_delete_prod(vdi->vnode_tx_prod);
if (vdi->vnode_ftx_prod != NULL)
vnode_mirror_delete_prod(vdi->vnode_ftx_prod);
if (vdi->vnode_ltx_prod != NULL)
vnode_mirror_delete_prod(vdi->vnode_ltx_prod);
FREE(vdi);
return 0;
}
/* 创建虚设备实例，应用使用 */
static struct vdev_instance * vdev_data_instance_create(struct _vdev * _vdev, const char * appsym,
unsigned int nr_rxstream, unsigned int nr_txstream)
{
struct vdev_instance * vdi = ZMALLOC(sizeof(struct vdev_instance));
MR_VERIFY_MALLOC(vdi);
vdi->vdev = &_vdev->vdev;
if (nr_rxstream > 0)
{
vdi->vnode_rx_cons = vnode_mirror_create_cons(_vdev->vnode_rx, appsym, nr_rxstream);
vdi->vnode_rx_cons_notify = vnode_mirror_notify_ctx_cons(vdi->vnode_rx_cons);
}
if (nr_txstream > 0)
{
vdi->vnode_tx_prod = vnode_mirror_create_prod(_vdev->vnode_tx, appsym, nr_txstream);
}
if (nr_txstream > 0)
{
vdi->vnode_ftx_prod = vnode_mirror_create_prod(_vdev->vnode_ftx, appsym, nr_txstream);
}
if (nr_txstream > 0)
{
vdi->vnode_ltx_prod = vnode_mirror_create_prod(_vdev->vnode_ltx, appsym, nr_txstream);
}
if (nr_rxstream > 0 && vdi->vnode_rx_cons == NULL)
{
MR_ERROR("Creating VDI for device %s failed: "
"Cannot create rx consumer, nr_rxstream = %d",
vdi->vdev->symbol, nr_rxstream);
goto errout;
}
if (nr_txstream > 0 && vdi->vnode_tx_prod == NULL)
{
MR_ERROR("Creating VDI for device %s failed: "
"Cannot create TX producer, nr_txstream = %d",
vdi->vdev->symbol, nr_txstream);
goto errout;
}
if (nr_txstream > 0 && vdi->vnode_ftx_prod == NULL)
{
MR_ERROR("Creating VDI for device %s failed: "
"Cannot create FTX producer, nr_txstream = %d",
vdi->vdev->symbol, nr_txstream);
goto errout;
}
if (nr_txstream > 0 && vdi->vnode_ltx_prod == NULL)
{
MR_ERROR("Creating VDI for device %s failed: "
"Cannot create LTX producer, nr_txstream = %d",
vdi->vdev->symbol, nr_txstream);
goto errout;
}
vdi->direct_pool = _vdev->direct_pool;
vdi->indirect_pool = _vdev->indirect_pool;
vdi->nr_rxstream = nr_rxstream;
vdi->nr_txstream = nr_txstream;
return vdi;
errout:
vdev_data_instance_destory(vdi);
return NULL;
}
static int vdev_data_destory(struct _vdev * _vdev)
{
return 0;
}
int vdev_data_create(struct vdev_main * v_main, const char * symbol, unsigned int nr_rxstream, unsigned int nr_txstream,
unsigned int sz_tunnel_rx_exclusive, unsigned int sz_tunnel_rx_shared, unsigned int sz_tunnel_tx,
unsigned int sz_buffer, unsigned int batch_interval_in_us, struct rte_mempool * direct_pool)
{
// 检查设备是否已经存在，不允许重复创建
struct vdev * vdev_info = vdev_lookup(v_main, symbol);
if (vdev_info != NULL)
{
MR_WARNING("vdev %s has been created. failed. \n", vdev_info->symbol);
return RT_IGNORE;
}
// 申请Info结构体的空间
struct _vdev * _vdev = ZMALLOC(sizeof(struct _vdev));
MR_VERIFY_MALLOC(_vdev);
/* 填充vdev信息 */
vdev_info = &_vdev->vdev;
/* 填充_vdev信息 */
snprintf(vdev_info->symbol, sizeof(vdev_info->symbol), "%s", symbol);
_vdev->nr_rxstream = nr_rxstream;
_vdev->nr_txstream = nr_txstream;
_vdev->sz_buffer = sz_buffer;
_vdev->sz_tunnel = sz_tunnel_rx_exclusive;
_vdev->direct_pool = direct_pool;
char vnode_sym_rx[MR_SYMBOL_MAX * 2];
snprintf(vnode_sym_rx, sizeof(vnode_sym_rx), "%s-rx", vdev_info->symbol);
char vnode_sym_tx[MR_SYMBOL_MAX * 2];
snprintf(vnode_sym_tx, sizeof(vnode_sym_tx), "%s-tx", vdev_info->symbol);
char vnode_sym_ftx[MR_SYMBOL_MAX * 2];
snprintf(vnode_sym_ftx, sizeof(vnode_sym_ftx), "%s-ftx", vdev_info->symbol);
char vnode_sym_ltx[MR_SYMBOL_MAX * 2];
snprintf(vnode_sym_ltx, sizeof(vnode_sym_ltx), "%s-ltx", vdev_info->symbol);
/* 创建VNODE */
_vdev->vnode_rx = vnode_mirror_create(vnode_sym_rx, sz_tunnel_rx_exclusive, sz_tunnel_rx_shared, sz_buffer, 1,
batch_interval_in_us);
_vdev->vnode_tx = vnode_mirror_create(vnode_sym_tx, sz_tunnel_tx, 0, sz_buffer, 0, 0);
_vdev->vnode_ftx = vnode_mirror_create(vnode_sym_ftx, sz_tunnel_tx, 0, 0, 0, 0);
_vdev->vnode_ltx = vnode_mirror_create(vnode_sym_ltx, sz_tunnel_tx, 0, 0, 0, 0);
#define ERR_VERIFY(x, ...) \
do \
{ \
if (x == NULL) \
{ \
MR_ERROR(__VA_ARGS__); \
goto errout; \
} \
} while (0)
/* 错误校验 */
ERR_VERIFY(_vdev->vnode_rx, "Create vdev %s rx vnode failed.", vdev_info->symbol);
ERR_VERIFY(_vdev->vnode_tx, "Create vdev %s tx vnode failed.", vdev_info->symbol);
ERR_VERIFY(_vdev->vnode_ftx, "Create vdev %s fast tx vnode failed. ", vdev_info->symbol);
ERR_VERIFY(_vdev->vnode_ltx, "Create vdev %s lock tx vnode failed. ", vdev_info->symbol);
_vdev->vnode_rx_prod = vnode_mirror_create_prod(_vdev->vnode_rx, "sv", nr_rxstream);
_vdev->vnode_tx_cons = vnode_mirror_create_cons(_vdev->vnode_tx, "sv", nr_txstream);
_vdev->vnode_ftx_cons = vnode_mirror_create_cons(_vdev->vnode_ftx, "sv", nr_txstream);
_vdev->vnode_ctrl_cons = vnode_mirror_create_cons(_vdev->vnode_ltx, "sv", nr_txstream);
/* 校验，申请VNODE、VNODE生产者、消费者是否成功 */
ERR_VERIFY(_vdev->vnode_rx_prod, "Create vdev %s rx vnode producer failed. ", vdev_info->symbol);
ERR_VERIFY(_vdev->vnode_tx_cons, "Create vdev %s tx vnode consumer failed. ", vdev_info->symbol);
ERR_VERIFY(_vdev->vnode_ftx_cons, "Create vdev %s fast tx vnode consumer failed. ", vdev_info->symbol);
ERR_VERIFY(_vdev->vnode_ctrl_cons, "Create vdev %s lock tx vnode consumer failed. ", vdev_info->symbol);
/* 注册回调函数 */
_vdev->dispatch = vdev_data_dispatch;
_vdev->collect = vdev_data_collect;
_vdev->idle_pool = vdev_data_idle_poll;
_vdev->destory = vdev_data_destory;
_vdev->stats_get = vdev_data_stats_get;
_vdev->vdi_create = vdev_data_instance_create;
_vdev->vdi_destory = vdev_data_instance_destory;
/* 加入虚设备列表 */
MR_VERIFY(v_main->vdev_max_idx <= RTE_DIM(v_main->_vdev_array));
vdev_info->port_id = v_main->vdev_max_idx;
v_main->_vdev_array[v_main->vdev_max_idx++] = _vdev;
return RT_SUCCESS;
errout:
if (_vdev != NULL)
vdev_data_destory(_vdev);
rte_free(_vdev);
return RT_ERR;
#undef ERR_VERIFY
}
|
