tests/test_mlx5_devx.py - linux-rdma/rdma-core - Git at Google

 # SPDX-License-Identifier: (GPL-2.0 OR Linux-OpenIB)
 # Copyright (c) 2021 Nvidia Inc. All rights reserved. See COPYING file

 """
 Test module for mlx5 DevX.
 """

 import resource
 import unittest
 import errno
 import os

 from tests.mlx5_base import Mlx5DevxRcResources, Mlx5DevxTrafficBase
 from tests.test_buf import alloc_buf, device_has_cc_dma_bounce, \
     make_cc_pd, register_buf_mr
 from pyverbs.providers.mlx5.mlx5dv import Mlx5Context, Mlx5DVContextAttr, \
     Mlx5DevxCmdComp, Mlx5DevxObj, Mlx5UMEM
 from pyverbs.providers.mlx5.mlx5_enums import mlx5dv_context_attr_flags
 from pyverbs.pyverbs_error import PyverbsRDMAError, PyverbsUserError
 import pyverbs.mem_alloc as mem
 from pyverbs.mr import MR
 from pyverbs.libibverbs_enums import ibv_access_flags, ibv_odp_transport_cap_bits
 import tests.utils as u


 class BufDevxRcResources(Mlx5DevxRcResources):
     """
     DevX RC resources for a Confidential Computing (CoCo) guest: every DevX
     UMEM and the data MR live in shared/unprotected memory allocated with
     Buf on a CC parent domain. Each UMEM is registered through the
     DevX dmabuf path using an FD exported from its Buf, so all NIC-DMA'd memory
     is shared as a DMA-bounce device requires.
     """
     def __init__(self, dev_name, ib_port, gid_index, msg_size=1024,
                  activate_port_state=False, send_dbr_mode=0):
         self.bufs = []
         self.dmabuf_fds = []
         self.base_pd = None
         super().__init__(dev_name, ib_port, gid_index, msg_size,
                          activate_port_state, send_dbr_mode)

     def create_pd(self):
         """Build a CC parent domain and derive the DevX pdn from it."""
         from pyverbs.providers.mlx5.mlx5dv_objects import Mlx5DvObj
         from pyverbs.providers.mlx5.mlx5_enums import mlx5dv_obj_type
         if not device_has_cc_dma_bounce(self.ctx):
             raise unittest.SkipTest('Device is not a CC DMA-bounce device')
         self.base_pd, self.pd = make_cc_pd(self.ctx)
         self.dv_pd = Mlx5DvObj(mlx5dv_obj_type.MLX5DV_OBJ_PD, pd=self.pd).dvpd

     def create_mr(self):
         """Register the data buffer as a shared Buf MR on the CC PD."""
         access = ibv_access_flags.IBV_ACCESS_REMOTE_WRITE | \
                  ibv_access_flags.IBV_ACCESS_LOCAL_WRITE | \
                  ibv_access_flags.IBV_ACCESS_REMOTE_READ
         buf = alloc_buf(self.pd, self.msg_size)
         self.mr = register_buf_mr(self.pd, buf, self.msg_size, access)

     def create_umem(self, size, access=ibv_access_flags.IBV_ACCESS_LOCAL_WRITE,
                     alignment=resource.getpagesize()):
         """Return a DevX UMEM backed by a Buf exported as a dmabuf FD."""
         from pyverbs.providers.mlx5.mlx5_enums import MLX5DV_UMEM_MASK_DMABUF
         page_size = resource.getpagesize()
         alloc_size = max(size, page_size)
         buf = alloc_buf(self.pd, alloc_size)
         mem.write(buf.addr, bytes(alloc_size), alloc_size)  # Zero-fill the buffer
         fd = self.export_buf_dmabuf_fd(buf)
         umem = Mlx5UMEM(self.ctx, alloc_size, addr=0, alignment=alignment, access=access,
                         pgsz_bitmap=page_size, comp_mask=MLX5DV_UMEM_MASK_DMABUF, dmabuf_fd=fd)
         umem.umem_addr = buf.addr
         self.bufs.append(buf)
         self.dmabuf_fds.append(fd)
         return umem

     def export_buf_dmabuf_fd(self, buf):
         """Export the Buf's dmabuf FD, skipping when it is not dmabuf-backed."""
         try:
             return buf.export_dmabuf_fd()
         except PyverbsRDMAError as ex:
             if ex.error_code == errno.ENODATA:
                 raise unittest.SkipTest('Buf is not dmabuf-backed')
             raise

     def close_resources(self):
         """Close the exported dmabuf FDs"""
         super().close_resources()
         for fd in self.dmabuf_fds:
             os.close(fd)
         self.dmabuf_fds = []


 class Mlx5DevxRcOdpRes(Mlx5DevxRcResources):
     @u.requires_odpv2
     def create_mr(self):
         self.with_odp = True
         self.user_addr = mem.mmap(length=self.msg_size,
                                   flags=mem.MAP_ANONYMOUS_ | mem.MAP_PRIVATE_)
         access = ibv_access_flags.IBV_ACCESS_LOCAL_WRITE | ibv_access_flags.IBV_ACCESS_REMOTE_READ | \
                  ibv_access_flags.IBV_ACCESS_ON_DEMAND
         self.mr = MR(self.pd, self.msg_size, access, self.user_addr)


 class Mlx5DevxRcTrafficTest(Mlx5DevxTrafficBase):
     """
     Test various functionality of mlx5 DevX objects
     """

     def test_devx_rc_qp_send_imm_traffic(self):
         """
         Creates two DevX RC QPs and modifies them to RTS state.
         Then does SEND_IMM traffic.
         """
         self.create_players(Mlx5DevxRcResources)
         # Send traffic
         self.send_imm_traffic()

     def test_devx_rc_qp_send_imm_buf_umem_traffic(self):
         """
         Run DevX RC SEND_IMM traffic where all NIC memory (QP, CQ, doorbell
         UMEMs and the data MR) is shared CoCo memory: buffers on a CC parent
         domain, registered as UMEMs via their dmabuf FD.
         """
         self.create_players(BufDevxRcResources)
         self.send_imm_traffic()

     def test_devx_rc_qp_send_imm_doorbell_less_traffic(self):
         """
         Creates two DevX RC QPs with dbr less ext and modifies them to RTS state.
         Then does SEND_IMM traffic.
         """
         from tests.mlx5_prm_structs import SendDbrMode

         self.create_players(Mlx5DevxRcResources, send_dbr_mode=SendDbrMode.NO_DBR_EXT)
         # Send traffic
         self.send_imm_traffic()

     @u.requires_odp('rc', ibv_odp_transport_cap_bits.IBV_ODP_SUPPORT_SEND | ibv_odp_transport_cap_bits.IBV_ODP_SUPPORT_RECV)
     def test_devx_rc_qp_odp_traffic(self):
         """
         Creates two DevX RC QPs using ODP enabled MKeys.
         Then does SEND_IMM traffic.
         """
         self.create_players(Mlx5DevxRcOdpRes)
         # Send traffic
         self.send_imm_traffic()


 class Mlx5DevxApiTest(Mlx5DevxTrafficBase):
     def setUp(self):
         super().setUp()
         self.devx_res = None

     def tearDown(self):
         super().tearDown()
         if self.devx_res:
             self.devx_res.close_resources()

     def _create_devx_ctx(self):
         try:
             attr = Mlx5DVContextAttr(mlx5dv_context_attr_flags.MLX5DV_CONTEXT_FLAGS_DEVX)
             return Mlx5Context(attr, self.dev_name)
         except PyverbsUserError as ex:
             raise unittest.SkipTest(f'Could not open mlx5 context ({ex})')
         except PyverbsRDMAError:
             raise unittest.SkipTest('Opening mlx5 DevX context is not supported')

     def test_devx_async_query(self):
         """
         Test DevX Async Query API.
         Creating a DevX QP and query it using DevX async query.
         """
         self.devx_res = Mlx5DevxRcResources(**self.dev_info)
         self.cmd_comp = Mlx5DevxCmdComp(self.devx_res.ctx)
         from tests.mlx5_prm_structs import QueryQpIn, QueryQpOut
         query_qp_in = QueryQpIn(qpn=self.devx_res.qpn)
         qp_wr_id = 100
         try:
             self.devx_res.qp.query_async(query_qp_in, len(QueryQpOut()), wr_id=qp_wr_id,
                                     cmd_comp=self.cmd_comp)
             wr_id, out_data = self.cmd_comp.get_async_cmd_comp()
         except PyverbsRDMAError as ex:
             if ex.error_code == errno.EOPNOTSUPP:
                 raise unittest.SkipTest('Async command completion is not supported')
             raise ex

         query_qp_out = QueryQpOut(out_data)
         self.assertTrue(query_qp_out.status == 0,
                         'Query Devx QP by Async Query API failed with non-zero status: '
                         f'{query_qp_out.status}')
         self.assertTrue(wr_id == qp_wr_id,
                         f'Mismatched work request ID. Expected: {qp_wr_id}, Actual: {wr_id}')
         self.assertTrue(query_qp_out.sw_qpc.log_rq_size == self.devx_res.log_rq_size,
                         f'Mismatched RQ size. Expected: {self.devx_res.log_rq_size}, '
                         f'Actual: {query_qp_out.sw_qpc.log_rq_size}')

     @u.skip_unsupported
     def test_umem_export_import(self):
         """
         Create UMEM, export, import, then close both.
         """
         with self._create_devx_ctx() as ctx:
             with Mlx5UMEM(ctx, size=resource.getpagesize()) as umem:
                 original_id = umem.umem_id
                 data = umem.export()
                 with Mlx5UMEM.import_umem(ctx, data) as imported_umem:
                     self.assertEqual(imported_umem.umem_id, original_id,
                                      f'Imported UMEM ID {imported_umem.umem_id}'
                                      f' does not match original {original_id}')

     @u.skip_unsupported
     def test_devx_obj_export_import(self):
         """
         Create a DevX flow counter object, export to opaque buffer,
         import from buffer, then close both.
         """
         from tests.mlx5_prm_structs import AllocFlowCounterIn, AllocFlowCounterOut

         with self._create_devx_ctx() as ctx:
             with Mlx5DevxObj(ctx, AllocFlowCounterIn(),
                              len(AllocFlowCounterOut())) as counter:
                 data = counter.export()
                 with Mlx5DevxObj.import_obj(ctx, data) as imported_counter:
                     self.assertIsNotNone(imported_counter,
                                          'Failed to import DevX object '
                                          'from exported data')
	# SPDX-License-Identifier: (GPL-2.0 OR Linux-OpenIB)
	# Copyright (c) 2021 Nvidia Inc. All rights reserved. See COPYING file

	"""
	Test module for mlx5 DevX.
	"""

	import resource
	import unittest
	import errno
	import os

	from tests.mlx5_base import Mlx5DevxRcResources, Mlx5DevxTrafficBase
	from tests.test_buf import alloc_buf, device_has_cc_dma_bounce, \
	make_cc_pd, register_buf_mr
	from pyverbs.providers.mlx5.mlx5dv import Mlx5Context, Mlx5DVContextAttr, \
	Mlx5DevxCmdComp, Mlx5DevxObj, Mlx5UMEM
	from pyverbs.providers.mlx5.mlx5_enums import mlx5dv_context_attr_flags
	from pyverbs.pyverbs_error import PyverbsRDMAError, PyverbsUserError
	import pyverbs.mem_alloc as mem
	from pyverbs.mr import MR
	from pyverbs.libibverbs_enums import ibv_access_flags, ibv_odp_transport_cap_bits
	import tests.utils as u


	class BufDevxRcResources(Mlx5DevxRcResources):
	"""
	DevX RC resources for a Confidential Computing (CoCo) guest: every DevX
	UMEM and the data MR live in shared/unprotected memory allocated with
	Buf on a CC parent domain. Each UMEM is registered through the
	DevX dmabuf path using an FD exported from its Buf, so all NIC-DMA'd memory
	is shared as a DMA-bounce device requires.
	"""
	def __init__(self, dev_name, ib_port, gid_index, msg_size=1024,
	activate_port_state=False, send_dbr_mode=0):
	self.bufs = []
	self.dmabuf_fds = []
	self.base_pd = None
	super().__init__(dev_name, ib_port, gid_index, msg_size,
	activate_port_state, send_dbr_mode)

	def create_pd(self):
	"""Build a CC parent domain and derive the DevX pdn from it."""
	from pyverbs.providers.mlx5.mlx5dv_objects import Mlx5DvObj
	from pyverbs.providers.mlx5.mlx5_enums import mlx5dv_obj_type
	if not device_has_cc_dma_bounce(self.ctx):
	raise unittest.SkipTest('Device is not a CC DMA-bounce device')
	self.base_pd, self.pd = make_cc_pd(self.ctx)
	self.dv_pd = Mlx5DvObj(mlx5dv_obj_type.MLX5DV_OBJ_PD, pd=self.pd).dvpd

	def create_mr(self):
	"""Register the data buffer as a shared Buf MR on the CC PD."""
	access = ibv_access_flags.IBV_ACCESS_REMOTE_WRITE \| \
	ibv_access_flags.IBV_ACCESS_LOCAL_WRITE \| \
	ibv_access_flags.IBV_ACCESS_REMOTE_READ
	buf = alloc_buf(self.pd, self.msg_size)
	self.mr = register_buf_mr(self.pd, buf, self.msg_size, access)

	def create_umem(self, size, access=ibv_access_flags.IBV_ACCESS_LOCAL_WRITE,
	alignment=resource.getpagesize()):
	"""Return a DevX UMEM backed by a Buf exported as a dmabuf FD."""
	from pyverbs.providers.mlx5.mlx5_enums import MLX5DV_UMEM_MASK_DMABUF
	page_size = resource.getpagesize()
	alloc_size = max(size, page_size)
	buf = alloc_buf(self.pd, alloc_size)
	mem.write(buf.addr, bytes(alloc_size), alloc_size) # Zero-fill the buffer
	fd = self.export_buf_dmabuf_fd(buf)
	umem = Mlx5UMEM(self.ctx, alloc_size, addr=0, alignment=alignment, access=access,
	pgsz_bitmap=page_size, comp_mask=MLX5DV_UMEM_MASK_DMABUF, dmabuf_fd=fd)
	umem.umem_addr = buf.addr
	self.bufs.append(buf)
	self.dmabuf_fds.append(fd)
	return umem

	def export_buf_dmabuf_fd(self, buf):
	"""Export the Buf's dmabuf FD, skipping when it is not dmabuf-backed."""
	try:
	return buf.export_dmabuf_fd()
	except PyverbsRDMAError as ex:
	if ex.error_code == errno.ENODATA:
	raise unittest.SkipTest('Buf is not dmabuf-backed')
	raise

	def close_resources(self):
	"""Close the exported dmabuf FDs"""
	super().close_resources()
	for fd in self.dmabuf_fds:
	os.close(fd)
	self.dmabuf_fds = []


	class Mlx5DevxRcOdpRes(Mlx5DevxRcResources):
	@u.requires_odpv2
	def create_mr(self):
	self.with_odp = True
	self.user_addr = mem.mmap(length=self.msg_size,
	flags=mem.MAP_ANONYMOUS_ \| mem.MAP_PRIVATE_)
	access = ibv_access_flags.IBV_ACCESS_LOCAL_WRITE \| ibv_access_flags.IBV_ACCESS_REMOTE_READ \| \
	ibv_access_flags.IBV_ACCESS_ON_DEMAND
	self.mr = MR(self.pd, self.msg_size, access, self.user_addr)


	class Mlx5DevxRcTrafficTest(Mlx5DevxTrafficBase):
	"""
	Test various functionality of mlx5 DevX objects
	"""

	def test_devx_rc_qp_send_imm_traffic(self):
	"""
	Creates two DevX RC QPs and modifies them to RTS state.
	Then does SEND_IMM traffic.
	"""
	self.create_players(Mlx5DevxRcResources)
	# Send traffic
	self.send_imm_traffic()

	def test_devx_rc_qp_send_imm_buf_umem_traffic(self):
	"""
	Run DevX RC SEND_IMM traffic where all NIC memory (QP, CQ, doorbell
	UMEMs and the data MR) is shared CoCo memory: buffers on a CC parent
	domain, registered as UMEMs via their dmabuf FD.
	"""
	self.create_players(BufDevxRcResources)
	self.send_imm_traffic()

	def test_devx_rc_qp_send_imm_doorbell_less_traffic(self):
	"""
	Creates two DevX RC QPs with dbr less ext and modifies them to RTS state.
	Then does SEND_IMM traffic.
	"""
	from tests.mlx5_prm_structs import SendDbrMode

	self.create_players(Mlx5DevxRcResources, send_dbr_mode=SendDbrMode.NO_DBR_EXT)
	# Send traffic
	self.send_imm_traffic()

	@u.requires_odp('rc', ibv_odp_transport_cap_bits.IBV_ODP_SUPPORT_SEND \| ibv_odp_transport_cap_bits.IBV_ODP_SUPPORT_RECV)
	def test_devx_rc_qp_odp_traffic(self):
	"""
	Creates two DevX RC QPs using ODP enabled MKeys.
	Then does SEND_IMM traffic.
	"""
	self.create_players(Mlx5DevxRcOdpRes)
	# Send traffic
	self.send_imm_traffic()


	class Mlx5DevxApiTest(Mlx5DevxTrafficBase):
	def setUp(self):
	super().setUp()
	self.devx_res = None

	def tearDown(self):
	super().tearDown()
	if self.devx_res:
	self.devx_res.close_resources()

	def _create_devx_ctx(self):
	try:
	attr = Mlx5DVContextAttr(mlx5dv_context_attr_flags.MLX5DV_CONTEXT_FLAGS_DEVX)
	return Mlx5Context(attr, self.dev_name)
	except PyverbsUserError as ex:
	raise unittest.SkipTest(f'Could not open mlx5 context ({ex})')
	except PyverbsRDMAError:
	raise unittest.SkipTest('Opening mlx5 DevX context is not supported')

	def test_devx_async_query(self):
	"""
	Test DevX Async Query API.
	Creating a DevX QP and query it using DevX async query.
	"""
	self.devx_res = Mlx5DevxRcResources(**self.dev_info)
	self.cmd_comp = Mlx5DevxCmdComp(self.devx_res.ctx)
	from tests.mlx5_prm_structs import QueryQpIn, QueryQpOut
	query_qp_in = QueryQpIn(qpn=self.devx_res.qpn)
	qp_wr_id = 100
	try:
	self.devx_res.qp.query_async(query_qp_in, len(QueryQpOut()), wr_id=qp_wr_id,
	cmd_comp=self.cmd_comp)
	wr_id, out_data = self.cmd_comp.get_async_cmd_comp()
	except PyverbsRDMAError as ex:
	if ex.error_code == errno.EOPNOTSUPP:
	raise unittest.SkipTest('Async command completion is not supported')
	raise ex

	query_qp_out = QueryQpOut(out_data)
	self.assertTrue(query_qp_out.status == 0,
	'Query Devx QP by Async Query API failed with non-zero status: '
	f'{query_qp_out.status}')
	self.assertTrue(wr_id == qp_wr_id,
	f'Mismatched work request ID. Expected: {qp_wr_id}, Actual: {wr_id}')
	self.assertTrue(query_qp_out.sw_qpc.log_rq_size == self.devx_res.log_rq_size,
	f'Mismatched RQ size. Expected: {self.devx_res.log_rq_size}, '
	f'Actual: {query_qp_out.sw_qpc.log_rq_size}')

	@u.skip_unsupported
	def test_umem_export_import(self):
	"""
	Create UMEM, export, import, then close both.
	"""
	with self._create_devx_ctx() as ctx:
	with Mlx5UMEM(ctx, size=resource.getpagesize()) as umem:
	original_id = umem.umem_id
	data = umem.export()
	with Mlx5UMEM.import_umem(ctx, data) as imported_umem:
	self.assertEqual(imported_umem.umem_id, original_id,
	f'Imported UMEM ID {imported_umem.umem_id}'
	f' does not match original {original_id}')

	@u.skip_unsupported
	def test_devx_obj_export_import(self):
	"""
	Create a DevX flow counter object, export to opaque buffer,
	import from buffer, then close both.
	"""
	from tests.mlx5_prm_structs import AllocFlowCounterIn, AllocFlowCounterOut

	with self._create_devx_ctx() as ctx:
	with Mlx5DevxObj(ctx, AllocFlowCounterIn(),
	len(AllocFlowCounterOut())) as counter:
	data = counter.export()
	with Mlx5DevxObj.import_obj(ctx, data) as imported_counter:
	self.assertIsNotNone(imported_counter,
	'Failed to import DevX object '
	'from exported data')