bucket_sampler.py 5.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162
  1. # modified from https://github.com/feng-yufei/shared_debugging_code/blob/main/bucketsampler.py
  2. import itertools
  3. import math
  4. import random
  5. from random import shuffle
  6. from typing import Iterator
  7. from typing import Optional
  8. from typing import TypeVar
  9. import torch
  10. import torch.distributed as dist
  11. from torch.utils.data import Dataset
  12. from torch.utils.data import Sampler
  13. __all__ = [
  14. "DistributedBucketSampler",
  15. ]
  16. T_co = TypeVar("T_co", covariant=True)
  17. class DistributedBucketSampler(Sampler[T_co]):
  18. r"""
  19. sort the dataset wrt. input length
  20. divide samples into buckets
  21. sort within buckets
  22. divide buckets into batches
  23. sort batches
  24. """
  25. def __init__(
  26. self,
  27. dataset: Dataset,
  28. num_replicas: Optional[int] = None,
  29. rank: Optional[int] = None,
  30. shuffle: bool = True,
  31. seed: int = 0,
  32. drop_last: bool = False,
  33. batch_size: int = 32,
  34. ) -> None:
  35. if num_replicas is None:
  36. if not dist.is_available():
  37. raise RuntimeError("Requires distributed package to be available")
  38. num_replicas = dist.get_world_size() if torch.cuda.is_available() else 1
  39. if rank is None:
  40. if not dist.is_available():
  41. raise RuntimeError("Requires distributed package to be available")
  42. rank = dist.get_rank() if torch.cuda.is_available() else 0
  43. if torch.cuda.is_available():
  44. torch.cuda.set_device(rank)
  45. if rank >= num_replicas or rank < 0:
  46. raise ValueError(
  47. "Invalid rank {}, rank should be in the interval"
  48. " [0, {}]".format(rank, num_replicas - 1)
  49. )
  50. self.dataset = dataset
  51. self.num_replicas = num_replicas
  52. self.rank = rank
  53. self.epoch = 0
  54. self.drop_last = drop_last
  55. # If the dataset length is evenly divisible by # of replicas, then there
  56. # is no need to drop any data, since the dataset will be split equally.
  57. if (
  58. self.drop_last and len(self.dataset) % self.num_replicas != 0
  59. ): # type: ignore[arg-type]
  60. # Split to nearest available length that is evenly divisible.
  61. # This is to ensure each rank receives the same amount of data when
  62. # using this Sampler.
  63. self.num_samples = math.ceil(
  64. (len(self.dataset) - self.num_replicas)
  65. / self.num_replicas # type: ignore[arg-type]
  66. )
  67. else:
  68. self.num_samples = math.ceil(
  69. len(self.dataset) / self.num_replicas
  70. ) # type: ignore[arg-type]
  71. self.total_size = self.num_samples * self.num_replicas
  72. self.shuffle = shuffle
  73. self.seed = seed
  74. self.batch_size = batch_size
  75. self.id_with_length = self._get_sample_lengths()
  76. self.id_buckets = self.make_buckets(bucket_width=2.0)
  77. def _get_sample_lengths(self):
  78. id_with_lengths = []
  79. for i in range(len(self.dataset)):
  80. id_with_lengths.append((i, self.dataset.get_sample_length(i)))
  81. id_with_lengths.sort(key=lambda x: x[1])
  82. return id_with_lengths
  83. def make_buckets(self, bucket_width: float = 2.0):
  84. buckets = []
  85. cur = []
  86. max_sec = bucket_width
  87. for id, sec in self.id_with_length:
  88. if sec < max_sec:
  89. cur.append(id)
  90. else:
  91. buckets.append(cur)
  92. cur = [id]
  93. max_sec += bucket_width
  94. if len(cur) > 0:
  95. buckets.append(cur)
  96. return buckets
  97. def __iter__(self) -> Iterator[T_co]:
  98. if self.shuffle:
  99. # deterministically shuffle based on epoch and seed
  100. g = torch.Generator()
  101. g.manual_seed(self.seed + self.epoch)
  102. random.seed(self.epoch + self.seed)
  103. shuffled_bucket = []
  104. for buc in self.id_buckets:
  105. buc_copy = buc.copy()
  106. shuffle(buc_copy)
  107. shuffled_bucket.append(buc_copy)
  108. grouped_batch_size = self.batch_size * self.num_replicas
  109. shuffled_bucket = list(itertools.chain(*shuffled_bucket))
  110. n_batch = int(math.ceil(len(shuffled_bucket) / grouped_batch_size))
  111. batches = [
  112. shuffled_bucket[b * grouped_batch_size : (b + 1) * grouped_batch_size]
  113. for b in range(n_batch)
  114. ]
  115. shuffle(batches)
  116. indices = list(itertools.chain(*batches))
  117. else:
  118. # type: ignore[arg-type]
  119. indices = list(range(len(self.dataset)))
  120. if not self.drop_last:
  121. # add extra samples to make it evenly divisible
  122. padding_size = self.total_size - len(indices)
  123. if padding_size <= len(indices):
  124. indices += indices[:padding_size]
  125. else:
  126. indices += (indices * math.ceil(padding_size / len(indices)))[
  127. :padding_size
  128. ]
  129. else:
  130. # remove tail of data to make it evenly divisible.
  131. indices = indices[: self.total_size]
  132. assert len(indices) == self.total_size
  133. # subsample
  134. indices = indices[self.rank : self.total_size : self.num_replicas]
  135. assert len(indices) == self.num_samples
  136. return iter(indices)
  137. def __len__(self) -> int:
  138. return self.num_samples
  139. def set_epoch(self, epoch: int) -> None:
  140. r"""
  141. Sets the epoch for this sampler. When :attr:`shuffle=True`, this ensures all replicas
  142. use a different random ordering for each epoch. Otherwise, the next iteration of this
  143. sampler will yield the same ordering.
  144. Args:
  145. epoch (int): Epoch number.
  146. """
  147. self.epoch = epoch