Lerking/OSACA
1
0
Fork 0
mirror of https://github.com/RRZE-HPC/OSACA.git synced 2026-01-05 10:40:06 +01:00
Code Issues Packages Projects Releases Wiki Activity

added tests for Kerncraft API

Browse Source
This commit is contained in:
JanLJL
2019-08-30 16:19:18 +02:00
parent 39004fce1a
commit 533f78f9cf
4 changed files with 149 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
osaca/api/kerncraft_interface.py
View File

@@ -49,4 +49,9 @@ class KerncraftAPI(object):
def get_latency(self, kernel):
kernel_graph = KernelDG(kernel, self.parser, self.machine_model)
return sum([x if x['latency'] is not None else 0 for x in kernel_graph])
return sum(
[
x['latency'] if x['latency'] is not None else 0
for x in kernel_graph.get_critical_path()
]
)

6
osaca/data/vulcan.yml
View File

@@ -62,6 +62,12 @@ instruction_forms:
throughput: 0.33333333
latency: 1.0 # 0 0DV 1 1DV 2 3 4 5
port_pressure: [0.33333333, 0.0, 0.33333333, 0.0, 0.33333333, 0.0, 0.0, 0.0]
- name: "b.ne"
operands:
- class: 'identifier'
throughput: 0.0
latency: 0.0 # 0 0DV 1 1DV 2 3 4 5
port_pressure: [0, 0, 0, 0, 0, 0, 0, 0]
- name: "bne"
operands:
- class: 'identifier'

82
osaca/semantics/semanticsAppender.py
View File

