Pipeline API

The zuspec-dataclasses pipeline API describes a synchronous, single-issue, in-order pipeline as a set of Python methods on a Component subclass. The synthesizer in zuspec-synth converts these methods into fully-synthesisable Verilog 2005.

Three decorators work together:

• @zdc.pipeline — Pipeline Root — marks the root data-flow method

• @zdc.stage — Stage Methods — marks each pipeline stage

• @zdc.sync — Synchronous FSMs — marks external synchronous FSMs that interact with stages

—

@zdc.pipeline — Pipeline Root

The pipeline root method describes a single transaction’s journey through all stages. The synthesizer repeats this path every cycle, inserting pipeline registers, stall logic, and forwarding muxes as needed.

@zdc.dataclass
class TwoStage(zdc.Component):
    clk:   zdc.clock
    rst_n: zdc.reset
    x:     zdc.u32

    @zdc.pipeline(clock="clk", reset="rst_n")
    def execute(self):
        (x,) = self.S1()
        self.S2(x)

    @zdc.stage
    def S1(self) -> (zdc.u32,):
        return (self.x,)

    @zdc.stage
    def S2(self, x: zdc.u32) -> ():
        pass

Each self.STAGE(args) call is a pipeline-stage invocation. Arguments are the live variables entering that stage; return values are the live variables leaving it. Variables that skip intermediate stages are auto-threaded — the synthesizer inserts a pipeline register at every stage boundary automatically.

@zdc.pipeline arguments

Argument	Type	Default	Description
clock	str	required	Clock field name on the component (e.g. "clk").
reset	str	required	Reset field name on the component (e.g. "rst_n").
forward	bool	True	Default hazard resolution: True = forward bypass mux, False = stall the pipeline.
no_forward	list[str]	[]	Per-signal override: signal names that should always use stall instead of forwarding, even when forward=True.

—

@zdc.stage — Stage Methods

Each stage is a def method decorated with @zdc.stage. The method receives data entering the stage and returns data leaving it. All register insertion, stall propagation, and valid-chain management is generated automatically.

@zdc.stage
def IF(self) -> (zdc.u32, zdc.u32):
    zdc.stage.stall(~self.imem_valid)   # hold stage until memory acks
    return (self.pc, self.imem_data)

@zdc.stage(no_forward=True)
def MEM(self, addr: zdc.u32, is_load: zdc.u1) -> (zdc.u32,):
    zdc.stage.stall(self.mem_req & ~self.mem_ack)
    return (self.mem_rdata,)

@zdc.stage arguments

Argument	Type	Default	Description
no_forward	bool	False	When True, exclude all of this stage’s outputs as forwarding sources. The synthesizer generates a stall for any hazard path that would otherwise produce a bypass mux. Use for stages whose output arrives late (e.g. a load from external memory).

All @zdc.stage methods must be plain def (not async def) and all parameters must carry type annotations so the synthesizer can determine pipeline register widths.

Stage DSL calls

These calls are synthesizer annotations — they are no-ops at Python runtime. The synthesizer recognises them by AST inspection and generates the corresponding hardware.

Call	Effect
zdc.stage.stall(cond)	Freeze this stage and all upstream stages while cond is true. The current stage stays valid (transaction is paused, not lost). The next stage receives a bubble.
zdc.stage.cancel(cond=True)	Discard the transaction in this stage (clear valid) without freezing upstream. Upstream stages continue to advance.
zdc.stage.flush(self.X, cond=True)	Invalidate the target stage X next cycle. Flush priority is higher than stall: a stalled stage can still be flushed.

stall always takes a positional condition. cancel and flush accept either a positional or cond= keyword:

# Inside a stage body
zdc.stage.stall(~self.imem_valid)          # positional
zdc.stage.stall(cond=~self.imem_valid)     # keyword — identical

if mispredicted:
    zdc.stage.flush(self.IF)               # cond=True implied by if-body
zdc.stage.flush(self.IF, cond=mispredicted)  # explicit — identical hardware

—

@zdc.sync — Synchronous FSMs

External FSMs that manage pipeline interactions (e.g. instruction-fetch, load-store units, interrupt controllers) are marked with @zdc.sync. They run every clock cycle and can observe stage-validity signals via DSL query functions.

@zdc.sync(clock="clk", reset="rst_n")
def fetch_ctrl(self):
    if zdc.stage.ready(self.IF):   # IF can accept a new instruction
        self.imem_req  = 1
        self.imem_addr = self.pc

@zdc.sync arguments

Argument	Type	Default	Description
clock	str	required	Clock field name.
reset	str	required	Reset field name.

Stage query functions (usable inside @zdc.sync and @zdc.stage bodies)

