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

---

fifo_slave.py

The read side of the FIFO. FIFOSlave drives read itself — with whatever delay profile you give it — and captures the data that comes out. Half driver, half monitor, which is why it inherits FIFOMonitorBase.

Overview

A FIFO consumer has two jobs: pull data out, and keep track of what it pulled. FIFOSlave does both. The read signal is actively driven with configurable or randomized timing, while the inherited monitoring path captures, unpacks, timestamps, and counts every transaction. Because the capture machinery is the same code the pure monitor uses, the slave and the monitor can't disagree about what a transaction looked like.

Key Features

• Active read control: drives read itself, with configurable or randomized delays
• Everything the monitor does: same capture, stats, and violation detection, inherited from FIFOMonitorBase
• Empty handling: read-while-empty attempts are detected and counted
• Memory integration: store or cross-check incoming data against a MemoryModel
• Two capture timings: fifo_mux (same-cycle) or fifo_flop (next-cycle), matching the DUT's output style
• Standard queue and callbacks: observed packets land in _recvQ, callbacks fire per transaction

Core Class

FIFOSlave

FIFO Slave component for reading transactions from FIFO interfaces.

Constructor

FIFOSlave(dut, title, prefix, clock, field_config,
          timeout_cycles=1000, mode='fifo_mux',
          bus_name='',
          pkt_prefix='',
          multi_sig=False,
          randomizer=None, log=None, super_debug=False,
          signal_map=None, **kwargs)

Parameters: - dut: Device under test - title: Component title/name - prefix: Bus prefix for signal naming - clock: Clock signal - field_config: Field configuration (FieldConfig object or dict) - timeout_cycles: Timeout for operations (default: 1000) - mode: FIFO mode ('fifo_mux', 'fifo_flop') - bus_name: Bus/channel name - pkt_prefix: Packet field prefix - multi_sig: Whether using multi-signal mode - randomizer: Optional randomizer for read delays - log: Logger instance - super_debug: Enable detailed debugging - signal_map: Optional manual signal mapping - **kwargs: Additional arguments for BusMonitor

The knob set mirrors FIFOMaster's, plus the read-side randomizer.

Example:

# Basic FIFO slave
slave = FIFOSlave(
    dut=dut,
    title="ReadSlave",
    prefix="",
    clock=clock,
    field_config=field_config
)

# Advanced configuration
slave = FIFOSlave(
    dut=dut,
    title="AdvancedSlave",
    prefix="",
    clock=clock,
    field_config=field_config,
    timeout_cycles=2000,
    mode='fifo_flop',
    multi_sig=True,
    randomizer=custom_randomizer,
    signal_map={'read': 'rd_en', 'empty': 'fifo_empty'}
)

Core Methods

Bus Control

async reset_bus()

Drop queued transactions and return the interface to idle.

# Reset slave and clear state
await slave.reset_bus()

async wait_cycles(cycles)

Exactly what it says.

Parameters: - cycles: Number of cycles to wait

# Wait for 5 cycles
await slave.wait_cycles(5)

Transaction Access

get_observed_packets(count=None)

Pull observed transactions out of the standard cocotb receive queue.

Parameters: - count: Number of packets to return (None = all)

Returns: List of observed FIFOPacket instances

# Get all observed transactions
all_packets = slave.get_observed_packets()

# Get last 5 transactions
recent_packets = slave.get_observed_packets(count=5)

# Process observed transactions
for packet in all_packets:
    print(f"Received: {packet.formatted()}")

clear_queue()

Empty the queue. Do it between test phases.

# Clear accumulated transactions
slave.clear_queue()

create_packet(**field_values)

Build a packet, usually for comparison against observed traffic.

Parameters: - **field_values: Field values to set in packet

Returns: FIFOPacket instance

# Create packet for comparison
expected = slave.create_packet(data=0x12345678)

Memory Operations

handle_memory_write(packet)

