Skip to content

Add domain-aware PCIe handling and improve pcie_common robustness #655

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
t2sharma wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Add domain-aware PCIe handling and improve pcie_common robustness #655

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1ebc138
Update pcie_common.py
t2sharma Apr 15, 2026
d1ac271
Apply suggestion from @bgallagher-nexthop
t2sharma Apr 17, 2026
ba6575d
Fix unit test to support domain-aware PCIe parsing (lspci -D)
t2sharma Apr 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
117 changes: 73 additions & 44 deletions sonic_platform_base/sonic_pcie/pcie_common.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,49 +35,70 @@ def load_config_file(self):

# load current PCIe device
def get_pcie_device(self):
pciDict = {}
pciList = []
p1 = "^(\w+):(\w+)\.(\w)\s(.*)\s*\(*.*\)*"
p2 = "^.*:.*:.*:(\w+)\s*\(*.*\)*"
command1 = ["sudo", "lspci"]
command2 = ["sudo", "lspci", "-n"]
# run command 1
proc1 = subprocess.Popen(command1, universal_newlines=True, stdout=subprocess.PIPE)
output1 = proc1.stdout.readlines()
(out, err) = proc1.communicate()
# run command 2
proc2 = subprocess.Popen(command2, universal_newlines=True, stdout=subprocess.PIPE)
output2 = proc2.stdout.readlines()
(out, err) = proc2.communicate()

if proc1.returncode > 0:
for line1 in output1:
print(line1.strip())
return
elif proc2.returncode > 0:
for line2 in output2:
print(line2.strip())
return
else:
for (line1, line2) in zip(output1, output2):
pciDict.clear()
match1 = re.search(p1, line1.strip())
match2 = re.search(p2, line2.strip())
if match1 and match2:
Bus = match1.group(1)
Dev = match1.group(2)
Fn = match1.group(3)
Name = match1.group(4)
Id = match2.group(1)
pciDict["name"] = Name
pciDict["bus"] = Bus
pciDict["dev"] = Dev
pciDict["fn"] = Fn
pciDict["id"] = Id
pciList.append(pciDict)
pciDict = deepcopy(pciDict)
else:
print("CAN NOT MATCH PCIe DEVICE")
seen = set()

# Domain-aware output:
# 0002:01:00.0 Ethernet controller: ...
p1 = r"^([0-9a-fA-F]{4}):([0-9a-fA-F]{2}):([0-9a-fA-F]{2})\.([0-7])\s+(.*)$"
# Numeric output:
# 0002:01:00.0 0200: 177d:a065 ...
p2 = r"^[0-9a-fA-F]{4}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}\.[0-7]\s+[0-9a-fA-F]{4}:\s*([0-9a-fA-F]{4}):([0-9a-fA-F]{4}).*$"

command1 = ["sudo", "lspci", "-D"]
command2 = ["sudo", "lspci", "-D", "-n"]

