classdef ExecutionTimeRealTimeScheduleController < coder.profile.ExecutionTimeScheduleController
%EXECUTIONTIMEREALTIMESCHEDULECONTROLLER generates the schedule from
%real-time simulations

% Copyright 2020-2021 The MathWorks, Inc.

properties (Constant=true, Access=private)
% core signals
SchedulingIdIdle = int32(0);
% thread signals
SchedulingNonActiveValue = int32(0);
SchedulingPreemptedValue = int32(2);
SchedulingRunning = int32(1);
end

properties(Access = private)
% Asyncqueue for core activity, core utilization and threads
AQCores;
AQThreads;
AQUsages;

% store number of cores and map between core ID from raw data and
% CPU asyncqueue index. There may be holes in the raw IDs and we
% don't want empty signals
NumCores;
CoreMapIDs;

% those are local variables which have class scope to avoid to pass
% them around as arguments
IsTaskActive;
Parents;
Stacks;
Usages;
LSEs;
end

methods
function obj = ExecutionTimeRealTimeScheduleController(modelname, timerData, tps)
obj = obj@coder.profile.ExecutionTimeScheduleController(modelname, timerData, tps);
% get raw data
[~, ~, lCoreNumStartVals, lCoreNumEndVals] = timerData.getTimerValues;
% process cores
% note: increment the core ID to prevent zero IDs which will be
% used as index in the map array
coreIDs = unique([lCoreNumStartVals, lCoreNumEndVals]) + 1;
obj.NumCores = length(coreIDs);
obj.CoreMapIDs = containers.Map('KeyType', 'uint32', 'ValueType', 'any');
for i = 1:length(coreIDs)
obj.CoreMapIDs(coreIDs(i)) = i;
end
end
end

methods(Access = protected)
function interMethods = getInterpolationMethod(this)
num = length(this.AQCores) + length(this.AQThreads);
interMethods1 = repmat({'zoh'}, 1, num);
interMethods2 = repmat({'linear'}, 1, length(this.AQUsages));
interMethods = [interMethods1(:)' interMethods2(:)'];
end

function [hQ, hC] = createAsyncQueues(this, names)
offset = 0;
% thread asyncqueues
[scT, this.AQThreads] = this.createThreadQueues(offset);
% core asyncqueues
offset = length(this.AQThreads);
[scC, this.AQCores] = this.createCoreQueues(names, offset);
% usage asyncqueues
offset = offset + length(this.AQCores);
[scU, this.AQUsages] = this.createUsageQueues(offset);
% Queue configuration
hQ = [this.AQCores this.AQThreads this.AQUsages];
hC = [scC scT scU];
end

