02229 Systems Optimization E20

Exercise: Assignment of real-time tasks to the multicores of an autonomous vehicle platform

Brief Abstract:

The prime objective of this assignment is to implement a metaheuristic solution over a decent neighbor function to optimize the task assigned to the cores of multicore processors. Starting with a small task assignment data, small.xml with 8 cores, and 9 tasks, we designed a basic neighbor function to implement and validate the simulation annealing method which improved the task scheduling to a great extent. However, the task assignment was pretty straightforward and there was no unschedulable job. To check the extent of our solution, we switched to medium.xml file which contains relatively larger task assignment data than small.xml. This gave us some insights to rectify and optimize the various aspects of our solution and subsequently achieved a valid and more promising solution. The same model is used to run over a large.xml file which contains 17 cores and approximately 250 tasks, and results were fairly auspicious. One important thing to note is that this large file with a significantly large number of tasks than cores initially showed some unschedulable tasks but later after the number of task shifts in between cores, there was a significant positive change in performance. This serves the purpose of this assignment and as an end solution, we have solution.xml which contains optimal placement of tasks corresponding to cores.

#All Imports import xml.etree.ElementTree as et import numpy as np import pandas as pd !pip install seaborn import seaborn as sns import scipy !pip install pandas_read_xml import pandas_read_xml as pdx import random import matplotlib.pyplot as plt from scipy.optimize import fsolve

Set-up

This section contains initial setup of variables, dataframes and extraction of data from xml files. DFapp contains the tasks with deadlines, id, period and WCET parameters whereas DFplat comprises of machines and cores with Id's and WCRT factor.

#Create two Pandas Dataframes for the applications and for the platforms in the XML DFapp = pd.DataFrame(columns=['Deadline', 'Id', 'Period', "WCET"]) DFplat= pd.DataFrame(columns=["MCPId","CoreId","WCETFactor"]) #Another Dataframe to structure the Solution Sol = pd.DataFrame(columns=["Task", "MCP", "Core", "WCRT"])

#Iterate through the XML file to get all the needed data in order to be able to append them to the dataframes tree = et.parse("large.xml") root = tree.getroot() for primary in root.iter('Model'): for types in primary.iter("Application"): for tasks in types.iter("Task"): DFapp = DFapp.append({'Deadline':int(tasks.get("Deadline")), 'Id':int(tasks.get("Id")), 'Period':int(tasks.get("Period")), 'WCET':int(tasks.get("WCET")) }, ignore_index=True) for types in primary.iter("Platform"): for mcps in types.iter("MCP"): for cores in mcps.iter("Core"): DFplat = DFplat.append({"MCPId":int(mcps.get("Id")), "CoreId":int(cores.get("Id")), "WCETFactor":float(cores.get("WCETFactor"))}, ignore_index=True)

Example of how the DFapp and DFplat dataframes will look like

DFapp.head()

DFplat['ID'] = DFplat.index DFplat.head()

Functions definitions

This section contains all the required functions which are used to calculate important parameters.

Mean laxity function:

Returns the mean laxity of the dataset to be used as cost function.

# Mean laxity def mean_laxity(DFapp): return (np.mean(np.array(DFapp['Laxity'])))

Laxity function:

Appends a column to the tasks dataset containing the new laxities calculated as:

$ L_i = D_i - WCRT_i $

Returns the updated tasks dataset.

# Laxity def Laxity2(DFapp): DFapp['Laxity'] = DFapp['Deadline'] - DFapp['WCRT'] return(DFapp)

Priority function:

Sets for each task an unique proprity represented as an integer. The lower the integer, the higher is the priority of the task (the task with priority 0 is the highest priority task).

# Priority function def priority(DFapp): DFapp["Deadline"] = DFapp["Deadline"] DFapp.sort_values(by="Deadline",inplace=True) DFapp.reset_index(drop=True, inplace=True) DFapp["Priority"] = DFapp.index.astype(int) return(DFapp) DFapp_prio = priority(DFapp) DFapp_prio.head()

