Skip to content

Commit

Permalink
Report app start column (#37)
Browse files Browse the repository at this point in the history
* Allow multiple instances of the same app to be run

* Try to fix test

* Add more tests

* Flake8

* More appropiate test name

* Make sure AppRunner has device, device is not found is previous call had an space issue

* Made code more pythonic

* Handle AppRunner object even if it fails

* Flake8

* Try to add start column

* Add starting column

* Flake8

* Update check

* Fix where assert is done

* Fix issue with counting

* Check that xbutil count is working
mariodruiz authored Jun 26, 2024

Verified

This commit was created on GitHub.com and signed with GitHub’s verified signature.
1 parent 1b24275 commit a2978ea
Showing 3 changed files with 81 additions and 58 deletions.
111 changes: 62 additions & 49 deletions npu/utils/xbutil.py
Original file line number Diff line number Diff line change
@@ -1,25 +1,30 @@
# Copyright (C) 2023 Advanced Micro Devices, Inc. All rights reserved.
# SPDX-License-Identifier: MIT

import os
from io import FileIO
from pathlib import Path
from typing import List, Dict, Set
import subprocess
import tempfile
import json
import shutil
import sys
import time

XBUTIL_DIR = Path("C:\Windows\System32\AMD")
XBUTIL_DIR = Path("C:\\Windows\\System32\\AMD")


def _map_list_to_list(appsmap: list) -> List[str]:
applist = []
for app in appsmap:
applist.append(list(app.keys())[0])
return applist


class XBUtil:

def __init__(self):
""" A class that wraps xbutil so that information can be
""" A class that wraps xbutil so that information can be
parsed about currently running applications on the NPU
device.
device.
Attributes
----------
@@ -36,120 +41,128 @@ def __init__(self):
self._check_devices()
self.devid = next(iter(self._devices))

def app_exists(self, name:str)->bool:
""" Returns true if an app with the given name is
def app_exists(self, name: str) -> bool:
""" Returns true if an app with the given name is
present on the NPU device """
riallto_pattern = "Riallto"
for f in self._get_loaded_functions():
for f in _map_list_to_list(self._get_loaded_functions()):
if f.endswith(riallto_pattern):
if f.startswith(name):
return True
return False

@property
def app_table(self)->str:
""" Returns a table of all the apps currently loaded onto the NPU device. """
def app_table(self) -> str:
"""Returns a table of the apps currently loaded onto the NPU device"""
s = "Currently loaded apps:\n"
for a in self.list_apps():
s += f"\t{a}\n"
for app in self.list_apps():
s += f"\t{list(app.keys())[0]}\n"
return s

def list_apps(self)->List[str]:
def list_apps(self) -> List[str]:
""" Lists all the apps running on the
NPU device """
l = []
wse_pattern = "IPUV1CNN"
applist = []
wse_pattern = "IPUV1CNN"
riallto_pattern = "Riallto"
for f in self._get_loaded_functions():
if f.endswith(riallto_pattern):
l.append(f)
elif f.endswith(wse_pattern):
l.append(f)
return l
name = list(f.keys())[0]
if name.endswith(riallto_pattern) or name.endswith(wse_pattern):
applist.append(f)
return applist

def _apps(self)->None:
def _apps(self) -> None:
""" displays all apps that are running on the NPU
device using an IPython widget """
import ipywidgets as widgets
from IPython.display import display
output1 = widgets.Output()
output2 = widgets.Output() # Multiple widgets to avoid flickering during update
output2 = widgets.Output() # Multiple widgets to avoid flickering during update
display(output1)
while True:
apps = self.list_apps()
if len(apps) > 0:
max_app_name = max(len(max(apps, key=len)), 10)
max_app_name = max(len(max([list(a.keys())[0] for a in apps],
key=len)), 10)
else:
max_app_name = 10
max_app_name = 10

s = f"Currently Running NPU apps: (update rate 1s)\n"
s = f"| {'app name': <{max_app_name}} | {'Num columns': <14} | {'start column': <14} |\n"
s = "Currently Running NPU apps: (update rate 1s)\n"
s = f"| {'app name': <{max_app_name}} | {'Num columns': <14} | "
s += f"{'start column': <14} |\n"
s += '-'*max_app_name + '-'*38 + '\n'

if len(apps) == 0:
s += ' '*( (int)((max_app_name+38)/2) - 12) + "No apps currently loaded\n"
s += ' '*((int)((max_app_name+38)/2) - 12)
s += "No apps currently loaded\n"
else:

for a in apps:
s += f" {a: <{max_app_name}} {1: <14} {'?': <14} \n"
appname = list(a.keys())[0]
s += f"| {appname: <{max_app_name}} | "
s += f"{ a[appname]['num_cols']: <14} | "
s += f"{ a[appname]['start_col']: <14} |\n"
output2.append_stdout(s)
output1.outputs = output2.outputs
output2.outputs = ()
time.sleep(1)

def apps(self)->None:
def apps(self) -> None:
""" Starts a thread displaying all the apps in an ipython widget. """
import threading
thread = threading.Thread(target=self._apps)
thread.start()


@property
def app_count(self)->int:
def app_count(self) -> int:
""" The total number of running apps """
return self.num_wse_streams + self.num_riallto_streams

@property
def num_wse_streams(self)->int:
def num_wse_streams(self) -> int:
""" Returns the current number of active WSE streams """
return self.loaded_functions.count('DPU_1x4:IPUV1CNN')

@property
def num_riallto_streams(self)->int:
def num_riallto_streams(self) -> int:
return sum(1 for app in self.loaded_functions if 'Riallto' in app)

def _get_loaded_functions(self)->List[str]:
def _get_loaded_functions(self) -> List[str]:
""" Returns a list of the loaded functions on the NPU from xbutil. """
l = []
d = self._cmd(['examine', '-d', self.devid, '-r', 'dynamic-regions'])
for f in d['devices'][0]['dynamic_regions'][0]['compute_units']:
l.append(f['name'])
return l
applist = []
d = self._cmd(['examine', '-d', self.devid, '-r', 'dynamic-regions',
'-r', 'aie-partitions'])

for f, col in zip(d['devices'][0]['dynamic_regions'][0]['compute_units'],
d['devices'][0]['aie_partitions']['partitions']):
applist.append({f['name']: {'start_col': str(col['start_col']),
'num_cols': str(col['num_cols'])}})
return applist

@property
def loaded_functions(self)->List[str]:
def loaded_functions(self) -> List[str]:
""" Returns a list of loaded functions on the NPU. Applications can have multiple functions."""
return self._get_loaded_functions()
return _map_list_to_list(self._get_loaded_functions())

def _check_xbutil_install(self):
""" Returns true if xbutil.exe is available. """
_ = self._cmd(['examine', '-d'])

def _cmd(self, cmdlist:List[str])->Dict:
def _cmd(self, cmdlist: List[str]) -> Dict:
""" Runs an xbutil.exe command to produce a json report that is parsed into a dict and returned. """
try:
tmp_dir = tempfile.mkdtemp()
_t = Path(tmp_dir) / "temp.json"
output = subprocess.check_output([self._xbutil]
output = subprocess.check_output([self._xbutil]
+ cmdlist
+ ['-f', 'json', '-o', _t.absolute(), '--force']
, stderr=subprocess.STDOUT)
+ ['-f', 'json', '-o',
_t.absolute(), '--force'],
stderr=subprocess.STDOUT)
return json.load(FileIO(_t.absolute()))
except subprocess.CalledProcessError as e:
print(f"xbutil command failed.\n\n {e.output.decode()}")
raise e

def _get_devices(self)->Set[str]:
def _get_devices(self) -> Set[str]:
""" Get the set of devices on this machine"""
t = []
examine_out = self._cmd(['examine', '-d'])
@@ -158,7 +171,7 @@ def _get_devices(self)->Set[str]:
t.append(d['bdf'])
return set(t)

def _check_devices(self)->None:
def _check_devices(self) -> None:
"""Raises an error if a problem is detected with the device. """
if len(self._devices) > 1:
raise RuntimeError("Unable to determine which device is the NPU, "
3 changes: 2 additions & 1 deletion tests/test_applications.py
Original file line number Diff line number Diff line change
@@ -300,7 +300,8 @@ def check_npu():
pytest.skip('Skipping test because the IPU device is not enabled on this device.')
xbu = XBUtil()
for app in xbu.list_apps():
if app.endswith("IPURiallto"):
appname = list(app.keys())[0]
if appname.endswith("IPURiallto"):
pytest.skip('Skipping test because the IPU is in an unstable state.')


25 changes: 17 additions & 8 deletions tests/test_multi_load_app.py
Original file line number Diff line number Diff line change
@@ -6,6 +6,7 @@
import os
from .test_applications import check_npu
from npu.runtime import AppRunner
from npu.utils.xbutil import XBUtil
from .test_applications import SimplePlusN


@@ -28,21 +29,26 @@ def test_double_load_custom_app():
assert app
app1 = AppRunner("SimplePlusN.xclbin")
assert app1
del app, app1
appsreport = XBUtil()
assert appsreport.app_count == 2
del app, app1, appsreport


@pytest.mark.parametrize('numapps', [2, 3, 4])
def test_videoapp_n_loads(numapps):
@pytest.mark.parametrize('numappsreport', [2, 3, 4])
def test_videoapp_n_loads(numappsreport):
"""Load N instances of the same app. Test should pass"""
check_npu()
appbin = _get_full_path("color_threshold_v2_720p.xclbin")
app = []
for _ in range(numapps):
for _ in range(numappsreport):
app.append(AppRunner(appbin))

for i in range(numapps):
appsreport = XBUtil()
assert appsreport.app_count == numappsreport

for i in range(numappsreport):
assert app[i]
del app
del app, appsreport


def test_videoapp_five_loads():
@@ -58,9 +64,12 @@ def test_videoapp_five_loads():
for i in range(4):
assert app[i]

appsreport = XBUtil()
assert appsreport.app_count == 4

with pytest.raises(RuntimeError) as verr:
app1 = AppRunner(appbin)
assert 'There is currently no free space on the NPU' in str(verr.value)
del app1
assert 'There is currently no free space on the NPU' in str(verr.value)

del app
del app, appsreport

0 comments on commit a2978ea

Please sign in to comment.