CocoTB Framework · Verification Infrastructure for RTL Testing GitHub · Documentation Index · MIT License

---

packet.py

The base packet class used across all protocols (GAXI, APB, FIFO, AXI4) — field management, validation, formatting, comparison, and FIFO packing, with a thread-safe cache behind the hot operations.

Overview

A packet in this framework is a bag of named fields whose widths, formats, and encodings come from a FieldConfig. The Packet class handles the work every protocol test needs: masking values to field width on assignment (so a 33-bit write to a 32-bit field is an accident you catch, not one you ship), translating between full field values and their FIFO representations when active_bits says only part of the field crosses the wire, and comparing packets while ignoring fields you don't care about. The expensive bits — masks, formatters, encodings — are cached behind a thread-safe lock, because packets get touched constantly and from everywhere.

Key Features

• Thread-safe caching behind field operations
• Automatic field validation and masking on assignment
• Protocol-agnostic — same class serves GAXI, FIFO, APB, AXI4
• FIFO packing/unpacking for signal-level interfaces
• Formatting and comparison built in
• Timing fields for latency analysis

Thread-Safe Caching System

_FieldCache

Internal thread-safe cache for field operations to improve performance in parallel testing environments.

Features

• Field masks cache: Caches field bit masks for validation
• Field bits cache: Caches field bit widths
• Field active bits cache: Caches active bit ranges
• Field formatters cache: Caches formatting functions
• Field encodings cache: Caches encoding dictionaries
• Thread safety: Uses RLock for concurrent access
• Performance tracking: Monitors cache hits and misses

Global Functions

get_field_cache_stats() -> Dict[str, Any]

Get field cache statistics (thread-safe).

stats = get_field_cache_stats()
print(f"Cache hit rate: {stats['hit_rate']:.1f}%")
print(f"Total operations: {stats['hits'] + stats['misses']}")

clear_field_cache()

Clear the field cache (thread-safe).

# Clear cache between test runs
clear_field_cache()

Core Class

Packet

Generic packet class for handling protocol transactions with thread-safe optimized performance.

Constructor

Packet(field_config: Union[FieldConfig, Dict[str, Dict[str, Any]]], 
       skip_compare_fields: Optional[List[str]] = None, 
       **kwargs)

Parameters: - field_config: Either a FieldConfig object or dictionary of field definitions - skip_compare_fields: List of field names to skip during comparison operations - **kwargs: Initial values for fields (e.g., addr=0x123, data=0xABC)

# Create packet with FieldConfig
config = FieldConfig()
config.add_field(FieldDefinition("addr", 32, format="hex"))
config.add_field(FieldDefinition("data", 32, format="hex"))

packet = Packet(config, addr=0x1000, data=0xDEADBEEF)

# Create packet with dictionary config
field_dict = {
    'addr': {'bits': 32, 'format': 'hex'},
    'data': {'bits': 32, 'format': 'hex'}
}
packet = Packet(field_dict, addr=0x2000, data=0x12345678)

Core Properties

• field_config: FieldConfig object defining packet structure
• fields: Dictionary containing current field values
• skip_compare_fields: Fields to skip during equality comparison
• start_time: Transaction start timestamp
• end_time: Transaction end timestamp

Field Access and Validation

Direct Attribute Access

Fields read and write as ordinary attributes. Assignment goes through validation and masking — this is the part that saves you from yourself:

packet.addr = 0x1000      # Sets addr field
packet.data = 0xDEADBEEF  # Sets data field

# Access field values
address = packet.addr     # Gets addr field value
data = packet.data        # Gets data field value

mask_field_value(value, field_name)

Mask a value to ensure it doesn't exceed the bit width of the specified field.

Parameters: - value: The value to mask - field_name: Name of the field whose bit width determines the mask

Returns: Value masked to fit within the field's bit width

# Field is 8 bits wide, value exceeds range
masked_value = packet.mask_field_value(0x1FF, "status")  # Returns 0xFF

FIFO Operations

These four methods handle the mismatch between "the field as the packet sees it" and "the bits that actually cross the FIFO interface" when a field has active_bits set. If a 32-bit address field only brings bits [31:5] across the wire, shift_for_fifo drops the low 5 bits on the way out and expand_from_fifo puts them back on the way in.

shift_for_fifo(value, field_name)

Convert a full field value to its FIFO representation by right-shifting based on active_bits.

Parameters: - value: The full field value - field_name: Name of the field

Returns: Value adjusted according to active_bits configuration for FIFO

# If addr[31:5] is 0x12345678, this returns 0x91A2B3 (shifted right by 5)
fifo_value = packet.shift_for_fifo(0x12345678, "addr")

expand_from_fifo(value, field_name)

Expand a FIFO value to its full field representation by left-shifting.

Parameters: - value: The FIFO field value - field_name: Name of the field

