| #!/usr/bin/python |
| # $Copyright: |
| # ---------------------------------------------------------------- |
| # This confidential and proprietary software may be used only as |
| # authorised by a licensing agreement from ARM Limited |
| # (C) COPYRIGHT 2015 ARM Limited |
| # ALL RIGHTS RESERVED |
| # The entire notice above must be reproduced on all authorised |
| # copies and copies may only be made to the extent permitted |
| # by a licensing agreement from ARM Limited. |
| # ---------------------------------------------------------------- |
| # File: test_cpu_power.py |
| # ---------------------------------------------------------------- |
| # $ |
| # |
| |
| import matplotlib |
| import pandas as pd |
| |
| from test_thermal import BaseTestThermal |
| import cr2 |
| import cpu_power |
| |
| class TestCpuPower(BaseTestThermal): |
| def __init__(self, *args, **kwargs): |
| super(TestCpuPower, self).__init__(*args, **kwargs) |
| self.map_label = {"00000000,00000039": "A53", "00000000,00000006": "A57"} |
| |
| def test_pivot_with_labels(self): |
| """Test pivot_with_labels()""" |
| map_label = {"000000f0": "A15", "0000000f": "A7"} |
| dfr_in = pd.DataFrame({'cpus': ["000000f0", "0000000f", "000000f0", "0000000f"], |
| 'freq': [1, 3, 2, 6]}) |
| |
| dfr_out = cpu_power.pivot_with_labels(dfr_in, "freq", "cpus", map_label) |
| |
| self.assertEquals(dfr_out["A15"].iloc[0], 1) |
| self.assertEquals(dfr_out["A15"].iloc[1], 1) |
| self.assertEquals(dfr_out["A15"].iloc[2], 2) |
| self.assertEquals(dfr_out["A7"].iloc[1], 3) |
| |
| def test_num_cpus_in_mask(self): |
| """num_cpus_in_mask() works with the masks we usually use""" |
| mask = "000000f0" |
| self.assertEquals(cpu_power.num_cpus_in_mask(mask), 4) |
| |
| mask = sorted(self.map_label)[0] |
| self.assertEquals(cpu_power.num_cpus_in_mask(mask), 2) |
| |
| mask = sorted(self.map_label)[1] |
| self.assertEquals(cpu_power.num_cpus_in_mask(mask), 4) |
| |
| def test_cpuoutpower_dataframe(self): |
| """Test that CpuOutPower() creates a proper data_frame""" |
| outp = cr2.Run().cpu_out_power |
| |
| self.assertEquals(outp.data_frame["power"].iloc[0], 1344) |
| self.assertTrue("cdev_state" in outp.data_frame.columns) |
| |
| def test_cpuoutpower_get_all_freqs(self): |
| """Test CpuOutPower.get_all_freqs()""" |
| dfr = cr2.Run().cpu_out_power.get_all_freqs(self.map_label) |
| |
| self.assertEquals(dfr["A57"].iloc[0], 1100) |
| self.assertEquals(dfr["A53"].iloc[1], 850) |
| |
| def test_cpuinpower_get_dataframe(self): |
| """Test that CpuInPower() creates a proper data_frame()""" |
| inp = cr2.Run().cpu_in_power |
| |
| self.assertTrue("load0" in inp.data_frame.columns) |
| self.assertEquals(inp.data_frame["load0"].iloc[0], 24) |
| |
| def test_cpuinpower_big_cpumask(self): |
| """CpuInPower()'s data_frame is not confused by 64-bit cpumasks""" |
| in_data = """ kworker/2:2-679 [002] 676.256284: thermal_power_cpu_get: cpus=00000000,0000000f freq=261888 cdev_state=5 power=12 |
| kworker/2:2-679 [002] 676.276200: thermal_power_cpu_get: cpus=00000000,00000030 freq=261888 cdev_state=5 power=0 |
| kworker/2:2-679 [002] 676.416202: thermal_power_cpu_get: cpus=00000000,0000000f freq=261888 cdev_state=5 power=0 |
| """ |
| with open("trace.txt", "w") as fout: |
| fout.write(in_data) |
| |
| inp = cr2.Run(normalize_time=False).cpu_in_power |
| self.assertEquals(round(inp.data_frame.index[0], 6), 676.256284) |
| self.assertEquals(inp.data_frame["cpus"].iloc[1], "00000000,00000030") |
| |
| def test_cpuinpower_data_frame_asymmetric_clusters(self): |
| """Test that CpuInPower()'s data_frame can handle asymmetric clusters |
| |
| That is 2 cpus in one cluster and 4 in another, like Juno |
| """ |
| in_data = """ |
| kworker/2:2-679 [002] 676.256261: thermal_power_cpu_get: cpus=00000000,00000030 freq=1900000 raw_cpu_power=1259 load={74 49} power=451 |
| kworker/2:2-679 [002] 676.256271: thermal_power_cpu_get: cpus=00000000,0000000f freq=450000 raw_cpu_power=36 load={1 2 1 3} power=9 |
| """ |
| |
| with open("trace.txt", "w") as fout: |
| fout.write(in_data) |
| |
| inp = cr2.Run(normalize_time=False).cpu_in_power |
| |
| self.assertEquals(inp.data_frame["load0"].iloc[0], 74) |
| self.assertEquals(inp.data_frame["load1"].iloc[0], 49) |
| self.assertEquals(inp.data_frame["load2"].iloc[0], 0) |
| self.assertEquals(inp.data_frame["load3"].iloc[0], 0) |
| self.assertEquals(inp.data_frame["load0"].iloc[1], 1) |
| self.assertEquals(inp.data_frame["load1"].iloc[1], 2) |
| self.assertEquals(inp.data_frame["load2"].iloc[1], 1) |
| self.assertEquals(inp.data_frame["load3"].iloc[1], 3) |
| |
| def test_cpuinpower_get_all_freqs(self): |
| """Test CpuInPower.get_all_freqs()""" |
| dfr = cr2.Run().cpu_in_power.get_all_freqs(self.map_label) |
| |
| self.assertEquals(dfr["A57"].iloc[0], 1100) |
| self.assertEquals(dfr["A53"].iloc[1], 850) |
| self.assertEquals(dfr["A57"].iloc[5], 1100) |
| |
| def test_cpuinpower_get_load_data(self): |
| """Test CpuInPower.get_load_data()""" |
| run = cr2.Run() |
| first_load = run.cpu_in_power.data_frame["load0"].iloc[0] |
| load_data = run.cpu_in_power.get_load_data(self.map_label) |
| |
| self.assertEquals(load_data["A57"].iloc[0], 24 + 19) |
| self.assertEquals(load_data["A53"].iloc[3], 32 + 28 + 46 + 44) |
| self.assertEquals(load_data["A57"].iloc[0], load_data["A57"].iloc[1]) |
| |
| self.assertEquals(run.cpu_in_power.data_frame["load0"].iloc[0], |
| first_load) |
| |
| def test_cpuinpower_get_normalized_load_data(self): |
| """Test CpuInPower.get_normalized_load_data()""" |
| run = cr2.Run() |
| first_load = run.cpu_in_power.data_frame["load0"].iloc[0] |
| load_data = run.cpu_in_power.get_normalized_load_data(self.map_label) |
| |
| # Ideally the trace should have an event in which the cpus are |
| # not running at maximum frequency |
| self.assertEquals(load_data["A57"].iloc[0], |
| (24. + 19) * 1100000 / (1100000 * 2)) |
| self.assertEquals(load_data["A53"].iloc[1], |
| (36. + 49 + 48 + 7) * 850000 / (850000 * 4)) |
| self.assertEquals(run.cpu_in_power.data_frame["load0"].iloc[0], |
| first_load) |