universal_checkpoint.py 5.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108
  1. """
  2. Copyright 2022 The Microsoft DeepSpeed Team
  3. """
  4. import os
  5. import torch
  6. import types
  7. from .constants import (FP32_WEIGHT_KEY,
  8. PARAM,
  9. VOCAB_DIVISIBILITY_PADDING_TENSOR,
  10. CAT_DIM)
  11. def load_hp_checkpoint_state(self, folder, tp_rank, tp_world_size):
  12. hp_mapping = self._hp_mapping
  13. optim_state_keys = hp_mapping.get_optim_state_keys()
  14. hp_keys = [FP32_WEIGHT_KEY] + optim_state_keys
  15. checkpoint_files = {key: os.path.join(folder, f"{key}.pt") for key in hp_keys}
  16. for file in checkpoint_files.values():
  17. assert os.path.isfile(file), f'{file} is not a valid file'
  18. for key in hp_keys:
  19. ckpt_file = checkpoint_files[key]
  20. ckpt_dict = torch.load(ckpt_file)
  21. full_hp_param = ckpt_dict[PARAM]
  22. # need to deal with slices that were averaged.
  23. # the opposite of averaging here becomes an exact copy of the first slice
  24. # I thought of 2 ways:
  25. # implementation a. find a way for a client to pass a dict with patterns
  26. # if any(re.search(pattern, folder) for pattern in WEIGHTS_TO_AVERAGE_PATTERNS):
  27. # tp_rank = 0
  28. # tp_world_size = 1
  29. # the other approach is to assume that the saved data is correct and if full_hp_param.shape ==
  30. # self.shape that means we automatically copy?
  31. # implementation b.
  32. # this version requires no additional data passed from the client
  33. # if the shapes already match it must be slices that were averaged - so we just hack around those
  34. if full_hp_param.shape == self.shape:
  35. tp_rank = 0
  36. tp_world_size = 1
  37. # special case for word_embeddings weights which get padded differently depending on TP degree.
  38. # the converter to universal currently strips the original padding completely so the saved
  39. # weight is padding-free and we just need to add new padding depending on the target TP
  40. # degree
  41. vocab_divisibility_padding_tensor = ckpt_dict.get(
  42. VOCAB_DIVISIBILITY_PADDING_TENSOR,
  43. None)
  44. if vocab_divisibility_padding_tensor is not None:
  45. # In the absence of data passed from the user wrt new padded vocab specific to tp degree
  46. # we can again derive that data by reverse engineering the target shapes like so:
  47. padded_target_vocab_size = self.shape[0] * tp_world_size
  48. if padded_target_vocab_size > full_hp_param.shape[0]:
  49. # Need to expand
  50. padding_size = padded_target_vocab_size - full_hp_param.shape[0]
  51. # Implement the following concat in efficient way using pad
  52. #full_hp_param = torch.cat((full_hp_param, padding_tensor), 0)
  53. full_hp_param = torch.nn.functional.pad(full_hp_param,
  54. (0,
  55. 0,
  56. 0,
  57. padding_size),
  58. "constant",
  59. 0)
  60. full_hp_param[:-padding_size, :] = vocab_divisibility_padding_tensor
  61. else:
  62. # Need to shrink or keep the same
  63. full_hp_param = full_hp_param[:padded_target_vocab_size, :]
  64. full_param_numel = full_hp_param.numel()
  65. tp_slice_numel = self.numel()
  66. # if key == FP32_WEIGHT_KEY and 'word_embeddings.weight' in folder:
  67. # print_rank_0(f'{full_hp_param[:10]=}', force=True)
  68. assert full_param_numel == tp_world_size * tp_slice_numel, \
  69. f'Loading {ckpt_file} full param numel {full_param_numel} != tensor slice numel {tp_slice_numel} * tp_world_size {tp_world_size}'
  70. dst_tensor = hp_mapping.hp_fragment if key == FP32_WEIGHT_KEY else hp_mapping.get_optim_state_fragment(
  71. key)
  72. # print(f"{full_hp_param.shape=} {full_param_numel=} {folder=}")
  73. # print(f"{dst_tensor.shape=} {dst_tensor.numel()=}{folder=}")
  74. # since when we do many to 1 on tp we cat sometimes on dim=0 and other times on dim=1 we have to do exactly the same in reverse
  75. chunk_dim = ckpt_dict.get(CAT_DIM, 0)
  76. # this performs the opposite of cat when merging TP slices
  77. tp_hp_slice = full_hp_param.chunk(tp_world_size, chunk_dim)[tp_rank]
  78. tp_hp_slice = tp_hp_slice.flatten()
  79. lp_frag_address = hp_mapping.lp_fragment_address
  80. tp_hp_fragment = tp_hp_slice.narrow(0,
  81. lp_frag_address.start,
  82. lp_frag_address.numel)
  83. assert dst_tensor.numel() == lp_frag_address.numel, \
  84. f'Load checkpoint {key} dst_tensor numel {dst_tensor.numel()} != src numel {lp_frag_address.numel}'
  85. # print(f"{key} SHAPE: {tp_hp_slice.shape=}")
  86. # print(f"{key} SHAPE: {dst_tensor.shape=}")
  87. # print(f"{key} SHAPE: {tp_hp_fragment.shape=}")
  88. dst_tensor.data.copy_(tp_hp_fragment.data)
  89. def enable_universal_checkpoint(param_list):
  90. for param in param_list:
  91. param.load_hp_checkpoint_state = types.MethodType(load_hp_checkpoint_state,
  92. param)