1 # SPDX-License-Identifier: Apache-2.0 2 # 3 # Copyright (C) 2017, ARM Limited and contributors. 4 # 5 # Licensed under the Apache License, Version 2.0 (the "License"); you may 6 # not use this file except in compliance with the License. 7 # You may obtain a copy of the License at 8 # 9 # http://www.apache.org/licenses/LICENSE-2.0 10 # 11 # Unless required by applicable law or agreed to in writing, software 12 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 13 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14 # See the License for the specific language governing permissions and 15 # limitations under the License. 16 # 17 18 from collections import OrderedDict, namedtuple 19 import json 20 import os 21 22 from perf_analysis import PerfAnalysis 23 from wlgen import RTA, Periodic, Ramp, Step 24 25 from test_wlgen import WlgenSelfBase 26 27 class RTABase(WlgenSelfBase): 28 """ 29 Common functionality for testing RTA 30 31 Doesn't have "Test" in the name so that nosetests doesn't try to run it 32 directly 33 """ 34 35 tools = ['rt-app'] 36 37 def get_expected_command(self, rta_wload): 38 """Return the rt-app command we should execute when `run` is called""" 39 rta_path = os.path.join(self.target.executables_directory, 'rt-app') 40 json_path = os.path.join(rta_wload.run_dir, rta_wload.json) 41 return '{} {} 2>&1'.format(rta_path, json_path) 42 43 def setUp(self): 44 super(RTABase, self).setUp() 45 46 # Can't calibrate rt-app because: 47 # - Need to set performance governor 48 # - Need to use SCHED_FIFO + high priority 49 # We probably don't have permissions so use a dummy calibration. 50 self.calibration = {c: 100 51 for c in range(len(self.target.cpuinfo.cpu_names))} 52 53 os.makedirs(self.host_out_dir) 54 55 def assert_output_file_exists(self, path): 56 """Assert that a file was created in host_out_dir""" 57 path = os.path.join(self.host_out_dir, path) 58 self.assertTrue(os.path.isfile(path), 59 'No output file {} from rt-app'.format(path)) 60 61 def assert_can_read_logfile(self, exp_tasks): 62 """Assert that the perf_analysis module understands the log output""" 63 pa = PerfAnalysis(self.host_out_dir) 64 self.assertSetEqual(set(exp_tasks), set(pa.tasks())) 65 66 class TestRTAProfile(RTABase): 67 def _do_test(self, task, exp_phases): 68 rtapp = RTA(self.target, name='test', calibration=self.calibration) 69 70 rtapp.conf( 71 kind = 'profile', 72 params = {'my_task': task.get()}, 73 run_dir=self.target_run_dir 74 ) 75 76 with open(rtapp.json) as f: 77 conf = json.load(f, object_pairs_hook=OrderedDict) 78 79 # Check that the configuration looks like we expect it to 80 phases = conf['tasks']['my_task']['phases'].values() 81 self.assertEqual(len(phases), len(exp_phases), 'Wrong number of phases') 82 for phase, exp_phase in zip(phases, exp_phases): 83 self.assertDictEqual(phase, exp_phase) 84 85 # Try running the workload and check that it produces the expected log 86 # files 87 rtapp.run(out_dir=self.host_out_dir) 88 89 rtapp_cmds = [c for c in self.target.executed_commands if 'rt-app' in c] 90 self.assertListEqual(rtapp_cmds, [self.get_expected_command(rtapp)]) 91 92 self.assert_output_file_exists('output.log') 93 self.assert_output_file_exists('test_00.json') 94 self.assert_output_file_exists('rt-app-my_task-0.log') 95 self.assert_can_read_logfile(exp_tasks=['my_task']) 96 97 def test_profile_periodic_smoke(self): 98 """ 99 Smoketest Periodic rt-app workload 100 101 Creates a workload using Periodic, tests that the JSON has the expected 102 content, then tests that it can be run. 103 """ 104 105 task = Periodic(period_ms=100, duty_cycle_pct=20, duration_s=1) 106 107 exp_phases = [ 108 { 109 'loop': 10, 110 'run': 20000, 111 'timer': { 112 'period': 100000, 113 'ref': 'my_task' 114 } 115 } 116 ] 117 118 self._do_test(task, exp_phases) 119 120 def test_profile_step_smoke(self): 121 """ 122 Smoketest Step rt-app workload 123 124 Creates a workload using Step, tests that the JSON has the expected 125 content, then tests that it can be run. 126 """ 127 128 task = Step(start_pct=100, end_pct=0, time_s=1) 129 130 exp_phases = [ 131 { 132 'run': 1000000, 133 'loop': 1 134 }, 135 { 136 'sleep': 1000000, 137 'loop': 1 138 }, 139 ] 140 141 self._do_test(task, exp_phases) 142 143 def test_composition(self): 144 """ 145 Test RTA task composition with __add__ 146 147 Creates a composed workload by +-ing RTATask objects, tests that the 148 JSON has the expected content, then tests running the workload 149 """ 150 light = Periodic(duty_cycle_pct=10, duration_s=1.0, period_ms=10) 151 152 start_pct = 10 153 end_pct = 90 154 delta_pct = 20 155 num_ramp_phases = ((end_pct - start_pct) / delta_pct) + 1 156 ramp = Ramp(start_pct=start_pct, end_pct=end_pct, delta_pct=delta_pct, 157 time_s=1, period_ms=50) 158 159 heavy = Periodic(duty_cycle_pct=90, duration_s=0.1, period_ms=100) 160 161 task = light + ramp + heavy 162 163 exp_phases = [ 164 # Light phase: 165 { 166 "loop": 100, 167 "run": 1000, 168 "timer": { 169 "period": 10000, 170 "ref": "my_task" 171 } 172 }, 173 # Ramp phases: 174 { 175 "loop": 20, 176 "run": 5000, 177 "timer": { 178 "period": 50000, 179 "ref": "my_task" 180 } 181 }, 182 { 183 "loop": 20, 184 "run": 15000, 185 "timer": { 186 "period": 50000, 187 "ref": "my_task" 188 } 189 }, 190 { 191 "loop": 20, 192 "run": 25000, 193 "timer": { 194 "period": 50000, 195 "ref": "my_task" 196 } 197 }, 198 { 199 "loop": 20, 200 "run": 35000, 201 "timer": { 202 "period": 50000, 203 "ref": "my_task" 204 } 205 }, 206 { 207 "loop": 20, 208 "run": 45000, 209 "timer": { 210 "period": 50000, 211 "ref": "my_task" 212 } 213 }, 214 # Heavy phase: 215 { 216 "loop": 1, 217 "run": 90000, 218 "timer": { 219 "period": 100000, 220 "ref": "my_task" 221 } 222 }] 223 224 225 self._do_test(task, exp_phases) 226 227 def test_invalid_composition(self): 228 """Test that you can't compose tasks with a delay in the second task""" 229 t1 = Periodic() 230 t2 = Periodic(delay_s=1) 231 232 # Should work fine if delayed task is the first one 233 try: 234 t3 = t2 + t1 235 except Exception as e: 236 raise AssertionError("Couldn't compose tasks: {}".format(e)) 237 238 # But not the other way around 239 with self.assertRaises(ValueError): 240 t3 = t1 + t2 241 242 243 class TestRTACustom(RTABase): 244 def _test_custom_smoke(self, calibration): 245 """ 246 Test RTA custom workload 247 248 Creates an rt-app workload using 'custom' and checks that the json 249 roughly matches the file we provided. If we have root, attempts to run 250 the workload. 251 """ 252 253 json_path = os.path.join(os.getenv('LISA_HOME'), 254 'assets', 'mp3-short.json') 255 rtapp = RTA(self.target, name='test', calibration=calibration) 256 257 # Configure this RTApp instance to: 258 rtapp.conf(kind='custom', params=json_path, duration=5, 259 run_dir=self.target_run_dir) 260 261 with open(rtapp.json) as f: 262 conf = json.load(f) 263 264 # Convert k to str because the json loader gives us unicode strings 265 tasks = set([str(k) for k in conf['tasks'].keys()]) 266 self.assertSetEqual( 267 tasks, 268 set(['AudioTick', 'AudioOut', 'AudioTrack', 269 'mp3.decoder', 'OMXCall'])) 270 271 # Would like to try running the workload but mp3-short.json has nonzero 272 # 'priority' fields, and we probably don't have permission for that 273 # unless we're root. 274 if self.target.is_rooted: 275 rtapp.run(out_dir=self.host_out_dir) 276 277 rtapp_cmds = [c for c in self.target.executed_commands 278 if 'rt-app' in c] 279 self.assertListEqual(rtapp_cmds, [self.get_expected_command(rtapp)]) 280 281 self.assert_output_file_exists('output.log') 282 self.assert_output_file_exists('test_00.json') 283 284 def test_custom_smoke_calib(self): 285 """Test RTA custom workload (providing calibration)""" 286 self._test_custom_smoke(self.calibration) 287 288 def test_custom_smoke_no_calib(self): 289 """Test RTA custom workload (providing no calibration)""" 290 self._test_custom_smoke(None) 291 292 293 DummyBlModule = namedtuple('bl', ['bigs']) 294 295 class TestRTACalibrationConf(RTABase): 296 """Test setting the "calibration" field of rt-app config""" 297 def _get_calib_conf(self, calibration): 298 rtapp = RTA(self.target, name='test', calibration=calibration) 299 300 rtapp.conf( 301 kind = 'profile', 302 params = {'t1': Periodic().get()}, 303 run_dir=self.target_run_dir 304 ) 305 306 with open(rtapp.json) as f: 307 return json.load(f)['global']['calibration'] 308 309 def test_calibration_conf_pload(self): 310 """Test that the smallest pload value is used, if provided""" 311 cpus = range(self.target.number_of_cpus) 312 conf = self._get_calib_conf(dict(zip(cpus, [c + 100 for c in cpus]))) 313 self.assertEqual(conf, 100, 314 'Calibration not set to minimum pload value') 315 316 def test_calibration_conf_bl(self): 317 """Test that a big CPU is used if big.LITTLE data is available""" 318 self.target.modules.append('bl') 319 self.target.bl = DummyBlModule([1, 2]) 320 conf = self._get_calib_conf(None) 321 self.assertIn(conf, ['CPU{}'.format(c) for c in self.target.bl.bigs], 322 'Calibration not set to use a big CPU') 323 324 def test_calibration_conf_nodata(self): 325 """Test that the last CPU is used if no data is available""" 326 conf = self._get_calib_conf(None) 327 cpu = self.target.number_of_cpus - 1 328 self.assertEqual(conf, 'CPU{}'.format(cpu), 329 'Calibration not set to highest numbered CPU') 330