Allocation Function

This function is used at the initialization to randomly allocate the tasks to cores.

# Allocation function def allocation(DFapp,DFplat): cores = [] WCETFactors = [] for task in DFapp['Id']: sample = DFplat.sample(1) cores.append(int(sample['ID'])) WCETFactors.append(float(sample['WCETFactor'])) DFapp['CoreIndex'] = cores DFapp['WCETFactor'] = WCETFactors return(DFapp) DFapp_allo = allocation(DFapp,DFplat)

Neighbor function

Generates a neighbor starting from the current dataset. The simplest option is to take a random task and move it from the current core to another random core.

def neighbor1(app,DFplat,N): tasks = random.sample(list(app['Id']),N) core_ids = [] for taskId in tasks: coreId = random.choice(list(DFplat['ID'])) core_ids.append(coreId) core_ids.append(int(app[DFapp['Id'] == taskId]['CoreIndex'])) app.loc[(app['Id'] == taskId),['CoreIndex']] = coreId app.loc[app['Id'] == taskId,'WCETFactor'] = float(DFplat[DFplat['ID'] == coreId]['WCETFactor']) return(app,np.unique(core_ids))

Schedulability function:

Checks if the whole set is schedulable by checking that all the laxities are positive. Returns True or False.

# Check schedulability(if task is schdulable or not) def check_schedulability(DFapp): return all(DFapp['Laxity'] > 0)

WCRT calculation

The longest response time is calculated as:

$$R_i = C_i + I_i$$

Where interference $I_i$ is:

$$Ii = \sum{j=1}^{i-1}\frac{R_i}{T_j}C_j$$

To calculate WCRT, the following formula is used:

$$Ri = C_i + \sum^{i - 1}{j = 1}\frac{R_i}{T_j} C_j$$

where $R_i$ is the longest response time of a task $\tau_i$, $C$ is the worst-case computation time and $T$ is the period.

The algorithm used is from 4.4.2, Hard-Real time periodic scheduling chapter.

### CORRECT FUNCTION FOR WCRT def compute_WCRT(DFapp,core_ids): core_ids = np.unique(core_ids) ## new WCRT is calculated only for the cores in which something has changed, since it's useless to calculate it each time for every core also if it's the same as before for core in core_ids: subset = DFapp[DFapp['CoreIndex'] == core] ## create a subset of the tasks allocated to the same core counter = 0 for taskId in subset['Id']: if counter == 0: ## if the element has the highest priority in the subset, the WCRT of task J is only the computation time R = subset.loc[subset['Id'] == taskId, 'WCET'] * subset.loc[subset['Id'] == taskId, 'WCETFactor'] else: ## if the element has not the highes priority in the subset, the WCRT of task J is calculated accounting for higher priority tasks P = int(subset.loc[subset['Id'] == taskId, 'Priority']) ## priority of core J D = int(subset.loc[subset['Id'] == taskId, 'Deadline']) ## deadline of core J Ct = float(subset.loc[subset['Id'] == taskId, 'WCET']) * float(subset.loc[subset['Id'] == taskId, 'WCETFactor']) ## computation time of task J (without interference) R = Ct ## the response time of core J is set initially equal to the computation time without interferences T = np.array(subset[subset['Priority'] < P]['Period']) ## array with periods of tasks with higher priority than J C = np.array(subset[subset['Priority'] < P]['WCET']) * np.array(subset[subset['Priority'] < P]['WCETFactor']) ## array with computation times of tasks with higher priority than J I = np.sum(np.ceil(R/T)*C) ## interference time control = True ## initialize value for while cycle while control == True: R = I + Ct ## update WCRT I = np.sum((np.ceil(R/T))*C) ## calculate interference with new WCRT if (I + Ct) > D: ## check if it's NOT schedulable R = 2*D ## if it's NOT schedulable, give a very high value of WCRT (so laxity is negative for the check) control = False if abs((I+Ct) - R) < 1: ## if it IS schedulable and the error on the WCRT is less than 1, R = WCRT and interrupt while loop control = False DFapp.loc[(DFapp['Id'] == taskId),['WCRT']] = np.ceil(R) ## update value of WCRT counter = counter + 1 return(DFapp)

