Part VII: Hardware Meets Software | Chapter 22

Capstone: A Hardware Accelerator

A streaming dot-product engine fed by DMA, measured against NumPy on the ARM cores; when acceleration pays and when it does not; where to go next.

1. The goal: an accelerator you can measure

This chapter puts everything together. You will design a streaming dot-product engine in Verilog and VHDL, verify it under random back-pressure, place it between the two channels of an AXI DMA, and drive it from a Jupyter notebook with NumPy arrays. Then you will do the most important thing an engineer can do with an accelerator: measure it honestly against the CPU — and understand why, for this particular kernel, the CPU may well win.

The dot product of two vectors \(\mathbf{x}, \mathbf{y}\) of length \(L\) is\(\;\sum_{i=0}^{L-1} x_i y_i\), one multiply-accumulate (MAC) per element pair. It is the heart of filters (Chapter 18), matrix products and neural-network layers, and it maps perfectly onto one DSP48E1 slice.

The capstone system. AXI DMA reads interleaved (x, y) pairs from DDR through HP0 and streams them (MM2S) into dot_engine; one 64-bit result per vector streams back (S2MM) into a second buffer. An AXI GPIO on GP0 sets the vector length. Python only sets up buffers and starts transfers.

The architecture decision, made once and kept: data in by MM2S, results out by S2MM, both as AXI-Stream. Results could instead be read from an AXI-Lite register, but that costs one slow MMIO access per vector (Chapter 21); streaming them back lets one DMA call process thousands of vectors.

2. Designing the engine

