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
 
# Hardware for the ChannelArbiter cosim integration tests. Builds two
# independent DUTs under a single top:
#
#   * ChannelArbiterTest ("arbiter_test"): NUM_INPUTS host-driven `from_host`
#     UInt(32) channels multiplexed into a single `to_host` channel. The host
#     checks that every value is delivered exactly once (no loss / duplication).
#     Instantiated twice: a power-of-two ("balanced") count and a
#     non-power-of-two ("unbalanced") count.
#
#   * ChannelArbiterListTest ("list_test"): two in-hardware list-window
#     producers (distinct `src` ids and message lengths) contend for the
#     arbiter. A hardware checker verifies that, once a message is granted, its
#     flits arrive contiguously (no interleaving) up to the flit tagged `last`,
#     and reports (sticky-error, src, count) per completed message on a
#     `to_host` channel.
#
#   * ChannelArbiterTokenTest ("token_test"): TOKEN_NUM_INPUTS in-hardware
#     zero-width (`i0`) token producers, each emitting exactly TOKENS_PER_INPUT
#     tokens, contend for the arbiter. A hardware counter reports a running
#     1-based ordinal per delivered token, so the host can confirm every token
#     is delivered exactly once (conservation) and every producer is served
#     (no starvation) -- exercising the credit-counter-as-buffer i0 path.
 
import sys
 
from pycde import (AppID, Clock, Input, Module, Output, Reset, System,
                   generator)
from pycde.constructs import Counter, Mux, Wire
from pycde.types import Bits, Channel, List as ListType, StructType, UInt, Window
from pycde import esi
 
from esiaccel.bsp import get_bsp
from esiaccel.components import ChannelArbiter
 
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.
TOKEN_NUM_INPUTS = 5  # input count for the zero-width (i0) token variant.
# Sustained-throughput probe: enough inputs that a per-sweep scheduler bubble
# would be clearly visible (it would cap throughput at n/(n+1) == 0.8) while
# keeping simulation time short.
THROUGHPUT_NUM_INPUTS = 4
THROUGHPUT_WINDOW = 1000  # measurement window, in cycles.
TOKENS_PER_INPUT = 8  # tokens each producer emits in the token test.
# A fan-in wide enough to exercise large-N arbitration. The BSP instantiates
# arbiters with ~31 inputs (one per host-memory write client), a regime none of
# the small counts above reach.
WIDE_NUM_INPUTS = 13
 
# A list-window payload: a struct with a `src` tag and a variable-length list.
# `Window.default_of` adds a per-flit `last` field to the lowered frame struct,
# which is exactly what `ChannelArbiter` uses to keep messages contiguous.
ListInto = StructType({'src': UInt(8), 'items': ListType(UInt(16))})
Flit = Window.default_of(ListInto)
FlitLowered = Flit.lowered_type  # struct<src: ui8, items: ui16, last: i1>
 
 
def HostMux(num_inputs: int,
            mux_pipeline_levels=None,
            pipelined_scheduler=False):
  """A host-driven single-flit multiplexer: `num_inputs` `from_host` UInt(32)
  channels muxed into a single `to_host` channel. `mux_pipeline_levels` pipelines
  the selection mux tree; `pipelined_scheduler` selects the decoupled
  grant-queue arbitration."""
 
  class HostMux(Module):
    clk = Clock()
    rst = Reset()
 
    @generator
    def build(ports):
      ins = [
          esi.ChannelService.from_host(AppID(f"in_{i}"), UInt(32))
          for i in range(num_inputs)
      ]
      out = ChannelArbiter(ins,
                           ports.clk,
                           ports.rst,
                           mux_pipeline_levels=mux_pipeline_levels,
                           pipelined_scheduler=pipelined_scheduler,
                           telemetry=False)
      esi.ChannelService.to_host(AppID("out"), out)
 
  HostMux.__name__ = (
      f"HostMux_{num_inputs}_{mux_pipeline_levels}_{pipelined_scheduler}")
  return HostMux
 
 
