Skip to content

Block throughput cost model

block_throughput_cost_model

This module defines an abstract base class BlockThroughputCostModel for estimating the throughput of a given Block.

It includes a concrete implementation, MCABlockThroughputCostModel, that uses the llvm-mca tool to perform this estimation.

BlockThroughputCostModel

Bases: ABC

An abstract base class for a throughput cost model.

This class provides an interface for estimating the throughput of a block of assembly-like operations for a specific microarchitecture.

Source code in xdsl/backend/block_throughput_cost_model.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class BlockThroughputCostModel(ABC):
    """
    An abstract base class for a throughput cost model.

    This class provides an interface for estimating the throughput of a block
    of assembly-like operations for a specific microarchitecture.
    """

    @abstractmethod
    def estimate_throughput(self, block: Block) -> float | None:
        """
        Estimates the throughput of a block of assembly-like operations.

        Throughput is defined as the number of cycles per iteration of a block.
        The throughput value gives an indication of the performance of the block.

        Args:
            block: The block of operations to analyze.

        Returns:
            The estimated throughput as a float, or None if the estimation fails.
        """
        pass

estimate_throughput(block: Block) -> float | None abstractmethod

Estimates the throughput of a block of assembly-like operations.

Throughput is defined as the number of cycles per iteration of a block. The throughput value gives an indication of the performance of the block.

Parameters:

Name Type Description Default
block Block

The block of operations to analyze.

 required

Returns:

Type Description
float | None

The estimated throughput as a float, or None if the estimation fails.
Source code in xdsl/backend/block_throughput_cost_model.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@abstractmethod
def estimate_throughput(self, block: Block) -> float | None:
    """
    Estimates the throughput of a block of assembly-like operations.

    Throughput is defined as the number of cycles per iteration of a block.
    The throughput value gives an indication of the performance of the block.

    Args:
        block: The block of operations to analyze.

    Returns:
        The estimated throughput as a float, or None if the estimation fails.
    """
    pass

ExternalBlockThroughputCostModel

Bases: BlockThroughputCostModel

An abstract base class for throughput cost models that use external command-line tools.

This class extends BlockThroughputCostModel and provides common functionality for models that rely on an external command-line tool to produce a report from a source file. Subclasses must implement methods to define the command to run (cmd) and name (tool_name), and to parse the tool's output (process_report).

Source code in xdsl/backend/block_throughput_cost_model.py 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
class ExternalBlockThroughputCostModel(BlockThroughputCostModel):
    """
    An abstract base class for throughput cost models that use external command-line tools.

    This class extends `BlockThroughputCostModel` and provides common functionality
    for models that rely on an external command-line tool to produce a report
    from a source file. Subclasses must implement methods to define the command
    to run (`cmd`) and name (`tool_name`), and to parse the tool's output (`process_report`).
    """

    target: str
    """
    The target triple.
    """
    arch: str
    """
    The architecture name of the target.
    """
    src_path: str

    def __init__(self, *, target: str, arch: str):
        self.target = target
        self.arch = arch

    @classmethod
    @abstractmethod
    def tool_name(cls) -> str:
        pass

    @abstractmethod
    def cmd(self) -> list[str]:
        pass

    def produce_report(self) -> str:
        cmd = self.cmd()
        with NamedTemporaryFile(mode="w+", suffix=".s", delete=False) as report_file:
            _ = subprocess.run(
                cmd, stdout=report_file, stderr=subprocess.DEVNULL, check=True
            )
            report_file.seek(0)
            return report_file.read()

    @abstractmethod
    def process_report(self, report: str) -> float | None:
        pass

    @classmethod
    def is_installed(cls) -> bool:
        return which(cls.tool_name()) is not None

    def estimate_throughput(self, block: Block) -> float | None:
        with NamedTemporaryFile(mode="w+", delete=False, suffix=".s") as tmp_file:
            self.src_path = tmp_file.name
            printer = AssemblyPrinter(stream=tmp_file)
            for op in block.walk():
                assert isinstance(op, AssemblyPrintable), (
                    f"Block operation {op} should be an assembly instruction"
                )
                op.print_assembly(printer)

        try:
            report = self.produce_report()
            return self.process_report(report)
        finally:
            if os.path.exists(self.src_path):
                os.remove(self.src_path)

