concepts

DataLoader

Status: Public API in netcl.data.dataloader.DataLoader

DataLoader is netcl's data-pipeline primitive. It is responsible for fetching the next batch of training examples from a dataset, applying any user-supplied transforms, optionally converting the batch to device tensors, and yielding it to the training loop.

netcl's DataLoader is worker-based and asynchronous: it spawns a persistent pool of worker processes (reused across epochs, not torn down between them) that load and preprocess batches ahead of consumption in a sliding-window ring. With use_shared_mem=True, workers write their finished batches directly into a shared-memory ring buffer instead of returning them through a pickled pipe; the main process reads batches out of the ring without ever blocking on a worker that is already done.

Overview

The standard pattern — plain NumPy batches, no GPU involvement in the loader itself:

from netcl.data.dataloader import DataLoader
from my_dataset import MyDataset

dataset = MyDataset(root="data/", train=True)
loader  = DataLoader(dataset, batch_size=128, shuffle=True,
                     num_workers=4, use_shared_mem=True,
                     drop_last=False)

for x, y in loader:
    # x, y are numpy arrays here
    ...

Each iteration of the for loop returns a tuple of NumPy arrays, one per dataset output, np.stacked along a new leading axis. To get device Tensors directly out of the loader — with the H2D copy overlapped with GPU compute instead of sitting on the training loop's critical path — call bind_queue(queue) once after GPU init; see GPU Prefetch below.

Where It Lives

File path: data/dataloader.py.
Module path: netcl.data.dataloader.
Public re-export: from netcl.data import DataLoader.
Sibling: data.gpu_batch (the GpuBatch container and the async-H2D upload helper used by bind_queue()), data.shared_batch.SharedBatchRingBuffer (the shared-memory ring used by the workers), data.augment (the CPU augmentation primitives), data.augment_gpu (the GPU augmentation primitives).

Diagram

How It Works

The DataLoader.__iter__ method lazily creates a persistent pool of worker processes on first use (fork on Linux, spawn on macOS/Windows — see the Fork-vs-Spawn section below) and submits batches ahead of consumption up to prefetch batches in flight. With use_shared_mem=True, each worker writes its finished batch directly into a slot of a SharedBatchRingBuffer instead of returning it through the pool's pickle pipe; the main process reads the slot's data out (as a Tensor upload when a queue is bound, or a plain np.copy otherwise) and immediately frees the slot back to the pool — except when a GPU upload from that slot is still in flight, in which case the slot is held until the copy completes (see GPU Prefetch).

The num_workers parameter controls the worker count. A good default is the number of CPU cores available to the process divided by the per-worker CPU cost; 4 is a safe starting point for most image-classification workloads. num_workers=0 runs everything on the calling thread/process — no pool, no ring — and is the simplest configuration for unit tests and small models.

The shuffle=True flag enables index shuffling at the dataset level (a plain numpy permutation of range(len(dataset)), reshuffled every epoch via the loader's own np.random.Generator); the workers themselves are agnostic to shuffling and just process whatever indices they are handed. Pass seed= for a reproducible shuffle order.

Code Example

A minimal DataLoader use:

from netcl.data.dataloader import DataLoader

class MNIST:
    def __init__(self, x, y):
        self.x, self.y = x, y
    def __len__(self):
        return len(self.x)
    def __getitem__(self, i):
        return self.x[i], self.y[i]

dataset = MNIST(x_train, y_train)   # x_train: (N,1,28,28), y_train: (N,)
loader  = DataLoader(dataset, batch_size=128, shuffle=True,
                     num_workers=4, use_shared_mem=True)

for x, y in loader:
    # x.shape == (128, 1, 28, 28), y.shape == (128,) — plain numpy arrays
    ...

GPU Prefetch

Calling bind_queue(queue) once, after the OpenCL device exists, switches the loader into GPU-prefetch mode: every yielded batch becomes a netcl.data.gpu_batch.GpuBatch, a tuple whose float32 fields have already been uploaded via a non-blocking cl.enqueue_copy on a pooled device buffer.

loader.bind_queue(queue)
for batch in loader:
    xb, yb = batch[0], batch[1]   # xb is a device Tensor; yb (int labels) stays numpy
    ...

This is deliberately not the same thing as a pin_memory=True flag: pinning only speeds up the host side of an H2D copy that still has to happen synchronously somewhere. GPU prefetch moves the copy itself off the critical path entirely, by doing it during the load phase — which, in a well-pipelined training loop, runs concurrently with the previous step's GPU backward pass. On the WIDER-FACE reference training script this turned a ~50 ms main-thread cost per batch into well under 1 ms.

The full behavioral contract (async copy semantics, SHM-slot deferral, the legacy device_queue/async_transfer constructor arguments, and the fallback-on-GPU-error path) is documented on the API page.

Custom batching logic (e.g. for variable-length sequences) goes into transforms=, which is called on the already-stacked batch (xb, yb) after the default collation:

def pad_transform(xb, yb):
    # xb is already a numpy array (B, max_len) after default stacking.
    # Trim trailing zeros or re-pad here if needed.
    return xb, yb

loader = DataLoader(dataset, batch_size=32, transforms=pad_transform)

GPU augmentations from augment_gpu are applied after the batch arrives from the loader. The available functions are flip_horizontal(x, mask), apply_color_jitter(x, brightness, contrast), and cutout(x, centers, size). They all take a Tensor (not a numpy array) and per-sample parameters as numpy arrays:

import numpy as np
from netcl.core.tensor import Tensor
from netcl.core.device import manager
from netcl.data.augment_gpu import flip_horizontal, apply_color_jitter

q = manager.default("auto").queue

for xb, yb in loader:          # xb, yb are numpy arrays from the loader
    x = Tensor.from_host(q, xb)
    mask = (np.random.rand(len(xb)) > 0.5).astype(np.uint8)
    x = flip_horizontal(x, mask)
    ...

Note: DataLoader does not have a collate_fn parameter. Use transforms= for per-batch CPU processing, or apply GPU augmentations manually after the batch transfer. If you also called bind_queue(), xb in the loop above is already a Tensor — skip the manual Tensor.from_host(q, xb) call and pass it straight to flip_horizontal.

Performance & Trade-offs

num_workers=0 runs the data loading in the calling process/thread with no pool and no ring. This is the simplest configuration and is fine for small models or unit tests; for production training, the default of 4 (or higher) is the right starting point.
bind_queue() is the cheapest single perf win once the loader itself is not the bottleneck. It moves the H2D copy off the training loop's critical path by doing it during the load phase with a non-blocking enqueue_copy on a pooled buffer — see GPU Prefetch above. This is strictly more than a pin_memory-style flag: pinning only makes a synchronous copy faster, bind_queue() overlaps the copy with GPU compute so it is (close to) free.
Worker pool lifetime spans the whole DataLoader, not one epoch. The pool is created lazily on the first __iter__ call and reused across every subsequent epoch as long as the dataset object's identity does not change — there is no per-epoch teardown/respawn pause. Call close() explicitly when you are done with the loader (or rely on __del__ for short scripts).
Shared-memory ring size is prefetch + 2 slots by default (prefetch itself defaults to 2 * num_workers). This is enough to absorb small bursts; for very uneven per-batch compute (e.g. one batch is 10x slower than the others), raise prefetch to give the main process more slack.