packet

Packetized data streams.

Dispatcher

Bases: Component

Dispatcher for packet streams.

The first matching output interface is selected for each input packet.

Source code in katsuo/stream/packet/dispatcher.py

class Dispatcher(wiring.Component):
    '''Dispatcher for packet streams.

    The first matching output interface is selected for each input packet.
    '''

    def __init__(self, packet: Packet):
        if not isinstance(packet, Packet):
            raise ValueError('packet must be an instance of Packet')

        super().__init__({
            'i': wiring.In(stream.Signature(packet)),
        })

        self._outputs = []
        self._frozen = False
        self._packet = packet

    def get_output(self, predicate):
        '''Returns a new input stream interface.

        Must be called before the component is elaborated.

        Args:
            predicate: A function that takes a packet and returns a signal indicating whether this output should handle the packet.
        '''

        if self._frozen:
            raise RuntimeError('Cannot get new input after elaboration')

        interface = stream.Signature(self._packet).create()
        self._outputs.append((interface, predicate))
        return interface

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

        self._frozen = True

        active_output = Signal(range(len(self._outputs)))

        with m.FSM() as fsm:
            with m.State('IDLE'):
                with m.If(self.i.valid):
                    with m.If(0):
                        pass
                    for i, (_, predicate) in enumerate(self._outputs):
                        with m.Elif(predicate(self.i.p)):
                            m.d.sync += active_output.eq(i)
                            m.next = 'ACTIVE'

            with m.State('ACTIVE'):
                with m.Switch(active_output):
                    for i, (output, _) in enumerate(self._outputs):
                        with m.Case(i):
                            wiring.connect(m, wiring.flipped(self.i), wiring.flipped(output))
                            with m.If(self.i.valid & self.i.ready & (self.i.p.last if self._packet.semantics.has_last else self.i.p.end)):
                                m.next = 'IDLE'

        return m

get_output(predicate)

Returns a new input stream interface.

Must be called before the component is elaborated.

Parameters:

Name	Type	Description	Default
predicate		A function that takes a packet and returns a signal indicating whether this output should handle the packet.	required

Source code in katsuo/stream/packet/dispatcher.py

def get_output(self, predicate):
    '''Returns a new input stream interface.

    Must be called before the component is elaborated.

    Args:
        predicate: A function that takes a packet and returns a signal indicating whether this output should handle the packet.
    '''

    if self._frozen:
        raise RuntimeError('Cannot get new input after elaboration')

    interface = stream.Signature(self._packet).create()
    self._outputs.append((interface, predicate))
    return interface

Packet

Bases: StructLayout

Payload shape for a packetized data stream.

Parameters:

Name	Type	Description	Default
data_shape	ShapeLike	Shape of a data token.	8
header_shape	ShapeLike | None	Shape of the optional header field.	None
semantics	Semantics	Semantics of the packetized data stream.	LAST

Source code in katsuo/stream/packet/__init__.py

class Packet(data.StructLayout):
    '''Payload shape for a packetized data stream.

    Args:
        data_shape: Shape of a data token.
        header_shape: Shape of the optional header field.
        semantics: Semantics of the packetized data stream.
    '''

    class Semantics(enum.Enum):
        '''Semantics of the packetized data stream.'''

        LAST = enum.auto()
        '''Payload has a `last` field that's asserted during the last data transfer of a packet.'''

        FIRST_LAST = enum.auto()
        '''Payload has a `first` field that's asserted during the first data transfer of a packet in addition to the `last` field.'''

        END = enum.auto()
        '''Payload has an `end` field that's asserted during a separate transfer after the last data transfer of a packet.'''

        FIRST_END = enum.auto()
        '''Payload has a `first` field that's asserted during the first data transfer of a packet in addition to the `end` field.'''

        @property
        def has_first(self):
            '''True if the semantics includes a `first` field.'''
            return self in {self.FIRST_LAST, self.FIRST_END}

        @property
        def has_last(self):
            '''True if the semantics includes a `last` field.'''
            return self in {self.LAST, self.FIRST_LAST}

        @property
        def has_end(self):
            '''True if the semantics includes an `end` field.'''
            return self in {self.END, self.FIRST_END}

    def __init__(self, data_shape: ShapeLike = 8, *, header_shape: ShapeLike | None = None, semantics: Semantics = Semantics.LAST):
        if not isinstance(semantics, self.Semantics):
            raise TypeError(f'semantics must be of type Packet.Semantics, not {type(semantics)}')

        self.semantics = semantics

        layout = {'data': data_shape}
        if header_shape is not None:
            layout['header'] = header_shape
        if semantics.has_first:
            layout['first'] = 1
        if semantics.has_last:
            layout['last'] = 1
        if semantics.has_end:
            layout['end'] = 1

        super().__init__(layout)

        self.data_shape = data_shape
        self.header_shape = header_shape

    def __call__(self, value):
        return PacketView(self, value)

