msi.py

"""PCIe MSI (Message Signaled Interrupt) controllers.

Ported from ``litepcie/litepcie/core/msi.py``.

Three variants:

- :class:`PCIeMSI` — Single-vector MSI, edge-triggered.
- :class:`PCIeMSIMultiVector` — Multi-vector MSI, priority-encoded.
- :class:`PCIeMSIX` — MSI-X via TLP writes to host memory.
"""

from amaranth import *
from amaranth.lib.wiring import Component, In, Out

from amaranth_stream import Signature as StreamSignature

from amaranth_pcie.common import (
    msi_layout,
    msi_signature,
    request_signature,
)
from amaranth_pcie.phy.common import PCIeMSIInterface


__all__ = [
    "PCIeMSI",
    "PCIeMSIMultiVector",
    "PCIeMSIX",
]


class PCIeMSI(Component):
    """Single-vector MSI controller, edge-triggered.

    Detects rising edges on ``irqs & enable``, generates MSI pulse on
    the ``source`` stream.

    When the PHY provides a :class:`PCIeMSIInterface` with ``extended=True``,
    the controller also drives the ``ack`` signal and reads ``status``
    for completion feedback.

    Parameters
    ----------
    width : :class:`int`
        Number of IRQ sources (default 32).
    phy_msi : :class:`PCIeMSIInterface` or ``None``
        Optional PHY MSI interface for extended mode support.

    Ports
    -----
    irqs : In(width)
        One bit per IRQ source.
    source : Out(msi_signature())
        MSI output stream.
    enable : In(width)
        Per-IRQ enable mask (directly driven by software via CSR).
    clear : In(width)
        Per-IRQ clear mask (directly driven by software via CSR).
    clear_strobe : In(1)
        Strobe signal indicating ``clear`` value is valid (pulse).
    vector : Out(width)
        Current pending IRQ vector (readable by software via CSR).
    """

    def __init__(self, width=32, phy_msi=None):
        self._width = width
        self._phy_msi = phy_msi
        super().__init__({
            "irqs":         In(width),
            "source":       Out(msi_signature()),
            "enable":       In(width),
            "clear":        In(width),
            "clear_strobe": In(1),
            "vector":       Out(width),
        })

    @property
    def width(self):
        return self._width

    def elaborate(self, platform):
        m = Module()

        width = self._width
        enable = self.enable
        irqs = self.irqs

        clear = Signal(width, name="clear_mask")
        vector = Signal(width, name="vector_reg")

        # Memorize and clear IRQ Vector
        with m.If(self.clear_strobe):
            m.d.comb += clear.eq(self.clear)

        m.d.comb += self.vector.eq(vector)
        m.d.sync += vector.eq(enable & ((vector & ~clear) | irqs))

        # Generate MSI
        msi = Signal(width, name="msi_pending")
        m.d.comb += self.source.valid.eq(msi != 0)
        m.d.sync += [
            msi.eq((msi | irqs) & enable),
        ]

        # Determine ready signal — use extended ack if available
        msi_ready = self.source.ready
        if (self._phy_msi is not None
                and isinstance(self._phy_msi, PCIeMSIInterface)
                and self._phy_msi.extended):
            # In extended mode, use ack as the ready signal
            msi_ready = self._phy_msi.ack

        with m.If(self.source.valid & msi_ready):
            m.d.sync += msi.eq(irqs & enable)

        return m


class PCIeMSIMultiVector(Component):
    """Multi-vector MSI controller, priority-encoded.

    Same as :class:`PCIeMSI` but ``source.payload.dat`` carries the IRQ
    number (priority-encoded, lower index = higher priority).

    Parameters
    ----------
    width : :class:`int`
        Number of IRQ sources (default 32).

    Ports
    -----
    irqs : In(width)
        One bit per IRQ source.
    source : Out(msi_signature())
        MSI output stream. ``dat`` = IRQ number.
    enable : In(width)
        Per-IRQ enable mask.
    vector : Out(width)
        Current pending IRQ vector.
    """

    def __init__(self, width=32):
        self._width = width
        super().__init__({
            "irqs":   In(width),
            "source": Out(msi_signature()),
            "enable": In(width),
            "vector": Out(width),
        })

    @property
    def width(self):
        return self._width

    def elaborate(self, platform):
        m = Module()

        width = self._width
        enable = self.enable
        irqs = self.irqs

        clear = Signal(width, name="clear_mask")
        vector = Signal(width, name="vector_reg")

        # Memorize and clear IRQ Vector
        m.d.sync += vector.eq(enable & ((vector & ~clear) | irqs))

        m.d.comb += self.vector.eq(vector)

        # Generate MSI — priority encoder (lower index = higher priority)
        # Use reversed iteration so lower indices override (last assignment wins)
        for i in reversed(range(width)):
            with m.If(vector[i]):
                m.d.comb += [
                    self.source.valid.eq(1),
                    self.source.payload.dat.eq(i),
                ]
                with m.If(self.source.valid & self.source.ready):
                    m.d.comb += clear.eq(1 << i)

        return m


