yolact
yolact copied to clipboard
dbolya
Cuda error
Hi,
Your work is very well. I have two issues, with the same origin CUDA. I am on win 10
I have installed CUDA 11.1, and all updated package but: if I do train.py --dataset ... I have : File "C:\Users\Guillaume\miniconda3\envs\ia\lib\site-packages\torch\utils\data\sampler.py", line 124, in iter yield from torch.randperm(n, generator=generator).tolist() RuntimeError: Expected a 'cuda' device type for generator but found 'cpu'
if I do train.py --cuda 1 --dataset... I have : RuntimeError: Expected a 'cuda' device type for generator but found 'cpu'
Can someone help me? (I have a RTX3090 and some version of cuda doesn't work very well like tensorflow)
1)
_Begin training!
Traceback (most recent call last):
File "C:\Users\Guillaume\Documents\Python\AutonomousCar\code\yolact-master\train.py", line 504, in
2)
_ (ia) C:\Users\Guillaume\Documents\Python\AutonomousCar\code\yolact-master>python train.py --config=yolact_base_config --dataset=my_custom_dataset --batch_size=2 --cuda 1 Scaling parameters by 0.25 to account for a batch size of 2. Per-GPU batch size is less than the recommended limit for batch norm. Disabling batch norm. loading annotations into memory... Done (t=29.70s) creating index... index created! loading annotations into memory... Done (t=5.69s) creating index... index created! C:\Users\Guillaume\miniconda3\envs\ia\lib\site-packages\torch\jit_recursive.py:221: UserWarning: 'lat_layers' was found in ScriptModule constants, but it is a non-constant submodule. Consider removing it. warnings.warn("'{}' was found in ScriptModule constants, " C:\Users\Guillaume\miniconda3\envs\ia\lib\site-packages\torch\jit_recursive.py:221: UserWarning: 'pred_layers' was found in ScriptModule constants, but it is a non-constant submodule. Consider removing it. warnings.warn("'{}' was found in ScriptModule constants, " C:\Users\Guillaume\miniconda3\envs\ia\lib\site-packages\torch\jit_recursive.py:221: UserWarning: 'downsample_layers' was found in ScriptModule constants, but it is a non-constant submodule. Consider removing it. warnings.warn("'{}' was found in ScriptModule constants, " Initializing weights... C:\Users\Guillaume\miniconda3\envs\ia\lib\site-packages\torch\nn\functional.py:718: UserWarning: Named tensors and all their associated APIs are an experimental feature and subject to change. Please do not use them for anything important until they are released as stable. (Triggered internally at ..\c10/core/TensorImpl.h:1156.) return torch.max_pool2d(input, kernel_size, stride, padding, dilation, ceil_mode) Begin training!
Traceback (most recent call last):
File "C:\Users\Guillaume\Documents\Python\AutonomousCar\code\yolact-master\train.py", line 504, in
I have pass the error with conda install pytorch torchvision torchaudio cudatoolkit=11.1 -c pytorch-lts -c conda-forge
(install the LTS 1.8 version not the 1.9)
But i have a lot of other problem to train on my own dataset
@dbolya @chongzhou96
Hi, I also got same issue ->RuntimeError: Expected a 'cuda' device type for generator but found 'cpu'. Could you please help us? Thank you
(I have: Ubuntu20,RTX3060,Cuda11.3,Torch1.9.0+cu111)
1)
_Begin training!
Traceback (most recent call last): File "C:\Users\Guillaume\Documents\Python\AutonomousCar\code\yolact-master\train.py", line 504, in train() File "C:\Users\Guillaume\Documents\Python\AutonomousCar\code\yolact-master\train.py", line 270, in train for datum in data_loader: File "C:\Users\Guillaume\miniconda3\envs\ia\lib\site-packages\torch\utils\data\dataloader.py", line 359, in iter return self._get_iterator() File "C:\Users\Guillaume\miniconda3\envs\ia\lib\site-packages\torch\utils\data\dataloader.py", line 305, in _get_iterator return _MultiProcessingDataLoaderIter(self) File "C:\Users\Guillaume\miniconda3\envs\ia\lib\site-packages\torch\utils\data\dataloader.py", line 944, in init self._reset(loader, first_iter=True) File "C:\Users\Guillaume\miniconda3\envs\ia\lib\site-packages\torch\utils\data\dataloader.py", line 975, in _reset self._try_put_index() File "C:\Users\Guillaume\miniconda3\envs\ia\lib\site-packages\torch\utils\data\dataloader.py", line 1209, in _try_put_index index = self._next_index() File "C:\Users\Guillaume\miniconda3\envs\ia\lib\site-packages\torch\utils\data\dataloader.py", line 512, in _next_index return next(self.sampler_iter) # may raise StopIteration File "C:\Users\Guillaume\miniconda3\envs\ia\lib\site-packages\torch\utils\data\sampler.py", line 226, in iter for idx in self.sampler: File "C:\Users\Guillaume\miniconda3\envs\ia\lib\site-packages\torch\utils\data\sampler.py", line 124, in iter yield from torch.randperm(n, generator=generator).tolist() RuntimeError: Expected a 'cuda' device type for generator but found 'cpu'
Even I have the same issue. Were you able to solve it?
I changed shuffle = True to shuffle = False in the train.py file and it worked for me. In other words, from this
data_loader = data.DataLoader(dataset, args.batch_size,
num_workers=args.num_workers,
shuffle=True, collate_fn=detection_collate,
pin_memory=True)
to this
data_loader = data.DataLoader(dataset, args.batch_size,
num_workers=args.num_workers,
shuffle=False, collate_fn=detection_collate,
pin_memory=True)
I changed shuffle = True to shuffle = False in the train.py file and it worked for me. In other words, from this
data_loader = data.DataLoader(dataset, args.batch_size, num_workers=args.num_workers, shuffle=True, collate_fn=detection_collate, pin_memory=True)to this
data_loader = data.DataLoader(dataset, args.batch_size, num_workers=args.num_workers, shuffle=False, collate_fn=detection_collate, pin_memory=True)
- Thank you so much for the fix. But this gives out a warning, which I don't know if it's safe to continue training:
[W pthreadpool-cpp.cc:90] Warning: Leaking Caffe2 thread-pool after fork. (function pthreadpool)
[W pthreadpool-cpp.cc:90] Warning: Leaking Caffe2 thread-pool after fork. (function pthreadpool)
[W pthreadpool-cpp.cc:90] Warning: Leaking Caffe2 thread-pool after fork. (function pthreadpool)
[W pthreadpool-cpp.cc:90] Warning: Leaking Caffe2 thread-pool after fork. (function pthreadpool)
I changed generator = torch.Generator() to generator =torch.Generator(device='cuda') in torch\utils\data\sampler.py in line 115. It worked for me.
I have meet the same question in other project, and find this issue. But change 'shuffle=False' in and change 'torch\utils\data\sampler.py' is not what I want.
Then I find the source code for the error, in line 124 in torch\utils\data\sampler.py.
yield from torch.randperm(n, generator=generator).tolist()
the return tensor device of torch.randperm is the current device (see 'device' of api, https://pytorch.org/docs/stable/generated/torch.randperm.html)
In my project, I use ' torch.set_default_tensor_type('torch.cuda.FloatTensor')', and when I delete it , my problem get solved.
I have meet the same question in other project, and find this issue
yield from torch.randperm(n, generator=generator).tolist() how to solve this error
I have meet the same question in other project, and find this issue. But change 'shuffle=False' in and change 'torch\utils\data\sampler.py' is not what I want.
Then I find the source code for the error, in line 124 in torch\utils\data\sampler.py.
yield from torch.randperm(n, generator=generator).tolist()the return tensor device of torch.randperm is the current device (see 'device' of api, https://pytorch.org/docs/stable/generated/torch.randperm.html)
In my project, I use ' torch.set_default_tensor_type('torch.cuda.FloatTensor')', and when I delete it , my problem get solved.
how to solve this...... even I have same error.....Can u say
@monisha21898 In my project, I use ' torch.set_default_tensor_type('torch.cuda.FloatTensor')', and when I delete it , my problem get solved.
The generator is created in cpu, and the device of return value of "torch.randperm" is follow the current device of project. I set torch.set_default_tensor_type('torch.cuda.FloatTensor') before my project, that's means the current device is cuda, so when I delete it, my problem get solved.
hi When I deleted this line torch.set_default_tensor_type('torch.cuda.FloatTensor') still I'm getting same error Then I find the source code for the error, in line 124 in torch\utils\data\sampler.py. yield from torch.randperm(n, generator=generator).tolist()
You may need to debug, see the device of 'n','generator', and the current device of your project. This should not be wrong if all three are the same
Hi Sir , Can u help us to solve this issue
Load Model and Summary
from torchsummary import summary torch.set_default_dtype(torch.float32)torch.set_default_tensor_type('torch.cuda.FloatTensor')torch.backends.cudnn.enabledmodel =modules.RadioWNet(phase="firstU")model.cuda()summary(model, input_size=(2, 256,256))
Training Loop
#Adapted from https://github.com/usuyama/pytorch-unet import torchimport torch.optim as optimfrom torch.optim import lr_schedulerimport timeimport copyfrom collections import defaultdictimport torch.nn.functional as Fimport torch.nn as nn def calc_loss_dense(pred, target, metrics): criterion = nn.MSELoss() loss = criterion(pred, target) metrics['loss'] += loss.data.cpu().numpy() * target.size(0)
return loss
def calc_loss_sparse(pred, target, samples, metrics, num_samples): criterion = nn.MSELoss() loss = criterion(samplespred, samplestarget)*(256**2)/num_samples metrics['loss'] += loss.data.cpu().numpy() * target.size(0)
return loss
def print_metrics(metrics, epoch_samples, phase): outputs1 = [] outputs2 = [] for k in metrics.keys(): outputs1.append("{}: {:4f}".format(k, metrics[k] / epoch_samples))
print("{}: {}".format(phase, ", ".join(outputs1)))
def train_model(model, optimizer, scheduler, num_epochs=50, WNetPhase="firstU", targetType="dense", num_samples=300): # WNetPhase: traine first U and freez second ("firstU"), or vice verse ("secondU"). # targetType: train against dense images ("dense") or sparse measurements ("sparse") best_model_wts = copy.deepcopy(model.state_dict()) best_loss = 1e10
for epoch in range(num_epochs):
print('Epoch {}/{}'.format(epoch, num_epochs - 1))
print('-' * 10)
since = time.time()
# Each epoch has a training and validation phase
for phase in ['train', 'val']:
if phase == 'train':
scheduler.step()
for param_group in optimizer.param_groups:
print("learning rate", param_group['lr'])
model.train() # Set model to training mode
else:
model.eval() # Set model to evaluate mode
metrics = defaultdict(float)
epoch_samples = 0
if targetType=="dense":
for inputs, targets in dataloaders[phase]:
inputs = inputs.to(device)
targets = targets.to(device)
# zero the parameter gradients
optimizer.zero_grad()
# forward
# track history if only in train
with torch.set_grad_enabled(phase == 'train'):
[outputs1,outputs2] = model(inputs)
if WNetPhase=="firstU":
loss = calc_loss_dense(outputs1, targets, metrics)
else:
loss = calc_loss_dense(outputs2, targets, metrics)
# backward + optimize only if in training phase
if phase == 'train':
loss.backward()
optimizer.step()
# statistics
epoch_samples += inputs.size(0)
elif targetType=="sparse":
for inputs, targets, samples in dataloaders[phase]:
inputs = inputs.to(device)
targets = targets.to(device)
samples = samples.to(device)
# zero the parameter gradients
optimizer.zero_grad()
# forward
# track history if only in train
with torch.set_grad_enabled(phase == 'train'):
[outputs1,outputs2] = model(inputs)
if WNetPhase=="firstU":
loss = calc_loss_sparse(outputs1, targets,
samples, metrics, num_samples) else: loss = calc_loss_sparse(outputs2, targets, samples, metrics, num_samples)
# backward + optimize only if in training phase
if phase == 'train':
loss.backward()
optimizer.step()
# statistics
epoch_samples += inputs.size(0)
print_metrics(metrics, epoch_samples, phase)
epoch_loss = metrics['loss'] / epoch_samples
# deep copy the model
if phase == 'val' and epoch_loss < best_loss:
print("saving best model")
best_loss = epoch_loss
best_model_wts = copy.deepcopy(model.state_dict())
time_elapsed = time.time() - since
print('{:.0f}m {:.0f}s'.format(time_elapsed // 60, time_elapsed % 60))
print('Best val loss: {:4f}'.format(best_loss))
# load best model weights
model.load_state_dict(best_model_wts)
return model
Training First UNet
import torchimport torch.optim as optimfrom torch.optim import lr_schedulerimport timeimport copy device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")print(device)
optimizer_ft = optim.Adam(filter(lambda p: p.requires_grad, model.parameters()), lr=1e-4) exp_lr_scheduler = lr_scheduler.StepLR(optimizer_ft, step_size=30, gamma=0.1) model = train_model(model, optimizer_ft, exp_lr_scheduler)
ERROR IN TRAINING MODEL:
Epoch 0/49
learning rate 0.0001
RuntimeError Traceback (most recent call last)
8 frames
/usr/local/lib/python3.7/dist-packages/torch/utils/data/dataloader.py https://localhost:8080/# in iter(self) 357 return self._iterator 358 else: --> 359 return self._get_iterator() 360 361 @property
/usr/local/lib/python3.7/dist-packages/torch/utils/data/dataloader.py https://localhost:8080/# in _get_iterator(self) 303 else: 304 self.check_worker_number_rationality() --> 305 return _MultiProcessingDataLoaderIter(self) 306 307 @property
/usr/local/lib/python3.7/dist-packages/torch/utils/data/dataloader.py https://localhost:8080/# in init(self, loader) 942 _utils.signal_handling._set_SIGCHLD_handler() 943 self._worker_pids_set = True --> 944 self._reset(loader, first_iter=True) 945 946 def _reset(self, loader, first_iter=True):
/usr/local/lib/python3.7/dist-packages/torch/utils/data/dataloader.py https://localhost:8080/# in _reset(self, loader, first_iter) 973 # prime the prefetch loop 974 for _ in range(self._prefetch_factor * self._num_workers): --> 975 self._try_put_index() 976 977 def _try_get_data(self, timeout=_utils.MP_STATUS_CHECK_INTERVAL):
/usr/local/lib/python3.7/dist-packages/torch/utils/data/dataloader.py https://localhost:8080/# in _try_put_index(self) 1207 1208 try: -> 1209 index = self._next_index() 1210 except StopIteration: 1211 return
/usr/local/lib/python3.7/dist-packages/torch/utils/data/dataloader.py https://localhost:8080/# in _next_index(self) 510 511 def _next_index(self): --> 512 return next(self._sampler_iter) # may raise StopIteration 513 514 def _next_data(self):
/usr/local/lib/python3.7/dist-packages/torch/utils/data/sampler.py https://localhost:8080/# in iter(self) 224 def iter(self) -> Iterator[List[int]]: 225 batch = [] --> 226 for idx in self.sampler: 227 batch.append(idx) 228 if len(batch) == self.batch_size:
/usr/local/lib/python3.7/dist-packages/torch/utils/data/sampler.py https://localhost:8080/# in iter(self) 122 yield from torch.randint(high=n, size=(self.num_samples % 32,), dtype=torch.int64, generator=generator).tolist() 123 else: --> 124 yield from torch.randperm(n, generator=generator).tolist() 125 126 def len(self) -> int:
RuntimeError: Expected a 'cuda' device type for generator but found 'cpu'
Sampler.py File
import torch from torch import Tensor
from typing import Iterator, Optional, Sequence, List, TypeVar, Generic, Sized
T_co = TypeVar('T_co', covariant=True)
class Sampler(Generic[T_co]): r"""Base class for all Samplers.
Every Sampler subclass has to provide an :meth:`__iter__` method,
providing a
way to iterate over indices of dataset elements, and a
:meth:__len__ method
that returns the length of the returned iterators.
.. note:: The :meth:`__len__` method isn't strictly required by
:class:`~torch.utils.data.DataLoader`, but is expected in any
calculation involving the length of a
:class:~torch.utils.data.DataLoader.
"""
def __init__(self, data_source: Optional[Sized]) -> None:
pass
def __iter__(self) -> Iterator[T_co]:
raise NotImplementedError
# NOTE [ Lack of Default `__len__` in Python Abstract Base Classes ]
#
# Many times we have an abstract class representing a collection/iterable of
# data, e.g., `torch.utils.data.Sampler`, with its subclasses optionally
# implementing a `__len__` method. In such cases, we must make sure to not
# provide a default implementation, because both straightforward default
# implementations have their issues:
#
# + `return NotImplemented`:
# Calling `len(subclass_instance)` raises:
# TypeError: 'NotImplementedType' object cannot be
interpreted as an integer
#
# + raise NotImplementedError():
# This prevents triggering some fallback behavior. E.g., the built-in
# list(X) tries to call len(X) first, and executes a different code
# path if the method is not found or NotImplemented is returned, while
# raising an NotImplementedError will propagate and and make the call
# fail where it could have use __iter__ to complete the call.
#
# Thus, the only two sensible things to do are
#
# + not provide a default __len__.
#
# + raise a TypeError instead, which is what Python uses when users call
# a method that is not defined on an object.
# @.*** verifies that this works on at least Python 3.7.)
class SequentialSampler(Sampler[int]): r"""Samples elements sequentially, always in the same order.
Args:
data_source (Dataset): dataset to sample from
"""
data_source: Sized
def __init__(self, data_source: Sized) -> None:
self.data_source = data_source
def __iter__(self) -> Iterator[int]:
return iter(range(len(self.data_source)))
def __len__(self) -> int:
return len(self.data_source)
class RandomSampler(Sampler[int]):
r"""Samples elements randomly. If without replacement, then sample
from a shuffled dataset.
If with replacement, then user can specify :attr:num_samples to draw.
Args:
data_source (Dataset): dataset to sample from
replacement (bool): samples are drawn on-demand with
replacement if True, default=False
num_samples (int): number of samples to draw,
default=len(dataset). This argument
is supposed to be specified only when replacement is True.
generator (Generator): Generator used in sampling.
"""
data_source: Sized
replacement: bool
def __init__(self, data_source: Sized, replacement: bool = False,
num_samples: Optional[int] = None, generator=None) -> None:
self.data_source = data_source
self.replacement = replacement
self._num_samples = num_samples
self.generator = generator
if not isinstance(self.replacement, bool):
raise TypeError("replacement should be a boolean value, but got "
"replacement={}".format(self.replacement))
if self._num_samples is not None and not replacement:
raise ValueError("With replacement=False, num_samples
should not be specified, " "since a random permute will be performed.")
if not isinstance(self.num_samples, int) or self.num_samples <= 0:
raise ValueError("num_samples should be a positive integer "
"value, but got
num_samples={}".format(self.num_samples))
@property
def num_samples(self) -> int:
# dataset size might change at runtime
if self._num_samples is None:
return len(self.data_source)
return self._num_samples
def __iter__(self) -> Iterator[int]:
n = len(self.data_source)
if self.generator is None:
generator = torch.Generator()
generator.manual_seed(int(torch.empty((),
dtype=torch.int64).random_().item())) else: generator = self.generator if self.replacement: for _ in range(self.num_samples // 32): yield from torch.randint(high=n, size=(32,), dtype=torch.int64, generator=generator).tolist() yield from torch.randint(high=n, size=(self.num_samples % 32,), dtype=torch.int64, generator=generator).tolist() else: yield from torch.randperm(n, generator=generator).tolist()
def __len__(self) -> int:
return self.num_samples
class SubsetRandomSampler(Sampler[int]): r"""Samples elements randomly from a given list of indices, without replacement.
Args:
indices (sequence): a sequence of indices
generator (Generator): Generator used in sampling.
"""
indices: Sequence[int]
def __init__(self, indices: Sequence[int], generator=None) -> None:
self.indices = indices
self.generator = generator
def __iter__(self) -> Iterator[int]:
return (self.indices[i] for i in
torch.randperm(len(self.indices), generator=self.generator))
def __len__(self) -> int:
return len(self.indices)
class WeightedRandomSampler(Sampler[int]):
r"""Samples elements from [0,..,len(weights)-1] with given
probabilities (weights).
Args:
weights (sequence) : a sequence of weights, not necessary
summing up to one
num_samples (int): number of samples to draw
replacement (bool): if True, samples are drawn with replacement.
If not, they are drawn without replacement, which means that when a
sample index is drawn for a row, it cannot be drawn again
for that row.
generator (Generator): Generator used in sampling.
Example:
>>> list(WeightedRandomSampler([0.1, 0.9, 0.4, 0.7, 3.0, 0.6],
5, replacement=True)) [4, 4, 1, 4, 5] >>> list(WeightedRandomSampler([0.9, 0.4, 0.05, 0.2, 0.3, 0.1], 5, replacement=False)) [0, 1, 4, 3, 2] """ weights: Tensor num_samples: int replacement: bool
def __init__(self, weights: Sequence[float], num_samples: int,
replacement: bool = True, generator=None) -> None:
if not isinstance(num_samples, int) or isinstance(num_samples,
bool) or
num_samples <= 0:
raise ValueError("num_samples should be a positive integer "
"value, but got
num_samples={}".format(num_samples))
if not isinstance(replacement, bool):
raise ValueError("replacement should be a boolean value, but got "
"replacement={}".format(replacement))
self.weights = torch.as_tensor(weights, dtype=torch.double)
self.num_samples = num_samples
self.replacement = replacement
self.generator = generator
def __iter__(self) -> Iterator[int]:
rand_tensor = torch.multinomial(self.weights,
self.num_samples, self.replacement, generator=self.generator) return iter(rand_tensor.tolist())
def __len__(self) -> int:
return self.num_samples
class BatchSampler(Sampler[List[int]]): r"""Wraps another sampler to yield a mini-batch of indices.
Args:
sampler (Sampler or Iterable): Base sampler. Can be any iterable object
batch_size (int): Size of mini-batch.
drop_last (bool): If ``True``, the sampler will drop the last batch if
its size would be less than ``batch_size``
Example:
>>> list(BatchSampler(SequentialSampler(range(10)),
batch_size=3, drop_last=False)) [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9]] >>> list(BatchSampler(SequentialSampler(range(10)), batch_size=3, drop_last=True)) [[0, 1, 2], [3, 4, 5], [6, 7, 8]] """
def __init__(self, sampler: Sampler[int], batch_size: int,
drop_last: bool) -> None:
# Since collections.abc.Iterable does not check for __getitem__, which
# is one way for an object to be an iterable, we don't do an
isinstance
# check here.
if not isinstance(batch_size, int) or isinstance(batch_size, bool) or
batch_size <= 0:
raise ValueError("batch_size should be a positive integer value, "
"but got batch_size={}".format(batch_size))
if not isinstance(drop_last, bool):
raise ValueError("drop_last should be a boolean value, but got "
"drop_last={}".format(drop_last))
self.sampler = sampler
self.batch_size = batch_size
self.drop_last = drop_last
def __iter__(self) -> Iterator[List[int]]:
batch = []
for idx in self.sampler:
batch.append(idx)
if len(batch) == self.batch_size:
yield batch
batch = []
if len(batch) > 0 and not self.drop_last:
yield batch
def __len__(self) -> int:
# Can only be called if self.sampler has __len__ implemented
# We cannot enforce this condition, so we turn off typechecking for the
# implementation below.
# Somewhat related: see NOTE [ Lack of Default `__len__` in
Python Abstract Base Classes ] if self.drop_last: return len(self.sampler) // self.batch_size # type: ignore[arg-type] else: return (len(self.sampler) + self.batch_size - 1) // self.batch_size # type: ignore[arg-type]
Dataloaders.py File*
r"""Definition of the DataLoader and associated iterators that subclass _BaseDataLoaderIter
To support these two classes, in ./_utils we define many utility methods and
functions to be run in multiprocessing. E.g., the data loading worker loop is
in ./_utils/worker.py.
"""
import os import threading import itertools import warnings import queue from typing import Any, Callable, TypeVar, Generic, Sequence, List, Optional
import multiprocessing as python_multiprocessing import torch import torch.multiprocessing as multiprocessing from torch._utils import ExceptionWrapper from torch._six import string_classes
from . import IterableDataset, Sampler, SequentialSampler, RandomSampler, BatchSampler, Dataset from . import _utils
T_co = TypeVar('T_co', covariant=True) T = TypeVar('T') _worker_init_fn_t = Callable[[int], None]
Ideally we would parameterize DataLoader by the return type of
collate_fn, but there is currently no way to have that
type parameter set to a default value if the user doesn't pass in a
custom 'collate_fn'.
See https://github.com/python/mypy/issues/3737.
_collate_fn_t = Callable[[List[T]], Any]
This function used to be defined in this file. However, it was moved to
_utils/collate.py. Although it is rather hard to access this from user land
(one has to explicitly directly import torch.utils.data.dataloader), there
probably is user code out there using it. This aliasing maintains BC in this
aspect.
default_collate: _collate_fn_t = _utils.collate.default_collate
get_worker_info = _utils.worker.get_worker_info
class _DatasetKind(object): Map = 0 Iterable = 1
@staticmethod
def create_fetcher(kind, dataset, auto_collation, collate_fn, drop_last):
if kind == _DatasetKind.Map:
return _utils.fetch._MapDatasetFetcher(dataset,
auto_collation, collate_fn, drop_last) else: return _utils.fetch._IterableDatasetFetcher(dataset, auto_collation, collate_fn, drop_last)
class _InfiniteConstantSampler(Sampler):
r"""Analogous to itertools.repeat(None, None).
Used as sampler for :class:~torch.utils.data.IterableDataset.
Args:
data_source (Dataset): dataset to sample from
"""
def __init__(self):
super(_InfiniteConstantSampler, self).__init__(None)
def __iter__(self):
while True:
yield None
class DataLoader(Generic[T_co]): r""" Data loader. Combines a dataset and a sampler, and provides an iterable over the given dataset.
The :class:`~torch.utils.data.DataLoader` supports both map-style and
iterable-style datasets with single- or multi-process loading, customizing
loading order and optional automatic batching (collation) and
memory pinning.
See :py:mod:`torch.utils.data` documentation page for more details.
Args:
dataset (Dataset): dataset from which to load the data.
batch_size (int, optional): how many samples per batch to load
(default: ``1``).
shuffle (bool, optional): set to ``True`` to have the data reshuffled
at every epoch (default: ``False``).
sampler (Sampler or Iterable, optional): defines the strategy to draw
samples from the dataset. Can be any ``Iterable`` with ``__len__``
implemented. If specified, :attr:`shuffle` must not be specified.
batch_sampler (Sampler or Iterable, optional): like :attr:`sampler`, but
returns a batch of indices at a time. Mutually exclusive with
:attr:`batch_size`, :attr:`shuffle`, :attr:`sampler`,
and :attr:`drop_last`.
num_workers (int, optional): how many subprocesses to use for data
loading. ``0`` means that the data will be loaded in the
main process.
(default: 0)
collate_fn (callable, optional): merges a list of samples to form a
mini-batch of Tensor(s). Used when using batched loading from a
map-style dataset.
pin_memory (bool, optional): If True, the data loader will
copy Tensors
into CUDA pinned memory before returning them. If your
data elements
are a custom type, or your :attr:collate_fn returns a
batch that is a custom type,
see the example below.
drop_last (bool, optional): set to True to drop the last
incomplete batch,
if the dataset size is not divisible by the batch size. If
False and
the size of dataset is not divisible by the batch size,
then the last batch
will be smaller. (default: False)
timeout (numeric, optional): if positive, the timeout value
for collecting a batch
from workers. Should always be non-negative. (default: 0)
worker_init_fn (callable, optional): If not None, this
will be called on each
worker subprocess with the worker id (an int in [0, num_workers - 1]) as
input, after seeding and before data loading. (default: None)
generator (torch.Generator, optional): If not None, this
RNG will be used
by RandomSampler to generate random indexes and
multiprocessing to generate
base_seed for workers. (default: None)
prefetch_factor (int, optional, keyword-only arg): Number of
samples loaded
in advance by each worker. 2 means there will be a total of
2 * num_workers samples prefetched across all workers.
(default: 2)
persistent_workers (bool, optional): If True, the data
loader will not shutdown
the worker processes after a dataset has been consumed
once. This allows to
maintain the workers Dataset instances alive. (default: False)
.. warning:: If the ``spawn`` start method is used, :attr:`worker_init_fn`
cannot be an unpicklable object, e.g., a lambda function. See
:ref:`multiprocessing-best-practices` on more details related
to multiprocessing in PyTorch.
.. warning:: ``len(dataloader)`` heuristic is based on the length
of the sampler used.
When :attr:dataset is an
:class:~torch.utils.data.IterableDataset,
it instead returns an estimate based on
len(dataset) / batch_size, with proper
rounding depending on :attr:drop_last, regardless
of multi-process loading
configurations. This represents the best guess
PyTorch can make because PyTorch
trusts user :attr:dataset code in correctly
handling multi-process
loading to avoid duplicate data.
However, if sharding results in multiple workers
having incomplete last batches,
this estimate can still be inaccurate, because (1) an
otherwise complete batch can
be broken into multiple ones and (2) more than one
batch worth of samples can be
dropped when :attr:drop_last is set. Unfortunately,
PyTorch can not detect such
cases in general.
See `Dataset Types`_ for more details on these two
types of datasets and how
:class:~torch.utils.data.IterableDataset interacts with
Multi-process data loading_.
.. warning:: See :ref:`reproducibility`, and
:ref:dataloader-workers-random-seed, and
:ref:data-loading-randomness notes for random seed
related questions.
"""
dataset: Dataset[T_co]
batch_size: Optional[int]
num_workers: int
pin_memory: bool
drop_last: bool
timeout: float
sampler: Sampler
prefetch_factor: int
_iterator : Optional['_BaseDataLoaderIter']
__initialized = False
def __init__(self, dataset: Dataset[T_co], batch_size: Optional[int] = 1,
shuffle: bool = False, sampler: Optional[Sampler[int]] = None,
batch_sampler: Optional[Sampler[Sequence[int]]] = None,
num_workers: int = 0, collate_fn:
Optional[_collate_fn_t] = None, pin_memory: bool = False, drop_last: bool = False, timeout: float = 0, worker_init_fn: Optional[_worker_init_fn_t] = None, multiprocessing_context=None, generator=None, *, prefetch_factor: int = 2, persistent_workers: bool = False): torch._C._log_api_usage_once("python.data_loader")
if num_workers < 0:
raise ValueError('num_workers option should be non-negative; '
'use num_workers=0 to disable multiprocessing.')
if timeout < 0:
raise ValueError('timeout option should be non-negative')
if num_workers == 0 and prefetch_factor != 2:
raise ValueError('prefetch_factor option could only be
specified in multiprocessing.' 'let num_workers > 0 to enable multiprocessing.') assert prefetch_factor > 0
if persistent_workers and num_workers == 0:
raise ValueError('persistent_workers option needs num_workers > 0')
self.dataset = dataset
self.num_workers = num_workers
self.prefetch_factor = prefetch_factor
self.pin_memory = pin_memory
self.timeout = timeout
self.worker_init_fn = worker_init_fn
self.multiprocessing_context = multiprocessing_context
# Arg-check dataset related before checking samplers because we want to
# tell users that iterable-style datasets are incompatible with custom
# samplers first, so that they don't learn that this combo doesn't work
# after spending time fixing the custom sampler errors.
if isinstance(dataset, IterableDataset):
self._dataset_kind = _DatasetKind.Iterable
# NOTE [ Custom Samplers and IterableDataset ]
#
# `IterableDataset` does not support custom `batch_sampler` or
# `sampler` since the key is irrelevant (unless we support
# generator-style dataset one day...).
#
# For `sampler`, we always create a dummy sampler. This is an
# infinite sampler even when the dataset may have an implemented
# finite `__len__` because in multi-process data loading, naive
# settings will return duplicated data (which may be desired), and
# thus using a sampler with length matching that of dataset will
# cause data lost (you may have duplicates of the first couple
# batches, but never see anything afterwards). Therefore,
# `Iterabledataset` always uses an infinite sampler, an instance of
# `_InfiniteConstantSampler` defined above.
#
# A custom `batch_sampler` essentially only controls the batch size.
# However, it is unclear how useful it would be since an
iterable-style # dataset can handle that within itself. Moreover, it is pointless # in multi-process data loading as the assignment order of batches # to workers is an implementation detail so users can not control # how to batchify each worker's iterable. Thus, we disable this # option. If this turns out to be useful in future, we can re-enable # this, and support custom samplers that specify the assignments to # specific workers. if shuffle is not False: raise ValueError( "DataLoader with IterableDataset: expected unspecified " "shuffle option, but got shuffle={}".format(shuffle)) elif sampler is not None: # See NOTE [ Custom Samplers and IterableDataset ] raise ValueError( "DataLoader with IterableDataset: expected unspecified " "sampler option, but got sampler={}".format(sampler)) elif batch_sampler is not None: # See NOTE [ Custom Samplers and IterableDataset ] raise ValueError( "DataLoader with IterableDataset: expected unspecified " "batch_sampler option, but got batch_sampler={}".format(batch_sampler)) else: self._dataset_kind = _DatasetKind.Map
if sampler is not None and shuffle:
raise ValueError('sampler option is mutually exclusive with '
'shuffle')
if batch_sampler is not None:
# auto_collation with custom batch_sampler
if batch_size != 1 or shuffle or sampler is not None or drop_last:
raise ValueError('batch_sampler option is mutually exclusive '
'with batch_size, shuffle, sampler, and '
'drop_last')
batch_size = None
drop_last = False
elif batch_size is None:
# no auto_collation
if drop_last:
raise ValueError('batch_size=None option disables
auto-batching ' 'and is mutually exclusive with drop_last')
if sampler is None: # give default samplers
if self._dataset_kind == _DatasetKind.Iterable:
# See NOTE [ Custom Samplers and IterableDataset ]
sampler = _InfiniteConstantSampler()
else: # map-style
if shuffle:
# Cannot statically verify that dataset is Sized
# Somewhat related: see NOTE [ Lack of Default
__len__ in Python Abstract Base Classes ]
sampler = RandomSampler(dataset,
generator=generator) # type: ignore[arg-type]
else:
sampler = SequentialSampler(dataset) # type:
ignore[arg-type]
if batch_size is not None and batch_sampler is None:
# auto_collation without custom batch_sampler
batch_sampler = BatchSampler(sampler, batch_size, drop_last)
self.batch_size = batch_size
self.drop_last = drop_last
self.sampler = sampler
self.batch_sampler = batch_sampler
self.generator = generator
if collate_fn is None:
if self._auto_collation:
collate_fn = _utils.collate.default_collate
else:
collate_fn = _utils.collate.default_convert
self.collate_fn = collate_fn
self.persistent_workers = persistent_workers
self.__initialized = True
self._IterableDataset_len_called = None # See NOTE [
IterableDataset and len ]
self._iterator = None
self.check_worker_number_rationality()
def _get_iterator(self) -> '_BaseDataLoaderIter':
if self.num_workers == 0:
return _SingleProcessDataLoaderIter(self)
else:
self.check_worker_number_rationality()
return _MultiProcessingDataLoaderIter(self)
@property
def multiprocessing_context(self):
return self.__multiprocessing_context
@multiprocessing_context.setter
def multiprocessing_context(self, multiprocessing_context):
if multiprocessing_context is not None:
if self.num_workers > 0:
if isinstance(multiprocessing_context, string_classes):
valid_start_methods =
multiprocessing.get_all_start_methods() if multiprocessing_context not in valid_start_methods: raise ValueError( ('multiprocessing_context option ' 'should specify a valid start method in {!r}, but got '
'multiprocessing_context={!r}').format(valid_start_methods, multiprocessing_context)) # error: Argument 1 to "get_context" has incompatible type "Union[str, bytes]"; expected "str" [arg-type] multiprocessing_context = multiprocessing.get_context(multiprocessing_context) # type: ignore[arg-type]
if not isinstance(multiprocessing_context,
python_multiprocessing.context.BaseContext): raise TypeError(('multiprocessing_context option should be a valid context ' 'object or a string specifying the start method, but got '
'multiprocessing_context={}').format(multiprocessing_context)) else: raise ValueError(('multiprocessing_context can only be used with ' 'multi-process loading (num_workers
0), but got ' 'num_workers={}').format(self.num_workers))
self.__multiprocessing_context = multiprocessing_context
def __setattr__(self, attr, val):
if self.__initialized and attr in (
'batch_size', 'batch_sampler', 'sampler', 'drop_last',
'dataset', 'persistent_workers'): raise ValueError('{} attribute should not be set after {} is ' 'initialized'.format(attr, self.class.name))
super(DataLoader, self).__setattr__(attr, val)
# We quote '_BaseDataLoaderIter' since it isn't defined yet and
the definition can't be moved up # since '_BaseDataLoaderIter' references 'DataLoader'. def iter(self) -> '_BaseDataLoaderIter': # When using a single worker the returned iterator should be # created everytime to avoid reseting its state # However, in the case of a multiple workers iterator # the iterator is only created once in the lifetime of the # DataLoader object so that workers can be reused if self.persistent_workers and self.num_workers > 0: if self._iterator is None: self._iterator = self._get_iterator() else: self._iterator._reset(self) return self._iterator else: return self._get_iterator()
@property
def _auto_collation(self):
return self.batch_sampler is not None
@property
def _index_sampler(self):
# The actual sampler used for generating indices for `_DatasetFetcher`
# (see _utils/fetch.py) to read data at each time. This would be
# `.batch_sampler` if in auto-collation mode, and `.sampler` otherwise.
# We can't change `.sampler` and `.batch_sampler` attributes for BC
# reasons.
if self._auto_collation:
return self.batch_sampler
else:
return self.sampler
def __len__(self) -> int:
if self._dataset_kind == _DatasetKind.Iterable:
# NOTE [ IterableDataset and __len__ ]
#
# For `IterableDataset`, `__len__` could be inaccurate
when one naively
# does multi-processing data loading, since the samples
will be duplicated.
# However, no real use case should be actually using that
behavior, so
# it should count as a user error. We should generally trust user
# code to do the proper thing (e.g., configure each
replica differently
# in __iter__), and give us the correct __len__ if
they choose to
# implement it (this will still throw if the dataset does
not implement
# a __len__).
#
# To provide a further warning, we track if __len__ was
called on the
# DataLoader, save the returned value in
self._len_called, and warn
# if the iterator ends up yielding more than this number of samples.
# Cannot statically verify that dataset is Sized
length = self._IterableDataset_len_called =
len(self.dataset) # type: ignore[assignment, arg-type] if self.batch_size is not None: # IterableDataset doesn't allow custom sampler or batch_sampler from math import ceil if self.drop_last: length = length // self.batch_size else: length = ceil(length / self.batch_size) return length else: return len(self._index_sampler)
def check_worker_number_rationality(self):
# This function check whether the dataloader's worker number
is rational based on
# current system's resource. Current rule is that if the
number of workers this
# Dataloader will create is bigger than the number of logical
cpus that is allowed to
# use, than we will pop up a warning to let user pay attention.
#
# eg. If current system has 2 physical CPUs with 16 cores
each. And each core support 2
# threads, then the total logical cpus here is 2 * 16 * 2
= 64. Let's say current
# DataLoader process can use half of them which is 32,
then the rational max number of
# worker that initiated from this process is 32.
# Now, let's say the created DataLoader has num_works =
40, which is bigger than 32.
# So the warning message is triggered to notify the user
to lower the worker number if
# necessary.
#
#
# [Note] Please note that this function repects cpuset only
when os.sched_getaffinity is
# available (available in most of Linux system, but not
OSX and Windows).
# When os.sched_getaffinity is not available,
os.cpu_count() is called instead, but
# it doesn't repect cpuset.
# We don't take threading into account since each
worker process is single threaded
# at this time.
#
# We don't set any threading flags (eg.
OMP_NUM_THREADS, MKL_NUM_THREADS, etc)
# other than torch.set_num_threads to 1 in the worker
process, if the passing
# in functions use 3rd party modules that rely on those
threading flags to determine
# how many thread to create (eg. numpy, etc), then it
is caller's responsibility to
# set those flags correctly.
def _create_warning_msg(num_worker_suggest,
num_worker_created, cpuset_checked):
suggested_max_worker_msg = ((
"Our suggested max number of worker in current system
is {}{}, which is smaller "
"than what this DataLoader is going to create.").format(
num_worker_suggest,
("" if cpuset_checked else " (cpuset is not
taken into account)"))
) if num_worker_suggest is not None else (
"DataLoader is not able to compute a suggested max
number of worker in current system.")
warn_msg = (
"This DataLoader will create {} worker processes in total. {} "
"Please be aware that excessive worker creation might
get DataLoader running slow or even freeze, " "lower the worker number to avoid potential slowness/freeze if necessary.").format( num_worker_created, suggested_max_worker_msg) return warn_msg
if not self.num_workers or self.num_workers == 0:
return
# try to compute a suggested max number of worker based on
system's resource max_num_worker_suggest = None cpuset_checked = False if hasattr(os, 'sched_getaffinity'): try: max_num_worker_suggest = len(os.sched_getaffinity(0)) cpuset_checked = True except Exception: pass if max_num_worker_suggest is None: # os.cpu_count() could return Optional[int] # get cpu count first and check None in order to satify mypy check cpu_count = os.cpu_count() if cpu_count is not None: max_num_worker_suggest = cpu_count
if max_num_worker_suggest is None:
warnings.warn(_create_warning_msg(
max_num_worker_suggest,
self.num_workers,
cpuset_checked))
return
if self.num_workers > max_num_worker_suggest:
warnings.warn(_create_warning_msg(
max_num_worker_suggest,
self.num_workers,
cpuset_checked))
class _BaseDataLoaderIter(object): def init(self, loader: DataLoader) -> None: self._dataset = loader.dataset self._dataset_kind = loader._dataset_kind self._IterableDataset_len_called = loader._IterableDataset_len_called self._auto_collation = loader._auto_collation self._drop_last = loader.drop_last self._index_sampler = loader._index_sampler self._num_workers = loader.num_workers self._prefetch_factor = loader.prefetch_factor self._pin_memory = loader.pin_memory and torch.cuda.is_available() self._timeout = loader.timeout self._collate_fn = loader.collate_fn self._sampler_iter = iter(self._index_sampler) self.base_seed = torch.empty((), dtype=torch.int64).random(generator=loader.generator).item() self._persistent_workers = loader.persistent_workers self._num_yielded = 0 self._profile_name = "enumerate(DataLoader)#{}.next".format(self.class.name)
def __iter__(self) -> '_BaseDataLoaderIter':
return self
def _reset(self, loader, first_iter=False):
self._sampler_iter = iter(self._index_sampler)
self._num_yielded = 0
self._IterableDataset_len_called = loader._IterableDataset_len_called
def _next_index(self):
return next(self._sampler_iter) # may raise StopIteration
def _next_data(self):
raise NotImplementedError
def __next__(self) -> Any:
with torch.autograd.profiler.record_function(self._profile_name):
if self._sampler_iter is None:
self._reset()
data = self._next_data()
self._num_yielded += 1
if self._dataset_kind == _DatasetKind.Iterable and \
self._IterableDataset_len_called is not None and \
self._num_yielded > self._IterableDataset_len_called:
warn_msg = ("Length of IterableDataset {} was reported
to be {} (when accessing len(dataloader)), but {} " "samples have been fetched. ").format(self._dataset, self._IterableDataset_len_called,
self._num_yielded) if self._num_workers > 0: warn_msg += ("For multiprocessing data-loading, this could be caused by not properly configuring the " "IterableDataset replica at each worker. Please see "
"https://pytorch.org/docs/stable/data.html#torch.utils.data.IterableDataset for examples.") warnings.warn(warn_msg) return data
next = __next__ # Python 2 compatibility
def __len__(self) -> int:
return len(self._index_sampler)
def __getstate__(self):
# TODO: add limited pickling support for sharing an iterator
# across multiple threads for HOGWILD.
# Probably the best way to do this is by moving the sample pushing
# to a separate thread and then just sharing the data queue
# but signalling the end is tricky without a non-blocking API
raise NotImplementedError("{} cannot be pickled",
self.class.name)
class _SingleProcessDataLoaderIter(_BaseDataLoaderIter): def init(self, loader): super(_SingleProcessDataLoaderIter, self).init(loader) assert self._timeout == 0 assert self._num_workers == 0
self._dataset_fetcher = _DatasetKind.create_fetcher(
self._dataset_kind, self._dataset, self._auto_collation,
self._collate_fn, self._drop_last)
def _next_data(self):
index = self._next_index() # may raise StopIteration
data = self._dataset_fetcher.fetch(index) # may raise StopIteration
if self._pin_memory:
data = _utils.pin_memory.pin_memory(data)
return data
class _MultiProcessingDataLoaderIter(_BaseDataLoaderIter): r"""Iterates once over the DataLoader's dataset, as specified by the sampler"""
# NOTE [ Data Loader Multiprocessing Shutdown Logic ]
#
# Preliminary:
#
# Our data model looks like this (queues are indicated with curly brackets):
#
# main process ||
# | ||
# {index_queue} ||
# | ||
# worker processes || DATA
# | ||
# {worker_result_queue} || FLOW
# | ||
# pin_memory_thread of main process || DIRECTION
# | ||
# {data_queue} ||
# | ||
# data output \/
#
# P.S. `worker_result_queue` and `pin_memory_thread` part may be omitted if
# `pin_memory=False`.
#
#
# Terminating multiprocessing logic requires very careful design. In
# particular, we need to make sure that
#
# 1. The iterator gracefully exits the workers when its last reference is
# gone or it is depleted.
#
# In this case, the workers should be gracefully exited because the
# main process may still need to continue to run, and we want cleaning
# up code in the workers to be executed (e.g., releasing GPU memory).
# Naturally, we implement the shutdown logic in `__del__` of
# DataLoaderIterator.
#
# We delay the discussion on the logic in this case until later.
#
# 2. The iterator exits the workers when the loader process and/or worker
# processes exits normally or with error.
#
# We set all workers and `pin_memory_thread` to have `daemon=True`.
#
# You may ask, why can't we make the workers non-daemonic, and
# gracefully exit using the same logic as we have in `__del__` when the
# iterator gets deleted (see 1 above)?
#
# First of all, `__del__` is **not** guaranteed to be called when
# interpreter exits. Even if it is called, by the time it executes,
# many Python core library resources may alreay be freed, and even
# simple things like acquiring an internal lock of a queue may hang.
# Therefore, in this case, we actually need to prevent `__del__` from
# being executed, and rely on the automatic termination of daemonic
# children.
#
# Thus, we register an `atexit` hook that sets a global flag
# `_utils.python_exit_status`. Since `atexit` hooks are executed in the
# reverse order of registration, we are guaranteed that this flag is
# set before library resources we use are freed (which, at least in
# CPython, is done via an `atexit` handler defined in
# `multiprocessing/util.py`
# https://github.com/python/cpython/blob/c606624af8d4cb3b4a052fb263bb983b3f87585b/Lib/multiprocessing/util.py#L320-L362
# registered when an object requiring this mechanism is first
# created, e.g., `mp.Queue`
# https://github.com/python/cpython/blob/c606624af8d4cb3b4a052fb263bb983b3f87585b/Lib/multiprocessing/context.py#L100-L103
# https://github.com/python/cpython/blob/c606624af8d4cb3b4a052fb263bb983b3f87585b/Lib/multiprocessing/queues.py#L29
# )
#
# So in `__del__`, we check if `_utils.python_exit_status` is set or
# `None` (freed), and perform no-op if so.
#
# However, simply letting library clean-up codes run can also be bad,
# because such codes (i.e., `multiprocessing.util._exit_function()`)
# include join putting threads for `mp.Queue`, which can be blocking.
# Hence, the main process putting threads are called with
# `cancel_join_thread` at creation. See later section
# [ 3b. A process won't hang when putting into a queue; ]
# for more details.
#
# Here are two example cases where library clean-up codes can run
# before `__del__` is called:
#
# 1. If we hold onto a reference to the iterator, it more often
# than not tries to do `multiprocessing` library cleaning before
# clearing the alive referenced objects
(https://github.com/pytorch/pytorch/issues/48666)
# and thus prevents our cleaning-up code to run first.
#
# 2. A similar issue araises when a DataLoader is used in
a subprocess.
# When a process ends, it shuts the all its daemonic children
# down with a SIGTERM (instead of joining them without a timeout).
# Simiarly for threads, but by a different mechanism. This fact,
# together with a few implementation details of
multiprocessing, forces
# us to make workers daemonic. All of our problems arise when a
# DataLoader is used in a subprocess, and are caused by
multiprocessing
# code which looks more or less like this:
#
# try:
# your_function_using_a_dataloader()
# finally:
# multiprocessing.util._exit_function()
#
# The joining/termination mentioned above happens inside
# _exit_function(). Now, if your_function_using_a_dataloader()
# throws, the stack trace stored in the exception will prevent the
# frame which uses DataLoaderIter to be freed. If the
frame has any
# reference to the DataLoaderIter (e.g., in a method
of the iter),
# its __del__, which starts the shutdown procedure, will not be
# called. That, in turn, means that workers aren't
notified. Attempting
# to join in _exit_function will then result in a hang.
#
# For context, _exit_function is also registered as an
atexit call.
# So it is unclear to me @.***) why this is needed in a
finally block.
# The code dates back to 2008 and there is no comment on
the original
# PEP 371 or patch https://bugs.python.org/issue3050
(containing both
# the finally block and the atexit registration) that
explains this.
#
#
# Finally, another choice is to just shutdown workers with logic in 1
# above whenever we see an error in next. This isn't ideal because
# a. It prevents users from using try-catch to resume data loading.
# b. It doesn't prevent hanging if users have references to the
# iterator.
#
# 3. All processes exit if any of them die unexpectedly by fatal signals.
#
# As shown above, the workers are set as daemonic children of the main
# process. However, automatic cleaning-up of such child processes only
# happens if the parent process exits gracefully (e.g., not via fatal
# signals like SIGKILL). So we must ensure that each process will exit
# even the process that should send/receive data to/from it were
# killed, i.e.,
#
# a. A process won't hang when getting from a queue.
#
# Even with carefully designed data dependencies (i.e., a put()
# always corresponding to a get()), hanging on get() can still
# happen when data in queue is corrupted (e.g., due to
# cancel_join_thread or unexpected exit).
#
# For child exit, we set a timeout whenever we try to get data
# from data_queue, and check the workers' status on each timeout
# and error.
# See _DataLoaderiter._get_batch() and
# _DataLoaderiter._try_get_data() for details.
#
# Additionally, for child exit on non-Windows platforms, we also
# register a SIGCHLD handler (which is supported on Windows) on
# the main process, which checks if any of the workers fail in the
# (Python) handler. This is more efficient and faster in detecting
# worker failures, compared to only using the above mechanism.
# See DataLoader.cpp and _utils/signal_handling.py
for details.
#
# For .get() calls where the sender(s) is not the workers, we
# guard them with timeouts, and check the status of the sender
# when timeout happens:
# + in the workers, the _utils.worker.ManagerWatchdog class
# checks the status of the main process.
# + if pin_memory=True, when getting from pin_memory_thread,
# check pin_memory_thread status periodically until .get()
# returns or see that pin_memory_thread died.
#
# b. A process won't hang when putting into a queue;
#
# We use mp.Queue which has a separate background thread to put
# objects from an unbounded buffer array. The background thread is
# daemonic and usually automatically joined when the process
# exits.
#
# In case that the receiver has ended abruptly while
# reading from the pipe, the join will hang forever. The usual
# solution for this in Python is calling q.cancel_join_thread,
# which prevents automatically joining it when finalizing
# (exiting).
#
# Nonetheless, cancel_join_thread must only be called when the
# queue is not going to be read from or write into by another
# process, because it may hold onto a lock or leave corrupted data
# in the queue, leading other readers/writers to hang.
#
# Hence,
# + For worker processes, we only do so (for their output
# queues, i.e., worker_result_queue) before exiting.
# + For pin_memory_thread, its output queue data_queue is a
# queue.Queue that does blocking put if the queue is full.
# So there is no above problem, but as a result, in
# _pin_memory_loop, we do need to wrap the put in a loop
# that breaks not only upon success, but also when the main
# process stops reading, i.e., is shutting down.
# + For loader process, we cancel_join_thread() for all
# _index_queues because the whole purpose of workers and
# pin_memory_thread is to serve the loader process. If
# loader process is already exiting, we don't really care if
# the queues are corrupted.
#
#
# Now let's get back to 1:
# how we gracefully exit the workers when the last reference to the
# iterator is gone.
#
# To achieve this, we implement the following logic along with the design
# choices mentioned above:
#
# workers_done_event:
# A multiprocessing.Event shared among the main process and all worker
# processes. This is used to signal the workers that the iterator is
# shutting down. After it is set, they will not send processed data to
# queues anymore, and only wait for the final None before exiting.
# done_event isn't strictly needed. I.e., we can just check for None
# from the input queue, but it allows us to skip wasting resources
# processing data if we are already shutting down.
#
# pin_memory_thread_done_event:
# A threading.Event for a similar purpose to that of
# workers_done_event, but is for the pin_memory_thread. The reason
# that separate events are needed is that pin_memory_thread reads from
# the output queue of the workers. But the workers, upon seeing that
# workers_done_event is set, only wants to see the final None, and is
# not required to flush all data in the output queue (e.g., it may call
# cancel_join_thread on that queue if its IterableDataset iterator
# happens to exhaust coincidentally, which is out of the control of the
# main process). Thus, since we will exit pin_memory_thread before the
# workers (see below), two separete events are used.
#
# NOTE: In short, the protocol is that the main process will set these
# done_events and then the corresponding processes/threads a None,
# and that they may exit at any time after receiving the None.
#
# NOTE: Using None as the final signal is valid, since normal data will
# always be a 2-tuple with the 1st element being the index of the data
# transferred (different from dataset index/key), and the 2nd being
# either the dataset key or the data sample (depending on which part
# of the data model the queue is at).
#
# [ worker processes ]
# While loader process is alive:
# Get from index_queue.
# If get anything else,
# Check workers_done_event.
# If set, continue to next iteration
# i.e., keep getting until see the None, then exit.
# Otherwise, process data:
# If is fetching from an IterableDataset and the iterator
# is exhausted, send an _IterableDatasetStopIteration
# object to signal iteration end. The main process, upon
# receiving such an object, will send None to this
# worker and not use the corresponding index_queue
# anymore.
# If timed out,
# No matter workers_done_event is set (still need to see None)
# or not, must continue to next iteration.
# (outside loop)
# If workers_done_event is set, (this can be False with
IterableDataset)
# data_queue.cancel_join_thread(). (Everything is ending here:
# main process won't read from it;
# other workers will also call
# cancel_join_thread.)
#
# [ pin_memory_thread ]
# # No need to check main thread. If this thread is alive, the main loader
# # thread must be alive, because this thread is set as daemonic.
# While pin_memory_thread_done_event is not set:
# Get from index_queue.
# If timed out, continue to get in the next iteration.
# Otherwise, process data.
# While pin_memory_thread_done_event is not set:
# Put processed data to data_queue (a queue.Queue with
blocking put)
# If timed out, continue to put in the next iteration.
# Otherwise, break, i.e., continuing to the out loop.
#
# NOTE: we don't check the status of the main thread because
# 1. if the process is killed by fatal signal, pin_memory_thread
# ends.
# 2. in other cases, either the cleaning-up in del or the
# automatic exit of daemonic thread will take care of it.
# This won't busy-wait either because .get(timeout) does not
# busy-wait.
#
# [ main process ]
# In the DataLoader Iter's __del__
# b. Exit pin_memory_thread
# i. Set pin_memory_thread_done_event.
# ii Put None in worker_result_queue.
# iii. Join the pin_memory_thread.
# iv. worker_result_queue.cancel_join_thread().
#
# c. Exit the workers.
# i. Set workers_done_event.
# ii. Put None in each worker's index_queue.
# iii. Join the workers.
# iv. Call .cancel_join_thread() on each worker's index_queue.
#
# NOTE: (c) is better placed after (b) because it may leave corrupted
# data in worker_result_queue, which pin_memory_thread
# reads from, in which case the pin_memory_thread can only
# happen at timeing out, which is slow. Nonetheless, same thing
# happens if a worker is killed by signal at unfortunate times,
# but in other cases, we are better off having a non-corrupted
# worker_result_queue for pin_memory_thread.
#
# NOTE: If pin_memory=False, there is no pin_memory_thread and (b)
# can be omitted
#
# NB: done_events isn't strictly needed. E.g., we can just check for
# None from index_queue, but it allows us to skip wasting resources
# processing indices already in index_queue if we are already shutting
# down.
def __init__(self, loader):
super(_MultiProcessingDataLoaderIter, self).__init__(loader)
assert self._num_workers > 0
assert self._prefetch_factor > 0
if loader.multiprocessing_context is None:
multiprocessing_context = multiprocessing
else:
multiprocessing_context = loader.multiprocessing_context
self._worker_init_fn = loader.worker_init_fn
self._worker_queue_idx_cycle = itertools.cycle(range(self._num_workers))
# No certainty which module multiprocessing_context is
self._worker_result_queue = multiprocessing_context.Queue() #
type: ignore[var-annotated] self._worker_pids_set = False self._shutdown = False self._workers_done_event = multiprocessing_context.Event()
self._index_queues = []
self._workers = []
for i in range(self.
I changed shuffle = True to shuffle = False in the train.py file and it worked for me. In other words, from this
data_loader = data.DataLoader(dataset, args.batch_size, num_workers=args.num_workers, shuffle=True, collate_fn=detection_collate, pin_memory=True)to this
data_loader = data.DataLoader(dataset, args.batch_size, num_workers=args.num_workers, shuffle=False, collate_fn=detection_collate, pin_memory=True)
- Thank you so much for the fix. But this gives out a warning, which I don't know if it's safe to continue training:
[W pthreadpool-cpp.cc:90] Warning: Leaking Caffe2 thread-pool after fork. (function pthreadpool) [W pthreadpool-cpp.cc:90] Warning: Leaking Caffe2 thread-pool after fork. (function pthreadpool) [W pthreadpool-cpp.cc:90] Warning: Leaking Caffe2 thread-pool after fork. (function pthreadpool) [W pthreadpool-cpp.cc:90] Warning: Leaking Caffe2 thread-pool after fork. (function pthreadpool)
This worked for me. Thank you!
I changed generator = torch.Generator() to generator =torch.Generator(device='cuda') in torch\utils\data\sampler.py in line 115. It worked for me.
This should be the correct answer, thanks.
I recommend to check the pytorch version. I came from this repo https://github.com/autonomousvision/graf and suffered from the same issue with pytorch1.9.1 but it has gone after I changed the torch version to 1.8.0!
I changed shuffle = True to shuffle = False in the train.py file and it worked for me. In other words, from this
data_loader = data.DataLoader(dataset, args.batch_size, num_workers=args.num_workers, shuffle=True, collate_fn=detection_collate, pin_memory=True)to this
data_loader = data.DataLoader(dataset, args.batch_size, num_workers=args.num_workers, shuffle=False, collate_fn=detection_collate, pin_memory=True)
Attention shuffle=False will make the final result worse, which cannot reproduce the paper's result. My solution is to implement the sampler in data.Dataloader with numpy to replace the default pytorch RandomSampler, and it works.
I changed generator = torch.Generator() to generator =torch.Generator(device='cuda') in torch\utils\data\sampler.py in line 115. It worked for me.
This should be the correct answer, thanks.
I'm having this same issue and it seems the torch.Generator() stuff seems to work. How do I locate the directory torch/utils/data/sampler.py in Google Colab?
I have meet the same question in other project, and find this issue. But change 'shuffle=False' in and change 'torch\utils\data\sampler.py' is not what I want.
Then I find the source code for the error, in line 124 in torch\utils\data\sampler.py.
yield from torch.randperm(n, generator=generator).tolist()the return tensor device of torch.randperm is the current device (see 'device' of api, https://pytorch.org/docs/stable/generated/torch.randperm.html)
In my project, I use ' torch.set_default_tensor_type('torch.cuda.FloatTensor')', and when I delete it , my problem get solved.
I'm having this same issue and it seems the torch.Generator() stuff seems to work. How do I locate the directory torch/utils/data/sampler.py in Google Colab?
I have meet the same question in other project, and find this issue. But change 'shuffle=False' in and change 'torch\utils\data\sampler.py' is not what I want.
Then I find the source code for the error, in line 124 in torch\utils\data\sampler.py.
yield from torch.randperm(n, generator=generator).tolist()the return tensor device of torch.randperm is the current device (see 'device' of api, https://pytorch.org/docs/stable/generated/torch.randperm.html)
In my project, I use ' torch.set_default_tensor_type('torch.cuda.FloatTensor')', and when I delete it , my problem get solved.
I'm having this same issue and it seems the torch.Generator() stuff seems to work. How do I locate the directory torch/utils/data/sampler.py in Google Colab? I can't find it in my directory.
data_loader = data.DataLoader(... generator=torch.Generator(device='cuda'))
this fix for me.
I changed generator = torch.Generator() to generator =torch.Generator(device='cuda') in torch\utils\data\sampler.py in line 115. It worked for me.
Ubuntu20.04, RTX3080Ti, Torch1.9.0+cu111 I fixed it by your way, thanks
data_loader = data.DataLoader(... generator=torch.Generator(device='cuda'))this fix for me.
This is the solution that works for PyTorch 1.10.0 Cu113. Mostly the proper way to resolve this as of now.
Note: Please dont set the shuffle parameters to False. This is not the actual way to solve this issue. As a bonus, it will make your training worse.
I have tried turning shuffle off as mentioned and that worked but its not correct way as it will make your training worse.
Keep the shuffle ON and follow below step, these would vary according to pytorch version:
- In file "site-packages/torch/utils/data/sampler.py" located in anaconda or wherever.
- [Modify line 116]:
generator = torch.Generator()- change to
generator = torch.Generator(device='cuda')
- change to
- [Modify line 126]:
yield from torch.randperm(n, generator=generator).tolist()- change to
yield from torch.randperm(n, generator=generator, device='cuda').tolist()
- change to
Line number could be different for different version but point to note is adding device='cuda' to functions.
I think one should really try to avoid modifying PyTorch's source code.
A simpler solution is that provided by @GatzZ .