Store the packet's data in the attached MemoryModel. Called automatically during transaction processing when a model is present; you can also call it directly.

Parameters: - packet: Packet to write to memory

Returns: True if successful, False otherwise

# Automatically called during transaction processing
# Can also be called manually
success = slave.handle_memory_write(packet)

handle_memory_read(packet)

Fetch data from the model at the packet's address.

Parameters: - packet: Packet with address to read from

Returns: Tuple of (success, data)

# Read from memory
success, data = slave.handle_memory_read(packet)
if success:
    log.info(f"Read data: 0x{data:X}")

Statistics and Monitoring

get_stats()

Monitor counters plus the base's. received_transactions is the slave-side count; transactions_observed is the monitor view of the same traffic.

Returns: Dictionary containing all statistics

stats = slave.get_stats()

# Monitor statistics
monitor_stats = stats['monitor_stats']
print(f"Transactions received: {monitor_stats['received_transactions']}")
print(f"Transactions observed: {monitor_stats['transactions_observed']}")
print(f"Protocol violations: {monitor_stats['protocol_violations']}")
print(f"Read while empty: {monitor_stats['read_while_empty']}")

# Component statistics
print(f"Observed packets: {stats['observed_packets']}")
print(f"Component type: {stats['component_type']}")

Operational Modes

FIFO_MUX Mode

Combinatorial read data: when read asserts and the FIFO isn't empty, the data is valid that same cycle, so the slave captures it immediately.

slave = FIFOSlave(dut, "MuxSlave", "", clock, field_config, mode='fifo_mux')

# Data captured immediately when read=1 and empty=0
# Timeline: read=1 -> data captured same cycle

FIFO_FLOP Mode

Registered read data: the data shows up the cycle after read asserts, so the slave waits a cycle before capturing.

slave = FIFOSlave(dut, "FlopSlave", "", clock, field_config, mode='fifo_flop')

# Data captured one cycle after read assertion
# Timeline: read=1 -> wait 1 cycle -> data captured

Pick the mode that matches the DUT. A mismatch here shifts every capture by a cycle, and the resulting errors look exactly like data corruption — this is the part that bites people.

Usage Patterns

Basic Read Operations

The slave works in the background once constructed. You just harvest the queue:

# Set up slave
field_config = FieldConfig.create_data_only(32)
slave = FIFOSlave(dut, "Slave", "", clock, field_config)

# Start monitoring (automatic in background)
# Slave automatically drives read signals and captures data

# Access observed transactions
await Timer(1000, 'ns')  # Let some transactions occur
packets = slave.get_observed_packets()
for packet in packets:
    print(f"Received data: 0x{packet.data:X}")

Multi-Field Transaction Processing

Same as the monitor: multi_sig=True, and packets come out fully unpacked:

# Configure multi-field packets
field_config = FieldConfig()
field_config.add_field(FieldDefinition("addr", 32, format="hex"))
field_config.add_field(FieldDefinition("data", 32, format="hex"))
field_config.add_field(FieldDefinition("cmd", 4, format="hex"))

# Create slave with multi-signal mode
slave = FIFOSlave(
    dut=dut,
    title="MultiFieldSlave",
    prefix="",
    clock=clock,
    field_config=field_config,
    multi_sig=True
)

# Process complex transactions
await Timer(1000, 'ns')
packets = slave.get_observed_packets()
for packet in packets:
    print(f"Command: {packet.cmd}, Address: 0x{packet.addr:X}, Data: 0x{packet.data:X}")

    # Process based on command
    if packet.cmd == 1:  # READ
        print("Processing read command")
    elif packet.cmd == 2:  # WRITE
        print("Processing write command")

Randomized Read Timing

The read_delay randomizer controls how eagerly the slave drains the FIFO. Mostly-fast for throughput runs; long tails when you want the DUT to sit full:

# Create randomizer for read delays
read_randomizer = FlexRandomizer({
    'read_delay': ([(0, 1), (2, 8), (9, 30)], [5, 2, 1])
})

# Create slave with randomized timing
slave = FIFOSlave(
    dut=dut,
    title="RandomSlave",
    prefix="",
    clock=clock,
    field_config=field_config,
    randomizer=read_randomizer
)

# Reads will have randomized delays between transactions

Memory-Integrated Processing

With a model attached, incoming data can be stored and cross-checked:

# Set up memory model
memory = MemoryModel(num_lines=256, bytes_per_line=4)

# Create slave with memory integration
slave = FIFOSlave(
    dut=dut,
    title="MemorySlave",
    prefix="",
    clock=clock,
    field_config=field_config,
    memory_model=memory
)

# Transactions automatically stored in memory
await Timer(2000, 'ns')

# Verify memory contents
packets = slave.get_observed_packets()
for packet in packets:
    if hasattr(packet, 'addr') and hasattr(packet, 'data'):
        # Verify memory write occurred
        success, stored_data = slave.handle_memory_read(packet)
        assert success and stored_data == packet.data

Real-Time Transaction Processing

add_callback fires per transaction — the place to put live protocol handling:

class ProcessingSlave:
    def __init__(self, dut, clock, field_config):
        self.slave = FIFOSlave(dut, "ProcessingSlave", "", clock, field_config)
        self.processed_count = 0

        # Add callback for real-time processing
        self.slave.add_callback(self.process_transaction)

    def process_transaction(self, packet):
        """Process transactions as they arrive"""
        self.processed_count += 1

        # Real-time processing
        if hasattr(packet, 'cmd'):
            if packet.cmd == 1:  # READ command
                self.handle_read_request(packet)
            elif packet.cmd == 2:  # WRITE command
                self.handle_write_request(packet)

        log.info(f"Processed transaction {self.processed_count}: {packet.formatted()}")

    def handle_read_request(self, packet):
        """Handle read request"""
        if hasattr(packet, 'addr'):
            log.info(f"Read request for address 0x{packet.addr:X}")
            # Process read...

    def handle_write_request(self, packet):
        """Handle write request"""
        if hasattr(packet, 'addr') and hasattr(packet, 'data'):
            log.info(f"Write request: 0x{packet.data:X} -> 0x{packet.addr:X}")
            # Process write...

# Usage
processor = ProcessingSlave(dut, clock, field_config)
# Transactions processed automatically as they arrive

Error Detection and Monitoring

Poll the violation counters over a window and timestamp what you find:

class MonitoringSlave:
    def __init__(self, dut, clock, field_config):
        self.slave = FIFOSlave(
            dut=dut,
            title="MonitoringSlave", 
            prefix="",
            clock=clock,
            field_config=field_config
        )
        self.error_log = []

    async def monitor_with_error_detection(self, duration_ns=10000):
        """Monitor for specified duration with error checking"""
        start_time = cocotb.utils.get_sim_time('ns')

        while cocotb.utils.get_sim_time('ns') - start_time < duration_ns:
            await Timer(100, 'ns')

            # Check for protocol violations
            stats = self.slave.get_stats()
            monitor_stats = stats['monitor_stats']

            if monitor_stats['protocol_violations'] > 0:
                self.error_log.append({
                    'type': 'protocol_violation',
                    'count': monitor_stats['protocol_violations'],
                    'time': cocotb.utils.get_sim_time('ns')
                })

            if monitor_stats['read_while_empty'] > 0:
                self.error_log.append({
                    'type': 'read_while_empty',
                    'count': monitor_stats['read_while_empty'],
                    'time': cocotb.utils.get_sim_time('ns')
                })

        return self.error_log

    def generate_error_report(self):
        """Generate comprehensive error report"""
        stats = self.slave.get_stats()

        report = f"""
        Slave Error Report:
        ==================
        Total Transactions: {stats['monitor_stats']['transactions_observed']}
        Protocol Violations: {stats['monitor_stats']['protocol_violations']}
        Read While Empty: {stats['monitor_stats']['read_while_empty']}
        X/Z Violations: {stats['monitor_stats']['x_z_violations']}

        Error Timeline:
        """

        for error in self.error_log:
            report += f"\n  {error['time']}ns: {error['type']} (count: {error['count']})"

        return report