Simulated annealing algorithm

Initialization

    - Generate a random schedule using the function.
    - Set initial temperature T and reduction parameter r.

Looping: apply the simulated annealing algorithm.

    - Generates a new neighbor and computes WCRT and schedulability.
    - Updates WCRT and laxity.
    - Check schedulability. If it's false, penalize the laxity with a factor 100000.
    - If (new laxity) > (old laxity), accept the solution. Otherwise, accept with certain probability.
    - Decrease the temperature: T = T(1-r)

## Simulated annealing # inizialization N = 1 DFplat['ID'] = DFplat.index T = 10000 r = 0.003 DFapp = priority(DFapp) ## assign priorities to tasks DFapp = allocation(DFapp,DFplat) ## initial allocation DFapp, core_ids = neighbor1(DFapp,DFplat,N) ## generate neighbor DFapp['WCRT'] = np.nan DFapp = compute_WCRT(DFapp,np.unique(DFplat['ID'])) ## initial computation of WCRT DFapp = Laxity2(DFapp) ## initial calculation of laxity schedulability = check_schedulability(DFapp) ## initial check of schedulability if schedulability == True: laxity = mean_laxity(DFapp) elif schedulability == False: laxity = mean_laxity(DFapp)/100000 ## penalize laxity if taskset is not schedulable all_costs = [] all_costs.append(laxity) # loop while (T>1): DFapp_new = DFapp.copy() # IMPORTANT!!! OTHERWISE ALSO DFAPP IS UPDATED DFapp_new, core_ids = neighbor1(DFapp_new,DFplat,1) ## generate neighbor DFapp_new = compute_WCRT(DFapp_new,core_ids) ## update WCRT DFapp_new = Laxity2(DFapp_new) ## update laxity schedulability_new = check_schedulability(DFapp_new) ## check schedulability if(schedulability_new == False): laxity_new = mean_laxity(DFapp_new)/100000 ## penalize the laxity if the task set is not schedulable else: laxity_new = mean_laxity(DFapp_new) if (laxity_new > laxity): all_costs.append(laxity_new) laxity = laxity_new DFapp = DFapp_new else: prob = np.exp((laxity_new - laxity)/T) ## probability to accept the new laxity if worse than the previous rand = np.random.rand(1) ## random number between 0 and 1 if(prob > rand): ## condition to accept a worse solution all_costs.append(laxity_new) laxity = laxity_new DFapp = DFapp_new else: ## condition to refuse a worse solution all_costs.append(laxity) T = T*(1-r) ## reduce the temperature print(T) # plot the results plt.plot(all_costs[1:])

DFapp[DFapp['CoreIndex'] == 1]

Showing total number of tasks allocated on each core

DFapp["CoreIndex"].value_counts().plot.bar()

Results

Get results in XML format

In order to get the Solution back in a XML format, the pandas dataframe will have to be converted into a string and then appended to the solution xml file

def xml_add(data,cores): xml = ['<solution>'] for index in range(len(data)): xml.append(' <Task Id="{0}" MCP="{1}" Core="{2}" WCRT="{3}" />'.format(data.loc[index]["Id"], int(cores.loc[data.loc[index]["CoreIndex"]]["MCPId"]), int(cores.loc[data.loc[index]["CoreIndex"]]["CoreId"]), data.loc[index]["WCRT"])) xml.append('</solution>') xml.append(''.format(all_costs[-1] * len(data))) return xml

xml = xml_add(DFapp,DFplat) with open("Solution.xml", "w") as txt_file: #txt.write(u'\uFEFF'.encode('UTF-8')) for line in xml: txt_file.write("".join(str(line)) + "\n")