test_channel_arbiter.py

Go to the documentation of this file.

#  Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
#  See https://llvm.org/LICENSE.txt for license information.
#  SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
"""Cosim integration tests for `esiaccel.components.ChannelArbiter`.
 
Runs under Verilator via the `@cosim_test` harness. Two DUTs (built by
`hw/channel_arbiter.py`) are exercised:
 
  * `arbiter_test` / `arbiter_test_odd`: host-driven single-flit multiplexers
    with a power-of-two ('balanced') and non-power-of-two ('unbalanced') input
    count. The host writes distinct tagged values to each input and checks that
    every value comes out of the single output exactly once (no loss /
    duplication) and that every input is served.
 
  * `list_test`: two in-hardware list-window producers contend for the
    arbiter; a hardware checker verifies message contiguity and streams a
    per-message report back to the host.
 
  * `token_test`: several in-hardware zero-width (`i0`) token producers, each
    emitting a fixed number of tokens, contend for the arbiter. A hardware
    counter reports a running ordinal per delivered token so the host can
    confirm every token is delivered exactly once and every producer is served.
"""
 
from __future__ import annotations
 
from collections import Counter
from pathlib import Path
 
import esiaccel
from esiaccel.accelerator import AcceleratorConnection
from esiaccel.cosim.pytest import cosim_test
 
HW_DIR = Path(__file__).resolve().parent / "hw"
 
NUM_INPUTS = 4  # power-of-two ("balanced") input count.
ODD_NUM_INPUTS = 3  # non-power-of-two ("unbalanced") input count.
PIPE_NUM_INPUTS = 6  # input count for the pipelined-mux-tree variant.
WIDE_NUM_INPUTS = 13  # wide fan-in; must match hw/channel_arbiter.py.
WIDE_FANIN_NUM_INPUTS = 17  # above the wide-fan-in threshold; must match hw.
TOKEN_NUM_INPUTS = 5  # input count for the zero-width (i0) token variant.
TOKENS_PER_INPUT = 8  # tokens each producer emits in the token test.
THROUGHPUT_NUM_INPUTS = 4  # must match hw/channel_arbiter.py.
THROUGHPUT_WINDOW = 1000  # measurement window in cycles; must match hw.
 
 
def _check_mux(conn: AcceleratorConnection,
               dut_name: str,
               num_inputs: int,
               max_in_flight: int = 2) -> None:
  """Drive an N-input host mux and check every value is delivered exactly
  once and every input is served."""
  acc = conn.build_accelerator()
  dut = acc.children[esiaccel.AppID(dut_name)]
  ins = [dut.ports[esiaccel.AppID(f"in_{i}")] for i in range(num_inputs)]
  out = dut.ports[esiaccel.AppID("out")]
  for p in ins:
    p.connect()
  out.connect()
 
  # The value encodes its source input in the upper bits (i << 16) and a round
  # counter in the low bits, so the received multiset uniquely identifies every message.
  rounds = 6
  writes = [
      ((i << 16) | r, ins[i]) for r in range(rounds) for i in range(num_inputs)
  ]
 
  # Keep the number of in-flight (written-but-not-yet-read) messages strictly
  # below the output FIFO depth. This keeps a couple of inputs backlogged at
  # once (so the round-robin arbiter has to choose between them) while
  # avoiding the write-a-burst-before-reading deadlock.
  sent: list[int] = []
  recv: list[int] = []
  wi = 0
  while len(recv) < len(writes):
    while wi < len(writes) and (len(sent) - len(recv)) < max_in_flight:
      value, port = writes[wi]
      port.write(value)
      sent.append(value)
      wi += 1
    recv.append(out.read().result())
 
  assert sorted(recv) == sorted(sent), \
      "arbiter dropped, duplicated or corrupted a value"
 
  # Every input (decoded by v >> 16) is served exactly `rounds` times.
  by_src = Counter(v >> 16 for v in recv)
  for i in range(num_inputs):
    assert by_src[i] == rounds, \
        f"input {i} served {by_src[i]} times, expected {rounds}"
 
 