class PCIeMSIX(Component):
    """MSI-X controller via TLP writes to host memory.

    Uses a memory-mapped table (per-vector address/data/mask) and issues
    TLP memory writes through a crossbar master port.

    This module requires an endpoint to be passed at construction time
    so it can allocate a master port on the crossbar.

    Parameters
    ----------
    width : :class:`int`
        Number of IRQ sources (default 32, max 64).

    Ports
    -----
    irqs : In(width)
        One bit per IRQ source.
    enable : In(width)
        Per-IRQ enable mask.
    pba : Out(width)
        Pending Bit Array (current pending vector).
    """

    def __init__(self, width=32):
        assert width <= 64
        self._width = width
        super().__init__({
            "irqs":   In(width),
            "enable": In(width),
            "pba":    Out(width),
        })

    @property
    def width(self):
        return self._width

    def elaborate(self, platform):
        m = Module()

        width = self._width
        enable = self.enable
        irqs = self.irqs

        clear = Signal(width, name="clear_mask")
        vector = Signal(width, name="vector_reg")

        # Memorize and clear IRQ Vector
        m.d.sync += vector.eq(enable & ((vector & ~clear) | irqs))
        m.d.comb += self.pba.eq(vector)

        # Priority encoder for MSI-X
        msix_valid = Signal(name="msix_valid")
        msix_num = Signal(range(max(width, 2)), name="msix_num")
        msix_clear = Signal(width, name="msix_clear")

        for i in reversed(range(width)):
            with m.If(vector[i]):
                m.d.comb += [
                    msix_valid.eq(1),
                    msix_num.eq(i),
                    msix_clear.eq(1 << i),
                ]

        # MSI-X Table (per-vector: address[32] + data[32] + mask[1])
        # Stored as 128-bit entries: [0:1]=mask, [32:64]=data, [96:128]=address
        # Using a simple register array for the table
        table_entries = Array([Signal(128, name=f"table_{i}", init=0b1)
                               for i in range(width)])

        # Table read port
        table_data = Signal(128, name="table_data")
        m.d.comb += table_data.eq(table_entries[msix_num])

        msix_adr = table_data[96:128]   # Lower Address
        msix_dat = table_data[32:64]    # Message Data
        msix_mask = table_data[0]       # Mask bit

        # FSM for issuing MSI-X TLP writes
        # Note: The actual TLP write port connection must be done externally
        # by connecting self.req_source to a crossbar master port.
        # Here we just generate the request stream.

        msix_clear_on_ready = Signal(width, name="msix_clear_on_ready")

        # Create a request-like output for MSI-X TLP writes
        # This is a simplified version — the full version would use
        # a crossbar master port. For now, expose the write request
        # as signals that can be connected externally.
        self.msix_wr_valid = Signal(name="msix_wr_valid")
        self.msix_wr_ready = Signal(name="msix_wr_ready")
        self.msix_wr_adr = Signal(32, name="msix_wr_adr")
        self.msix_wr_dat = Signal(32, name="msix_wr_dat")

        m.d.comb += [
            self.msix_wr_adr.eq(msix_adr),
            self.msix_wr_dat.eq(msix_dat),
        ]

        with m.FSM(name="msix_fsm"):
            with m.State("IDLE"):
                with m.If(msix_valid):
                    m.d.sync += msix_clear_on_ready.eq(msix_clear)
                    m.next = "ISSUE-WRITE"

            with m.State("ISSUE-WRITE"):
                m.d.comb += self.msix_wr_valid.eq(~msix_mask)
                with m.If(self.msix_wr_ready | msix_mask):
                    m.d.comb += clear.eq(msix_clear_on_ready)
                    m.next = "IDLE"

        return m