Input format. Each 32-bit beat carries one element pair: \(x_i\) in bits [15:0] and\(y_i\) in bits [31:16], both signed 16-bit (two's complement, Chapter 1). Interleaving the two vectors means one DMA channel is enough and the engine never waits for one operand while holding the other.

Where does a vector end? AXI-Stream marks the end of a packet with TLAST. The AXI DMA raises TLAST on the last beat of each transfer, so one transfer per vector works — but costs the full Python overhead per vector. To batch \(M\) vectors in one transfer, the engine also accepts a length vlen from a register: a vector ends after vlen beats or on TLAST. With vlen = 0, only TLAST counts.

Pipeline. Stage 1 multiplies; stage 2 adds the product to a 48-bit accumulator. On the last beat of a vector, stage 2 outputs the final sum on the master stream and clears the accumulator, so the next vector starts without a gap. Both stages map onto one DSP48E1 slice, which has exactly a 25×18 multiplier, a pipeline register after it, and a 48-bit accumulator.

Inside dot_engine. All registers advance together when adv = !m_tvalid || m_tready; when the output is stalled, the whole pipeline (and s_tready) freezes, so no data is lost and nothing changes on a stalled output.

How wide must the accumulator be? The largest product magnitude is\((-2^{15})^2 = 2^{30}\), so a vector of length \(L\) can reach \(L \cdot 2^{30}\):

\[ L \cdot 2^{30} \le 2^{47} - 1 \quad\Longrightarrow\quad L < 2^{17} = 131\,072 \text{ elements without overflow in 48 bits} \]

A 32-bit accumulator would overflow after only two worst-case products. The result is sign-extended to 64 bits so that NumPy can read it directly as int64.

3. The dot-product engine in Verilog and VHDL

dot_engine — streaming multiply-accumulate with TLAST/vlen framing

Verilog

// Streaming dot-product engine.
// Input : one 32-bit AXI-Stream beat per element pair, x = tdata[15:0],
//         y = tdata[31:16], both signed 16-bit.
// A vector ends on s_axis_tlast, or after vlen beats if vlen != 0
// (AXI DMA raises TLAST only at the end of a whole transfer, so a batch of
// many vectors in one transfer needs vlen).
// Output: one 64-bit beat per vector, the 48-bit sum sign-extended;
//         m_axis_tlast is set on the result of the vector that ended with
//         s_axis_tlast, which closes the S2MM transfer.
module dot_engine (
    input  wire        clk,
    input  wire        rst,
    input  wire [15:0] vlen,              // 0 = use TLAST only
    input  wire        s_axis_tvalid,
    output wire        s_axis_tready,
    input  wire [31:0] s_axis_tdata,
    input  wire        s_axis_tlast,
    output reg         m_axis_tvalid = 1'b0,
    input  wire        m_axis_tready,
    output reg  [63:0] m_axis_tdata  = 64'd0,
    output reg         m_axis_tlast  = 1'b0
);
    // The whole pipeline moves only when the output slot is free or being
    // emptied; otherwise everything holds (AXI rule: keep data until taken).
    wire adv     = !m_axis_tvalid || m_axis_tready;
    wire in_xfer = s_axis_tvalid && adv;
    assign s_axis_tready = adv;

    wire signed [15:0] x = s_axis_tdata[15:0];
    wire signed [15:0] y = s_axis_tdata[31:16];

    reg [15:0] cnt = 16'd0;               // beats so far in this vector
    wire vec_end = s_axis_tlast || (vlen != 16'd0 && cnt == vlen - 16'd1);

    // stage 1: multiply (maps to the DSP48E1 multiplier + its M register)
    reg signed [31:0] prod    = 32'sd0;
    reg               p_valid = 1'b0, p_end = 1'b0, p_last = 1'b0;
    // stage 2: accumulate (48 bits, like the DSP48E1 accumulator)
    reg  signed [47:0] acc = 48'sd0;
    wire signed [47:0] sum = acc + prod;

    always @(posedge clk) begin
        if (rst) begin
            cnt           <= 16'd0;
            p_valid       <= 1'b0;
            acc           <= 48'sd0;
            m_axis_tvalid <= 1'b0;
        end else if (adv) begin
            p_valid <= in_xfer;
            prod    <= x * y;
            p_end   <= vec_end;
            p_last  <= s_axis_tlast;
            if (in_xfer)
                cnt <= vec_end ? 16'd0 : cnt + 16'd1;

            m_axis_tvalid <= 1'b0;
            if (p_valid) begin
                if (p_end) begin
                    m_axis_tdata  <= {{16{sum[47]}}, sum};
                    m_axis_tvalid <= 1'b1;
                    m_axis_tlast  <= p_last;
                    acc           <= 48'sd0;
                end else
                    acc <= sum;
            end
        end
    end
endmodule

VHDL

-- Streaming dot-product engine.
-- Input : one 32-bit AXI-Stream beat per element pair, x = tdata(15:0),
--         y = tdata(31:16), both signed 16-bit.
-- A vector ends on s_axis_tlast, or after vlen beats if vlen /= 0
-- (AXI DMA raises TLAST only at the end of a whole transfer, so a batch of
-- many vectors in one transfer needs vlen).
-- Output: one 64-bit beat per vector, the 48-bit sum sign-extended;
--         m_axis_tlast is set on the result of the vector that ended with
--         s_axis_tlast, which closes the S2MM transfer.
library ieee;
use ieee.std_logic_1164.all;
use ieee.numeric_std.all;

entity dot_engine is
    port (
        clk           : in  std_logic;
        rst           : in  std_logic;
        vlen          : in  std_logic_vector(15 downto 0);   -- 0 = use TLAST only
        s_axis_tvalid : in  std_logic;
        s_axis_tready : out std_logic;
        s_axis_tdata  : in  std_logic_vector(31 downto 0);
        s_axis_tlast  : in  std_logic;
        m_axis_tvalid : out std_logic;
        m_axis_tready : in  std_logic;
        m_axis_tdata  : out std_logic_vector(63 downto 0);
        m_axis_tlast  : out std_logic
    );
end entity;

architecture rtl of dot_engine is
    signal adv, in_xfer, vec_end : std_logic;
    signal x, y    : signed(15 downto 0);
    signal cnt     : unsigned(15 downto 0) := (others => '0');
    -- stage 1: multiply (maps to the DSP48E1 multiplier + its M register)
    signal prod    : signed(31 downto 0) := (others => '0');
    signal p_valid, p_end, p_last : std_logic := '0';
    -- stage 2: accumulate (48 bits, like the DSP48E1 accumulator)
    signal acc     : signed(47 downto 0) := (others => '0');
    signal sum     : signed(47 downto 0);
    signal o_valid, o_last : std_logic := '0';
    signal o_data  : signed(63 downto 0) := (others => '0');
begin
    -- The whole pipeline moves only when the output slot is free or being
    -- emptied; otherwise everything holds (AXI rule: keep data until taken).
    adv     <= (not o_valid) or m_axis_tready;
    in_xfer <= s_axis_tvalid and adv;
    s_axis_tready <= adv;

    x <= signed(s_axis_tdata(15 downto 0));
    y <= signed(s_axis_tdata(31 downto 16));

    vec_end <= '1' when s_axis_tlast = '1'
                    or (unsigned(vlen) /= 0 and cnt = unsigned(vlen) - 1) else '0';
    sum <= acc + resize(prod, 48);

    process (clk)
    begin
        if rising_edge(clk) then
            if rst = '1' then
                cnt     <= (others => '0');
                p_valid <= '0';
                acc     <= (others => '0');
                o_valid <= '0';
            elsif adv = '1' then
                p_valid <= in_xfer;
                prod    <= x * y;
                p_end   <= vec_end;
                p_last  <= s_axis_tlast;
                if in_xfer = '1' then
                    if vec_end = '1' then
                        cnt <= (others => '0');
                    else
                        cnt <= cnt + 1;
                    end if;
                end if;

                o_valid <= '0';
                if p_valid = '1' then
                    if p_end = '1' then
                        o_data  <= resize(sum, 64);
                        o_valid <= '1';
                        o_last  <= p_last;
                        acc     <= (others => '0');
                    else
                        acc <= sum;
                    end if;
                end if;
            end if;
        end if;
    end process;

    m_axis_tvalid <= o_valid;
    m_axis_tdata  <= std_logic_vector(o_data);
    m_axis_tlast  <= o_last;
end architecture;

The flow control is the simplest correct scheme for a short pipeline: one signal adv enables every register. When the output holds a result the sink has not taken, adv is low, s_axis_tready is low, and nothing moves. This makes s_axis_tready depend combinationally on m_axis_tready — fine for a block this small; if timing became a problem, the skid buffer of Chapter 19 on the input or output would cut that path without losing throughput.

Key idea: a stream engine is a pipeline plus framing

The arithmetic is one line. Almost all the design effort is in the framing — where vectors begin and end, what happens when either side stalls, and what the downstream DMA needs (a TLAST to finish its transfer). Get those right in simulation first; then the board works the first time.

4. Verification under back-pressure

The testbench sends twelve vectors through the engine while a 32-bit LFSR inserts random idle cycles on the input (about 25 %) and drops m_axis_tready about half the time. It computes every expected dot product itself in 64-bit arithmetic and checks value, order and m_axis_tlast. The vectors cover the corner cases: lengths 1, 2 and 3 (shorter than the pipeline), a 300-element vector of\(-32768 \times -32768\) whose sum \(300 \cdot 2^{30} \approx 3.2\times10^{11}\) needs more than 32 bits, and five vectors of length 8 batched in one packet with vlen = 8 and TLAST only at the very end.

Self-checking testbench with random gaps and back-pressure

Verilog

`timescale 1ns/1ps
// Sends 12 vectors with random gaps and random back-pressure:
//   phase 1 (vlen = 0): 6 vectors of lengths 1, 2, 3, 5, 16, 100, each with TLAST,
//                       then one of 300 beats with x = y = -32768 (overflow test)
//   phase 2 (vlen = 8): 5 vectors of 8 in ONE packet, TLAST only on the last beat
// The testbench computes every expected dot product itself and checks results,
// order and m_axis_tlast. Data come from a 32-bit LFSR, identical in VHDL.
module dot_engine_tb;
    reg clk = 0, rst = 1;
    always #5 clk = ~clk;                          // 100 MHz, like FCLK_CLK0

    reg  [15:0] vlen = 0;
    reg         s_tvalid = 0, s_tlast = 0, m_tready = 0;
    reg  [31:0] s_tdata = 0;
    wire        s_tready, m_tvalid, m_tlast;
    wire [63:0] m_tdata;

    dot_engine dut (.clk(clk), .rst(rst), .vlen(vlen),
        .s_axis_tvalid(s_tvalid), .s_axis_tready(s_tready), .s_axis_tdata(s_tdata), .s_axis_tlast(s_tlast),
        .m_axis_tvalid(m_tvalid), .m_axis_tready(m_tready), .m_axis_tdata(m_tdata), .m_axis_tlast(m_tlast));

    function [31:0] lfsr32(input [31:0] v);        // Galois, taps 0x80200003
        lfsr32 = {1'b0, v[31:1]} ^ (v[0] ? 32'h80200003 : 32'h0);
    endfunction

    reg [31:0] data_lfsr = 32'h1234_5678, gap_lfsr = 32'hCAFE_F00D, rdy_lfsr = 32'h0BAD_BEEF;
    reg signed [63:0] expected [0:15];
    reg               exp_last [0:15];
    integer n_exp = 0, n_got = 0, errors = 0;
    reg signed [63:0] checksum = 0;

    // one beat: x, y from the LFSR (or forced), random idle cycles first
    task beat(input last, input force_min);
        reg signed [15:0] x, y;
        begin
            while (gap_lfsr[1:0] == 2'b00) begin       // ~25 % idle cycles
                @(negedge clk) s_tvalid = 0;
                gap_lfsr = lfsr32(gap_lfsr);
            end
            gap_lfsr = lfsr32(gap_lfsr);
            data_lfsr = lfsr32(data_lfsr);
            s_tdata = force_min ? 32'h8000_8000 : data_lfsr;
            @(negedge clk);
            s_tvalid = 1; s_tlast = last;
            @(posedge clk);
            while (!s_tready) @(posedge clk);         // hold until accepted
        end
    endtask

    // a vector of n beats; tlast_at_end = 1 marks its last beat with TLAST
    task vector(input integer n, input tlast_at_end, input result_last, input force_min);
        integer i;
        reg signed [63:0] acc;
        begin
            acc = 0;
            for (i = 0; i < n; i = i + 1) begin
                beat(tlast_at_end && i == n - 1, force_min);
                acc = acc + $signed(s_tdata[15:0]) * $signed(s_tdata[31:16]);
            end
            expected[n_exp] = acc; exp_last[n_exp] = result_last; n_exp = n_exp + 1;
        end
    endtask

    // sink: random TREADY (~50 %), check each result in order
    always @(posedge clk) begin
        if (m_tvalid && m_tready) begin
            if (m_tdata !== expected[n_got] || m_tlast !== exp_last[n_got]) begin
                $display("FAIL: result %0d = %0d (last=%b), expected %0d (last=%b)",
                         n_got, $signed(m_tdata), m_tlast, expected[n_got], exp_last[n_got]);
                errors = errors + 1;
            end
            checksum = checksum + $signed(m_tdata);
            n_got = n_got + 1;
        end
    end
    always @(negedge clk) begin
        rdy_lfsr = lfsr32(rdy_lfsr);
        m_tready = rdy_lfsr[0];
    end

    integer k;
    initial begin
        repeat (3) @(posedge clk);
        rst = 0;
        vector(1, 1, 1, 0); vector(2, 1, 1, 0); vector(3, 1, 1, 0);
        vector(5, 1, 1, 0); vector(16, 1, 1, 0); vector(100, 1, 1, 0);
        vector(300, 1, 1, 1);                            // 300 * 2^30: needs > 32 bits
        @(negedge clk) s_tvalid = 0;
        wait (n_got == n_exp);
        vlen = 8;                                        // phase 2: batched vectors
        for (k = 0; k < 5; k = k + 1)
            vector(8, k == 4, k == 4, 0);
        @(negedge clk) s_tvalid = 0;
        wait (n_got == n_exp);
        repeat (2) @(posedge clk);
        if (errors == 0)
            $display("PASS: %0d dot products correct, checksum = 0x%h, last = %0d",
                     n_got, checksum, expected[n_exp - 1]);
        else
            $display("FAIL: %0d errors", errors);
        $finish;
    end
endmodule

VHDL

-- Sends 12 vectors with random gaps and random back-pressure:
--   phase 1 (vlen = 0): 6 vectors of lengths 1, 2, 3, 5, 16, 100, each with TLAST,
--                       then one of 300 beats with x = y = -32768 (overflow test)
--   phase 2 (vlen = 8): 5 vectors of 8 in ONE packet, TLAST only on the last beat
-- The testbench computes every expected dot product itself and checks results,
-- order and m_axis_tlast. Data come from a 32-bit LFSR, identical in Verilog.
library ieee;
use ieee.std_logic_1164.all;
use ieee.numeric_std.all;

entity dot_engine_tb is
end entity;

architecture sim of dot_engine_tb is
    signal clk  : std_logic := '0';
    signal rst  : std_logic := '1';
    signal done : boolean := false;
    signal vlen : std_logic_vector(15 downto 0) := (others => '0');
    signal s_tvalid, s_tlast, m_tready : std_logic := '0';
    signal s_tdata  : std_logic_vector(31 downto 0) := (others => '0');
    signal s_tready, m_tvalid, m_tlast : std_logic;
    signal m_tdata  : std_logic_vector(63 downto 0);

    type results is array (0 to 15) of signed(63 downto 0);
    type flags   is array (0 to 15) of std_logic;
    signal expected : results := (others => (others => '0'));
    signal exp_last : flags := (others => '0');
    signal n_exp, n_got, errors : natural := 0;
    signal checksum : signed(63 downto 0) := (others => '0');

    function lfsr32(v : std_logic_vector(31 downto 0)) return std_logic_vector is
    begin                                           -- Galois, taps 0x80200003
        if v(0) = '1' then
            return ('0' & v(31 downto 1)) xor x"80200003";
        else
            return '0' & v(31 downto 1);
        end if;
    end function;
begin
    clk <= not clk after 5 ns when not done;        -- 100 MHz, like FCLK_CLK0

    dut : entity work.dot_engine
        port map (clk => clk, rst => rst, vlen => vlen,
                  s_axis_tvalid => s_tvalid, s_axis_tready => s_tready,
                  s_axis_tdata => s_tdata, s_axis_tlast => s_tlast,
                  m_axis_tvalid => m_tvalid, m_axis_tready => m_tready,
                  m_axis_tdata => m_tdata, m_axis_tlast => m_tlast);

    -- sink: random TREADY (~50 %)
    ready_gen : process (clk)
        variable rdy_lfsr : std_logic_vector(31 downto 0) := x"0BADBEEF";
    begin
        if falling_edge(clk) then
            rdy_lfsr := lfsr32(rdy_lfsr);
            m_tready <= rdy_lfsr(0);
        end if;
    end process;

    -- check each result in order
    checker : process (clk)
    begin
        if rising_edge(clk) and m_tvalid = '1' and m_tready = '1' then
            if signed(m_tdata) /= expected(n_got) or m_tlast /= exp_last(n_got) then
                report "FAIL: result " & integer'image(n_got) & " = "
                    & to_hstring(m_tdata) & ", expected " & to_hstring(expected(n_got));
                errors <= errors + 1;
            end if;
            checksum <= checksum + signed(m_tdata);
            n_got <= n_got + 1;
        end if;
    end process;

    stimulus : process
        variable data_lfsr : std_logic_vector(31 downto 0) := x"12345678";
        variable gap_lfsr  : std_logic_vector(31 downto 0) := x"CAFEF00D";
        variable n         : natural := 0;

        -- one beat: x, y from the LFSR (or forced), random idle cycles first
        procedure beat(last, force_min : boolean) is
        begin
            while gap_lfsr(1 downto 0) = "00" loop  -- ~25 % idle cycles
                wait until falling_edge(clk);
                s_tvalid <= '0';
                gap_lfsr := lfsr32(gap_lfsr);
            end loop;
            gap_lfsr  := lfsr32(gap_lfsr);
            data_lfsr := lfsr32(data_lfsr);
            wait until falling_edge(clk);
            if force_min then s_tdata <= x"80008000"; else s_tdata <= data_lfsr; end if;
            s_tvalid <= '1';
            if last then s_tlast <= '1'; else s_tlast <= '0'; end if;
            wait until rising_edge(clk);
            while s_tready /= '1' loop               -- hold until accepted
                wait until rising_edge(clk);
            end loop;
        end procedure;

        -- a vector of len beats; tlast_at_end marks its last beat with TLAST
        procedure vector(len : natural; tlast_at_end, result_last, force_min : boolean) is
            variable acc : signed(63 downto 0) := (others => '0');
        begin
            for i in 0 to len - 1 loop
                beat(tlast_at_end and i = len - 1, force_min);
                acc := acc + resize(signed(s_tdata(15 downto 0)) * signed(s_tdata(31 downto 16)), 64);
            end loop;
            expected(n) <= acc;
            if result_last then exp_last(n) <= '1'; else exp_last(n) <= '0'; end if;
            n := n + 1;
            n_exp <= n;
        end procedure;
    begin
        for i in 1 to 3 loop
            wait until rising_edge(clk);
        end loop;
        rst <= '0';
        vector(1, true, true, false);  vector(2, true, true, false);  vector(3, true, true, false);
        vector(5, true, true, false);  vector(16, true, true, false); vector(100, true, true, false);
        vector(300, true, true, true);                    -- 300 * 2^30: needs > 32 bits
        wait until falling_edge(clk);
        s_tvalid <= '0';
        wait until n_got = n;
        vlen <= std_logic_vector(to_unsigned(8, 16));     -- phase 2: batched vectors
        for k in 0 to 4 loop
            vector(8, k = 4, k = 4, false);
        end loop;
        wait until falling_edge(clk);
        s_tvalid <= '0';
        wait until n_got = n;
        for i in 1 to 2 loop
            wait until rising_edge(clk);
        end loop;
        if errors = 0 then
            report "PASS: " & integer'image(n_got) & " dot products correct, checksum = 0x"
                & to_hstring(checksum) & ", last = " & integer'image(to_integer(expected(n - 1)));
        else
            report "FAIL: " & integer'image(errors) & " errors" severity error;
        end if;
        done <= true;
        wait;
    end process;
end architecture;

Run it

iverilog -g2012 -o tb dot_engine.v dot_engine_tb.v && vvp tb
# PASS: 12 dot products correct, checksum = 0x0000004a67967c8f, last = 339160294

ghdl -a --std=08 dot_engine.vhd dot_engine_tb.vhd
ghdl -e --std=08 dot_engine_tb && ghdl -r --std=08 dot_engine_tb
# PASS: 12 dot products correct, checksum = 0x0000004A67967C8F, last = 339160294   (@6135 ns)

Both simulations finish at 6135 ns with the same checksum over all twelve results: the two languages describe the same hardware, cycle for cycle.

5. The block design

On the board: building the dot overlay

Start from the Chapter 20 flow (Vivado version matching your PYNQ image). Create a block design named dot with the ZYNQ7 Processing System; run block automation. In the PS configuration, enable S_AXI_HP0.
Bring in dot_engine, either by packaging it as IP (Chapter 20; associate clk with both stream interfaces and declare rst as an active-high reset) or by dragging the source file into the block design as a module reference. Vivado infers s_axis and m_axis as AXI-Stream interfaces from the port names.
Add AXI Direct Memory Access. Configure it: scatter-gather off (simple mode, which the PYNQ DMA driver uses), both read (MM2S) and write (S2MM) channels on, buffer-length register width 26 bits, MM2S stream width 32 bits, S2MM stream width 64 bits.
Connect DMA M_AXIS_MM2S → dot_engine/s_axis and dot_engine/m_axis → DMA S_AXIS_S2MM. Our engine has no TKEEP; the DMA's unconnected TKEEP input should default to “all bytes valid”, which is what we want — check for a warning about it when you validate the design.
Add AXI GPIO, one channel, 16 bits, all outputs; connect its output to vlen. Connect dot_engine/rst to the active-high peripheral_reset output of the Processor System Reset block.
Run connection automation: the DMA's and GPIO's control ports go to M_AXI_GP0, the DMA's memory ports (M_AXI_MM2S, M_AXI_S2MM) to S_AXI_HP0, all on FCLK_CLK0 (100 MHz here).
Validate, create the HDL wrapper, generate the bitstream, and copy dot.bit + dot.hwh to the board. Check the timing summary: with one DSP and short paths, 100 MHz should close easily.

Common mistake: a stream that never ends

If the engine never raises m_axis_tlast, the S2MM channel never completes and recvchannel.wait() hangs forever. Likewise if you ask S2MM for more bytes than the engine will send. In our design the last input beat carries TLAST from MM2S, which produces TLAST on the last result. Size the receive buffer as exactly (number of vectors × 8) bytes.

6. The notebook: send vectors, compare with NumPy

The notebook packs \(M = 1024\) pairs of random int16 vectors of length \(L = 256\) into one contiguous buffer (1 MB), sets vlen = 256, runs one DMA transfer in each direction, and compares the 1024 results with NumPy computed in 64-bit integers — exact, so the comparison must be bit-for-bit equal.

notebook — run the accelerator (runs on the board)

import numpy as np
from pynq import Overlay, allocate

ol   = Overlay('dot.bit')          # dot.bit + dot.hwh, same folder
dma  = ol.axi_dma_0
vlen = ol.axi_gpio_0               # AXI GPIO driving dot_engine's vlen input

M, L = 1024, 256                   # 1024 vectors of 256 elements, one transfer
rng = np.random.default_rng(0)
X = rng.integers(-2**15, 2**15, size=(M, L), dtype=np.int16)
Y = rng.integers(-2**15, 2**15, size=(M, L), dtype=np.int16)

inbuf  = allocate(shape=(M * L,), dtype=np.uint32)   # one beat = (y << 16) | x
outbuf = allocate(shape=(M,), dtype=np.int64)        # one 64-bit result per vector

def pack(X, Y, buf):
    buf[:] = (X.view(np.uint16).ravel().astype(np.uint32)
              | (Y.view(np.uint16).ravel().astype(np.uint32) << 16))

def run_hw():
    dma.recvchannel.transfer(outbuf)   # arm S2MM first
    dma.sendchannel.transfer(inbuf)
    dma.sendchannel.wait()
    dma.recvchannel.wait()

vlen.write(0x0, L)                 # GPIO data register at offset 0: vlen = 256
pack(X, Y, inbuf)
run_hw()

ref = np.einsum('ij,ij->i', X.astype(np.int64), Y.astype(np.int64))
print('match:', np.array_equal(outbuf, ref))

Before trusting it on the board, check the packing on your PC: unpack the 32-bit words into \(x\) and\(y\) with the same bit fields as the HDL, compute the sums, and compare with einsum. A mistake here (for example, swapping the halves or forgetting the sign) gives wrong results that look like a hardware bug.

7. Measuring speed-up honestly

A speed-up is a ratio of two times, and both must be measured fairly. The rules:

Time the whole job. From data in a NumPy array to results in a NumPy array: packing into the DMA buffer, both transfers, waiting, and (if needed) unpacking. Timing only the hardware hides the costs that usually dominate.
Warm up. The first call of anything includes one-time costs (page faults, caches, lazy initialisation). Run both versions a few times before timing.
Repeat and take the median. Linux schedules other work; single measurements on the board vary noticeably. Report the median of many runs, and look at the spread.
Compare against good software. NumPy's vectorised routines, not a Python forloop. Beating a Python loop by 100× proves nothing about the hardware.
Check correctness every time. A fast wrong answer is not a speed-up.

notebook — fair benchmark (runs on the board)

import time

def bench(fn, reps=50, warmup=5):
    """Median wall-clock time of fn() in seconds, after warm-up runs."""
    for _ in range(warmup):
        fn()
    t = []
    for _ in range(reps):
        t0 = time.perf_counter()
        fn()
        t.append(time.perf_counter() - t0)
    return float(np.median(t))

Xf, Yf = X.astype(np.float32), Y.astype(np.float32)
t_np_int   = bench(lambda: np.einsum('ij,ij->i', X.astype(np.int64), Y.astype(np.int64)))
t_np_float = bench(lambda: np.einsum('ij,ij->i', Xf, Yf))      # inputs already converted
t_hw       = bench(run_hw)                                     # DMA in + compute + DMA out
t_hw_e2e   = bench(lambda: (pack(X, Y, inbuf), run_hw()))      # including packing in Python

n_mac = M * L
for name, t in [('NumPy int64', t_np_int), ('NumPy float32', t_np_float),
                ('FPGA, data already packed', t_hw), ('FPGA, end to end', t_hw_e2e)]:
    print(f'{name:28s} {t*1e3:8.2f} ms   {n_mac / t / 1e6:7.1f} M MAC/s')
print(f'speed-up vs NumPy int64 (end to end): {t_np_int / t_hw_e2e:.2f}x')

What should you expect? The engine does one MAC per cycle: \(10^8\) MAC/s at 100 MHz, if the DMA keeps it fed at 400 MB/s. The ARM Cortex-A9 with NumPy runs a dot product at a rate of the same order of magnitude, and the end-to-end hardware time also includes the packing, which is itself a NumPy pass over the data. Do not be surprised if the speed-up is around 1× — or below. That is not a failed lab; it is the lesson of the next section.

8. Why a dot product is memory-bound

The key quantity is arithmetic intensity: how many operations a kernel performs per byte it has to move.

\[ I = \frac{\text{operations}}{\text{bytes moved}}, \qquad P_\text{attainable} = \min\!\left(P_\text{peak},\; I \times B\right) \]

This is the roofline model. For our dot product each 4-byte beat brings one multiply and one add:\(I = 2/4 = 0.5\) op/B. With \(B = 400\) MB/s, at most 0.2 GOP/s — no matter how many DSPs you throw at it. The PL's 220 DSP slices could do \(220 \times 2 \times 100\,\text{MHz} = 44\) GOP/s, but only for kernels with \(I \gtrsim 110\) op/B. And the ARM reads the same DDR at a comparable rate, so for a low-intensity kernel the two are bound by the same memory.

Then there is Amdahl's law. If a fraction \(f\) of a program's runtime is accelerated by a factor \(s\), the overall speed-up is

\[ S = \frac{1}{(1-f) + f/s} \;\le\; \frac{1}{1-f} \]

Even an infinitely fast accelerator cannot beat \(1/(1-f)\). Data packing and transfer that the software version did not need count as part of \(1-f\), or even make the program slower. The model below plots both ideas. Its bandwidth and ARM numbers are assumptions (labelled in the code); the DSP count is from the datasheet. Change them and see which conclusions survive.

Python

script.py70 lines

import numpy as np
import matplotlib.pyplot as plt

# Roofline model of the PYNQ-Z2. Every number marked ASSUMPTION is a round
# figure in the plausible range, not a measurement: replace it with your own.
F_PL      = 100e6        # PL clock (FCLK_CLK0), Hz              -- ASSUMPTION
N_DSP     = 220          # DSP48E1 slices on the XC7Z020          -- datasheet
PEAK_PL   = 2 * N_DSP * F_PL          # 1 multiply + 1 add per DSP per cycle
BW_PL     = 400e6        # B/s, DMA over one HP port, 32 bit @ 100 MHz -- ASSUMPTION
PEAK_ARM  = 1.0e9        # op/s NumPy can sustain on the Cortex-A9 -- ASSUMPTION
BW_ARM    = 1.0e9        # B/s the ARM can stream from DDR         -- ASSUMPTION

# Arithmetic intensity I = operations per byte that must cross the DMA link
kernels = {
    'vector add (int16)':           1 / 6,          # 1 op, 2 in + 1 out samples of 2 B
    'dot product (int16)':          2 / 4,          # mul + add per 4-byte beat
    '3x3 convolution (8-bit image)': 18 / 2,        # 9 MACs per pixel, 1 B in + 1 B out
    '64-tap FIR (int16)':           128 / 4,        # 64 MACs per sample, 2 B in + 2 B out
    '64x64 matrix multiply (int8)': 2 * 64**3 / (3 * 64**2),   # 2n^3 ops / 3n^2 bytes
}

def attainable(I, peak, bw):
    return np.minimum(peak, I * bw)

I = np.logspace(-1.5, 3, 300)
plt.style.use('dark_background')
fig, (ax, ax2) = plt.subplots(1, 2, figsize=(13, 5.4), gridspec_kw={'width_ratios': [1.35, 1]})
fig.patch.set_facecolor('#0a0a0a')
for a in (ax, ax2):
    a.set_facecolor('#0f0f0f'); a.grid(alpha=0.15)
    for sp in a.spines.values(): sp.set_edgecolor('#374151')

ax.loglog(I, attainable(I, PEAK_PL, BW_PL) / 1e9, color='#fb923c', lw=2.4, label='PL: 220 DSP, DMA-fed')
ax.loglog(I, attainable(I, PEAK_ARM, BW_ARM) / 1e9, color='#38bdf8', lw=2.4, label='ARM + NumPy')
ax.loglog(I, attainable(I, 2 * F_PL, BW_PL) / 1e9, color='#fbbf24', lw=1.4, ls='--',
          label='PL: our 1-DSP dot_engine')
ridge = PEAK_PL / BW_PL
ax.axvline(ridge, color='#fb923c', ls=':', lw=1)
ax.text(ridge * 1.1, 0.012, f'PL ridge\n{ridge:.0f} op/B', color='#fb923c', fontsize=8.5)
cols = ['#f87171', '#f87171', '#4ade80', '#4ade80', '#4ade80']
offs = [(-10, -16), (-30, 12), (8, -12), (8, -14), (-150, 8)]   # label positions (points)
print(f"{'kernel':32s} {'I (op/B)':>9s} {'PL GOP/s':>9s} {'ARM GOP/s':>10s} {'PL/ARM':>7s}")
for (name, Ik), c, off in zip(kernels.items(), cols, offs):
    p_pl, p_arm = attainable(Ik, PEAK_PL, BW_PL), attainable(Ik, PEAK_ARM, BW_ARM)
    ax.plot(Ik, p_pl / 1e9, 'o', color=c, ms=7, mec='white', mew=0.6)
    ax.annotate(name, (Ik, p_pl / 1e9), textcoords='offset points', xytext=off,
                fontsize=8.5, color=c)
    print(f"{name:32s} {Ik:9.2f} {p_pl/1e9:9.2f} {p_arm/1e9:10.2f} {p_pl/p_arm:7.2f}")
ax.set_xlabel('arithmetic intensity  (operations per byte moved)')
ax.set_ylabel('attainable performance  (GOP/s)')
ax.set_ylim(0.01, 200)
ax.set_title('Roofline: PYNQ-Z2 PL vs ARM (assumed bandwidths)', color='#fdba74', fontsize=11)
ax.legend(fontsize=8.5, framealpha=0.2, loc='upper left')

# Amdahl: overall speed-up when a fraction f of the runtime is accelerated by s
f = np.linspace(0, 1, 201)
for s, c in ((2, '#38bdf8'), (10, '#fbbf24'), (100, '#fb923c')):
    ax2.plot(f, 1 / ((1 - f) + f / s), color=c, lw=2, label=f'kernel {s}x faster')
ax2.axhline(1, color='#9ca3af', lw=0.8, ls=':')
ax2.set_yscale('log'); ax2.set_ylim(0.9, 120)
ax2.set_xlabel('fraction f of the original runtime that is accelerated')
ax2.set_ylabel('overall speed-up')
ax2.set_title("Amdahl's law", color='#fdba74', fontsize=11)
ax2.legend(fontsize=8.5, framealpha=0.2, loc='upper left')
plt.tight_layout()
plt.savefig('output.png', dpi=120, facecolor='#0a0a0a')

print(f"\nPL peak  : {PEAK_PL/1e9:.0f} GOP/s (220 DSP x 2 op x {F_PL/1e6:.0f} MHz), "
      f"ridge at {ridge:.0f} op/B")
print(f"Amdahl   : f = 0.9, kernel 100x faster -> overall {1/(0.1 + 0.9/100):.1f}x")

Click Run to execute the Python code

Code will be executed with Python 3 on the server

Kernel	Ops per byte	Bound on the PYNQ-Z2	Worth accelerating?
vector add	≈ 0.17	bandwidth	no — the ARM is as fast
dot product	0.5	bandwidth	rarely (only to free the CPU)
3×3 convolution	≈ 9	bandwidth, but above the ARM	yes, a few times faster
64-tap FIR	≈ 32	bandwidth, far above the ARM	yes
matrix multiply (n = 64)	≈ 43, grows with n	approaches compute	yes, the classic case

9. What does benefit from the FPGA

The dot product is a fine teaching vehicle precisely because it does not flatter the hardware. Kernels that do pay off share at least one of these properties:

High arithmetic intensity. Data are reused many times once on chip: matrix products, convolutions, long FIR filters (Chapter 18). Block RAM keeps the reused data next to the DSPs.
Bit-level operations. CRCs, hashing, error-correcting codes, bit permutations, population counts, cryptographic rounds: a CPU spends many instructions per bit; LUTs do them in one level of logic.
Custom precision. A 9-bit or 4-bit multiply, or binary neural networks, use a fraction of a DSP or only LUTs. Many small operators fit where a CPU still uses its 32-bit datapath.
Streaming with low, fixed latency. Data that arrive from pins — an ADC, a camera, a network PHY — can be processed sample by sample with a latency of a few clock cycles and no jitter, without ever going through DDR. A CPU under Linux cannot promise that.
Offloading. Even with no speed-up, hardware that runs alone frees the ARM for other work, and may use less energy per operation.

10. Where next

You can now go from a blank file to Python driving your own verified hardware. Some directions from here:

High-level synthesis (HLS). Vitis HLS compiles annotated C/C++ into RTL with AXI interfaces generated for you. It is very productive for loops and arithmetic; knowing RTL, as you now do, is what lets you understand its reports (initiation interval, latency, resource use) and fix the result when it is not what you expected.
Open-source flows. Yosys (synthesis) and nextpnr (place and route) are mature for Lattice iCE40 and ECP5 FPGAs. Support for Xilinx 7-series, including Zynq-7000, comes from Project X-Ray and the openXC7 toolchain and is still experimental: good for learning and small PL designs, not yet a replacement for Vivado on a full Zynq system with the PS.
Bigger boards. The Kria KV260 (Zynq UltraScale+ MPSoC, quad-core Cortex-A53, more DSPs and bandwidth) runs PYNQ too, and is aimed at vision. RFSoC boards add multi-gigasample ADCs and DACs on the same chip — the ultimate streaming-from-pins platform, also with PYNQ support.
Keep learning. Read the AMBA AXI specification once (it is shorter than you fear); study the DSP48E1 user guide to learn what one slice can do; read other people's well-verified RTL; and keep the habit from this course: simulate first, measure honestly, and only then believe the board.

The whole course in one sentence

Describe hardware precisely, prove it in simulation, let the processor do what software does well, and give the FPGA the work that is parallel, bit-level, or must happen at a fixed time — then measure to make sure it paid off.

Practice Problems

Progress: 0/5 (0%)

⭐ Easy

The dot engine accepts one 32-bit beat per clock at 100 MHz. What input bandwidth does it need to run at full speed, and how many multiply-accumulates per second does it then perform?

⭐ Easy

A program spends 70 % of its time in a kernel. You make the kernel 20× faster. What is the overall speed-up? What is the limit if the kernel became infinitely fast?

⭐⭐ Medium

The accumulator is 48 bits wide and inputs are signed 16-bit. What is the longest vector for which the sum can never overflow? Is the 16-bit vlen register a safe limit?

⭐⭐ Medium

A student reports a 15× speed-up: they timed only dma.sendchannel.wait() after starting the transfer in an earlier cell, and compared with the first call of a NumPy function. List the three things wrong with this measurement.

⭐⭐⭐ Hard

Using the roofline numbers of Section 8 (PL peak 44 GOP/s, DMA bandwidth 400 MB/s; ARM 1 GOP/s, 1 GB/s — all assumptions), (a) what arithmetic intensity does a kernel need for the PL to beat the ARM at all? (b) A 3×3 convolution on an 8-bit image does 9 MACs per pixel and moves 2 bytes per pixel. What speed-up does the model predict, and how many DSP slices would the PL design actually need to reach it at 100 MHz?

Problem Set Statistics

Total Problems

Attempted

Completion

Share:X Reddit LinkedIn

← 21. Python Drives Hardware: PYNQ Overlays Course overview →