Fix sampling error with many pipelines when allow_repeats=False

[caciolai]

What does this PR do? Please describe:
A summary of the change or the issue that is fixed.

Fixes #1472

Does your PR introduce any breaking changes? If yes, please list them:
List of all backwards-incompatible changes.

Check list:

• Was the content of this PR discussed and approved via a GitHub issue? (no need for typos or documentation improvements)
• Did you read the contributor guideline?
• Did you make sure that your PR does only one thing instead of bundling different changes together?
• Did you make sure to update the documentation with your changes? (if necessary)
• Did you write any new necessary tests?
• Did you verify new and existing tests pass locally with your changes?
• Did you update the CHANGELOG? (no need for typos, documentation, or minor internal changes)

[caciolai]

Problematic example

nb = 21
weights = np.random.RandomState(0).rand(nb)
sizes = np.random.RandomState(0).randint(0, 100_000, nb)
pipelines = []
for pipeline_size in sizes:
    pipeline_idx = len(pipelines)
    items = [pipeline_idx] * pipeline_size
    pipelines.append(read_sequence(items).and_return())

builder = DataPipeline.sample(pipelines, weights, seed=123, allow_repeats=False)

pipeline = builder.and_return()

all_items = []
sampling_history = []
pipeline_num_samples = dict()

for step, item in enumerate(iter(pipeline)):
    all_items.append(item)
    pipeline_idx = item
    pipeline_num_samples[pipeline_idx] = pipeline_num_samples.get(pipeline_idx, 0) + 1
    sampling_history.append({
        "step": step,
        "pipeline_idx": pipeline_idx,
        "sample_so_far": pipeline_num_samples[pipeline_idx],
        "remaining_in_pipeline": sizes[pipeline_idx] - pipeline_num_samples[pipeline_idx],
    })

assert len(all_items) == sum(sizes), f"Expected {sum(sizes)} samples, got {len(all_items)}"

data = []
for pipeline_idx in range(len(sizes)):
    expected_size = sizes[pipeline_idx]
    actual_size = pipeline_num_samples.get(pipeline_idx, 0)
    data.append({
        "pipeline_idx": pipeline_idx,
        "type": "sampled",
        "num_samples": actual_size,
    })
    data.append({
        "pipeline_idx": pipeline_idx,
        "type": "expected",
        "num_samples": expected_size,
    })

Before patch

After patch

Tested up to 100 pipelines, with no failing assertions

[cbalioglu]

Thanks for the fix! As we discussed offline, we can slightly simplify the code by reusing is_epoch_done_ for the purpose of checking blocked pipelines. Other than that, I only have a few nit comments.

[cbalioglu]

LGTM! Thanks for the fix!