Semantics

Bases: Enum

Semantics of the packetized data stream.

Source code in katsuo/stream/packet/__init__.py

class Semantics(enum.Enum):
    '''Semantics of the packetized data stream.'''

    LAST = enum.auto()
    '''Payload has a `last` field that's asserted during the last data transfer of a packet.'''

    FIRST_LAST = enum.auto()
    '''Payload has a `first` field that's asserted during the first data transfer of a packet in addition to the `last` field.'''

    END = enum.auto()
    '''Payload has an `end` field that's asserted during a separate transfer after the last data transfer of a packet.'''

    FIRST_END = enum.auto()
    '''Payload has a `first` field that's asserted during the first data transfer of a packet in addition to the `end` field.'''

    @property
    def has_first(self):
        '''True if the semantics includes a `first` field.'''
        return self in {self.FIRST_LAST, self.FIRST_END}

    @property
    def has_last(self):
        '''True if the semantics includes a `last` field.'''
        return self in {self.LAST, self.FIRST_LAST}

    @property
    def has_end(self):
        '''True if the semantics includes an `end` field.'''
        return self in {self.END, self.FIRST_END}

END = enum.auto() class-attribute instance-attribute

Payload has an end field that's asserted during a separate transfer after the last data transfer of a packet.

FIRST_END = enum.auto() class-attribute instance-attribute

Payload has a first field that's asserted during the first data transfer of a packet in addition to the end field.

FIRST_LAST = enum.auto() class-attribute instance-attribute

Payload has a first field that's asserted during the first data transfer of a packet in addition to the last field.

LAST = enum.auto() class-attribute instance-attribute

Payload has a last field that's asserted during the last data transfer of a packet.

has_end property

True if the semantics includes an end field.

has_first property

True if the semantics includes a first field.

has_last property

True if the semantics includes a last field.

PacketQueue

Bases: Component

FIFO queue for packetized data streams.

A packet will only be released on the output side once it is fully received on the input side. If the input semantics includes the first signal, a partially received packet will be discarded when a new packet starts. If the max_inflight parameter is set, the queue includes acknowledgment and replay functionality to allow retransmission of unacknowledged packets.

This makes the queue suitable for the following use cases: - Ingress buffering where packets may be dropped partway on error conditions. - Egress buffering where packets must be sent as contiguous bursts. - Egress buffering with retransmission capabilities.

Behavior is undefined if a packet larger than the queue depth is received.

Parameters:

Name	Type	Description	Default
shape	ShapeLike	Shape of the packetized data stream.	required
depth	int	Depth of the FIFO queue.	required
i_semantics	Semantics	Packet semantics of the input stream.	required
o_semantics	Semantics	Packet semantics of the output stream.	required
max_inflight	int | None	Maximum number of in-flight packets for replay functionality.	None

Attributes:

Name	Type	Description
i	stream	Input stream.
o	stream	Output stream.
ack	in	Acknowledgment signal for completed packets (if max_inflight is set).
replay	in	Replay signal to resend unacknowledged packets (if max_inflight is set).

Source code in katsuo/stream/packet/queue.py

