Source code for qblox_scheduler.backends.qblox.control_flow_graph
# Repository: https://gitlab.com/qblox/packages/software/qblox-scheduler
# Licensed according to the LICENSE file on the main branch
#
# Copyright 2026, Qblox B.V.
"""Control-flow graph definition."""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import TYPE_CHECKING
from qblox_scheduler.backends.qblox.helpers import to_grid_time
from qblox_scheduler.backends.qblox.operation_handling.base import IOperationStrategy
from qblox_scheduler.backends.qblox.operation_handling.virtual import (
ConditionalStrategy,
ControlFlowReturnStrategy,
LoopStrategy,
)
from qblox_scheduler.helpers.linked_list import DLinkedList
if TYPE_CHECKING:
from collections.abc import Iterable, Iterator
[docs]
class BasicBlock(DLinkedList[IOperationStrategy]):
"""
A sequence of instructions that does not branch.
The BasicBlock is constructed with pointers to its first and its last instruction in a
`DLinkedList` of instructions.
A basic block does not contain any control-flow between its first and last instruction, meaning
the flow of execution can only enter at the first instruction, and exit at the last.
"""
def __init__(self, unfinished_basic_block: _UnfinishedBasicBlock) -> None:
super().__init__(unfinished_basic_block)
[docs]
self.start_time_ns = unfinished_basic_block.start_time_ns
"""
The start time of this block. This is helpful information as we do not have explicit wait
operations.
"""
[docs]
self.end_time_ns = unfinished_basic_block.end_time_ns
"""
The end time of this block. This is helpful information as we do not have explicit wait
operations.
If the block is a loop, this is the time after 1 repetition finished, not the whole loop.
"""
[docs]
self.successors: list[BasicBlock] = []
"""The basic blocks that can directly succeed this block."""
@dataclass
[docs]
class _UnfinishedBasicBlock(DLinkedList):
"""An unfinished basic block, that is used during control-flow graph construction."""
def __init__(self, start_time_ns: int, predecessors: list[_UnfinishedBasicBlock]) -> None:
super().__init__()
[docs]
self.successors: list[_UnfinishedBasicBlock] = []
[docs]
self.predecessors: list[_UnfinishedBasicBlock] = predecessors
[docs]
self.start_time_ns: int = start_time_ns
[docs]
self.end_time_ns: int | None = None
[docs]
class InstructionsIteratorFactory:
"""Iterator factory for instructions built from an iterator of basic blocks."""
def __init__(self, basic_blocks: Iterable[BasicBlock]) -> None:
[docs]
self._basic_blocks = basic_blocks
def __iter__(self) -> Iterator[IOperationStrategy]:
for basic_block in self._basic_blocks:
yield from basic_block
@dataclass
[docs]
class ControlFlowGraph:
"""
An intermediate representation of the instructions for one sequencer, with explicit control-flow
information.
It consists of two doubly linked lists: one for all instructions in chronological order, and one
for basic blocks with pointers into the list of instructions.
"""
[docs]
basic_blocks: list[BasicBlock] = field(default_factory=list)
@property
[docs]
def instructions(self) -> Iterable[IOperationStrategy]:
"""Iterator for instructions."""
return InstructionsIteratorFactory(self.basic_blocks)
@property
[docs]
def instructions_len(self) -> int:
"""Number of instructions."""
return sum(len(b) for b in self.basic_blocks)
@classmethod
[docs]
def __parse_operations_recursively(
cls,
instructions_iter: Iterator[IOperationStrategy],
current_block: _UnfinishedBasicBlock,
unfinished_blocks: list[_UnfinishedBasicBlock],
) -> _UnfinishedBasicBlock:
start_basic_block = current_block
while (instr := next(instructions_iter, None)) is not None:
instr_start_time_ns = to_grid_time(instr.operation_info.timing)
if isinstance(instr, (LoopStrategy, ConditionalStrategy)):
# Finish off block, if this is not the very first instruction.
next_blocks_predecessors = []
if len(current_block) != 0:
if current_block.end_time_ns is None:
current_block.end_time_ns = instr_start_time_ns
unfinished_blocks.append(current_block)
next_blocks_predecessors.append(current_block)
# Create new unfinished block.
current_block = _UnfinishedBasicBlock(
start_time_ns=instr_start_time_ns,
predecessors=next_blocks_predecessors,
)
current_block.append(instr)
current_block = cls.__parse_operations_recursively(
instructions_iter=instructions_iter,
current_block=current_block,
unfinished_blocks=unfinished_blocks,
)
elif isinstance(instr, ControlFlowReturnStrategy):
# Process the control-flow we're ending.
cf_start_instr = start_basic_block[0]
next_block_start_time_ns = instr_start_time_ns
if isinstance(cf_start_instr, LoopStrategy):
# Create loopback link: add to successors.
current_block.successors.append(start_basic_block)
loop_body_duration = instr_start_time_ns - to_grid_time(
cf_start_instr.operation_info.timing
)
reps = cf_start_instr.operation_info.data["repetitions"]
next_block_start_time_ns += loop_body_duration * (reps - 1)
# Finish off block.
current_block.append(instr)
current_block.end_time_ns = instr_start_time_ns
unfinished_blocks.append(current_block)
# Create new unfinished block.
current_block = _UnfinishedBasicBlock(
start_time_ns=next_block_start_time_ns,
predecessors=[current_block],
)
return current_block
else:
current_block.append(instr)
return current_block
@classmethod
[docs]
def from_instructions(cls, instructions: list[IOperationStrategy]) -> ControlFlowGraph:
"""
Construct a control-flow graph from a chronologically ordered `IOperationStrategy` list.
Parameters
----------
instructions : list[IOperationStrategy]
A chronologically ordered list of `IOperationStrategy` objects.
Returns
-------
ControlFlowGraph
The control-flow graph representation of the instructions.
"""
unfinished_blocks = []
current_block = _UnfinishedBasicBlock(
start_time_ns=to_grid_time(instructions[0].operation_info.timing), predecessors=[]
)
current_block = cls.__parse_operations_recursively(
iter(instructions), current_block, unfinished_blocks
)
if len(current_block) != 0 and (
len(unfinished_blocks) == 0 or unfinished_blocks[-1] is not current_block
):
# Handle last basic block
# Could only happen if we were not in a loop or conditional
instr = current_block[-1]
current_block.end_time_ns = to_grid_time(
instr.operation_info.timing + instr.operation_info.duration
)
unfinished_blocks.append(current_block)
return cls.__finish_basic_blocks(unfinished_blocks)
@classmethod
[docs]
def __finish_basic_blocks(
cls, unfinished_blocks: list[_UnfinishedBasicBlock]
) -> ControlFlowGraph:
"""
The unfinished blocks reference other unfinished blocks in the predecessors and successors.
These must be corrected to point to already finished block objects.
"""
cfg = cls()
unfinished_to_finished_map: dict[int, BasicBlock] = {}
for unfinished_block in unfinished_blocks:
if unfinished_block.start_time_ns is None or unfinished_block.end_time_ns is None:
raise RuntimeError("Error creating control-flow graph.")
finished_block = BasicBlock(unfinished_block)
cfg.basic_blocks.append(finished_block)
unfinished_to_finished_map[id(unfinished_block)] = finished_block
for unfinished_block in unfinished_blocks:
finished_block = unfinished_to_finished_map[id(unfinished_block)]
for unfinished_successor in unfinished_block.successors:
finished_successor = unfinished_to_finished_map[id(unfinished_successor)]
if finished_successor not in finished_block.successors:
finished_block.successors.append(finished_successor)
for unfinished_predecessor in unfinished_block.predecessors:
finished_predecessor = unfinished_to_finished_map[id(unfinished_predecessor)]
if unfinished_block not in finished_predecessor.successors:
finished_predecessor.successors.append(finished_block)
return cfg
