1#!/usr/bin/env python3

2#

3# Uses the cyclictest binary to run cyclictest and measure process latency.

4# This test needs to run for at least 24 hours in order to provide useful data.

5#

6# The output from the test, as well as any errors, are then dumped into a file

7# so the second part can check the results. A passing test will depend on the

8# maximum latency observed and whether that is an acceptable number. This

9# value depends on the DUT's platform and CPU. These values are provided by

10# Intel as part of the Real Time Key Performance Indicator (RT KPI) document.

11#

12# Pass conditions:

13# * Max latency is equal to or lower than the expected value for the DUT's HW

14# * 0 overflows (caused by a process waiting too long. also called a "miss")

15#

16# Fail conditions:

17# * Any errors from the test.

18# * Unacceptable values for Max Latency or Overflows.

19

20 import argparse

21 import subprocess

22

23

24 def run_it(duration):

25 """

26 This function runs the cyclictest command, and captures output

27

28 Inputs:

29 duration - the desired duration of the test in seconds.

30 """

31

32 # timeout must be larger than the test duration, otherwise the test will be

33 # interrupted

34 test_timeout = duration + 1

35

36 cmd = [

37 "chrt",

38 "-f",

39 "99",

40 "cyclictest",

41 "-a2-3",

42 "-t2",

43 "-m",

44 "-p99",

45 "-i250",

46 "-h700",

47 "-q",

48 "-D",

49 str(duration),

50 ]

51

52 try:

53 result = subprocess.run(

54 cmd,

55 capture_output=True,

56 text=True,

57 timeout=test_timeout,

58 check=True,

59 )

60 except Exception as err:

61 raise SystemExit(err)

62

63 # Note: A returncode of 0 simply means the command executed successfully.

64 # It does NOT mean the test passed (or failed.)

65 # That will be determined by a separate function.

66 return result

67

68

69 def lookup_max_latency():

70 """

71 The expected maximum latency is based on the HW platform and CPU.

72 For instance, ADL-S with an i7-12700 has an expected maximum latency of

73 11ms

74

75 However, some hardware may not proper identification yet.

76 In this case just return the largest latency listed in the document.

77 """

78

79 # In case we cannot determine the HW or the CPU of the DUT, use the largest

80 # value from the table of expected maximum latency.

81 MAX_ALLOWED_LATENCY = 100

82

83 # TODO: Implement the methods to determine the platform and CPU, and use

84 # that to lookup the expected maximum latency.

85 #

86 # For now, just return MAX_ALLOWED_LATENCY

87

88 return MAX_ALLOWED_LATENCY

89

90

91 def verify_cyclictest_results(result):

92 """

93 This function takes output from the call to subprocess containing the

94 output from a run of cyclictest and looks at std out to verify two things:

95 1. The maximum latency from each thread did not exceed the maximum latency

96 specified by Intel for the DUT's HW Platform and CPU.

97

98 2. The number of overflows (scheduling misses) was 0 for each thread.

99

100 Both conditions must be true for this function to return success (0)

101

102 Sample output for these lines looks like:

103 # Max Latencies: 00041 00046

104 # Histogram Overflows: 00000 00000

105 The values after these strings are the results from each thread.

106

107 All other lines are ignored.

108 """

109 max_latency = lookup_max_latency()

110 print("Maximum latency allowed: " + str(max_latency))

111 # list of latencies, from each thread

112 latencies = []

113

114 # list of overflows (scheduling misses) from each thread

115 overflows = []

116

117 return_code = 0

118

119 std_output = result.stdout.split("\n")

120 for line in std_output:

121

122 if "# Max Latencies" in line:

123 latencies = line.split(":")[1].split()

124 for latency in latencies:

125 if int(latency) > max_latency:

126 print(

127 "Test fails.\tLatency : "

128 + latency

129 + "\t-- Maximum Latency value of: "

130 + str(max_latency)

131 + " exceeded."

132 )

133 return_code = 1

134 else:

135 print("Test passes.\tLatency : " + latency + " -- OK.")

136

137 elif "# Histogram Overflows" in line:

138 overflows = line.split(":")[1].split()

139 for overflow in overflows:

140 if int(overflow) > 0:

141 print(

142 "Test fails.\tOverflow: "

143 + overflow

144 + " -- 0 Overflows expected."

145 )

146 return_code = 1

147 else:

148 print("Test passes.\tOverflow: " + overflow + " -- OK.")

149 continue

150

151 return return_code

152

153

154 def main():

155 return_code = 0

156

157 parser = argparse.ArgumentParser()

158 # default duration is 86400 seconds = 24hr

159 parser.add_argument("--duration", type=int, default=86400)

160 args = parser.parse_args()

161

162 result = run_it(args.duration)

163 print("stdout: {}".format(result.stdout))

164 print("stderr: {}\n".format(result.stderr))

165

166 return_code = verify_cyclictest_results(result)

167 return return_code

168

169

170 if __name__ == "__main__":

171 raise SystemExit(main())