experiments.py

#!/usr/bin/python3

"""
Example of a script that uses SimSo.
"""

import sys
import random
from numpy import arange
from context import simso
from simso.core import Model
from simso.configuration import Configuration
from simso.generator.task_generator import gen_periods_uniform, UUniFastDiscard, gen_tasksets, gen_faults, gen_arrivals

import pickle



def main(argv):
    # Manual configuration:
    schedulers = ["./simso/schedulers/RM.py", "./simso/schedulers/EDF.py","./simso/schedulers/P_RM.py","./simso/schedulers/P_EDF.py"] 
    # schedulers = []
    configuration = Configuration()

    configuration.cycles_per_ms = 1000

    configuration.etm = "wcet"

    configuration.duration = 1000 * configuration.cycles_per_ms

    num_processors = 8 
    # Add a processor:
    for np in range(0, num_processors):
        configuration.add_processor(name=f"CPU {np}" , identifier=np)


    nsets = 20 

    for u in arange(1,num_processors +0.1, 0.5):
        print(f"Utilization = {u}")
        utilizations = UUniFastDiscard(20, u, nsets)
        periods = gen_periods_uniform(20, nsets, 2, 100)
        tasksets = gen_tasksets(utilizations, periods)

        missed_deadlines = {}
        for sched in schedulers:
            missed_deadlines[sched] = 0

        for n in range(nsets):
            configuration._task_info_list = []

            for t in tasksets[n]:
                identifier = tasksets[n].index(t)
                configuration.add_task(name="t" + str(identifier), identifier=identifier,period=t[1],wcet=t[0],deadline=t[1])

            faults = gen_faults(n_faults=5,task_info_list=configuration._task_info_list,processor_info_list=configuration._proc_info_list, sim_duration=configuration.duration_ms)

            for f in faults:
                configuration.add_fault(fault_type=f['fault_type'], start_time=f['start_time'], end_time=f['end_time'], 
                                        fault_target=f['fault_target'], parameter=f.get('parameter'), value=f.get('value'))

            for sched in schedulers:
                configuration.scheduler_info.filename = sched 

                # Check the config before trying to run it.
                configuration.check_all()

                # Init a model from the configuration.
                model = Model(configuration)

                # Execute the simulation.
                try:
                    model.run_model()
                except RuntimeError as e:
                    if e.args[0] != "Bin packing failed!":
                        raise e 
                    missed_deadlines[sched] += 1 
                except AssertionError:
                    missed_deadlines[sched] += 1
                except ValueError:
                    with open('failed_configuration.pkl', 'wb') as out:
                        pickle.dump(configuration, out, pickle.HIGHEST_PROTOCOL)

                if model.results is not None and model.results.total_exceeded_count > 0:
                    missed_deadlines[sched] += 1

        for sched in schedulers:
            print(f"{sched}: {missed_deadlines[sched]}")
            

        #     # # Print logs.
        #     # for log in model.logs:
        #     #     print(log)


main(sys.argv)