(names_to_check: dict[str, bool], socket_paths: list[tuple[str, str]])
| 103 | names_to_check[name] = True |
| 104 | |
| 105 | def check_weights(names_to_check: dict[str, bool], socket_paths: list[tuple[str, str]]): |
| 106 | socket_paths = dict(socket_paths) |
| 107 | update_weights_from_ipc( |
| 108 | _zmq_ctx, |
| 109 | socket_paths[device_uuid], |
| 110 | device_id=rank, |
| 111 | run=lambda weights: check(names_to_check, weights), |
| 112 | post_hook=lambda: device_manager.device_module.synchronize(), |
| 113 | ) |
| 114 | device_manager.device_module.synchronize() |
| 115 | device_manager.device_module.empty_cache() |
| 116 | mem_info = device_manager.device_module.mem_get_info() |
| 117 | memory_usage = mem_info[1] - mem_info[0] |
| 118 | memory_history.append(memory_usage) |
| 119 | assert all(names_to_check.values()) |
| 120 | |
| 121 | while True: |
| 122 | socket_paths: list[tuple[str, str]] = queue.get() |
no test coverage detected