# Usage
monitor = MonitoringSlave(dut, clock, field_config)
await monitor.monitor_with_error_detection(duration_ns=50000)
print(monitor.generate_error_report())

High-Performance Data Collection

For soak tests, drain the queue in batches and keep the per-sample work minimal:

class HighPerformanceSlave:
    def __init__(self, dut, clock, field_config):
        # Configure for maximum performance
        self.slave = FIFOSlave(
            dut=dut,
            title="HighPerfSlave",
            prefix="",
            clock=clock,
            field_config=field_config,
            super_debug=False  # Disable debug for performance
        )
        self.collection_buffer = []

    async def fast_collection(self, target_count=1000):
        """High-speed data collection"""
        collected = 0
        start_time = time.time()

        while collected < target_count:
            await Timer(10, 'ns')  # Minimal wait

            # Check for new transactions
            current_packets = self.slave.get_observed_packets()
            if len(current_packets) > collected:
                # New transactions available
                new_packets = current_packets[collected:]
                self.collection_buffer.extend(new_packets)
                collected = len(current_packets)

        end_time = time.time()
        duration = end_time - start_time
        rate = target_count / duration

        log.info(f"Collected {target_count} transactions in {duration:.2f}s ({rate:.1f} TPS)")
        return self.collection_buffer

    def analyze_collected_data(self):
        """Analyze collected transaction data"""
        if not self.collection_buffer:
            return {}

        data_values = [p.data for p in self.collection_buffer if hasattr(p, 'data')]

        analysis = {
            'count': len(self.collection_buffer),
            'unique_data_values': len(set(data_values)),
            'data_range': (min(data_values), max(data_values)) if data_values else (0, 0),
            'average_data': sum(data_values) / len(data_values) if data_values else 0
        }

        return analysis

# Usage
collector = HighPerformanceSlave(dut, clock, field_config)
await collector.fast_collection(target_count=5000)
analysis = collector.analyze_collected_data()
print(f"Collection analysis: {analysis}")

Custom Signal Mapping

rd_en / fifo_empty / dout instead of the defaults? Map them — don't rename the RTL:

# For non-standard FIFO interfaces
def create_custom_slave(dut, clock, custom_signals):
    """Create slave with custom signal naming"""

    # Map custom signals to expected names
    signal_map = {
        'read': custom_signals.get('read_enable', 'read'),
        'empty': custom_signals.get('empty_flag', 'empty'),
        'data': custom_signals.get('read_data', 'data')
    }

    # Create field configuration
    field_config = FieldConfig.create_data_only(32)

    # Create slave with custom mapping
    slave = FIFOSlave(
        dut=dut,
        title="CustomSlave",
        prefix="",
        clock=clock,
        field_config=field_config,
        signal_map=signal_map,
        super_debug=True  # Enable for signal mapping debugging
    )

    return slave

# Usage with custom interface
custom_signals = {
    'read_enable': 'rd_en',
    'empty_flag': 'fifo_empty',
    'read_data': 'dout'
}
slave = create_custom_slave(dut, clock, custom_signals)

Protocol Violation Detection

The slave counts these as it runs; you read the counters.

Read While Empty

Counted every time read asserts against an empty FIFO. If this fires, either the testbench got ahead of the DUT or the DUT's empty is lying — both are worth knowing:

# Automatically detected and logged
stats = slave.get_stats()
read_while_empty_count = stats['monitor_stats']['read_while_empty']

if read_while_empty_count > 0:
    log.warning(f"Detected {read_while_empty_count} read-while-empty violations")