def ListProducer(src_id: int, length: int):
  """A module which continuously emits back-to-back list messages of `length`
  flits tagged with `src_id`. `items` counts 0..length-1 and `last` is set on
  the final flit. Always valid, so two of these contend for the arbiter."""
 
  class ListProducer(Module):
    clk = Clock()
    rst = Reset()
    out = Output(Channel(Flit))
 
    @generator
    def build(ports):
      i = Wire(UInt(8))
      last = i == UInt(8)(length - 1)
      st = FlitLowered({
          'src': UInt(8)(src_id),
          'items': i.as_uint(16),
          'last': last,
      })
      chan, ready = Channel(Flit).wrap(Flit.wrap(st), Bits(1)(1))
      ports.out = chan
      # valid is constant 1, so a transaction happens whenever `ready`.
      nxt = Mux(last, (i + UInt(8)(1)).as_uint(8), UInt(8)(0))
      i.assign(nxt.reg(ports.clk, ports.rst, ce=ready, rst_value=0))
 
  ListProducer.__name__ = f"ListProducer_src{src_id}_len{length}"
  return ListProducer
 
 
class ListChecker(Module):
  """Consumes the muxed list-window stream and verifies message contiguity.
 
  Emits one UInt(32) report per completed message:
    bit  24    : sticky interleave-error flag (should stay 0)
    bits 23:16 : src of the completed message
    bits 15:0  : running completed-message count
  """
 
  clk = Clock()
  rst = Reset()
  in_ = Input(Channel(Flit))
  report = Output(Channel(UInt(32)))
 
  @generator
  def build(ports):
    active = Wire(Bits(1))
    err = Wire(Bits(1))
 
    in_ready = Wire(Bits(1))
    win, valid = ports.in_.unwrap(in_ready)
    st = win.unwrap()
    src = st['src']
    last = st['last']
 
    # A message boundary: the current flit completes a message.
    is_completing = valid & last
    # Emit a report exactly when a message completes; back-pressure the input
    # on that flit until the report is accepted.
    report_valid = is_completing
 
    # src of the in-progress message, latched at its first flit.
    start_any = valid & in_ready & ~active
    cur_src = src.reg(ports.clk, ports.rst, ce=start_any, rst_value=0)
 
    # Interleave error: a flit whose src differs from the owner mid-message.
    interleave_err = (valid & in_ready) & active & (src != cur_src)
    err.assign((err | interleave_err).reg(ports.clk, ports.rst, rst_value=0))
 
    # `active` tracks whether we are mid-message (past the first flit, before
    # `last`).
    xact = valid & in_ready
    begin_multi = xact & ~active & ~last
    end_msg = xact & last
    active_next = Mux(end_msg, Mux(begin_multi, active, Bits(1)(1)), Bits(1)(0))
    active.assign(active_next.reg(ports.clk, ports.rst, rst_value=0))
 
    # Completed-message counter.
    msg_count = Counter(16)(clk=ports.clk,
                            rst=ports.rst,
                            clear=Bits(1)(0),
                            increment=end_msg)
 
    report_data = ((err.as_uint(32) * UInt(32)(0x1000000)).as_uint(32) +
                   (cur_src.as_uint(32) * UInt(32)(0x10000)).as_uint(32) +
                   msg_count.out.as_uint(32)).as_uint(32)
    report_chan, report_ready = Channel(UInt(32)).wrap(report_data,
                                                       report_valid)
    ports.report = report_chan
 
    # Accept every non-completing flit; on a completing flit, only accept when
    # the report is accepted so no completion is dropped.
    in_ready.assign(Mux(is_completing, Bits(1)(1), report_ready))
 
 
def ChannelArbiterListTestMod(pipelined_scheduler: bool):
  """Contending list producers -> arbiter -> contiguity checker. Message
  atomicity is the property most at risk from any arbitration change, so it is
  covered for both arbitration modes."""
 
  class ChannelArbiterListTest(Module):
    clk = Clock()
    rst = Reset()
 
    @generator
    def build(ports):
      # More producers than the grant-queue depth, so the scheduled variant
      # exercises a queue that actually fills.
      prods = [
          ListProducer(src, 3 + (src % 3))(clk=ports.clk, rst=ports.rst)
          for src in range(1, 7)
      ] if pipelined_scheduler else [
          ListProducer(1, 3)(clk=ports.clk, rst=ports.rst),
          ListProducer(2, 4)(clk=ports.clk, rst=ports.rst),
      ]
      muxed = ChannelArbiter([p.out for p in prods],
                             ports.clk,
                             ports.rst,
                             pipelined_scheduler=pipelined_scheduler,
                             telemetry=False)
      chk = ListChecker(clk=ports.clk, rst=ports.rst, in_=muxed)
      esi.ChannelService.to_host(AppID("report"), chk.report)
 
  ChannelArbiterListTest.__name__ = (
      f"ChannelArbiterListTest_{pipelined_scheduler}")
  return ChannelArbiterListTest
 
 