function [startTime, lastEnd] = plotScheduling(this, nameIds, ~, ~)
% get raw data
[timerStartValues, timerEndValues, lCoreNumStartVals, lCoreNumEndVals,...
nodeNumbers] = this.TimerData.getTimerValues;
% init data
startTime = double(timerStartValues(1));
lastEnd = startTime;
this.initSharedData(startTime);
this.initQueues(startTime);
nStart = length(timerStartValues);
% check if data must be limited
pointCounter = 0;
limitPlot = false;
if ~isinf(this.MaxNumPointsToProcess) && ...
length(timerStartValues) > this.MaxNumPointsToProcess
limitPlot = true;
MSLDiagnostic('CoderProfile:ExecutionTime:ProfilingSRLTcheduleTooManyPoints').reportAsWarning;
end
idxStart = 1;
endValStack = [];
endIdxStack = [];
% iterate over all the sorted scheduling events
while idxStart <= nStart || ~isempty(endValStack)
% get next start event
if idxStart <= nStart
valStart = double(timerStartValues(idxStart));
else
valStart = Inf;
end
% get next end event
if isempty(endValStack)
valEnd = Inf;
else
[valEnd, idxPosEnd] = min(endValStack);
idxEnd = endIdxStack(idxPosEnd);
valEnd = double(valEnd);
end
% check which one must be processed first
% don't use <=, give priority to ends
if valStart < valEnd
% extract start sample
actNodeId = nodeNumbers(idxStart) - 1;
nodeId = this.RealNodes(actNodeId);
if nodeId == 0 || this.Tasks(nodeId)
% process the sample if it is either a task or a
% preemption section (nodeId==0). Ignore functions
idxCore = this.CoreMapIDs(lCoreNumStartVals(idxStart) + 1);
itsValEnd = timerEndValues(idxStart);
% process sample
this.processStartSample(nodeId, actNodeId, valStart, startTime, idxCore, nameIds);
% update internal data
endValStack(end+1) = itsValEnd; %#ok<AGROW>
endIdxStack(end+1) = idxStart; %#ok<AGROW>
end
idxStart = idxStart + 1;
else
% extract end sample
lastEnd = valEnd;
actNodeId = nodeNumbers(idxEnd) - 1;
nodeId = this.RealNodes(actNodeId);
if nodeId == 0 || this.Tasks(nodeId)
% process the sample if it is either a task or a
% preemption section (nodeId==0). Ignore functions
dstCore = this.CoreMapIDs(lCoreNumEndVals(idxEnd) + 1);
srcCore = this.CoreMapIDs(lCoreNumStartVals(idxEnd) + 1);
% process sample
this.processEndSample(nodeId, actNodeId, valEnd, srcCore, dstCore, nameIds);
% update internal data
endValStack(idxPosEnd) = []; %#ok<AGROW>
endIdxStack(idxPosEnd) = []; %#ok<AGROW>
end
end
% check num point limit
if limitPlot
pointCounter = pointCounter + 1;
if pointCounter > this.MaxNumPointsToProcess && isempty(endValStack)
break;
end
end
end
% add final values to make all signals stop at last event
this.termQueues(lastEnd);
end
end

methods(Access = private)
function processStartSample(this, nodeId, actNodeId, t, startTime, idxCore, nameIds)
% functions aren't processed
tScaled = double(t) / this.TimeScaleFactor;
if nodeId > 0
% task element (not a preemption section)

% task AQ: mark it as running
qThreadIdx = find(nodeId == find(this.Tasks));
this.AQThreads(qThreadIdx).write(tScaled, this.SchedulingRunning);
this.IsTaskActive(nodeId) = true;
% usage AQ
if ~isempty(this.Stacks{idxCore})
% the core isn't idle so, it was active since
% last scheduling event
this.Usages(idxCore) = this.Usages(idxCore) + t - this.LSEs(idxCore);
util = this.Usages(idxCore) / t;
this.AQUsages(idxCore).write(tScaled, util)
else
% the core is idle since last scheduling event
if t == startTime
% this is the first execution on the core since the
% simulation start: the core is fully utilizied
util = 1.0;
else
util = this.Usages(idxCore) / t;
end
this.AQUsages(idxCore).write(tScaled, util)
end
this.LSEs(idxCore) = t;
% core AQ: show the active task as running
this.AQCores(idxCore).write(tScaled, int32(nameIds(nodeId)));
% update task stack for the core
this.Stacks{idxCore}(end+1) = nodeId;
else
% this is a fictious preemption section. Mark the parent
% node as preempted - don't need to update stacks, next
% task will update the internal data structures
% when internal functions are profiled, we need to find the
% task the preemption refers to so, the direct parent isn't
% enough
do = true;
while do
parent = this.RealNodes(this.Parents(actNodeId));
do = ~this.Tasks(parent);
actNodeId = this.Parents(actNodeId);
end
qThreadIdx = find(parent == find(this.Tasks));
this.AQThreads(qThreadIdx).write(tScaled, this.SchedulingPreemptedValue);
end
end

