# modified from https://github.com/feng-yufei/shared_debugging_code/blob/main/bucketsampler.py import itertools import math import random from random import shuffle from typing import Iterator from typing import Optional from typing import TypeVar import torch import torch.distributed as dist from torch.utils.data import Dataset from torch.utils.data import Sampler __all__ = [ "DistributedBucketSampler", ] T_co = TypeVar("T_co", covariant=True) class DistributedBucketSampler(Sampler[T_co]): r""" sort the dataset wrt. input length divide samples into buckets sort within buckets divide buckets into batches sort batches """ def __init__( self, dataset: Dataset, num_replicas: Optional[int] = None, rank: Optional[int] = None, shuffle: bool = True, seed: int = 0, drop_last: bool = False, batch_size: int = 32, ) -> None: if num_replicas is None: if not dist.is_available(): raise RuntimeError("Requires distributed package to be available") num_replicas = dist.get_world_size() if torch.cuda.is_available() else 1 if rank is None: if not dist.is_available(): raise RuntimeError("Requires distributed package to be available") rank = dist.get_rank() if torch.cuda.is_available() else 0 if torch.cuda.is_available(): torch.cuda.set_device(rank) if rank >= num_replicas or rank < 0: raise ValueError( "Invalid rank {}, rank should be in the interval" " [0, {}]".format(rank, num_replicas - 1) ) self.dataset = dataset self.num_replicas = num_replicas self.rank = rank self.epoch = 0 self.drop_last = drop_last # If the dataset length is evenly divisible by # of replicas, then there # is no need to drop any data, since the dataset will be split equally. if ( self.drop_last and len(self.dataset) % self.num_replicas != 0 ): # type: ignore[arg-type] # Split to nearest available length that is evenly divisible. # This is to ensure each rank receives the same amount of data when # using this Sampler. self.num_samples = math.ceil( (len(self.dataset) - self.num_replicas) / self.num_replicas # type: ignore[arg-type] ) else: self.num_samples = math.ceil( len(self.dataset) / self.num_replicas ) # type: ignore[arg-type] self.total_size = self.num_samples * self.num_replicas self.shuffle = shuffle self.seed = seed self.batch_size = batch_size self.id_with_length = self._get_sample_lengths() self.id_buckets = self.make_buckets(bucket_width=2.0) def _get_sample_lengths(self): id_with_lengths = [] for i in range(len(self.dataset)): id_with_lengths.append((i, self.dataset.get_sample_length(i))) id_with_lengths.sort(key=lambda x: x[1]) return id_with_lengths def make_buckets(self, bucket_width: float = 2.0): buckets = [] cur = [] max_sec = bucket_width for id, sec in self.id_with_length: if sec < max_sec: cur.append(id) else: buckets.append(cur) cur = [id] max_sec += bucket_width if len(cur) > 0: buckets.append(cur) return buckets def __iter__(self) -> Iterator[T_co]: if self.shuffle: # deterministically shuffle based on epoch and seed g = torch.Generator() g.manual_seed(self.seed + self.epoch) random.seed(self.epoch + self.seed) shuffled_bucket = [] for buc in self.id_buckets: buc_copy = buc.copy() shuffle(buc_copy) shuffled_bucket.append(buc_copy) grouped_batch_size = self.batch_size * self.num_replicas shuffled_bucket = list(itertools.chain(*shuffled_bucket)) n_batch = int(math.ceil(len(shuffled_bucket) / grouped_batch_size)) batches = [ shuffled_bucket[b * grouped_batch_size : (b + 1) * grouped_batch_size] for b in range(n_batch) ] shuffle(batches) indices = list(itertools.chain(*batches)) else: # type: ignore[arg-type] indices = list(range(len(self.dataset))) if not self.drop_last: # add extra samples to make it evenly divisible padding_size = self.total_size - len(indices) if padding_size <= len(indices): indices += indices[:padding_size] else: indices += (indices * math.ceil(padding_size / len(indices)))[ :padding_size ] else: # remove tail of data to make it evenly divisible. indices = indices[: self.total_size] assert len(indices) == self.total_size # subsample indices = indices[self.rank : self.total_size : self.num_replicas] assert len(indices) == self.num_samples return iter(indices) def __len__(self) -> int: return self.num_samples def set_epoch(self, epoch: int) -> None: r""" Sets the epoch for this sampler. When :attr:`shuffle=True`, this ensures all replicas use a different random ordering for each epoch. Otherwise, the next iteration of this sampler will yield the same ordering. Args: epoch (int): Epoch number. """ self.epoch = epoch