(self,
tokenizer_type: str = "llama",
thread_counts: List[int] = [1, 2, 4, 8, 16, 32],
text_size_mb: int = 1024, # 1GB by default
iterations_per_thread: int = 10)
| 65 | """Multithreaded throughput benchmark suite.""" |
| 66 | |
| 67 | def __init__(self, |
| 68 | tokenizer_type: str = "llama", |
| 69 | thread_counts: List[int] = [1, 2, 4, 8, 16, 32], |
| 70 | text_size_mb: int = 1024, # 1GB by default |
| 71 | iterations_per_thread: int = 10): |
| 72 | self.src_dir = Path(__file__).parent.parent / "src" |
| 73 | self.test_dir = Path(__file__).parent |
| 74 | self.tokenizer_type = tokenizer_type.lower() |
| 75 | self.thread_counts = thread_counts |
| 76 | self.text_size_mb = text_size_mb |
| 77 | self.iterations_per_thread = iterations_per_thread |
| 78 | self.results: List[ThroughputResult] = [] |
| 79 | |
| 80 | # Validate tokenizer type |
| 81 | if self.tokenizer_type not in ["llama", "mistral"]: |
| 82 | raise ValueError(f"Invalid tokenizer type: {tokenizer_type}. Must be 'llama' or 'mistral'") |
| 83 | |
| 84 | print(f"Throughput benchmark configuration:") |
| 85 | print(f" Tokenizer: {self.tokenizer_type}") |
| 86 | print(f" Thread counts: {self.thread_counts}") |
| 87 | print(f" Text size: {self.text_size_mb} MB") |
| 88 | print(f" Iterations per thread: {self.iterations_per_thread}") |
| 89 | |
| 90 | def load_llama_config(self) -> Tuple[str, List[Dict[str, Any]], Dict[str, int]]: |
| 91 | """ |
nothing calls this directly
no outgoing calls
no test coverage detected