Method benchmark_throughput

tests/throughput_test.py:394–449

Benchmark tokenizer throughput with specified thread count.

(self, tokenizer, tokenizer_name: str, thread_count: int, test_text: str)

        )
    
    def benchmark_throughput(self, tokenizer, tokenizer_name: str, thread_count: int, test_text: str) -> ThroughputResult:
        """Benchmark tokenizer throughput with specified thread count."""
        print(f"  Testing {tokenizer_name} with {thread_count} threads...")
        
        # Split text into chunks for processing
        chunk_size = len(test_text) // (thread_count * self.iterations_per_thread)
        text_chunks = []
        
        for i in range(thread_count * self.iterations_per_thread):
            start_idx = i * chunk_size
            if i == (thread_count * self.iterations_per_thread) - 1:
                # Last chunk gets remaining text
                chunk = test_text[start_idx:]
            else:
                end_idx = (i + 1) * chunk_size
                chunk = test_text[start_idx:end_idx]
            text_chunks.append(chunk)
        
        # Benchmark the tokenizer
        start_time = time.perf_counter()
        
        try:
            token_results = tokenizer.encode_batch(text_chunks, num_threads=thread_count)
            success = True
        except Exception as e:
            print(f"    ERROR: {tokenizer_name} encode_batch failed: {e}")
            return None
        
        end_time = time.perf_counter()
        
        if not success or not token_results:
            print(f"    ERROR: No successful tokenizations for {tokenizer_name}")
            return None
        
        # Calculate statistics
        total_time = end_time - start_time
        total_bytes = sum(len(chunk.encode('utf-8')) for chunk in text_chunks)
        total_tokens = sum(len(tokens) for tokens in token_results)
        total_mb = total_bytes / (1024 * 1024)
        
        throughput_mb_per_sec = total_mb / total_time
        throughput_tokens_per_sec = total_tokens / total_time
        avg_latency_ms = (total_time / len(text_chunks)) * 1000
        
        print(f"    {tokenizer_name}: {throughput_mb_per_sec:.2f} MB/s, {throughput_tokens_per_sec:.0f} tokens/s")
        
        return ThroughputResult(
            thread_count=thread_count,
            tokenizer_name=tokenizer_name,
            total_text_size_mb=total_mb,
            total_tokens=total_tokens,
            total_time_seconds=total_time,
            throughput_mb_per_sec=throughput_mb_per_sec,
            throughput_tokens_per_sec=throughput_tokens_per_sec,
            avg_latency_ms=avg_latency_ms
        )
    
    def run_throughput_benchmarks(self):