function processEndSample(this, nodeId, actNodeId, t, srcCore, dstCore, nameIds)
tScaled = double(t) / this.TimeScaleFactor;
if nodeId > 0
% mark thread as not active
qThreadIdx = find(nodeId == find(this.Tasks));
this.AQThreads(qThreadIdx).write(tScaled, this.SchedulingNonActiveValue);
this.IsTaskActive(nodeId) = false;

if dstCore == srcCore
% the task started and ended on the same core, no
% migration detected

% we should assert that running task on the core is the
% current one: this.Stacks{dstCore}(end) == nodeId
% however, let us try to be as robust as possible
if this.Stacks{dstCore}(end) == nodeId
% the current task is running, remove it
this.Stacks{dstCore}(end) = [];
% update core AQ: if the stack is empty, mark as
% idle otherwise active the task at the stack top
this.writeTaskTerminationOnCore(t, dstCore, nameIds);
% update core usage - the core was already active
this.Usages(dstCore) = this.Usages(dstCore) + t - this.LSEs(dstCore);
util = this.Usages(dstCore) / t;
this.AQUsages(dstCore).write(tScaled, util);
this.LSEs(dstCore) = t;
else
% the current task isn't the just terminated one.
% remove it from the queue anyway
idToRemove = this.Stacks{dstCore} == nodeId;
this.Stacks{dstCore}(idToRemove) = [];
end
else
% the task was migrated during its execution

% update original core by removing the current task
% from its call stack
idxInOriginalStackCore = find(this.Stacks{srcCore} == nodeId);
assert(length(idxInOriginalStackCore) == 1);
this.Stacks{srcCore}(idxInOriginalStackCore) = [];
% update the original core AQ to show that the
% migrated task terminated. If the stack is empty, mark
% as idle otherwise run the task at the stack top
this.writeTaskTerminationOnCore(t, srcCore, nameIds);
% update destination core to show that the current task
% terminated on it
nextItem = int32(nameIds(nodeId));
this.AQCores(dstCore).write(tScaled, nextItem);
this.writeTaskTerminationOnCore(t, dstCore, nameIds);
% update the usage of the src core. No need to update
% the destination code because it executed only an
% unknown interval which we assume it is negligible
this.Usages(srcCore) = this.Usages(srcCore) + t - this.LSEs(srcCore);
db = this.Usages(srcCore) / t;
this.AQUsages(srcCore).write(tScaled, db)
this.LSEs(srcCore) = t;
end
else
% this is a fictious preemption section. Mark the parent
% node as running on the core
% when internal functions are profiled, we need to find the
% task the preemption refers to so, the direct parent isn't
% enough
do = true;
while do
parent = this.RealNodes(this.Parents(actNodeId));
do = ~this.Tasks(parent);
actNodeId = this.Parents(actNodeId);
end
qThreadIdx = find(parent == find(this.Tasks));
this.AQThreads(qThreadIdx).write(tScaled, this.SchedulingRunning);
if ~this.IsTaskActive(parent)
% special case if the parent task is terminated already
% in the case of migration
this.AQThreads(qThreadIdx).write(tScaled, this.SchedulingNonActiveValue);
end
end
end

function initSharedData(this, startTime)
numNodesWithoutRoot = this.TimerData.getNumNodes() - 1;
this.Parents = zeros(1, numNodesWithoutRoot);
this.IsTaskActive = false(1, numNodesWithoutRoot);
for i = 1:numNodesWithoutRoot
this.Parents(i) = this.TimerData.getParentNumber(i);
end
% this contains the ready queue for each core. The top/end is
% the running task
this.Stacks = cell(1, this.NumCores);
% CPU utilization for each core
this.Usages = zeros(1, this.NumCores);
% provide initial values for the AQs at start time
% Last Scheduling Event on each core
this.LSEs = zeros(1, this.NumCores) + startTime;
end

function initQueues(this, startTime)
% provide an initial value to all the asyncqueues
tScaled = double(startTime) / this.TimeScaleFactor;
for idxCore = 1:this.NumCores
this.AQCores(idxCore).write(tScaled, this.SchedulingIdIdle);
this.AQUsages(idxCore).write(tScaled, 0);
end
for idTask = 1:this.NumTasks
this.AQThreads(idTask).write(tScaled, this.SchedulingNonActiveValue);
end
end