def TokenProducer(count: int):
  """Emits exactly `count` zero-width (`i0`) tokens then idles: `valid` stays
  high until `count` tokens have been accepted. There is no payload -- only the
  valid/ready handshake carries information."""
 
  class TokenProducer(Module):
    clk = Clock()
    rst = Reset()
    out = Output(Channel(Bits(0)))
 
    @generator
    def build(ports):
      xact = Wire(Bits(1))
      sent = Counter(16)(clk=ports.clk,
                         rst=ports.rst,
                         clear=Bits(1)(0),
                         increment=xact)
      valid = sent.out < UInt(16)(count)
      chan, ready = Channel(Bits(0)).wrap(Bits(0)(0), valid)
      ports.out = chan
      xact.assign(valid & ready)
 
  TokenProducer.__name__ = f"TokenProducer_{count}"
  return TokenProducer
 
 
class TokenChecker(Module):
  """Consumes the muxed zero-width token stream and emits one UInt(32) report
  per delivered token carrying its 1-based ordinal (1, 2, 3, ...). Backpressure
  from the report channel is fed to the arbiter, exercising its credit buffer.
  """
 
  clk = Clock()
  rst = Reset()
  in_ = Input(Channel(Bits(0)))
  report = Output(Channel(UInt(32)))
 
  @generator
  def build(ports):
    in_ready = Wire(Bits(1))
    _tok, valid = ports.in_.unwrap(in_ready)  # zero-width payload: ignore data.
 
    # Running count of delivered tokens; this token's ordinal is count + 1.
    count = Counter(32)(clk=ports.clk,
                        rst=ports.rst,
                        clear=Bits(1)(0),
                        increment=valid & in_ready)
    report_data = (count.out + UInt(32)(1)).as_uint(32)
    report_chan, report_ready = Channel(UInt(32)).wrap(report_data, valid)
    ports.report = report_chan
    # Accept a token exactly when its report is consumed by the host.
    in_ready.assign(report_ready)
 
 
class ChannelArbiterTokenTest(Module):
  """`TOKEN_NUM_INPUTS` bounded zero-width token producers -> arbiter -> token
  counter. Each producer emits exactly `TOKENS_PER_INPUT` tokens, so the host
  must see exactly TOKEN_NUM_INPUTS * TOKENS_PER_INPUT tokens: no loss or
  duplication (conservation), and every producer served (no starvation, since
  the total can only be reached if each producer's tokens all get through)."""
 
  clk = Clock()
  rst = Reset()
 
  @generator
  def build(ports):
    prods = [
        TokenProducer(TOKENS_PER_INPUT)(clk=ports.clk, rst=ports.rst)
        for _ in range(TOKEN_NUM_INPUTS)
    ]
    muxed = ChannelArbiter([p.out for p in prods],
                           ports.clk,
                           ports.rst,
                           telemetry=False)
    chk = TokenChecker(clk=ports.clk, rst=ports.rst, in_=muxed)
    esi.ChannelService.to_host(AppID("token_report"), chk.report)
 
 
def AlwaysValidProducer(tag: int):
  """Never idles: `valid` is tied high, so the only thing limiting the
  arbiter's delivery rate is the arbiter itself."""
 
  class AlwaysValidProducer(Module):
    clk = Clock()
    rst = Reset()
    out = Output(Channel(UInt(32)))
 
    @generator
    def build(ports):
      chan, _ready = Channel(UInt(32)).wrap(UInt(32)(tag), Bits(1)(1))
      ports.out = chan
 
  AlwaysValidProducer.__name__ = f"AlwaysValidProducer_{tag}"
  return AlwaysValidProducer
 
 