Returns: Value expanded according to active_bits configuration

# If addr[31:5] in FIFO is 0x91A2B3, this returns 0x12345660 (shifted left by 5)
full_value = packet.expand_from_fifo(0x91A2B3, "addr")

pack_for_fifo()

Pack the packet into a dictionary suitable for FIFO transmission.

Returns: Dictionary with field names and FIFO-adjusted values

fifo_data = packet.pack_for_fifo()
print(fifo_data)  # {'addr': 0x91A2B3, 'data': 0xDEADBEEF}

unpack_from_fifo(fifo_data)

Unpack FIFO data into full field values, applying appropriate bit field expansions.

Parameters: - fifo_data: Dictionary with field values from FIFO, or a single integer value

Returns: Self for chaining

# Unpack dictionary of FIFO values
fifo_data = {'addr': 0x91A2B3, 'data': 0xDEADBEEF}
packet.unpack_from_fifo(fifo_data)

# Unpack single value to 'data' field
packet.unpack_from_fifo(0x12345678)

Formatting and Display

formatted(compact=False, show_fifo=False)

Return a formatted string representation.

Parameters: - compact: If True, return a more compact representation - show_fifo: If True, show FIFO values instead of full field values (only applies with compact=True; the detailed form ignores this flag)

Returns: Formatted string representation

# Detailed formatting
print(packet.formatted())

# Compact formatting
print(packet.formatted(compact=True))

# Show FIFO values (compact mode)
print(packet.formatted(compact=True, show_fifo=True))

__str__()

Provide a detailed string representation with all fields displayed in definition order.

print(packet)
# Output:
# Packet:
#   Address        : 0x00001000
#   Data value     : 0xDEADBEEF
#   Start Time: 1000 ns
#   End Time: 2000 ns
#   Duration: 1000 ns

Comparison and Copying

__eq__(other)

Compare packets for equality, skipping fields in skip_compare_fields.

Parameters: - other: Another packet to compare with

Returns: True if all non-skipped fields match and have defined values

packet1 = Packet(config, addr=0x1000, data=0xDEADBEEF)
packet2 = Packet(config, addr=0x1000, data=0xDEADBEEF)

assert packet1 == packet2  # True

# Undefined values (X/Z represented as -1) cause comparison to fail.
# Note: -1 comes from monitors writing fields directly; assigning -1 via
# an attribute or constructor kwarg would be masked to all-ones instead.
packet3 = Packet(config, addr=0x1000)
packet3.fields['data'] = -1
assert packet1 != packet3  # True - undefined data

