Is there any way to solve hanging of IterableDataset using split by node + filtering during inference

### Describe the bug

I am using an iterable dataset in a multi-node setup, trying to do training/inference while filtering the data on the fly. I usually do not use `split_dataset_by_node` but it is very slow using the IterableDatasetShard in `accelerate` and `transformers`. When I filter after applying `split_dataset_by_node`, it results in shards that are not equal sizes due to unequal samples filtered from each one.

The distributed process hangs when trying to accomplish this. Is there any way to resolve this or is it impossible to implement?

### Steps to reproduce the bug

Here is a toy example of what I am trying to do that reproduces the behavior

```
# torchrun --nproc-per-node 2 file.py

import os

import pandas as pd
import torch
from accelerate import Accelerator
from datasets import Features, Value, load_dataset
from datasets.distributed import split_dataset_by_node
from torch.utils.data import DataLoader

accelerator = Accelerator(device_placement=True, dispatch_batches=False)
if accelerator.is_main_process:
if not os.path.exists("scratch_data"):
os.mkdir("scratch_data")

n_shards = 4
for i in range(n_shards):
df = pd.DataFrame({"id": list(range(10 * i, 10 * (i + 1)))})
df.to_parquet(f"scratch_data/shard_{i}.parquet")

world_size = accelerator.num_processes
local_rank = accelerator.process_index

def collate_fn(examples):
input_ids = []
for example in examples:
input_ids.append(example["id"])
return torch.LongTensor(input_ids)

dataset = load_dataset(
"parquet", data_dir="scratch_data", split="train", streaming=True
)
dataset = (
split_dataset_by_node(dataset, rank=local_rank, world_size=world_size)
.filter(lambda x: x["id"] < 35)
.shuffle(seed=42, buffer_size=100)
)

batch_size = 2
train_dataloader = DataLoader(
dataset,
batch_size=batch_size,
collate_fn=collate_fn,
num_workers=2
)

for x in train_dataloader:
x = x.to(accelerator.device)
print({"rank": local_rank, "id": x})

y = accelerator.gather_for_metrics(x)
if accelerator.is_main_process:
print("gathered", y)

```

### Expected behavior

Is there any way to continue training/inference on the GPUs that have remaining data left without waiting for the others? Is it impossible to filter when

### Environment info

- `datasets` version: 2.18.0
- Platform: Linux-5.10.209-198.812.amzn2.x86_64-x86_64-with-glibc2.31
- Python version: 3.10.13
- `huggingface_hub` version: 0.21.3
- PyArrow version: 15.0.0
- Pandas version: 2.2.1
- `fsspec` version: 2023.6.0