| import os |
| import json |
| import tempfile |
| import numpy as np |
| import torch |
| import time |
| import subprocess |
| import torch.distributed as dist |
| import torch.utils.data as data |
|
|
|
|
| def is_dist_avail_and_initialized(): |
| return dist.is_available() and dist.is_initialized() |
|
|
|
|
| def get_world_size(): |
| if not is_dist_avail_and_initialized(): |
| return 1 |
| return dist.get_world_size() |
|
|
|
|
| def get_rank(): |
| if not is_dist_avail_and_initialized(): |
| return 0 |
| return dist.get_rank() |
|
|
|
|
| def is_main_process(): |
| return get_rank() == 0 |
|
|
|
|
| def init_distributed_mode(): |
| from helpers.ddp_utils import init_distributed_mode as _init_distributed_mode |
| return _init_distributed_mode() |
|
|
|
|
| def safe_barrier(): |
| if is_dist_avail_and_initialized(): |
| dist.barrier() |
|
|
|
|
|
|
| def allreduce(x, average): |
| if mpi_size() > 1: |
| dist.all_reduce(x, dist.ReduceOp.SUM) |
| return x / mpi_size() if average else x |
|
|
|
|
| def get_cpu_stats_over_ranks(stat_dict): |
| keys = sorted(stat_dict.keys()) |
| allreduced = allreduce(torch.stack([torch.as_tensor(stat_dict[k]).detach().cpu().float() for k in keys]), average=True).cpu() |
| return {k: allreduced[i].item() for (i, k) in enumerate(keys)} |
|
|
|
|
| class Hyperparams(dict): |
| def __getattr__(self, attr): |
| try: |
| return self[attr] |
| except KeyError: |
| return None |
|
|
| def __setattr__(self, attr, value): |
| self[attr] = value |
|
|
|
|
| def logger(log_prefix): |
| 'Prints the arguments out to stdout, .txt, and .jsonl files' |
|
|
| jsonl_path = f'{log_prefix}.jsonl' |
| txt_path = f'{log_prefix}.txt' |
|
|
| def log(*args, pprint=False, **kwargs): |
| if mpi_rank() != 0: |
| return |
| t = time.ctime() |
| argdict = {'time': t} |
| if len(args) > 0: |
| argdict['message'] = ' '.join([str(x) for x in args]) |
| argdict.update(kwargs) |
|
|
| txt_str = [] |
| args_iter = sorted(argdict) if pprint else argdict |
| for k in args_iter: |
| val = argdict[k] |
| if isinstance(val, torch.Tensor): |
| val = val.item() if val.dim() == 0 else val.tolist() |
| elif isinstance(val, np.ndarray): |
| val = val.tolist() |
| elif isinstance(val, np.integer): |
| val = int(val) |
| elif isinstance(val, np.floating): |
| val = float(val) |
| argdict[k] = val |
| if isinstance(val, float): |
| val = f'{val:.5f}' |
| txt_str.append(f'{k}: {val}') |
| txt_str = ', '.join(txt_str) |
|
|
| if pprint: |
| json_str = json.dumps(argdict, sort_keys=True) |
| txt_str = json.dumps(argdict, sort_keys=True, indent=4) |
| else: |
| json_str = json.dumps(argdict) |
|
|
| print(txt_str, flush=True) |
|
|
| with open(txt_path, "a+") as f: |
| print(txt_str, file=f, flush=True) |
| with open(jsonl_path, "a+") as f: |
| print(json_str, file=f, flush=True) |
|
|
| return log |
|
|
|
|
| def maybe_download(path, filename=None): |
| '''If a path is a gsutil path, download it and return the local link, |
| otherwise return link''' |
| if not path.startswith('gs://'): |
| return path |
| if filename: |
| local_dest = f'/tmp/' |
| out_path = f'/tmp/{filename}' |
| if os.path.isfile(out_path): |
| return out_path |
| subprocess.check_output(['gsutil', '-m', 'cp', '-R', path, out_path]) |
| return out_path |
| else: |
| local_dest = tempfile.mkstemp()[1] |
| subprocess.check_output(['gsutil', '-m', 'cp', path, local_dest]) |
| return local_dest |
|
|
|
|
| def tile_images(images, d1=4, d2=4, border=1): |
| id1, id2, c = images[0].shape |
| out = np.ones([d1 * id1 + border * (d1 + 1), |
| d2 * id2 + border * (d2 + 1), |
| c], dtype=np.uint8) |
| out *= 255 |
| if len(images) != d1 * d2: |
| raise ValueError('Wrong num of images') |
| for imgnum, im in enumerate(images): |
| num_d1 = imgnum // d2 |
| num_d2 = imgnum % d2 |
| start_d1 = num_d1 * id1 + border * (num_d1 + 1) |
| start_d2 = num_d2 * id2 + border * (num_d2 + 1) |
| out[start_d1:start_d1 + id1, start_d2:start_d2 + id2, :] = im |
| return out |
|
|
|
|
| def mpi_size(): |
| return 0 |
|
|
|
|
| def mpi_rank(): |
| return 0 |
|
|
|
|
| def num_nodes(): |
| nn = mpi_size() |
| if nn % 8 == 0: |
| return nn // 8 |
| return nn // 8 + 1 |
|
|
|
|
| def gpus_per_node(): |
| size = mpi_size() |
| if size > 1: |
| return max(size // num_nodes(), 1) |
| return 1 |
|
|
|
|
| def local_mpi_rank(): |
| return mpi_rank() % gpus_per_node() |
|
|
|
|
| |
| |
| |
| |
| |
| |
| |
| |
|
|
|
|
| class ZippedDataset(data.Dataset): |
|
|
| def __init__(self, *datasets): |
| assert all(len(datasets[0]) == len(dataset) for dataset in datasets) |
| self.datasets = datasets |
|
|
| def __getitem__(self, index): |
| |
| return tuple(dataset[index] for dataset in self.datasets), index |
|
|
| def __len__(self): |
| return len(self.datasets[0]) |
|
|
|
|
| def configure_inductor_for_low_memory_compile(): |
| """Lower peak VRAM during ``torch.compile`` / Inductor warmup. |
| |
| Inductor's kernel autotune (``benchmark_all_configs``) can allocate |
| multi-gigabyte temporaries on top of an already full training graph, |
| which commonly OOMs large CIFAR runs on a single GPU. |
| """ |
| try: |
| import torch._inductor.config as inductor_config |
|
|
| inductor_config.max_autotune = False |
| if hasattr(inductor_config, "coordinate_descent_tuning"): |
| inductor_config.coordinate_descent_tuning = False |
| if hasattr(inductor_config, "max_autotune_pointwise"): |
| inductor_config.max_autotune_pointwise = False |
| except Exception: |
| pass |
|
|
|
|
