Method test_ring_buffer_performance

test/test_ring_buffer.py:246–289

Test ring buffer performance characteristics

(self)

        assert "n_threads" in repr_str, "String representation should contain n_threads"

    def test_ring_buffer_performance(self):
        """Test ring buffer performance characteristics"""
        ring_name = "/test/ring_buffer_perf"
        test_entries = 100

        # Step 1: Create ring buffer
        cli_cmd = f"test stats ring-buffer-gen {ring_name} {test_entries} {INTERVAL_USEC} {RING_SIZE}"
        result = self.vapi.cli(cli_cmd)
        assert "Generated" in result, f"CLI failed: {result}"

        # Step 2: Get ring buffer
        ring_buffer = self.statistics.get_ring_buffer(f"{ring_name}")

        # Step 3: Test individual consume performance
        start_time = time.time()
        individual_count = 0
        for _ in range(100):  # 100 operations
            data = ring_buffer.consume_data(thread_index=0, max_entries=1)
            individual_count += len(data)
        individual_time = time.time() - start_time

        # Step 4: Reset and test batch consume performance
        ring_buffer.local_tails[0] = 0
        ring_buffer.last_sequences[0] = None

        start_time = time.time()
        batch_count = 0
        for _ in range(10):  # 10 batch operations
            data = ring_buffer.consume_data_batch(thread_index=0, max_entries=10)
            batch_count += len(data)
        batch_time = time.time() - start_time

        # Step 5: Calculate performance metrics
        individual_throughput = 100 / individual_time if individual_time > 0 else 0
        batch_throughput = 100 / batch_time if batch_time > 0 else 0

        print(f"Individual consume: {individual_throughput:.0f} ops/sec")
        print(f"Batch consume: {batch_throughput:.0f} ops/sec")

        # Step 6: Validate performance (basic sanity checks)
        assert individual_time > 0, "Individual operations should take some time"
        assert batch_time > 0, "Batch operations should take some time"
        assert individual_count >= 0, "Individual count should be non-negative"
        assert batch_count >= 0, "Batch count should be non-negative"

    def test_ring_buffer_multiple_threads(self):
        """Test ring buffer access across multiple threads"""