torchcell.loader.cpu_experiment_loader.CpuExperimentLoaderMultiprocessing

class CpuExperimentLoaderMultiprocessing(dataset: Sequence[Any], batch_size: int, num_workers: int)

Bases: object

Process-based loader that prefetches dataset batches into a queue.

static worker_function(load_queue: Queue[int | None], data_queue: Queue[list[Any]], dataset: Sequence[Any], batch_size: int) → None

Pull batch indices, slice the dataset, and push batches until stopped.

close() → None

Tear down workers once (idempotent); never block indefinitely.

On the normal path workers are parked on load_queue.get() and exit on the None sentinel. On an error path the consumer may have abandoned iteration with data_queue full and workers blocked on put – those never reach the sentinel, so a plain join() would hang forever. We drain data_queue to unblock them within a bounded budget, then force-terminate any straggler so close() itself can never deadlock.