@cosim_test(HW_DIR / "channel_arbiter.py")
class TestChannelArbiterCosim:
 
  def test_mux_correctness(self, conn: AcceleratorConnection) -> None:
    """Balanced (power-of-two) input count: every value appears once."""
    _check_mux(conn, "arbiter_test", NUM_INPUTS)
 
  def test_mux_correctness_unbalanced(self,
                                      conn: AcceleratorConnection) -> None:
    """Unbalanced (non-power-of-two) input count: the array-indexed mux over
    N < 2**clog2(N) elements and the round-robin wrap still deliver every
    value exactly once."""
    _check_mux(conn, "arbiter_test_odd", ODD_NUM_INPUTS)
 
  def test_mux_correctness_pipelined(self, conn: AcceleratorConnection) -> None:
    """Pipelined selection mux tree: the multi-cycle mux latency (absorbed by a
    deeper output FIFO + credit counter) must still deliver every value exactly
    once."""
    _check_mux(conn, "arbiter_test_pipe", PIPE_NUM_INPUTS)
 
  def test_mux_correctness_wide(self, conn: AcceleratorConnection) -> None:
    """Wide fan-in: the BSP instantiates arbiters with ~31 inputs, a regime the
    small counts above never reach. Every value must still be delivered exactly
    once and every input served."""
    _check_mux(conn, "arbiter_test_wide", WIDE_NUM_INPUTS)
 
  def test_mux_correctness_scheduled(self, conn: AcceleratorConnection) -> None:
    """Decoupled grant-queue scheduler at wide fan-in: grants are chosen ahead
    of time and buffered, so this exercises the queue, the sweep reload and the
    stale-entry skip. Delivery must still be exactly-once and every input
    served."""
    _check_mux(conn, "arbiter_test_sched", WIDE_NUM_INPUTS)
 
  def test_mux_correctness_scheduled_unbalanced(
      self, conn: AcceleratorConnection) -> None:
    """The scheduler with a non-power-of-two input count: the one-hot->index
    encode and the sweep must not produce an out-of-range grant."""
    _check_mux(conn, "arbiter_test_sched_odd", ODD_NUM_INPUTS)
 
  def test_mux_correctness_wide_fanin(self,
                                      conn: AcceleratorConnection) -> None:
    """`wide_fanin` structures, with a throttled output so credits run out."""
    _check_mux(conn, "arbiter_test_widefanin", WIDE_FANIN_NUM_INPUTS, 8)
 
  def test_mux_correctness_wide_fanin_scheduled(
      self, conn: AcceleratorConnection) -> None:
    """As above, under the grant-queue scheduler."""
    _check_mux(conn, "arbiter_test_widefanin_sched", WIDE_FANIN_NUM_INPUTS, 8)
 
  def test_mux_correctness_wide_fanin_forced_on(
      self, conn: AcceleratorConnection) -> None:
    """`wide_fanin=True` below the threshold."""
    _check_mux(conn, "arbiter_test_forced_on", NUM_INPUTS)
 
  def test_mux_correctness_wide_fanin_forced_off(
      self, conn: AcceleratorConnection) -> None:
    """`wide_fanin=False` above the threshold."""
    _check_mux(conn, "arbiter_test_forced_off", WIDE_FANIN_NUM_INPUTS)
 
  def test_list_contiguity(self, conn: AcceleratorConnection) -> None:
    """Contending multi-flit list messages are never interleaved."""
    self._check_contiguity(conn, "list_test", {1, 2})
 
  def test_list_contiguity_scheduled(self, conn: AcceleratorConnection) -> None:
    """Message atomicity under the decoupled grant-queue scheduler, with more
    contending producers than the grant queue is deep."""
    self._check_contiguity(conn, "list_test_sched", set(range(1, 7)))
 
  def test_list_contiguity_wide_fanin(self,
                                      conn: AcceleratorConnection) -> None:
    """Contiguity with `wide_fanin` forced on."""
    self._check_contiguity(conn, "list_test_widefanin", {1, 2})
 
  def test_list_contiguity_wide_fanin_scheduled(
      self, conn: AcceleratorConnection) -> None:
    """As above, under the grant-queue scheduler."""
    self._check_contiguity(conn, "list_test_widefanin_sched", set(range(1, 7)))
 
  @staticmethod
  def _check_contiguity(conn: AcceleratorConnection, dut_name: str,
                        expected_src: set[int]) -> None:
    acc = conn.build_accelerator()
    dut = acc.children[esiaccel.AppID(dut_name)]
    report = dut.ports[esiaccel.AppID("report")]
    report.connect()
 
    seen_src: set[int] = set()
    # Enough reports to see every producer served several times over.
    num_reports = 20 * len(expected_src)
    for _ in range(num_reports):
      value = report.read().result()
      err = (value >> 24) & 0x1
      src = (value >> 16) & 0xff
      assert err == 0, \
          f"hardware detected interleaved list flits (report {value:#010x})"
      seen_src.add(src)
 
    # Every contending producer must get through -- i.e. no starvation.
    assert seen_src == expected_src, \
        f"expected sources {sorted(expected_src)}, saw {sorted(seen_src)}"
 
  def test_token_conservation(self, conn: AcceleratorConnection) -> None:
    """Zero-width (`i0`) token payloads: the credit-counter-as-buffer path
    delivers every token exactly once (no loss/duplication/reorder) and serves
    every producer -- the total is only reachable if no producer is starved."""
    acc = conn.build_accelerator()
    dut = acc.children[esiaccel.AppID("token_test")]
    report = dut.ports[esiaccel.AppID("token_report")]
    report.connect()
 
    total = TOKEN_NUM_INPUTS * TOKENS_PER_INPUT
    for expected in range(1, total + 1):
      value = report.read().result()
      assert value == expected, \
          f"token {expected} arrived as {value} (loss / duplication / reorder)"
 
  def test_throughput_flat(self, conn: AcceleratorConnection) -> None:
    """The flat round-robin arbiter sustains ~one beat per cycle."""
    self._check_throughput(conn, "throughput_test")
 
  def test_throughput_scheduled(self, conn: AcceleratorConnection) -> None:
    """The decoupled grant-queue scheduler must sustain ~one beat per cycle
    too. Regression test: reloading the sweep snapshot a cycle after it drains
    (rather than on the cycle the last entry is queued) costs one idle cycle
    per sweep, which caps throughput at `n/(n+1)` -- 0.8 here. Correctness
    tests do not notice that, only this one does.
 
    The producers emit **single-flit** messages deliberately: with multi-flit
    lists the datapath keeps streaming while the sweep refills, which hides the
    bubble entirely (measured: no loss at list length >= 2)."""
    self._check_throughput(conn, "throughput_test_sched")
 
  @staticmethod
  def _check_throughput(conn: AcceleratorConnection, dut_name: str) -> None:
    acc = conn.build_accelerator()
    dut = acc.children[esiaccel.AppID(dut_name)]
    report = dut.ports[esiaccel.AppID("throughput_report")]
    report.connect()
 
    beats = report.read().result()
    throughput = beats / THROUGHPUT_WINDOW
    # Allow for pipeline fill at the start of the window; well above the 0.8
    # a per-sweep bubble would produce with this input count.
    assert throughput >= 0.95, (
        f"{dut_name}: {beats} beats in {THROUGHPUT_WINDOW} cycles "
        f"({throughput:.3f}/cycle); expected >= 0.95. A throughput of about "
        f"{THROUGHPUT_NUM_INPUTS / (THROUGHPUT_NUM_INPUTS + 1):.3f} means the "
        "scheduler is losing a cycle per sweep.")