class ThroughputProbe(Module):
  """Drains the arbiter at full rate (`ready` tied high, so the host can never
  backpressure it) and counts delivered beats over a fixed cycle window. Holds
  the tally on its report channel once the window closes.
 
  Draining in hardware is the point: the host-driven tests are rate-limited by
  the cosim DPI, so they cannot observe sustained throughput at all."""
 
  clk = Clock()
  rst = Reset()
  in_ = Input(Channel(UInt(32)))
  report = Output(Channel(UInt(32)))
 
  @generator
  def build(ports):
    _data, valid = ports.in_.unwrap(Bits(1)(1))  # never backpressure.
    cycles = Counter(32)(clk=ports.clk,
                         rst=ports.rst,
                         clear=Bits(1)(0),
                         increment=Bits(1)(1))
    running = cycles.out < UInt(32)(THROUGHPUT_WINDOW)
    beats = Counter(32)(clk=ports.clk,
                        rst=ports.rst,
                        clear=Bits(1)(0),
                        increment=valid & running)
    # Report only once the window has closed; the value is then stable, so the
    # host can read it whenever it gets around to it.
    chan, _ready = Channel(UInt(32)).wrap(beats.out, ~running)
    ports.report = chan
 
 
def ChannelArbiterThroughputTestMod(pipelined_scheduler: bool):
  """`THROUGHPUT_NUM_INPUTS` never-idle producers -> arbiter -> throughput
  probe. Pins the arbiter's sustained delivery rate: a scheduler which needs a
  refill/turnaround cycle between grants shows up here as a throughput well
  below one beat per cycle, while correctness tests stay green."""
 
  class ChannelArbiterThroughputTest(Module):
    clk = Clock()
    rst = Reset()
 
    @generator
    def build(ports):
      prods = [
          AlwaysValidProducer(i)(clk=ports.clk, rst=ports.rst)
          for i in range(THROUGHPUT_NUM_INPUTS)
      ]
      muxed = ChannelArbiter([p.out for p in prods],
                             ports.clk,
                             ports.rst,
                             pipelined_scheduler=pipelined_scheduler,
                             telemetry=False)
      probe = ThroughputProbe(clk=ports.clk, rst=ports.rst, in_=muxed)
      esi.ChannelService.to_host(AppID("throughput_report"), probe.report)
 
  ChannelArbiterThroughputTest.__name__ = (
      f"ChannelArbiterThroughputTest_{pipelined_scheduler}")
  return ChannelArbiterThroughputTest
 
 
class Top(Module):
  clk = Clock()
  rst = Reset()
 
  @generator
  def construct(ports):
    HostMux(NUM_INPUTS)(clk=ports.clk,
                        rst=ports.rst,
                        appid=AppID("arbiter_test"))
    HostMux(ODD_NUM_INPUTS)(clk=ports.clk,
                            rst=ports.rst,
                            appid=AppID("arbiter_test_odd"))
    HostMux(PIPE_NUM_INPUTS,
            mux_pipeline_levels=1)(clk=ports.clk,
                                   rst=ports.rst,
                                   appid=AppID("arbiter_test_pipe"))
    HostMux(WIDE_NUM_INPUTS)(clk=ports.clk,
                             rst=ports.rst,
                             appid=AppID("arbiter_test_wide"))
    HostMux(WIDE_NUM_INPUTS,
            pipelined_scheduler=True)(clk=ports.clk,
                                      rst=ports.rst,
                                      appid=AppID("arbiter_test_sched"))
    HostMux(ODD_NUM_INPUTS,
            pipelined_scheduler=True)(clk=ports.clk,
                                      rst=ports.rst,
                                      appid=AppID("arbiter_test_sched_odd"))
    ChannelArbiterListTestMod(False)(clk=ports.clk,
                                     rst=ports.rst,
                                     appid=AppID("list_test"))
    ChannelArbiterListTestMod(True)(clk=ports.clk,
                                    rst=ports.rst,
                                    appid=AppID("list_test_sched"))
    ChannelArbiterTokenTest(clk=ports.clk,
                            rst=ports.rst,
                            appid=AppID("token_test"))
    ChannelArbiterThroughputTestMod(False)(clk=ports.clk,
                                           rst=ports.rst,
                                           appid=AppID("throughput_test"))
    ChannelArbiterThroughputTestMod(True)(clk=ports.clk,
                                          rst=ports.rst,
                                          appid=AppID("throughput_test_sched"))
 
 
if __name__ == "__main__":
  bsp = get_bsp(sys.argv[2] if len(sys.argv) > 2 else None)
  s = System(bsp(Top), name="ChannelArbiterTest", output_directory=sys.argv[1])
  s.compile()
  s.package()