proc1 = subprocess.Popen(command1, universal_newlines=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
output1, err1 = proc1.communicate()

proc2 = subprocess.Popen(command2, universal_newlines=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
output2, err2 = proc2.communicate()

if proc1.returncode != 0:
print(output1)
print(err1)
return []
if proc2.returncode != 0:
print(output2)
print(err2)
return []

lines1 = output1.splitlines()
lines2 = output2.splitlines()

for line1, line2 in zip(lines1, lines2):
match1 = re.search(p1, line1.strip())
match2 = re.search(p2, line2.strip())
if not (match1 and match2):
print("CAN NOT MATCH PCIe DEVICE")
print("lspci -D :", line1.strip())
print("lspci -D -n:", line2.strip())
continue

domain = match1.group(1).lower()
bus = match1.group(2).lower()
dev = match1.group(3).lower()
fn = match1.group(4).lower()
name = match1.group(5).strip()
vendor = match2.group(1).lower()
device = match2.group(2).lower()

key = (domain, bus, dev, fn)
if key in seen:
continue
seen.add(key)

pciDict = {
"name": name,
"domain": domain,
"bus": bus,
"dev": dev,
"fn": fn,
"id": device,
"vendor": vendor
}
pciList.append(deepcopy(pciDict))

return pciList

# check the sysfs tree for each PCIe device
Expand All @@ -91,10 +112,17 @@ def check_pcie_sysfs(self, domain=0, bus=0, device=0, func=0):
def get_pcie_check(self):
self.load_config_file()
for item_conf in self.confInfo:
domain_conf = item_conf.get("domain", "0000")
bus_conf = item_conf["bus"]
dev_conf = item_conf["dev"]
fn_conf = item_conf["fn"]
if self.check_pcie_sysfs(bus=int(bus_conf, base=16), device=int(dev_conf, base=16), func=int(fn_conf, base=16)):

if self.check_pcie_sysfs(
domain=int(domain_conf, base=16),
bus=int(bus_conf, base=16),
device=int(dev_conf, base=16),
func=int(fn_conf, base=16)
):
item_conf["result"] = "Passed"
else:
item_conf["result"] = "Failed"
Expand Down Expand Up @@ -142,5 +170,6 @@ def dump_conf_yaml(self):
conf_rev = "_{}".format(self._conf_rev) if self._conf_rev else ""
config_file = "{}/pcie{}.yaml".format(self.config_path, conf_rev)
with open(config_file, "w") as conf_file:
yaml.dump(curInfo, conf_file, default_flow_style=False)
yaml.dump(curInfo, conf_file, default_flow_style=False, sort_keys=False)
return

47 changes: 24 additions & 23 deletions tests/pcie_common_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,17 +14,16 @@
pcie_config_file = os.path.join(tests_dir, 'pcie.yaml')

lspci_output = '''\
00:01.0 PCI A
00:02.0 PCI B
00:02.1 PCI C
01:00.0 PCI D
0000:00:01.0 PCI A
0000:00:02.0 PCI B
0000:00:02.1 PCI C0000:01:00.0 PCI D
'''

lspci_ID_output = '''\
00:01.0 0001: 0000:000a
00:02.0 0002: 0000:000b
00:02.1 0003: 0000:000c
01:00.0 0004: 0000:000d
0000:00:01.0 0001: 0000:000a
0000:00:02.0 0002: 0000:000b
0000:00:02.1 0003: 0000:000c
0000:01:00.0 0004: 0000:000d
'''

pci_sysfs_paths = [
Expand All @@ -35,17 +34,17 @@
]

pcie_device_list = [
{'bus': '00', 'dev': '01', 'fn': '0', 'id': '000a', 'name': 'PCI A'},
{'bus': '00', 'dev': '02', 'fn': '0', 'id': '000b', 'name': 'PCI B'},
{'bus': '00', 'dev': '02', 'fn': '1', 'id': '000c', 'name': 'PCI C'},
{'bus': '01', 'dev': '00', 'fn': '0', 'id': '000d', 'name': 'PCI D'},
{'domain': '0000', 'bus': '00', 'dev': '01', 'fn': '0', 'id': '000a', 'vendor': '0000', 'name': 'PCI A'},
{'domain': '0000', 'bus': '00', 'dev': '02', 'fn': '0', 'id': '000b', 'vendor': '0000', 'name': 'PCI B'},
{'domain': '0000', 'bus': '00', 'dev': '02', 'fn': '1', 'id': '000c', 'vendor': '0000', 'name': 'PCI C'},
{'domain': '0000', 'bus': '01', 'dev': '00', 'fn': '0', 'id': '000d', 'vendor': '0000', 'name': 'PCI D'},
]

pcie_check_output = [
{'bus': '00', 'dev': '01', 'fn': '0', 'id': '000a', 'name': 'PCI A', 'result': 'Passed'},
{'bus': '00', 'dev': '02', 'fn': '0', 'id': '000b', 'name': 'PCI B', 'result': 'Passed'},
{'bus': '00', 'dev': '02', 'fn': '1', 'id': '000c', 'name': 'PCI C', 'result': 'Passed'},
{'bus': '01', 'dev': '00', 'fn': '0', 'id': '000d', 'name': 'PCI D', 'result': 'Passed'}
{'domain': '0000', 'bus': '00', 'dev': '01', 'fn': '0', 'id': '000a', 'vendor': '0000', 'name': 'PCI A', 'result': 'Passed'},
{'domain': '0000', 'bus': '00', 'dev': '02', 'fn': '0', 'id': '000b', 'vendor': '0000', 'name': 'PCI B', 'result': 'Passed'},
{'domain': '0000', 'bus': '00', 'dev': '02', 'fn': '1', 'id': '000c', 'vendor': '0000', 'name': 'PCI C', 'result': 'Passed'},
{'domain': '0000', 'bus': '01', 'dev': '00', 'fn': '0', 'id': '000d', 'vendor': '0000', 'name': 'PCI D', 'result': 'Passed'}
]

pcie_aer_correctable_content = '''\
Expand Down Expand Up @@ -131,16 +130,18 @@ class TestPcieCommon:
def test_get_pcie_devices(self, subprocess_popen_mock):

def subprocess_popen_side_effect(*args, **kwargs):
if args[0] == ['sudo', 'lspci']:
output = lspci_output.splitlines()
elif args[0] == ['sudo', 'lspci', '-n']:
output = lspci_ID_output.splitlines()
if args[0] == ['sudo', 'lspci', '-D']:
communicate_output = (lspci_output, '')
elif args[0] == ['sudo', 'lspci', '-D', '-n']:
communicate_output = (lspci_ID_output, '')
else:
raise AssertionError("Unexpected command: {}".format(args[0]))

popen_mock = mock.Mock()
popen_attributes = {
'returncode': 0,
'communicate.return_value': ('', ''),
'stdout.readlines.return_value': output
'communicate.return_value': communicate_output,
'stdout.readlines.return_value': communicate_output[0].splitlines()
}
popen_mock.configure_mock(**popen_attributes)
return popen_mock
Expand Down Expand Up @@ -211,4 +212,4 @@ def test_dump_conf_yaml(self):
def teardown_class(cls):
# Cleanup generated config
if os.path.isfile(pcie_config_file):
os.remove(pcie_config_file)
os.remove(pcie_config_file)
Loading