mosaicml / mosaicml/streaming

ValueError: cannot reshape array of size 24 into shape (8,newaxis,8) in Dataloader

Open
#535 16 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

Dominant language
Python
Stars
1.6k
Forks
206
PR merge metrics
No merged PRs in 30d

Description

Hi, looks like some new version of llm-foundry (updated from master) have lags in last week-two.
I have error like this

train_loader:
  dataset:
    max_seq_len: 2048
    shuffle: true
    shuffle_seed: 17
    streams:
      stream:
        local: 1
        remote: 2
        repeat: 1.0
        split: train
  drop_last: true
  name: text
  num_workers: 8
/usr/lib/python3/dist-packages/composer/trainer/trainer.py:1886 in fit       │
│                                                                              │
│   1883 │   │   │   self.state.scaler = ClosureGradScaler() if self._use_clos │
│   1884 │   │                                                                 │
│   1885 │   │   self.first_batch_complete = False                             │
│ ❱ 1886 │   │   self._train_loop()                                            │
│   1887 │                                                                     │
│   1888 │   def close(self):                                                  │
│   1889 │   │   """Shutdown the trainer.                                      │
│                                                                              │
│ /usr/lib/python3/dist-packages/composer/trainer/trainer.py:2007 in           │
│ _train_loop                                                                  │
│                                                                              │
│   2004 │   │   use_grad_scaling = self._use_grad_scaling(self.state.precisio │
│   2005 │   │                                                                 │
│   2006 │   │   if self.spin_dataloaders:                                     │
│ ❱ 2007 │   │   │   self._spin_dataloaders_to_cur_epoch()                     │
│   2008 │   │                                                                 │
│   2009 │   │   if self.state.timestamp.batch_in_epoch == 0 and self._rng_sta │
│   2010 │   │   │   # Only restore the rng state here if the step in the curr │
│                                                                              │
│ /usr/lib/python3/dist-packages/composer/trainer/trainer.py:1957 in           │
│ _spin_dataloaders_to_cur_epoch                                               │
│                                                                              │
│   1954 │   │   │   if isinstance(dataloader, DataLoader) and isinstance(data │
│   1955 │   │   │   │   dataloader.sampler.set_epoch(0)                       │
│   1956 │   │   │   if evaluator.label not in eval_state:                     │
│ ❱ 1957 │   │   │   │   for _ in dataloader:                                  │
│   1958 │   │   │   │   │   break                                             │
│   1959 │   │                                                                 │
│   1960 │   │   # spin the train dataloader's sampler to get to the state of  │
│                                                                              │
│ /usr/lib/python3/dist-packages/torch/utils/data/dataloader.py:630 in         │
│ __next__                                                                     │
│                                                                              │
│    627 │   │   │   if self._sampler_iter is None:                            │
│    628 │   │   │   │   # TODO(https://github.com/pytorch/pytorch/issues/7675 │
│    629 │   │   │   │   self._reset()  # type: ignore[call-arg]               │
│ ❱  630 │   │   │   data = self._next_data()                                  │
│    631 │   │   │   self._num_yielded += 1                                    │
│    632 │   │   │   if self._dataset_kind == _DatasetKind.Iterable and \      │
│    633 │   │   │   │   │   self._IterableDataset_len_called is not None and  │
│                                                                              │
│ /usr/lib/python3/dist-packages/torch/utils/data/dataloader.py:1345 in        │
│ _next_data                                                                   │
│                                                                              │
│   1342 │   │   │   │   self._task_info[idx] += (data,)                       │
│   1343 │   │   │   else:                                                     │
│   1344 │   │   │   │   del self._task_info[idx]                              │
│ ❱ 1345 │   │   │   │   return self._process_data(data)                       │
│   1346 │                                                                     │
│   1347 │   def _try_put_index(self):                                         │
│   1348 │   │   assert self._tasks_outstanding < self._prefetch_factor * self │
│                                                                              │
│ /usr/lib/python3/dist-packages/torch/utils/data/dataloader.py:1371 in        │
│ _process_data                                                                │
│                                                                              │
│   1368 │   │   self._rcvd_idx += 1                                           │
│   1369 │   │   self._try_put_index()                                         │
│   1370 │   │   if isinstance(data, ExceptionWrapper):                        │
│ ❱ 1371 │   │   │   data.reraise()                                            │
│   1372 │   │   return data                                                   │
│   1373 │                                                                     │
│   1374 │   def _mark_worker_as_unavailable(self, worker_id, shutdown=False): │
│                                                                              │
│ /usr/lib/python3/dist-packages/torch/_utils.py:694 in reraise                │
│                                                                              │
│   691 │   │   │   # If the exception takes multiple arguments, don't try to  │
│   692 │   │   │   # instantiate since we don't know how to                   │
│   693 │   │   │   raise RuntimeError(msg) from None                          │
│ ❱ 694 │   │   raise exception                                                │
│   695                                                                        │
│   696                                                                        │
│   697 def _get_available_device_type():                                      │
╰──────────────────────────────────────────────────────────────────────────────╯
ValueError: Caught ValueError in DataLoader worker process 0.
Original Traceback (most recent call last):
  File "/usr/lib/python3/dist-packages/torch/utils/data/_utils/worker.py", line 
308, in _worker_loop
    data = fetcher.fetch(index)
  File "/usr/lib/python3/dist-packages/torch/utils/data/_utils/fetch.py", line 
32, in fetch
    data.append(next(self.dataset_iter))
  File "/usr/lib/python3/dist-packages/streaming/base/dataset.py", line 1392, in
__iter__
    sample_ids = self._get_work(world, epoch, sample_in_epoch)
  File "/usr/lib/python3/dist-packages/streaming/base/dataset.py", line 934, in 
_get_work
    epoch_sample_ids = generate_work(self.batching_method, self, world, epoch,
  File "/usr/lib/python3/dist-packages/streaming/base/batching/__init__.py", 
line 43, in generate_work
    return get(dataset, world, epoch, sample_in_epoch)
  File "/usr/lib/python3/dist-packages/streaming/base/batching/random.py", line 
54, in generate_work_random_batching
    big_ids = get_partitions(dataset.partition_algo, dataset.epoch_size,
  File "/usr/lib/python3/dist-packages/streaming/base/partition/__init__.py", 
line 55, in get_partitions
    return get(num_samples, num_canonical_nodes, num_physical_nodes, 
ranks_per_node,
  File "/usr/lib/python3/dist-packages/streaming/base/partition/relaxed.py", 
line 65, in get_partitions_relaxed
    return get_partitions_orig(num_samples, num_canonical_nodes, 
num_physical_nodes,
  File "/usr/lib/python3/dist-packages/streaming/base/partition/orig.py", line 
133, in get_partitions_orig
    ids = ids.reshape(num_physical_nodes, -1, ranks_per_node)
ValueError: cannot reshape array of size 24 into shape (8,newaxis,8)

Changing drop_last to False doesn't helps
If I will change num_workers to some value will I lose some efficiency? Always launched with 8.

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start with streaming/base/partition/orig.py at get_partitions_orig and trace its callers in streaming/base/partition/relaxed.py and streaming/base/batching/random.py. Reproduce the shown configuration with eight workers and verify that partition generation no longer raises the reshape error; no specific test file is named in the issue.

Written by the indexing model from the issue text.

Assessment

Tech stack
python, pytorch
Domain
data-engineering, machine-learning
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.