The undefined-value rule deserves a second read: if any compared field holds -1 (the framework's marker for X/Z), equality is False. That's intentional — "this field was floating" is not the same as "this field matched," and scoreboard checks that treat them the same pass tests that should fail.

copy()

Create a copy of this packet.

Returns: New packet with the same field values

original = Packet(config, addr=0x1000, data=0xDEADBEEF)
copy_packet = original.copy()

# Modify copy without affecting original
copy_packet.data = 0x12345678
assert original.data == 0xDEADBEEF  # Original unchanged

Utility Methods

get_total_bits()

Calculate the total number of bits in the packet.

Returns: Total number of bits across all fields

total_bits = packet.get_total_bits()
print(f"Packet size: {total_bits} bits")

Usage Patterns

Basic Packet Creation and Usage

# Define field configuration
config = FieldConfig()
config.add_field(FieldDefinition("cmd", 4, format="hex", encoding={
    0x0: "NOP", 0x1: "READ", 0x2: "WRITE", 0x3: "BURST"
}))
config.add_field(FieldDefinition("addr", 32, format="hex"))
config.add_field(FieldDefinition("data", 32, format="hex"))

# Create packet with initial values
packet = Packet(config, cmd=0x2, addr=0x1000, data=0xDEADBEEF)

# Access and modify fields
print(f"Command: {packet.cmd}")  # 2
packet.addr = 0x2000
packet.data = 0x12345678

# Display packet
print(packet)

Protocol-Specific Packet Usage

class GAXIWritePacket(Packet):
    """GAXI-specific write packet"""

    def __init__(self, **kwargs):
        # Define GAXI write fields
        config = FieldConfig()
        config.add_field(FieldDefinition("awid", 4, format="hex"))
        config.add_field(FieldDefinition("awaddr", 32, format="hex"))
        config.add_field(FieldDefinition("awlen", 8, format="dec"))
        config.add_field(FieldDefinition("awsize", 3, format="hex"))
        config.add_field(FieldDefinition("awburst", 2, format="hex", encoding={
            0: "FIXED", 1: "INCR", 2: "WRAP"
        }))

        super().__init__(config, **kwargs)

    def is_single_transfer(self):
        return self.awlen == 0

    def get_burst_length(self):
        return self.awlen + 1

    def calculate_address_range(self):
        bytes_per_beat = 1 << self.awsize
        total_bytes = self.get_burst_length() * bytes_per_beat
        return (self.awaddr, self.awaddr + total_bytes - 1)

# Usage
gaxi_packet = GAXIWritePacket(
    awid=0x5,
    awaddr=0x1000,
    awlen=3,        # 4 beats
    awsize=2,       # 4 bytes per beat
    awburst=1       # INCR
)

print(f"Burst type: {gaxi_packet.formatted()}")
print(f"Address range: 0x{gaxi_packet.calculate_address_range()[0]:X} - 0x{gaxi_packet.calculate_address_range()[1]:X}")

Subclassing Packet per protocol, as above, is the intended pattern: the base class handles field plumbing, your subclass adds the protocol semantics (awlen + 1 beats, address-range math) that the raw fields don't know about.

FIFO Interface Usage

class FIFOInterface:
    def __init__(self, packet_config):
        self.config = packet_config

    def write_to_fifo(self, packet):
        """Convert packet to FIFO format and write"""
        fifo_data = packet.pack_for_fifo()

        # Write each field to appropriate FIFO signal
        for field_name, value in fifo_data.items():
            signal = getattr(self, f"{field_name}_sig")
            signal.value = value

    def read_from_fifo(self):
        """Read from FIFO and create packet"""
        packet = Packet(self.config)

        # Read from FIFO signals
        fifo_data = {}
        for field_name in self.config.field_names():
            signal = getattr(self, f"{field_name}_sig")
            fifo_data[field_name] = int(signal.value)

        # Unpack FIFO data to packet
        packet.unpack_from_fifo(fifo_data)
        return packet

Advanced Field Manipulation

class AdvancedPacket(Packet):
    """Extended packet with additional functionality"""

    def set_timestamp(self, timestamp=None):
        """Set timing information"""
        if timestamp is None:
            timestamp = cocotb.utils.get_sim_time()
        self.start_time = timestamp

    def complete_transaction(self, end_timestamp=None):
        """Mark transaction as complete"""
        if end_timestamp is None:
            end_timestamp = cocotb.utils.get_sim_time()
        self.end_time = end_timestamp

    def get_duration(self):
        """Get transaction duration"""
        if self.start_time and self.end_time:
            return self.end_time - self.start_time
        return None

    def validate_fields(self):
        """Validate all field values"""
        errors = []

        for field_name in self.field_config.field_names():
            field_def = self.field_config.get_field(field_name)
            value = getattr(self, field_name)

            # Check for undefined values
            if value == -1:
                errors.append(f"Field '{field_name}' has undefined value")

            # Check field-specific constraints
            max_value = (1 << field_def.bits) - 1
            if value > max_value:
                errors.append(f"Field '{field_name}' value {value} exceeds maximum {max_value}")

        return errors

    def apply_random_values(self, randomizer):
        """Apply random values using a randomizer"""
        if hasattr(randomizer, 'next'):
            values = randomizer.next()
            for field_name, value in values.items():
                if hasattr(self, field_name):
                    setattr(self, field_name, value)

Performance-Critical Usage

class HighPerformancePacketProcessor:
    """Optimized packet processing for high-throughput scenarios"""

    def __init__(self, packet_config):
        self.config = packet_config
        self.packet_pool = []
        self.pool_size = 100

        # Pre-allocate packet pool
        for _ in range(self.pool_size):
            self.packet_pool.append(Packet(packet_config))

    def get_packet(self):
        """Get packet from pool (avoids allocation overhead)"""
        if self.packet_pool:
            packet = self.packet_pool.pop()
            self._reset_packet(packet)
            return packet
        else:
            # Pool exhausted, create new packet
            return Packet(self.config)

    def return_packet(self, packet):
        """Return packet to pool"""
        if len(self.packet_pool) < self.pool_size:
            self.packet_pool.append(packet)

    def _reset_packet(self, packet):
        """Reset packet to default values"""
        for field_name, field_def in self.config.items():
            setattr(packet, field_name, field_def.default)
        packet.start_time = 0
        packet.end_time = 0

    def process_transaction_batch(self, raw_data_list):
        """Process multiple transactions efficiently"""
        processed_packets = []

        for raw_data in raw_data_list:
            packet = self.get_packet()
            packet.unpack_from_fifo(raw_data)

            # Process packet
            self._validate_and_transform(packet)
            processed_packets.append(packet)

        return processed_packets

    def _validate_and_transform(self, packet):
        """Validate and transform packet data"""
        # Apply any necessary transformations
        # Validate protocol-specific constraints
        pass

Packet pooling is worth the trouble only when you're allocating thousands of packets per test — which, if you're running long constrained-random tests, you are. Allocation churn in the monitor path shows up in wall-clock time faster than you'd expect.

Test Framework Integration

@cocotb.test()
def packet_comparison_test(dut):
    """Test using packet comparison for validation"""

    # Create expected packets
    expected_packets = []
    for i in range(10):
        packet = Packet(config, addr=0x1000 + i*4, data=i*0x100)
        expected_packets.append(packet)

    # Monitor actual packets
    actual_packets = []

    # Run transactions
    for expected in expected_packets:
        # Drive transaction
        yield drive_packet_to_dut(dut, expected)

        # Capture result
        actual = yield capture_packet_from_dut(dut)
        actual_packets.append(actual)

    # Compare expected vs actual
    for expected, actual in zip(expected_packets, actual_packets):
        assert expected == actual, f"Packet mismatch: expected {expected}, got {actual}"

    # Performance analysis
    cache_stats = get_field_cache_stats()
    cocotb.log.info(f"Cache performance: {cache_stats['hit_rate']:.1f}% hit rate")

@cocotb.coroutine
def drive_packet_to_dut(dut, packet):
    """Drive packet to DUT"""
    fifo_data = packet.pack_for_fifo()

    for field_name, value in fifo_data.items():
        signal = getattr(dut, f"{field_name}_i")
        signal.value = value

    yield RisingEdge(dut.clk)

@cocotb.coroutine
def capture_packet_from_dut(dut):
    """Capture packet from DUT"""
    # Wait for valid output
    yield RisingEdge(dut.valid_o)

    # Capture values
    fifo_data = {}
    for field_name in config.field_names():
        signal = getattr(dut, f"{field_name}_o")
        fifo_data[field_name] = int(signal.value)

    # Create packet
    packet = Packet(config)
    packet.unpack_from_fifo(fifo_data)
    return packet

Thread Safety and Performance

Cache Performance

The packet system includes thread-safe caching for optimal performance:

# Monitor cache performance
def monitor_cache_performance():
    stats = get_field_cache_stats()
    print(f"Cache Statistics:")
    print(f"  Hits: {stats['hits']}")
    print(f"  Misses: {stats['misses']}")
    print(f"  Hit Rate: {stats['hit_rate']:.1f}%")
    print(f"  Cache Sizes: {stats['cache_size']}")

# Clear cache between test runs
def cleanup_between_tests():
    clear_field_cache()

Thread-Safe Usage

The caching layer is lock-protected, so packets can be built and inspected from multiple threads without corrupting shared state:

import threading

def worker_thread(packet_config, thread_id, num_packets):
    """Worker thread that processes packets"""
    for i in range(num_packets):
        packet = Packet(packet_config, addr=thread_id*1000 + i, data=i)

        # Thread-safe field access and manipulation
        fifo_data = packet.pack_for_fifo()
        packet.unpack_from_fifo(fifo_data)

        # Process packet...

# Create multiple worker threads
threads = []
for thread_id in range(4):
    thread = threading.Thread(target=worker_thread, args=(config, thread_id, 1000))
    threads.append(thread)
    thread.start()

# Wait for completion
for thread in threads:
    thread.join()

Best Practices

1. Use Appropriate Field Configurations

The packet is only as good as its FieldConfig — give fields real formats and descriptions so the dumps are readable:

# Define meaningful field configurations
config = FieldConfig()
config.add_field(FieldDefinition("addr", 32, format="hex", description="Memory address"))
config.add_field(FieldDefinition("data", 32, format="hex", description="Data payload"))

2. Handle Undefined Values

-1 means X/Z. Check for it before you do arithmetic you can't undo:

# Check for undefined values before processing
if packet.data != -1:  # -1 indicates X/Z value
    process_valid_data(packet.data)
else:
    handle_undefined_data()

3. Use FIFO Operations for Signal Interface

Don't hand-roll the shifting. If active_bits is set anywhere in the config, pack_for_fifo()/unpack_from_fifo() are the only correct way across the signal boundary:

# Convert to FIFO format for signal driving
fifo_data = packet.pack_for_fifo()
drive_signals(fifo_data)

# Convert from FIFO format when receiving
packet.unpack_from_fifo(captured_fifo_data)

4. Leverage Packet Comparison for Validation

__eq__ already skips your skip_compare_fields and refuses to match on X/Z. Use it instead of field-by-field asserts:

# Use packet equality for test validation
assert expected_packet == actual_packet

5. Monitor Cache Performance

A low hit rate usually means you're constructing one-off FieldConfigs instead of reusing them:

# Check cache performance periodically
if test_count % 1000 == 0:
    stats = get_field_cache_stats()
    if stats['hit_rate'] < 90:
        log.warning(f"Low cache hit rate: {stats['hit_rate']:.1f}%")

Packet is the data structure everything else in the framework touches — scoreboards, monitors, memory models, randomizers. The masking on assignment and the FIFO translation are the two features that will save you the most debugging time; the cache is what makes using them on every transaction affordable.

---