PyTorch Dataloader讀取時，如何在進(jìn)程之間傳輸數(shù)據(jù)？

while watchdog.is_alive():    try:        r = index_queue.get(timeout=MP_STATUS_CHECK_INTERVAL)    except queue.Empty:        continue    if isinstance(r, _ResumeIteration):        # Acknowledge the main process        data_queue.put((r, None))        iteration_end = False        # Recreate the fetcher for worker-reuse policy        fetcher = _DatasetKind.create_fetcher(            dataset_kind, dataset, auto_collation, collate_fn, drop_last)        continue    elif r is None:        # Received the final signal        assert done_event.is_set() or iteration_end        break    elif done_event.is_set() or iteration_end:        # `done_event` is set. But I haven't received the final signal        # (None) yet. I will keep continuing until get it, and skip the        # processing steps.        continue    idx, index = r    data: Union[_IterableDatasetStopIteration, ExceptionWrapper]    if init_exception is not None:        data = init_exception        init_exception = None    else:        try:            data = fetcher.fetch(index)        except Exception as e:            if isinstance(e, StopIteration) and dataset_kind == _DatasetKind.Iterable:                data = _IterableDatasetStopIteration(worker_id)                # Set `iteration_end`                #   (1) to save future `next(...)` calls, and                #   (2) to avoid sending multiple `_IterableDatasetStopIteration`s.                iteration_end = True            else:                # It is important that we don't store exc_info in a variable.                # `ExceptionWrapper` does the correct thing.                # See NOTE [ Python Traceback Reference Cycle Problem ]                data = ExceptionWrapper(                    where="in DataLoader worker process {}".format(worker_id))    data_queue.put((idx, data))????del?data,?idx,?index,?r??#?save?memory

def reduce_storage(storage):    from . import get_sharing_strategy    if storage.is_cuda:        raise RuntimeError("Cannot pickle CUDA storage; try pickling a CUDA tensor instead")    elif get_sharing_strategy() == 'file_system':        metadata = storage._share_filename_()        cache_key = metadata[1]        rebuild = rebuild_storage_filename        storage._shared_incref()    elif storage.size() == 0:        # This is special cased because Empty tensors        # (with size 0) cannot be mmapped.        return (rebuild_storage_empty, (type(storage),))    else:        fd, size = storage._share_fd_()        df = multiprocessing.reduction.DupFd(fd)        cache_key = fd_id(fd)        metadata = (df, size)        rebuild = rebuild_storage_fd  # type: ignore[assignment]
    shared_cache[cache_key] = StorageWeakRef(storage)    return (rebuild, (type(storage),) + metadata)

if isinstance(elem, torch.Tensor):    out = None    if torch.utils.data.get_worker_info() is not None:        # If we're in a background process, concatenate directly into a        # shared memory tensor to avoid an extra copy        numel = sum([x.numel() for x in batch])        storage = elem.storage()._new_shared(numel)        out = elem.new(storage)????return?torch.stack(batch,?0,?out=out)