Query	Meaning / generated hardware
zdc.stage.valid(self.X)	Stage X holds a live transaction → X_valid flip-flop
zdc.stage.ready(self.X)	Stage X can accept a new transaction → (~X_valid | ~X_stalled) combinational wire
zdc.stage.stalled(self.X)	Stage X is currently stalled → X_stalled combinational wire

Flush from @zdc.sync

zdc.stage.flush(target, cond) may also be called from a @zdc.sync body. This lets interrupt controllers or exception handlers discard all in-flight pipeline transactions:

@zdc.sync(clock="clk", reset="rst_n")
def irq_ctrl(self):
    take_irq = self.irq & ~self.irq_masked
    if take_irq:
        zdc.stage.flush(self.S1)   # cond=True implied by if-body
        zdc.stage.flush(self.S2)
        zdc.stage.flush(self.S3)

—

Migration Guide — Old Sentinel API → New Method API

The original pipeline API used zdc.stage() sentinel objects inside the pipeline body:

# OLD API (deprecated) — sentinel-based
@zdc.pipeline(clock=lambda s: s.clk, reset=lambda s: s.rst_n,
              stages=["IF", "EX"])
def execute(self):
    IF = zdc.stage()
    pc = self.pc
    EX = zdc.stage()
    result = pc + 1

Replace it with the new method-per-stage API:

# NEW API — method-per-stage
@zdc.pipeline(clock="clk", reset="rst_n")
def execute(self):
    (pc,) = self.IF()
    self.EX(pc)

@zdc.stage
def IF(self) -> (zdc.u32,):
    return (self.pc,)

@zdc.stage
def EX(self, pc: zdc.u32) -> ():
    pass

Key changes:

• clock/reset change from lambda s: s.clk → "clk" (string field name).

• Remove stages= parameter; stage order is expressed by the call sequence.

• Each IF = zdc.stage() sentinel becomes a @zdc.stage method.

• Variables passed between stages become method parameters and return values.

• zdc.forward(...) / zdc.no_forward(...) helpers are replaced by the no_forward argument on @zdc.stage and @zdc.pipeline.

The old PipelineAnnotationPass in zuspec-synth is deprecated. The new PipelineFrontendPass handles components using the method-per-stage API.

—

Explicit Ports and Clock Domain (Recommended)

The preferred API for synthesisable pipelines uses explicit ingress/egress ports and a first-class ClockDomain field. This maps cleanly to RTL module ports and makes clock/reset relationships unambiguous.

Key building blocks

zdc.ClockDomain / zdc.clock_domain()

A field that bundles a clock and its reset. Pipelines use it with the clock_domain= decorator kwarg. In behavioral simulation it provides wait_cycle() / wait_cycles(n) for cycle-accurate modelling.

zdc.InPort[T] / zdc.in_port()

An ingress port. The pipeline calls await self.port.get() in the first stage to receive a value for each pipeline token.

zdc.OutPort[T] / zdc.out_port()

An egress port. The pipeline calls await self.port.put(value) in the last stage to emit the result.

@zdc.pipeline(clock_domain=lambda s: s.clk)

Binds the pipeline to a ClockDomain field instead of separate clock= / reset= lambdas.

Minimal example

import zuspec.dataclasses as zdc

@zdc.dataclass
class Adder(zdc.Component):
    # One clock domain drives the whole component
    clk: zdc.ClockDomain = zdc.clock_domain()

    # Ingress: two operands arrive together each cycle
    a_in: zdc.InPort[zdc.u32] = zdc.in_port()
    b_in: zdc.InPort[zdc.u32] = zdc.in_port()

    # Egress: sum leaves the pipeline after the last stage
    sum_out: zdc.OutPort[zdc.u32] = zdc.out_port()

    @zdc.pipeline(clock_domain=lambda s: s.clk)
    async def add(self):
        async with zdc.pipeline.stage() as FETCH:
            a = await self.a_in.get()
            b = await self.b_in.get()

        async with zdc.pipeline.stage() as EXEC:
            result: zdc.u32 = a + b

        async with zdc.pipeline.stage() as WB:
            await self.sum_out.put(result)

The synthesiser maps:

• clk → RTL module ports clk + rst_n

• a_in / b_in → input ports a_in / b_in

• sum_out → output port sum_out

• Each async with zdc.pipeline.stage() block → one pipeline stage register

Pattern constraints

The pipeline body must follow the GET … PUT pattern:

1. All await self.PORT.get() calls appear in the first stage (or early stages).

2. All await self.PORT.put(value) calls appear in the last stage.

3. No top-level while loop — the decorator implicitly repeats the body each cycle.

4. await self.clk.wait_cycle() (ClockDomain.wait_cycle()) can be used inside a stage to insert an explicit pipeline bubble.