function termQueues(this, lastEnd)
% provide a final value to all the asyncqueues and terminate them
tScaled = double(lastEnd) / this.TimeScaleFactor;
for idxCore = 1:this.NumCores
if this.LSEs(idxCore) < lastEnd
this.AQCores(idxCore).write(tScaled, this.SchedulingIdIdle);
if ~isempty(this.Stacks{idxCore})
this.Usages(idxCore) = this.Usages(idxCore) + lastEnd - this.LSEs(idxCore);
end
this.AQUsages(idxCore).write(tScaled, double(this.Usages(idxCore)/lastEnd));
end
end
for idTask = 1:this.NumTasks
this.AQThreads(idTask).write(tScaled, this.SchedulingNonActiveValue);
end
end

function writeTaskTerminationOnCore(this, t, idxCore, nameIds)
tScaled = double(t) / this.TimeScaleFactor;
% update core AQ to notify a task end
if isempty(this.Stacks{idxCore})
% no pending execution: the core is idle
this.AQCores(idxCore).write(tScaled, this.SchedulingIdIdle);
else
% run first element in queue
nextItem = int32(nameIds(this.Stacks{idxCore}(end)));
this.AQCores(idxCore).write(tScaled, nextItem);
end
end

function [sc, queues] = createCoreQueues(this, names, offset)
% create data type for core signals. A core can be idle or
% running one of the tasks
states = [{'idle'} names(:)'];
vals = [this.SchedulingIdIdle, int32(1:length(names))];
dtCores = Simulink.AsyncQueue.DataType.createEnum('ProfilingCoreResults', 'Simulink.IntEnumType', vals, states);
% create a pair (client and asyncqueue) for each core
queues = Simulink.AsyncQueue.Queue.empty(0, this.NumCores);
sc = Simulink.AsyncQueue.SignalClient.empty(0, this.NumCores);
for idCore = 1:this.NumCores
% create signal
sigName = ['Core ' num2str(idCore)];
[sc(idCore), queues(idCore)] = this.createAQueue(sigName, dtCores, idCore + offset);
end
end

function [sc, queues] = createThreadQueues(this, offset)
% create data type for thread signals. A thread can be not
% active, preempted or running. We aren't able to detect when
% tasks are ready
states = cell(1, 3);
states{1} = 'notActive';
states{2} = 'preempted';
states{3} = 'running';
vals = zeros(1, 3, 'int32');
vals(1) = this.SchedulingNonActiveValue;
vals(2) = this.SchedulingPreemptedValue;
vals(3) = this.SchedulingRunning;
dtThreads = Simulink.AsyncQueue.DataType.createEnum('ProfilingThreadResults', 'Simulink.IntEnumType', vals, states);
queues = Simulink.AsyncQueue.Queue.empty(0, this.NumTasks);
sc = Simulink.AsyncQueue.SignalClient.empty(0, this.NumTasks);
num = 0;
for idTask = 1:this.NumSections
if ~this.Tasks(idTask)
continue;
end
% create signal
sigName = ['Task: ' this.TaskNames{idTask}];
num = num + 1;
[sc(num), queues(num)] = this.createAQueue(sigName, dtThreads, offset + num);
end
end

function [sc, queues] = createUsageQueues(this, offset)
dtCPUUsage = Simulink.AsyncQueue.DataType;
sc = Simulink.AsyncQueue.SignalClient.empty(0, this.NumCores);
queues = Simulink.AsyncQueue.Queue.empty(0, this.NumCores);
for idCore = 1:this.NumCores
% create signal
sigName = ['Usage Core ' num2str(idCore)];
[sc(idCore), queues(idCore)] = this.createAQueue(sigName, dtCPUUsage, idCore + offset);
end
end
end
end