Work loop for the benchmarking subprocess.
(request_queue: Queue[Any], response_queue: Queue[Any])
| 106 | |
| 107 | @staticmethod |
| 108 | def workloop(request_queue: Queue[Any], response_queue: Queue[Any]) -> None: |
| 109 | """ |
| 110 | Work loop for the benchmarking subprocess. |
| 111 | """ |
| 112 | while True: |
| 113 | obj = request_queue.get() |
| 114 | |
| 115 | if obj is None: |
| 116 | break # None is a sentinel for the child to terminate |
| 117 | elif isinstance(obj, Ping): |
| 118 | response_queue.put(Pong()) |
| 119 | elif isinstance(obj, BenchmarkRequest): |
| 120 | response_queue.put(obj.benchmark()) |
| 121 | else: |
| 122 | raise RuntimeError(f"Invalid request type {type(obj)}") |
| 123 | |
| 124 | def valid(self) -> bool: |
| 125 | """ |
no test coverage detected