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
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
@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
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 is_available(self) -> bool:
        if not self.is_installed():
            return False
        # Probe with a trivial instruction fed through stdin, so an unsupported
        # target fails at target lookup while a supported one exits 0.
        self.src_path = "-"
        result = subprocess.run(
            self.cmd(),
            input="nop\n",
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            text=True,
        )
        return result.returncode == 0

    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(tmp_file.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
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
@classmethod
@abstractmethod
def tool_name(cls) -> str:
    pass

cmd() -> list[str] abstractmethod

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

produce_report() -> str

Source code in xdsl/backend/block_throughput_cost_model.py
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
@abstractmethod
def process_report(self, report: str) -> float | None:
    pass

is_installed() -> bool classmethod

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

is_available() -> bool

Source code in xdsl/backend/block_throughput_cost_model.py
def is_available(self) -> bool:
    if not self.is_installed():
        return False
    # Probe with a trivial instruction fed through stdin, so an unsupported
    # target fails at target lookup while a supported one exits 0.
    self.src_path = "-"
    result = subprocess.run(
        self.cmd(),
        input="nop\n",
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        text=True,
    )
    return result.returncode == 0

estimate_throughput(block: Block) -> float | None

Source code in xdsl/backend/block_throughput_cost_model.py
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(tmp_file.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
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
@classmethod
def tool_name(cls) -> str:
    return "llvm-mca"

cmd() -> list[str]

Source code in xdsl/backend/block_throughput_cost_model.py
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
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