(self, opcode, log_size, entry_sz, owner_off, extra_sz=0, **ctx_kw)
| 229 | if MLX_DEBUG >= 1: print(f"mlx5: QP 0x{self.qp_info['qpn']:x} (EQ={self.eq_info['eq_number']} CQ=0x{self.cq_info['cqn']:x})") |
| 230 | |
| 231 | def create_queue(self, opcode, log_size, entry_sz, owner_off, extra_sz=0, **ctx_kw): |
| 232 | mem, paddrs = self.dev.pci_dev.alloc_sysmem((n := ceildiv((1 << log_size) * entry_sz + extra_sz, 0x1000)) * 0x1000) |
| 233 | return mem, paddrs, self.dev.cmd.exec(opcode, payload=struct.pack(f'>{n}Q', *paddrs), **ctx_kw) |
| 234 | |
| 235 | def qp_op(self, opcode, qpc_args=None, addr_args=None, **kwargs): |
| 236 | qpc_args = dict(st=0, pm_state=3, pd=self.dev.pd, cqn_snd=self.cq_info['cqn'], cqn_rcv=self.cq_info['cqn'], **(qpc_args or {})) |
no test coverage detected