/
test_cqex.py
122 lines (102 loc) · 3.94 KB
/
test_cqex.py
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
from tests.base import RCResources, UDResources, XRCResources, RDMATestCase, \
PyverbsAPITestCase
from pyverbs.pyverbs_error import PyverbsRDMAError
from pyverbs.cq import CqInitAttrEx, CQEX
import pyverbs.enums as e
from pyverbs.mr import MR
import tests.utils as u
import unittest
import errno
def create_ex_cq(res):
"""
Create an Extended CQ using res's context and assign it to res's cq member.
IBV_WC_STANDARD_FLAGS is used for WC flags to avoid support differences
between devices.
:param res: An instance of TrafficResources
"""
wc_flags = e.IBV_WC_STANDARD_FLAGS
cia = CqInitAttrEx(cqe=2000, wc_flags=wc_flags)
try:
res.cq = CQEX(res.ctx, cia)
except PyverbsRDMAError as ex:
if ex.error_code == errno.EOPNOTSUPP:
raise unittest.SkipTest('Create Extended CQ is not supported')
raise ex
class CqExUD(UDResources):
def create_cq(self):
create_ex_cq(self)
def create_mr(self):
self.mr = MR(self.pd, self.msg_size + self.GRH_SIZE,
e.IBV_ACCESS_LOCAL_WRITE)
class CqExRC(RCResources):
def create_cq(self):
create_ex_cq(self)
class CqExXRC(XRCResources):
def create_cq(self):
create_ex_cq(self)
class CqExTestCase(RDMATestCase):
"""
Run traffic over the existing UD, RC and XRC infrastructure, but use
ibv_cq_ex instead of legacy ibv_cq
"""
def setUp(self):
super().setUp()
self.iters = 100
def test_ud_traffic_cq_ex(self):
self.create_players(CqExUD)
u.traffic(**self.traffic_args, is_cq_ex=True)
def test_rc_traffic_cq_ex(self):
self.create_players(CqExRC)
u.traffic(**self.traffic_args, is_cq_ex=True)
def test_xrc_traffic_cq_ex(self):
self.create_players(CqExXRC)
u.xrc_traffic(self.client, self.server, is_cq_ex=True)
class CQEXAPITest(PyverbsAPITestCase):
"""
Test the API of the CQEX class.
"""
def setUp(self):
super().setUp()
self.max_cqe = self.attr.max_cqe
def test_create_cq_ex(self):
"""
Test ibv_create_cq_ex()
"""
cq_init_attrs_ex = CqInitAttrEx(cqe=10, wc_flags=0, comp_mask=0, flags=0)
if self.attr_ex.raw_packet_caps & e.IBV_RAW_PACKET_CAP_CVLAN_STRIPPING:
cq_init_attrs_ex.wc_flags = e.IBV_WC_EX_WITH_CVLAN
CQEX(self.ctx, cq_init_attrs_ex)
for flag in list(e.ibv_create_cq_wc_flags):
cq_init_attrs_ex.wc_flags = flag
try:
cq_ex = CQEX(self.ctx, cq_init_attrs_ex)
cq_ex.close()
except PyverbsRDMAError as ex:
if ex.error_code != errno.EOPNOTSUPP:
raise ex
cq_init_attrs_ex.wc_flags = 0
cq_init_attrs_ex.comp_mask = e.IBV_CQ_INIT_ATTR_MASK_FLAGS
attr_flags = list(e.ibv_create_cq_attr_flags)
for flag in attr_flags:
cq_init_attrs_ex.flags = flag
try:
cq_ex = CQEX(self.ctx, cq_init_attrs_ex)
cq_ex.close()
except PyverbsRDMAError as ex:
if ex.error_code != errno.EOPNOTSUPP:
raise ex
def test_create_cq_ex_bad_flow(self):
"""
Test ibv_create_cq_ex() with wrong comp_vector / number of cqes
"""
cq_attrs_ex = CqInitAttrEx(cqe=self.max_cqe + 1, wc_flags=0, comp_mask=0, flags=0)
with self.assertRaises(PyverbsRDMAError) as ex:
CQEX(self.ctx, cq_attrs_ex)
if ex.exception.error_code == errno.EOPNOTSUPP:
raise unittest.SkipTest('Create Extended CQ is not supported')
self.assertEqual(ex.exception.error_code, errno.EINVAL)
cq_attrs_ex = CqInitAttrEx(10, wc_flags=0, comp_mask=0, flags=0)
cq_attrs_ex.comp_vector = self.ctx.num_comp_vectors + 1
with self.assertRaises(PyverbsRDMAError) as ex:
CQEX(self.ctx, cq_attrs_ex)
self.assertEqual(ex.exception.error_code, errno.EINVAL)