(socket_paths: list[tuple[str, str]])
| 60 | _zmq_ctx = zmq.Context() |
| 61 | |
| 62 | def trigger_error(socket_paths: list[tuple[str, str]]): |
| 63 | socket_paths = dict(socket_paths) |
| 64 | update_weights_from_ipc( |
| 65 | _zmq_ctx, |
| 66 | socket_paths[device_uuid], |
| 67 | device_id=rank, |
| 68 | run=error_run, |
| 69 | post_hook=lambda: device_manager.device_module.synchronize(), |
| 70 | ) |
| 71 | |
| 72 | def error_run(weights: list[tuple[str, torch.Tensor]]): |
| 73 | _ = weights # Do some fake processing |
no test coverage detected