src_path: str instance-attribute

target: str = target instance-attribute

The target triple.

arch: str = arch instance-attribute

The architecture name of the target.

__init__(*, target: str, arch: str)

Source code in xdsl/backend/block_throughput_cost_model.py 
64
65
66
def __init__(self, *, target: str, arch: str):
    self.target = target
    self.arch = arch

tool_name() -> str abstractmethod classmethod

Source code in xdsl/backend/block_throughput_cost_model.py 
68
69
70
71
@classmethod
@abstractmethod
def tool_name(cls) -> str:
    pass

cmd() -> list[str] abstractmethod

Source code in xdsl/backend/block_throughput_cost_model.py 
73
74
75
@abstractmethod
def cmd(self) -> list[str]:
    pass

produce_report() -> str

Source code in xdsl/backend/block_throughput_cost_model.py 
77
78
79
80
81
82
83
84
def produce_report(self) -> str:
    cmd = self.cmd()
    with NamedTemporaryFile(mode="w+", suffix=".s", delete=False) as report_file:
        _ = subprocess.run(
            cmd, stdout=report_file, stderr=subprocess.DEVNULL, check=True
        )
        report_file.seek(0)
        return report_file.read()

process_report(report: str) -> float | None abstractmethod

Source code in xdsl/backend/block_throughput_cost_model.py 
86
87
88
@abstractmethod
def process_report(self, report: str) -> float | None:
    pass

is_installed() -> bool classmethod

Source code in xdsl/backend/block_throughput_cost_model.py 
90
91
92
@classmethod
def is_installed(cls) -> bool:
    return which(cls.tool_name()) is not None

estimate_throughput(block: Block) -> float | None

Source code in xdsl/backend/block_throughput_cost_model.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def estimate_throughput(self, block: Block) -> float | None:
    with NamedTemporaryFile(mode="w+", delete=False, suffix=".s") as tmp_file:
        self.src_path = tmp_file.name
        printer = AssemblyPrinter(stream=tmp_file)
        for op in block.walk():
            assert isinstance(op, AssemblyPrintable), (
                f"Block operation {op} should be an assembly instruction"
            )
            op.print_assembly(printer)

    try:
        report = self.produce_report()
        return self.process_report(report)
    finally:
        if os.path.exists(self.src_path):
            os.remove(self.src_path)

MCABlockThroughputCostModel

Bases: ExternalBlockThroughputCostModel

A throughput cost model that uses the llvm-mca tool.

Source code in xdsl/backend/block_throughput_cost_model.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
class MCABlockThroughputCostModel(ExternalBlockThroughputCostModel):
    """
    A throughput cost model that uses the `llvm-mca` tool.
    """

    @classmethod
    def tool_name(cls) -> str:
        return "llvm-mca"

    def cmd(self) -> list[str]:
        return [
            self.tool_name(),
            f"-mtriple={self.target}",
            f"-mcpu={self.arch}",
            self.src_path,
        ]

    def process_report(self, report: str) -> float | None:
        cycles = None
        iterations = None
        assert report is not None

        for l in report.splitlines():
            if l.startswith("Iterations"):
                iterations = float(l.split(":")[1])
            elif l.startswith("Total Cycles"):
                cycles = float(l.split(":")[1])

        if iterations is not None and cycles is not None:
            return cycles / iterations

        return None

tool_name() -> str classmethod

Source code in xdsl/backend/block_throughput_cost_model.py 
117
118
119
@classmethod
def tool_name(cls) -> str:
    return "llvm-mca"

cmd() -> list[str]

Source code in xdsl/backend/block_throughput_cost_model.py 
121
122
123
124
125
126
127
def cmd(self) -> list[str]:
    return [
        self.tool_name(),
        f"-mtriple={self.target}",
        f"-mcpu={self.arch}",
        self.src_path,
    ]

process_report(report: str) -> float | None

Source code in xdsl/backend/block_throughput_cost_model.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def process_report(self, report: str) -> float | None:
    cycles = None
    iterations = None
    assert report is not None

    for l in report.splitlines():
        if l.startswith("Iterations"):
            iterations = float(l.split(":")[1])
        elif l.startswith("Total Cycles"):
            cycles = float(l.split(":")[1])

    if iterations is not None and cycles is not None:
        return cycles / iterations

    return None