class PacketQueue(wiring.Component):
    '''FIFO queue for packetized data streams.

    A packet will only be released on the output side once it is fully received on the input side.
    If the input semantics includes the `first` signal, a partially received packet will be discarded when a new packet starts.
    If the `max_inflight` parameter is set, the queue includes acknowledgment and replay functionality to allow retransmission of unacknowledged packets.

    This makes the queue suitable for the following use cases:
    - Ingress buffering where packets may be dropped partway on error conditions.
    - Egress buffering where packets must be sent as contiguous bursts.
    - Egress buffering with retransmission capabilities.

    Behavior is undefined if a packet larger than the queue depth is received.

    Args:
        shape: Shape of the packetized data stream.
        depth: Depth of the FIFO queue.
        i_semantics: Packet semantics of the input stream.
        o_semantics: Packet semantics of the output stream.
        max_inflight: Maximum number of in-flight packets for replay functionality.

    Attributes:
        i (stream): Input stream.
        o (stream): Output stream.
        ack (in): Acknowledgment signal for completed packets (if `max_inflight` is set).
        replay (in): Replay signal to resend unacknowledged packets (if `max_inflight` is set).
    '''

    def __init__(self, shape: ShapeLike, *, depth: int, i_semantics: Packet.Semantics, o_semantics: Packet.Semantics, max_inflight: int | None = None):
        signature = {
            'i': wiring.In(stream.Signature(Packet(shape, semantics = i_semantics))),
            'o': wiring.Out(stream.Signature(Packet(shape, semantics = o_semantics))),
        }

        if (depth  & (depth - 1) != 0) or depth <= 0:
            raise ValueError('depth must be a power of two')

        if max_inflight is None:
            pass
        elif (max_inflight & (max_inflight - 1) != 0) or max_inflight <= 0:
            raise ValueError('max_inflight must be a power of two')
        else:
            signature['ack'] = wiring.In(range(max_inflight + 1))
            signature['replay'] = wiring.In(1)

        super().__init__(signature)

        self._shape = shape
        self._depth = depth
        self._max_inflight = max_inflight

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

        addr_width = ceil_log2(self._depth)

        m.submodules.mem = mem = memory.Memory(shape = Packet(self._shape, semantics = Packet.Semantics.LAST), depth = self._depth, init = [])

        m.submodules.input_logic = input_logic = _InputLogic(shape = self._shape, semantics = self.i.payload.shape().semantics, addr_width = addr_width)
        m.submodules.output_logic = output_logic = _OutputLogic(shape = self._shape, semantics = self.o.payload.shape().semantics, addr_width = addr_width, max_inflight = self._max_inflight)

        wiring.connect(m, wiring.flipped(self.i), input_logic.i)
        wiring.connect(m, wiring.flipped(self.o), output_logic.o)
        wiring.connect(m, input_logic.w_port, mem.write_port())
        wiring.connect(m, output_logic.r_port, mem.read_port())
        wiring.connect(m, input_logic.ptrs, output_logic.ptrs)

        if self._max_inflight is not None:
            m.d.comb += [
                output_logic.ack.eq(self.ack),
                output_logic.replay.eq(self.replay),
            ]

        return m

PacketView

Bases: View

View of a packetized data stream payload.

Source code in katsuo/stream/packet/__init__.py

class PacketView(data.View):
    '''View of a packetized data stream payload.'''

    @property
    def h(self):
        '''Shorthand for `.header`.'''
        return self.header

    @property
    def d(self):
        '''Shorthand for `.data`.'''
        return self.data

d property

Shorthand for .data.

h property

Shorthand for .header.

PriorityArbiter

Bases: Component

Arbiter for packet streams with fixed priority.

Priority is determined by the order in which input interfaces are requested. The first requested interface has the highest priority.

Source code in katsuo/stream/packet/arbiter.py

class PriorityArbiter(wiring.Component):
    '''Arbiter for packet streams with fixed priority.

    Priority is determined by the order in which input interfaces are requested.
    The first requested interface has the highest priority.
    '''

    def __init__(self, packet: Packet):
        if not isinstance(packet, Packet):
            raise ValueError('packet must be an instance of Packet')

        super().__init__({
            'o': wiring.Out(stream.Signature(packet)),
        })

        self._inputs = []
        self._frozen = False
        self._packet = packet

    def get_input(self):
        '''Returns a new input stream interface.

        Must be called before the component is elaborated.
        '''

        if self._frozen:
            raise RuntimeError('Cannot get new input after elaboration')

        interface = stream.Signature(self._packet).flip().create()
        self._inputs.append(interface)
        return interface

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

        self._frozen = True

        active_input = Signal(range(len(self._inputs)))
        busy = Signal()

        with m.If(self.o.valid & self.o.ready):
            m.d.sync += busy.eq(~(self.o.p.last if self._packet.semantics.has_last else self.o.p.end))

        with m.If(0):
            pass

        for i, input in enumerate(self._inputs):
            with m.Elif((busy & (active_input == i)) | (~busy & input.valid)):
                wiring.connect(m, wiring.flipped(input), wiring.flipped(self.o))
                m.d.sync += active_input.eq(i)

        return m

get_input()

Returns a new input stream interface.

Must be called before the component is elaborated.

Source code in katsuo/stream/packet/arbiter.py

def get_input(self):
    '''Returns a new input stream interface.

    Must be called before the component is elaborated.
    '''

    if self._frozen:
        raise RuntimeError('Cannot get new input after elaboration')

    interface = stream.Signature(self._packet).flip().create()
    self._inputs.append(interface)
    return interface