@@ -13,8 +13,8 @@ class INSTR_FLAGS:
Flags used for unknown or special instructions
"""
TP_UNKWN = 'tp_unkown'
LT_UNKWN = 'lt_unkown'
TP_UNKWN = 'tp_unknown'
LT_UNKWN = 'lt_unknown'
NOT_BOUND = 'not_bound'
HIDDEN_LD = 'hidden_load'
HAS_LD = 'performs_load'
@@ -76,7 +76,9 @@ class SemanticsAppender(object):
if INSTR_FLAGS.HIDDEN_LD not in load_instr['flags']
]
)
load = [instr for instr in kernel if instr['line_number'] == min_distance_load[1]][0]
load = [instr for instr in kernel if instr['line_number'] == min_distance_load[1]][
0
]
# Hide load
load['flags'] += [INSTR_FLAGS.HIDDEN_LD]
load['port_pressure'] = self._nullify_data_ports(load['port_pressure'])
@@ -86,43 +88,49 @@ class SemanticsAppender(object):
def assign_tp_lt(self, instruction_form):
flags = []
port_number = len(self._machine_model['ports'])
instruction_data = self._machine_model.get_instruction(
instruction_form['instruction'], instruction_form['operands']
)
if instruction_data:
# instruction form in DB
throughput = instruction_data['throughput']
port_pressure = instruction_data['port_pressure']
try:
assert isinstance(port_pressure, list)
assert len(port_pressure) == port_number
instruction_form['port_pressure'] = port_pressure
if sum(port_pressure) == 0 and throughput is not None:
# port pressure on all ports 0 --> not bound to a port
flags.append(INSTR_FLAGS.NOT_BOUND)
except AssertionError:
warnings.warn(
'Port pressure could not be imported correctly from database. '
+ 'Please check entry for:\n {}'.format(instruction_form)
)
instruction_form['port_pressure'] = [0.0 for i in range(port_number)]
flags.append(INSTR_FLAGS.TP_UNKWN)
if throughput is None:
# assume 0 cy and mark as unknown
throughput = 0.0
flags.append(INSTR_FLAGS.TP_UNKWN)
latency = instruction_data['latency']
if latency is None:
# assume 0 cy and mark as unknown
latency = 0.0
flags.append(INSTR_FLAGS.LT_UNKWN)
else:
# instruction could not be found in DB
# --> mark as unknown and assume 0 cy for latency/throughput
if instruction_form['instruction'] is None:
# No instruction (label, comment, ...) --> ignore
throughput = 0.0
latency = 0.0
instruction_form['port_pressure'] = [0.0 for i in range(port_number)]
flags += [INSTR_FLAGS.TP_UNKWN, INSTR_FLAGS.LT_UNKWN]
else:
instruction_data = self._machine_model.get_instruction(
instruction_form['instruction'], instruction_form['operands']
)
if instruction_data:
# instruction form in DB
throughput = instruction_data['throughput']
port_pressure = instruction_data['port_pressure']
try:
assert isinstance(port_pressure, list)
assert len(port_pressure) == port_number
instruction_form['port_pressure'] = port_pressure
if sum(port_pressure) == 0 and throughput is not None:
# port pressure on all ports 0 --> not bound to a port
flags.append(INSTR_FLAGS.NOT_BOUND)
except AssertionError:
warnings.warn(
'Port pressure could not be imported correctly from database. '
+ 'Please check entry for:\n {}'.format(instruction_form)
)
instruction_form['port_pressure'] = [0.0 for i in range(port_number)]
flags.append(INSTR_FLAGS.TP_UNKWN)
if throughput is None:
# assume 0 cy and mark as unknown
throughput = 0.0
flags.append(INSTR_FLAGS.TP_UNKWN)
latency = instruction_data['latency']
if latency is None:
# assume 0 cy and mark as unknown
latency = 0.0
flags.append(INSTR_FLAGS.LT_UNKWN)
else:
# instruction could not be found in DB
# --> mark as unknown and assume 0 cy for latency/throughput
throughput = 0.0
latency = 0.0
instruction_form['port_pressure'] = [0.0 for i in range(port_number)]
flags += [INSTR_FLAGS.TP_UNKWN, INSTR_FLAGS.LT_UNKWN]
# flatten flag list
flags = list(set(flags))
if 'flags' not in instruction_form:

92
tests/test_kerncraftAPI.py Executable file
View File

@@ -0,0 +1,92 @@
#!/usr/bin/env python3
"""
Unit tests for OSACA Kerncraft API
"""
import os
import unittest
from collections import OrderedDict
from osaca.api import KerncraftAPI
from osaca.parser import ParserAArch64v81, ParserX86ATT
class TestKerncraftAPI(unittest.TestCase):
@classmethod
def setUpClass(self):
# set up parser and kernels
self.parser_x86 = ParserX86ATT()
self.parser_AArch64 = ParserAArch64v81()
with open(self._find_file('triad-x86-iaca.s')) as f:
self.code_x86 = f.read()
with open(self._find_file('triad-arm-iaca.s')) as f:
self.code_AArch64 = f.read()
###########
# Tests
###########
def test_kerncraft_API_x86(self):
kapi = KerncraftAPI('csx')
kernel = kapi.analyze_code(self.code_x86)
kapi.create_output(kernel)
self.assertEqual(kapi.get_unmatched_instruction_ratio(kernel), 0.0)
port_occupation = OrderedDict(
[
('0', 1.25),
('0DV', 0.0),
('1', 1.25),
('2', 2.0),
('2D', 1.5),
('3', 2.0),
('3D', 1.5),
('4', 1.0),
('5', 0.75),
('6', 0.75),
('7', 0.0),
]
)
self.assertEqual(kapi.get_port_occupation_cycles(kernel), port_occupation)
self.assertEqual(kapi.get_total_throughput(kernel), 2.0)
self.assertEqual(kapi.get_latency(kernel), 10.0)
def test_kerncraft_API_AArch64(self):
kapi = KerncraftAPI('vulcan')
kernel = kapi.analyze_code(self.code_AArch64)
kapi.create_output(kernel)
self.assertEqual(kapi.get_unmatched_instruction_ratio(kernel), 0.0)
port_occupation = OrderedDict(
[
('0', 34.0),
('0DV', 0.0),
('1', 34.0),
('1DV', 0.0),
('2', 2.0),
('3', 64.0),
('4', 64.0),
('5', 32.0),
]
)
self.assertEqual(kapi.get_port_occupation_cycles(kernel), port_occupation)
self.assertEqual(kapi.get_total_throughput(kernel), 64.0)
# TODO add missing latency values
# self.assertEqual(kapi.get_latency(kernel), 20.0)
##################
# Helper functions
##################
@staticmethod
def _find_file(name):
testdir = os.path.dirname(__file__)
name = os.path.join(testdir, 'test_files', name)
assert os.path.exists(name)
return name
if __name__ == '__main__':
suite = unittest.TestLoader().loadTestsFromTestCase(TestKerncraftAPI)
unittest.TextTestRunner(verbosity=2).run(suite)