-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathrun.py
150 lines (142 loc) · 5.42 KB
/
run.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
from progress.bar import Bar
from taskset import generate_taskset, calc_total_utilization
from rta import verify_no_migration, verify_model_1, verify_model_2, verify_model_3
from plot import plot_data
from joblib import Parallel, delayed
import config
import copy
import sys
def create_chart (results, x_label, y_label, filename):
data_to_plot = []
if config.CHECK_NO_MIGRATION:
data_to_plot.append({'label': 'No Migration', 'data': results[0]})
if config.CHECK_MODEL_1:
data_to_plot.append({'label': 'Model 1', 'data': results[1]})
if config.CHECK_MODEL_2:
data_to_plot.append({'label': 'Model 2', 'data': results[2]})
if config.CHECK_MODEL_3:
data_to_plot.append({'label': 'Model 3', 'data': results[3]})
plot_data(
data_to_plot,
x_label,
y_label,
config.RESULTS_DIR + filename)
# Run one instance of the tests (this functions is necessary for parallelism)
def run_instance (n, p, f, U):
taskset = generate_taskset(n, p, f, U)
# print (taskset)
# sys.exit()
taskset_utilization = calc_total_utilization(taskset)
taskset_schedulability = [False, False, False, False]
if config.CHECK_NO_MIGRATION and verify_no_migration(copy.deepcopy(taskset)):
taskset_schedulability[0] = True
if config.CHECK_MODEL_1 and verify_model_1(copy.deepcopy(taskset)):
taskset_schedulability[1] = True
if config.CHECK_MODEL_2 and verify_model_2(copy.deepcopy(taskset)):
taskset_schedulability[2] = True
if config.CHECK_MODEL_3 and verify_model_3(copy.deepcopy(taskset)):
taskset_schedulability[3] = True
return taskset_schedulability, taskset_utilization
# First test: check percentage of schedulable tasksets with different utilizations
def run_first_test ():
res_global = [[], [], [], []]
# Starting and final utilization values
U = 3.2
finish_U = 4.6
# Utilization step
step = 0.028
first_test_bar = Bar('First test', max=51)
while U <= finish_U:
res_local = [[U, 0], [U, 0], [U, 0], [U, 0]]
results = Parallel(n_jobs=config.PARALLEL_JOBS)(delayed(run_instance)(24, 0.5, 2, U) for _ in range(config.NUMBER_OF_TESTS))
for result in results:
for i in range(4):
if result[0][i]:
res_local[i][1] += 1
for i in range(4):
res_local[i][1] = res_local[i][1] * 100 / config.NUMBER_OF_TESTS
res_global[i].append(res_local[i])
U += step
first_test_bar.next()
first_test_bar.finish()
create_chart(res_global, 'Utilization', 'Schedulable Tasksets', 'result_1.png')
# This test is similar to "run_first_test" but keeps track of total utilization vs. total schedulable utilization
# n -> Taskset size
# p -> Percentage of HI-crit tasks
# f -> Criticality factor
def check_utilization_total_schedulability (n, p, f):
# Keep track of the sum of all tasksets' utilizations
total_utilizations = 0
# Keep track of the sum of schedulable tasksets' utilizations (for every model)
total_schedulable_utilizations = [0, 0, 0, 0]
# Starting utilization value
U = 3.2
# Utilization step
step = 0.028
while U <= 4.6:
results = Parallel(n_jobs=config.PARALLEL_JOBS)(delayed(run_instance)(n, p, f, U) for _ in range(config.NUMBER_OF_TESTS))
for result in results:
for i in range(4):
if result[0][i]:
total_schedulable_utilizations[i] += result[1]
total_utilizations += result[1]
U += step
return total_utilizations, total_schedulable_utilizations
def run_second_test ():
res_global = [[], [], [], []]
# Starting and final Criticality Factor values
f = 1.25
finish_f = 3.75
f_step = 0.25
second_test_bar = Bar('Second test', max=11)
while f <= finish_f:
total_utilizations, total_schedulable_utilizations = check_utilization_total_schedulability(24, 0.5, f)
for i in range(4):
res_global[i].append([f, total_schedulable_utilizations[i] / total_utilizations])
f += f_step
second_test_bar.next()
second_test_bar.finish()
create_chart(res_global, 'Criticality Factor', 'Weighted Schedulability', 'result_2.png')
def run_third_test ():
res_global = [[], [], [], []]
# Starting and final Proportion of HI-crit tasks values
p = 0.1
finish_p = 0.9
p_step = 0.1
third_test_bar = Bar('Third test', max=9)
while p <= finish_p:
total_utilizations, total_schedulable_utilizations = check_utilization_total_schedulability(24, p, 2)
for i in range(4):
res_global[i].append([p, total_schedulable_utilizations[i] / total_utilizations])
p += p_step
third_test_bar.next()
third_test_bar.finish()
create_chart(res_global, 'Proportion of HI-crit tasks', 'Weighted Schedulability', 'result_3.png')
def run_fourth_test ():
res_global = [[], [], [], []]
# Starting and final Tasksets' size values
n = 32
final_n = 144
n_step = 16
fourth_test_bar = Bar('Fourth test', max=8)
while n <= final_n:
total_utilizations, total_schedulable_utilizations = check_utilization_total_schedulability(n, 0.5, 2)
for i in range(4):
res_global[i].append([n, total_schedulable_utilizations[i] / total_utilizations])
n += n_step
fourth_test_bar.next()
fourth_test_bar.finish()
create_chart(res_global, 'Taskset size', 'Weighted Schedulability', 'result_4')
if config.RUN_FIRST_TEST:
print('>>> Running first test')
run_first_test()
if config.RUN_SECOND_TEST:
print('>>> Running second test')
run_second_test()
if config.RUN_THIRD_TEST:
print('>>> Running third test')
run_third_test()
if config.RUN_FOURTH_TEST:
print('>>> Running fourth test')
run_fourth_test()
print('>>> Done')