X/Z Signal Values

Unresolvable values on the control or data signals get counted too. This is usually a reset or initialization problem upstream:

# X/Z values in signals automatically detected
stats = slave.get_stats()
xz_violations = stats['monitor_stats']['x_z_violations']

if xz_violations > 0:
    log.warning(f"Detected {xz_violations} X/Z signal violations")

Timing Analysis

Read Timing Patterns

If you need inter-transaction intervals, timestamp the queue growth:

class TimingAnalyzer:
    def __init__(self, slave):
        self.slave = slave
        self.timing_data = []

    async def analyze_read_timing(self, duration_ns=10000):
        """Analyze read signal timing patterns"""
        start_time = cocotb.utils.get_sim_time('ns')
        last_packet_count = 0

        while cocotb.utils.get_sim_time('ns') - start_time < duration_ns:
            current_time = cocotb.utils.get_sim_time('ns')
            current_packets = len(self.slave.get_observed_packets())

            if current_packets > last_packet_count:
                # New transaction detected
                self.timing_data.append({
                    'time': current_time,
                    'transaction_count': current_packets
                })
                last_packet_count = current_packets

            await Timer(50, 'ns')

        return self.analyze_intervals()

    def analyze_intervals(self):
        """Analyze time intervals between transactions"""
        if len(self.timing_data) < 2:
            return {}

        intervals = []
        for i in range(1, len(self.timing_data)):
            interval = self.timing_data[i]['time'] - self.timing_data[i-1]['time']
            intervals.append(interval)

        return {
            'total_transactions': len(self.timing_data),
            'average_interval_ns': sum(intervals) / len(intervals),
            'min_interval_ns': min(intervals),
            'max_interval_ns': max(intervals),
            'intervals': intervals
        }

# Usage
analyzer = TimingAnalyzer(slave)
timing_analysis = await analyzer.analyze_read_timing(duration_ns=20000)
print(f"Timing analysis: {timing_analysis}")

Best Practices

1. Use Appropriate Mode for Interface

Combinatorial read data → fifo_mux; registered read data → fifo_flop. Wrong mode, every capture off by a cycle:

# For combinatorial FIFO outputs
slave = FIFOSlave(dut, "CombSlave", "", clock, field_config, mode='fifo_mux')

# For registered FIFO outputs  
slave = FIFOSlave(dut, "RegSlave", "", clock, field_config, mode='fifo_flop')

2. Monitor Error Conditions

Assert on the violation counters. They turn "something felt off" into a failing test with a number in it:

# Regular error checking
async def check_slave_health(slave):
    stats = slave.get_stats()
    monitor_stats = stats['monitor_stats']

    assert monitor_stats['protocol_violations'] == 0, "Protocol violations detected"
    assert monitor_stats['x_z_violations'] == 0, "X/Z signal violations detected"

3. Use Memory Integration for Verification

When data integrity is what you're checking, attach the model:

# Enable memory model for data verification
memory = MemoryModel(num_lines=1024, bytes_per_line=4)
slave = FIFOSlave(dut, "VerifySlave", "", clock, field_config, memory_model=memory)

4. Configure Appropriate Randomizers

Keep two profiles: realistic, and mean:

# For realistic read patterns
realistic_randomizer = FlexRandomizer({
    'read_delay': ([(0, 1), (2, 5)], [8, 2])  # Mostly quick reads
})

# For stress testing
stress_randomizer = FlexRandomizer({
    'read_delay': ([(0, 10), (20, 100)], [1, 1])  # Variable delays
})

5. Clear Queues Between Test Phases

# Reset state between test phases
await slave.reset_bus()
slave.clear_queue()

The slave is the monitor with a job to do: it drains the FIFO so the interesting behavior — full, watermarks, backpressure — actually happens. Give it a realistic delay profile and it will find corner cases your directed stimulus walked right past.