Skip to content
Snippets Groups Projects

Qt6 fixes and scheduler class started

Merged Qt6 fixes and scheduler class started
add-list-scheduler into master
Merged Oscar Gustafsson requested to merge add-list-scheduler into master
Viewing commit 97342c36
Prev Next
Show latest version
2 files
+ 274
− 120
Compare changes
  • Side-by-side
  • Inline
Files
2
b_asic/schedule.py
+ 266
− 120
@@ -36,6 +36,7 @@ from b_asic.resources import ProcessCollection
from b_asic.signal_flow_graph import SFG
from b_asic.special_operations import Delay, Input, Output
from b_asic.types import TypeName
from b_asic.scheduling_algorithm import SchedAlg
# Need RGB from 0 to 1
_EXECUTION_TIME_COLOR: Tuple[float, ...] = tuple(
@@ -68,13 +69,8 @@ class Schedule:
algorithm.
cyclic : bool, default: False
If the schedule is cyclic.
algorithm : {'ASAP', 'ALAP', 'provided'}, default: 'ASAP'
The scheduling algorithm to use. The following algorithm are available:
* ``'ASAP'``: As-soon-as-possible scheduling.
* ``'ALAP'``: As-late-as-possible scheduling.
If 'provided', use provided *start_times* and *laps* dictionaries.
algorithm : SchedulingAlgorithm, default: 'ASAP'
The scheduling algorithm to use.
start_times : dict, optional
Dictionary with GraphIDs as keys and start times as values.
Used when *algorithm* is 'provided'.
@@ -100,7 +96,7 @@ class Schedule:
sfg: SFG,
schedule_time: Optional[int] = None,
cyclic: bool = False,
algorithm: Literal["ASAP", "ALAP", "provided"] = "ASAP",
algorithm: SchedAlg = "ASAP",
start_times: Optional[Dict[GraphID, int]] = None,
laps: Optional[Dict[GraphID, int]] = None,
max_resources: Optional[Dict[TypeName, int]] = None,
@@ -115,10 +111,14 @@ class Schedule:
self._cyclic = cyclic
self._y_locations = defaultdict(_y_locations_default)
self._schedule_time = schedule_time
self.scheduler = Scheduler(self)
if algorithm == "ASAP":
self._schedule_asap()
self.scheduler.schedule_asap()
elif algorithm == "ALAP":
self._schedule_alap()
self.scheduler.schedule_alap()
elif algorithm == "earliest_deadline":
self.scheduler.schedule_earliest_deadline()
elif algorithm == "provided":
if start_times is None:
raise ValueError("Must provide start_times when using 'provided'")
@@ -797,116 +797,116 @@ class Schedule:
new_sfg = new_sfg.insert_operation_before(op, Delay(), port)
return new_sfg()
def _schedule_alap(self) -> None:
"""Schedule the operations using as-late-as-possible scheduling."""
precedence_list = self._sfg.get_precedence_list()
self._schedule_asap()
max_end_time = self.get_max_end_time()
if self.schedule_time is None:
self._schedule_time = max_end_time
elif self.schedule_time < max_end_time:
raise ValueError(f"Too short schedule time. Minimum is {max_end_time}.")
for output in self._sfg.find_by_type_name(Output.type_name()):
output = cast(Output, output)
self.move_operation_alap(output.graph_id)
for step in reversed(precedence_list):
graph_ids = {
outport.operation.graph_id
for outport in step
if not isinstance(outport.operation, Delay)
}
for graph_id in graph_ids:
self.move_operation_alap(graph_id)
def _schedule_asap(self) -> None:
"""Schedule the operations using as-soon-as-possible scheduling."""
precedence_list = self._sfg.get_precedence_list()
if len(precedence_list) < 2:
raise ValueError("Empty signal flow graph cannot be scheduled.")
non_schedulable_ops = set()
for outport in precedence_list[0]:
operation = outport.operation
if operation.type_name() not in [Delay.type_name()]:
if operation.graph_id not in self._start_times:
# Set start time of all operations in the first iter to 0
self._start_times[operation.graph_id] = 0
else:
non_schedulable_ops.add(operation.graph_id)
for outport in precedence_list[1]:
operation = outport.operation
if operation.graph_id not in self._start_times:
# Set start time of all operations in the first iter to 0
self._start_times[operation.graph_id] = 0
for outports in precedence_list[2:]:
for outport in outports:
operation = outport.operation
if operation.graph_id not in self._start_times:
# Schedule the operation if it does not have a start time yet.
op_start_time = 0
for current_input in operation.inputs:
if len(current_input.signals) != 1:
raise ValueError(
"Error in scheduling, dangling input port detected."
)
if current_input.signals[0].source is None:
raise ValueError(
"Error in scheduling, signal with no source detected."
)
source_port = current_input.signals[0].source
if source_port.operation.graph_id in non_schedulable_ops:
source_end_time = 0
else:
source_op_time = self._start_times[
source_port.operation.graph_id
]
if source_port.latency_offset is None:
raise ValueError(
f"Output port {source_port.index} of"
" operation"
f" {source_port.operation.graph_id} has no"
" latency-offset."
)
source_end_time = (
source_op_time + source_port.latency_offset
)
if current_input.latency_offset is None:
raise ValueError(
f"Input port {current_input.index} of operation"
f" {current_input.operation.graph_id} has no"
" latency-offset."
)
op_start_time_from_in = (
source_end_time - current_input.latency_offset
)
op_start_time = max(op_start_time, op_start_time_from_in)
self._start_times[operation.graph_id] = op_start_time
for output in self._sfg.find_by_type_name(Output.type_name()):
output = cast(Output, output)
source_port = cast(OutputPort, output.inputs[0].signals[0].source)
if source_port.operation.graph_id in non_schedulable_ops:
self._start_times[output.graph_id] = 0
else:
if source_port.latency_offset is None:
raise ValueError(
f"Output port {source_port.index} of operation"
f" {source_port.operation.graph_id} has no"
" latency-offset."
)
self._start_times[output.graph_id] = self._start_times[
source_port.operation.graph_id
] + cast(int, source_port.latency_offset)
self._remove_delays()
# def _schedule_alap(self) -> None:
# """Schedule the operations using as-late-as-possible scheduling."""
# precedence_list = self._sfg.get_precedence_list()
# self._schedule_asap()
# max_end_time = self.get_max_end_time()
# if self.schedule_time is None:
# self._schedule_time = max_end_time
# elif self.schedule_time < max_end_time:
# raise ValueError(f"Too short schedule time. Minimum is {max_end_time}.")
# for output in self._sfg.find_by_type_name(Output.type_name()):
# output = cast(Output, output)
# self.move_operation_alap(output.graph_id)
# for step in reversed(precedence_list):
# graph_ids = {
# outport.operation.graph_id
# for outport in step
# if not isinstance(outport.operation, Delay)
# }
# for graph_id in graph_ids:
# self.move_operation_alap(graph_id)
# def _schedule_asap(self) -> None:
# """Schedule the operations using as-soon-as-possible scheduling."""
# precedence_list = self._sfg.get_precedence_list()
# if len(precedence_list) < 2:
# raise ValueError("Empty signal flow graph cannot be scheduled.")
# non_schedulable_ops = set()
# for outport in precedence_list[0]:
# operation = outport.operation
# if operation.type_name() not in [Delay.type_name()]:
# if operation.graph_id not in self._start_times:
# # Set start time of all operations in the first iter to 0
# self._start_times[operation.graph_id] = 0
# else:
# non_schedulable_ops.add(operation.graph_id)
# for outport in precedence_list[1]:
# operation = outport.operation
# if operation.graph_id not in self._start_times:
# # Set start time of all operations in the first iter to 0
# self._start_times[operation.graph_id] = 0
# for outports in precedence_list[2:]:
# for outport in outports:
# operation = outport.operation
# if operation.graph_id not in self._start_times:
# # Schedule the operation if it does not have a start time yet.
# op_start_time = 0
# for current_input in operation.inputs:
# if len(current_input.signals) != 1:
# raise ValueError(
# "Error in scheduling, dangling input port detected."
# )
# if current_input.signals[0].source is None:
# raise ValueError(
# "Error in scheduling, signal with no source detected."
# )
# source_port = current_input.signals[0].source
# if source_port.operation.graph_id in non_schedulable_ops:
# source_end_time = 0
# else:
# source_op_time = self._start_times[
# source_port.operation.graph_id
# ]
# if source_port.latency_offset is None:
# raise ValueError(
# f"Output port {source_port.index} of"
# " operation"
# f" {source_port.operation.graph_id} has no"
# " latency-offset."
# )
# source_end_time = (
# source_op_time + source_port.latency_offset
# )
# if current_input.latency_offset is None:
# raise ValueError(
# f"Input port {current_input.index} of operation"
# f" {current_input.operation.graph_id} has no"
# " latency-offset."
# )
# op_start_time_from_in = (
# source_end_time - current_input.latency_offset
# )
# op_start_time = max(op_start_time, op_start_time_from_in)
# self._start_times[operation.graph_id] = op_start_time
# for output in self._sfg.find_by_type_name(Output.type_name()):
# output = cast(Output, output)
# source_port = cast(OutputPort, output.inputs[0].signals[0].source)
# if source_port.operation.graph_id in non_schedulable_ops:
# self._start_times[output.graph_id] = 0
# else:
# if source_port.latency_offset is None:
# raise ValueError(
# f"Output port {source_port.index} of operation"
# f" {source_port.operation.graph_id} has no"
# " latency-offset."
# )
# self._start_times[output.graph_id] = self._start_times[
# source_port.operation.graph_id
# ] + cast(int, source_port.latency_offset)
# self._remove_delays()
def _get_memory_variables_list(self) -> List[MemoryVariable]:
ret: List[MemoryVariable] = []
@@ -1227,3 +1227,149 @@ class Schedule:
# SVG is valid HTML. This is useful for e.g. sphinx-gallery
_repr_html_ = _repr_svg_
class Scheduler():
def __init__(self, schedule: Schedule) -> None:
self.schedule = schedule
def schedule_asap(self) -> None:
"""Schedule the operations using as-soon-as-possible scheduling."""
sched = self.schedule
prec_list = sched.sfg.get_precedence_list()
if len(prec_list) < 2:
raise ValueError("Empty signal flow graph cannot be scheduled.")
# handle the first set in precedence graph (input and delays)
non_schedulable_ops = set()
for outport in prec_list[0]:
operation = outport.operation
if operation.type_name() == Delay.type_name():
non_schedulable_ops.add(operation.graph_id)
elif operation.graph_id not in sched._start_times:
sched._start_times[operation.graph_id] = 0
# handle second set in precedence graph (first operations)
for outport in prec_list[1]:
operation = outport.operation
if operation.graph_id not in sched._start_times:
sched._start_times[operation.graph_id] = 0
# handle the remaining sets
for outports in prec_list[2:]:
for outport in outports:
operation = outport.operation
if operation.graph_id not in sched._start_times:
op_start_time = 0
for current_input in operation.inputs:
if len(current_input.signals) != 1:
raise ValueError(
"Error in scheduling, dangling input port detected."
)
if current_input.signals[0].source is None:
raise ValueError(
"Error in scheduling, signal with no source detected."
)
source_port = current_input.signals[0].source
if source_port.operation.graph_id in non_schedulable_ops:
source_end_time = 0
else:
source_op_time = sched._start_times[
source_port.operation.graph_id
]
if source_port.latency_offset is None:
raise ValueError(
f"Output port {source_port.index} of"
" operation"
f" {source_port.operation.graph_id} has no"
" latency-offset."
)
source_end_time = (
source_op_time + source_port.latency_offset
)
if current_input.latency_offset is None:
raise ValueError(
f"Input port {current_input.index} of operation"
f" {current_input.operation.graph_id} has no"
" latency-offset."
)
op_start_time_from_in = (
source_end_time - current_input.latency_offset
)
op_start_time = max(op_start_time, op_start_time_from_in)
sched._start_times[operation.graph_id] = op_start_time
# handle output and remove delays
for output in sched._sfg.find_by_type_name(Output.type_name()):
output = cast(Output, output)
source_port = cast(OutputPort, output.inputs[0].signals[0].source)
if source_port.operation.graph_id in non_schedulable_ops:
sched._start_times[output.graph_id] = 0
else:
if source_port.latency_offset is None:
raise ValueError(
f"Output port {source_port.index} of operation"
f" {source_port.operation.graph_id} has no"
" latency-offset."
)
sched._start_times[output.graph_id] = sched._start_times[
source_port.operation.graph_id
] + cast(int, source_port.latency_offset)
sched._remove_delays()
def schedule_alap(self) -> None:
"""Schedule the operations using as-late-as-possible scheduling."""
self.schedule_asap()
sched = self.schedule
max_end_time = sched.get_max_end_time()
if sched.schedule_time is None:
sched.set_schedule_time(max_end_time)
elif sched.schedule_time < max_end_time:
raise ValueError(f"Too short schedule time. Minimum is {max_end_time}.")
# move all outputs ALAP before operations
for output in sched.sfg.find_by_type_name(Output.type_name()):
output = cast(Output, output)
sched.move_operation_alap(output.graph_id)
# move all operations ALAP
for step in reversed(sched.sfg.get_precedence_list()):
for outport in step:
if not isinstance(outport.operation, Delay):
sched.move_operation_alap(outport.operation.graph_id)
def schedule_earliest_deadline(self) -> None:
"""Schedule the operations using earliest deadline scheduling."""
pass
# sched = self.schedule
# prec_list = sched.sfg.get_precedence_list()
# if len(prec_list) < 2:
# raise ValueError("Empty signal flow graph cannot be scheduled.")
# # handle the first set in precedence graph (input and delays)
# non_schedulable_ops = set()
# for outport in prec_list[0]:
# operation = outport.operation
# if operation.type_name() == Delay.type_name():
# non_schedulable_ops.add(operation.graph_id)
# elif operation.graph_id not in sched._start_times:
# sched._start_times[operation.graph_id] = 0
# # handle second set in precedence graph (first operations)
# for outport in prec_list[1]:
# operation = outport.operation
# if operation.graph_id not in sched._start_times:
# sched._start_times[operation.graph_id] = 0
# # handle the remaining sets
# for outports in prec_list[2:]:
# for outport in outports:
# operation = outport.operation
# if operation.graph_id not in sched._start_times:
# pass
\ No newline at end of file