`header()` causes DataLoader with MPRS to hang

[wchang-apixio]

🐛 Describe the bug

Using header() on an IterDataPipe causes DataLoader with MPRS to hang on the second time thru.

from torchdata.dataloader2 import DataLoader2
from torchdata.dataloader2 import MultiProcessingReadingService
from torchdata.datapipes.iter import IterableWrapper
from torch.utils.data.datapipes.iter.sharding import SHARDING_PRIORITIES

dp = IterableWrapper([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
dp = dp.sharding_round_robin_dispatch(SHARDING_PRIORITIES.MULTIPROCESSING)
dp = dp.header(3)

rs = MultiProcessingReadingService(
    num_workers=2,
    multiprocessing_context='fork')
dl = DataLoader2(dp, reading_service=rs)

print(list(dl))
print(list(dl))

I'd expect:

[1, 2, 3, 4, 5, 6]
[1, 2, 3, 4, 5, 6]

But it hangs after just the first line appears.

Versions

Collecting environment information... PyTorch version: 2.0.0+cu117 Is debug build: False CUDA used to build PyTorch: 11.7 ROCM used to build PyTorch: N/A

OS: Ubuntu 20.04.4 LTS (x86_64) GCC version: (Ubuntu 9.4.0-1ubuntu1~20.04.1) 9.4.0 Clang version: Could not collect CMake version: version 3.26.3 Libc version: glibc-2.31

Python version: 3.8.10 | packaged by conda-forge | (default, May 11 2021, 07:01:05) [GCC 9.3.0] (64-bit runtime) Python platform: Linux-5.4.196-108.356.amzn2.x86_64-x86_64-with-glibc2.10 Is CUDA available: True CUDA runtime version: 11.2.152 CUDA_MODULE_LOADING set to: LAZY GPU models and configuration: GPU 0: Tesla V100-SXM2-16GB Nvidia driver version: 470.57.02 cuDNN version: Probably one of the following: /usr/lib/x86_64-linux-gnu/libcudnn.so.8.1.0 /usr/lib/x86_64-linux-gnu/libcudnn_adv_infer.so.8.1.0 /usr/lib/x86_64-linux-gnu/libcudnn_adv_train.so.8.1.0 /usr/lib/x86_64-linux-gnu/libcudnn_cnn_infer.so.8.1.0 /usr/lib/x86_64-linux-gnu/libcudnn_cnn_train.so.8.1.0 /usr/lib/x86_64-linux-gnu/libcudnn_ops_infer.so.8.1.0 /usr/lib/x86_64-linux-gnu/libcudnn_ops_train.so.8.1.0 HIP runtime version: N/A MIOpen runtime version: N/A Is XNNPACK available: True

CPU: Architecture: x86_64 CPU op-mode(s): 32-bit, 64-bit Byte Order: Little Endian Address sizes: 46 bits physical, 48 bits virtual CPU(s): 32 On-line CPU(s) list: 0-31 Thread(s) per core: 2 Core(s) per socket: 16 Socket(s): 1 NUMA node(s): 1 Vendor ID: GenuineIntel CPU family: 6 Model: 79 Model name: Intel(R) Xeon(R) CPU E5-2686 v4 @ 2.30GHz Stepping: 1 CPU MHz: 2701.336 CPU max MHz: 3000.0000 CPU min MHz: 1200.0000 BogoMIPS: 4600.03 Hypervisor vendor: Xen Virtualization type: full L1d cache: 512 KiB L1i cache: 512 KiB L2 cache: 4 MiB L3 cache: 45 MiB NUMA node0 CPU(s): 0-31 Vulnerability Itlb multihit: KVM: Vulnerable Vulnerability L1tf: Mitigation; PTE Inversion Vulnerability Mds: Vulnerable: Clear CPU buffers attempted, no microcode; SMT Host state unknown Vulnerability Meltdown: Mitigation; PTI Vulnerability Spec store bypass: Vulnerable Vulnerability Spectre v1: Mitigation; usercopy/swapgs barriers and __user pointer sanitization Vulnerability Spectre v2: Mitigation; Retpolines, STIBP disabled, RSB filling Vulnerability Srbds: Not affected Vulnerability Tsx async abort: Vulnerable: Clear CPU buffers attempted, no microcode; SMT Host state unknown Flags: fpu vme de pse tsc msr pae mce cx8 apic sep mtrr pge mca cmov pat pse36 clflush mmx fxsr sse sse2 ht syscall nx pdpe1gb rdtscp lm constant_tsc rep_good nopl xtopology nonstop_tsc cpuid aperfmperf pni pclmulqdq ssse3 fma cx16 pcid sse4_1 sse4_2 x2apic movbe popcnt tsc_deadline_timer aes xsave avx f16c rdrand hypervisor lahf_lm abm 3dnowprefetch cpuid_fault invpcid_single pti fsgsbase bmi1 hle avx2 smep bmi2 erms invpcid rtm rdseed adx xsaveopt

Versions of relevant libraries: [pip3] mypy-boto3-s3==1.26.0.post1 [pip3] numpy==1.24.3 [pip3] pytorch-lightning==2.0.2 [pip3] torch==2.0.0 [pip3] torchdata==0.6.0 [pip3] torchmetrics==0.11.4 [pip3] triton==2.0.0 [conda] numpy 1.23.5 pypi_0 pypi [conda] torchdata 0.6.0 pypi_0 pypi

[ejguan]

Thanks for reporting. Seems like a serious bug, needs to investigate.

[JohnHBrock]

@wchang-apixio Did you ever find a solution to this?

[wchang-apixio]

Not really. The best I've done is use .enumerate() to add an index and then filter on it.