Distributed neural training

Aner separates model definition, Dataset policy and execution placement. The native aner.parallel module trains supported Dense classifiers on a local CPU or accelerator, or on an explicitly enrolled group of CPU, Metal/MPS and CUDA workers. Researchers retain the same architecture, loss, initialization and fixed global batch while changing the execution context.

This guide targets the current development engine and VS Code extension 0.1.33. Compatible engines must be installed on participating operating systems. A version label alone does not establish source/protocol compatibility. Convolutional and recurrent layers, arbitrary remote source functions, model sharding and optimized GPU collectives remain separate work.

Execution model

Open the native Dense training diagram at full size

Every worker owns a complete model replica. At each update, the coordinator freezes one canonical parameter version, selects a fixed global mini batch and assigns disjoint row slices. Workers run the supported forward pass and differentiation on their actual declared device. They return gradient sums, loss sums and valid row counts. The coordinator accumulates those sums in Float64, divides by the valid denominator and converts the mean gradient to the selected group dtype, then performs one canonical CPU optimizer update in that dtype. The next update broadcasts the resulting parameter version.

GPU placement therefore describes replica forward/backward computation. Dataset preparation, training only normalization, canonical SGD/Adam updates, validation and final fitted model evaluation use the CPU. Portable exchange stages supported numeric buffers through host memory and protected local pipes or authenticated SSH. Request parameter/feature/target values are encoded in the selected Float32 or Float64 dtype. Detached gradient sum replies use Float64; decoded host exchange vectors also use Float64. Changing compute precision therefore changes typed request bytes but does not halve all transport or host exchange storage. It does not use a direct Metal: CUDA collective, shared process pointers or a shared differentiation graph.

Begin with a simple fit

import aner.dataset
import aner.nn
import aner.ml
import aner.parallel

let parts = dataset.iris().partition(validation: 0.2, test: 0.2, seed: 42)
let model = ml.classifier(nn.mlp(4, List(8), 3, 42),
    standardize: true, optimizer: "adam")
let config = ml.training(epochs: 40, batch: 32, rate: 0.01, seed: 42)
    .validation(parts.validation())
let fitted = parallel.fit(model: model, data: parts.train(), config: config)
print(fitted.evaluate(parts.test()).summary())

The three argument fit follows the active resolved NN execution profile. Its ordinary default is CPU Float64 with preserve precision policy. Explicit source/CLI profile settings can change that resolved local placement, and an existing preserve policy fallback remains reported. The protected test rows remain outside training and validation selection. Iris is Fisher (1936), distributed by UCI under CC BY 4.0; its source attribution is retained.

Select precision and placement explicitly

let execution = parallel.device(device: "cpu", dtype: "Float64")
let fitted = parallel.fit(model: model, data: parts.train(),
    config: config, execution: execution, progress: true)

parallel.device requires both strings. Use "cuda", "Float64" for native high precision CUDA, or "mps", "Float32" for native MPS. CPU and CUDA also accept Float32. An explicit unavailable backend or MPS Float64 request fails before training; this strict context never silently changes precision or placement. No driver linking or command queue setup belongs in source programs: the native engine owns that connection after its backend is compiled and available. CUDA currently selects device 0, and MPS selects the system default Metal device. Configuring multiple accelerator ordinals within one node remains separate work.

Replica compute dtype and Float64 host accumulation are different facts. A Float32 forward/backward pass has Float32 rounding even when its gradients are later accumulated in Float64. The mean gradient is narrowed to Float32 for a Float32 group; canonical CPU parameters and optimizer state also use Float32. A Float64 group retains Float64 canonical state and updates. Parameters are placed on each replica device before its next forward pass. Faster execution and acceptable numerical error must be measured for the actual experiment.

These fit contexts do not convert Monte Carlo into a GPU workload. The supported CPU task API retains its separate executor/chunk contract in native parallel computing.

Enroll a group

The same scientific call can use an operator maintained private configuration:

let execution = parallel.cluster(config: "private.neural.cluster")
let fitted = parallel.fit(model: model, data: parts.train(),
    config: config, execution: execution)

A version 2 configuration declares the common precision and every selected replica backend. The following is a template, not an installed cluster. Replace placeholders with private verified connection targets and compatible executable paths. On POSIX, save the configuration as an owner only regular file (chmod 600).

ANER_CLUSTER 2
precision Float32
worker "Atlas-01" local "-" "/path/to/aner" 1 mps 1
worker "Atlas-02" ssh "USER@HOST" "/path/to/aner" 1 cuda 1
worker "Atlas-03" ssh "USER@OTHER_HOST" "/path/to/aner" 1 cuda 2

The worker fields are public logical alias, transport, private endpoint, absolute compatible executable path, CPU slot allowance, replica backend and optional positive relative batch weight. The omitted weight is 1. Weights divide one global batch; they do not multiply its size or change a sample's statistical weight. The runtime retains the partial final batch and validates actual contribution counts. A selected worker must perform admitted scientific work; merely listing a machine is not proof that it participated.

Version 1 CPU configurations remain supported for earlier CPU tasks. Version 2 uses one declared group dtype. An explicit benchmark CLI dtype must agree with the configuration rather than silently overriding it. A Float64 group cannot contain an MPS replica. SSH enrollment uses existing authentication and strict host key verification, opens no new listener, and never discovers/enrolls arbitrary machines. Failed capability or compatibility admission aborts the fit.

Download the local CPU Float64 program and configured group program. The cluster source requires a private configuration and installed compatible workers.

Preserve the scientific experiment

batch is global. A global batch of 1,536 split across three workers is still one batch of 1,536 and one optimizer update. Changing worker count must retain the Dataset source fingerprint, eligible row identities, protected roles, initialization, loss, optimizer settings, shuffle seed, global batch and stopping budget.

For worker means gᵢ over nᵢ valid rows, the global mean is:

global_gradient = sum(nᵢ × gᵢ) / sum(nᵢ)

Aner exchanges the corresponding sums and denominator directly. Means 2 and 4 over 32 and 224 rows combine to 3.75. Averaging those worker means without row weighting yields 3.0 and changes the algorithm. MSE also retains its class averaging convention. Local microbatches hold the same frozen parameter version before the global update. Retry or stale results cannot become duplicate statistical contributions.

Dataset partitions define row/group isolation, stratification, folds and lazy batches. Fitting and normalization reject protected validation/test roles. Validation is evaluated after an epoch and can select a best model when explicitly configured; final test evaluation stays outside that selection. Distributed execution does not relax these rules. Folds remain separate experiments, not shards of one update.

Add fine model control

The same fit accepts explicit Sequential dimensions and initial parameter tensors:

let network = nn.sequential(seed: 42)
    .dense(inputs: 4, outputs: 8,
        weights: nn.random(4, 8, seed: 7, scale: 0.1),
        bias: nn.zeros(1, 8))
    .tanh()
    .dense(inputs: 8, outputs: 3, initialization: "xavier")
let model = ml.classifier(network, standardize: true,
    loss: "cross_entropy", optimizer: "adam")

Dense shapes and finite values are checked before work is delegated. Researchers can add supported stages, choose initialization, change the supported objective/optimizer, attach validation policy and inspect fitted parameters without changing the execution API.

Classes can organize this configuration locally:

class ResearchClassifier {
    let model: ClassifierPipeline
    let config: TrainingConfig

    fn train(training: Dataset, execution: ParallelContext) {
        return parallel.fit(model: model, data: training,
            config: config, execution: execution)
    }
}

The complete class based research example uses the same native fit. Its local object is not serialized as a remote callback. Arbitrary custom class forward methods and unrestricted objectives are not distributed by this API. Existing explicit Tensor programming remains available for local research; additional distributed model families need their native checked forward/backward contracts.

Development and production

aner examples/parallel/neural_local.aner --dev neural-development
aner examples/parallel/neural_local.aner --prod

Use a fresh Dev directory. Dev records accepted global loss/progress, canonical fitted state and public safe replica placement history. Captured differentiation graphs are CPU reference graphs; they are not the remote worker gradient graphs. Individual worker graph/value inspection is currently unavailable. Production omits Dev graph/value reports while retaining bounded training history, the same Dataset checks, precision contract, weighting and optimizer updates. Progress is bounded and fixed horizon ETA is an estimate based on observed completed work.

Version 6 fitted checkpoints retain canonical parameters/preprocessing, Adam state where relevant, protected provenance/schedule metadata and public safe replica history. Existing version 1 to 5 checkpoints remain readable. Prediction and evaluation of a distributed fitted result use the canonical CPU representation and its recorded dtype. A fitted checkpoint is not a durable distributed job restart mechanism; worker reconnection, committed group checkpoints and coordinator recovery remain separate work.

Controlled Dense fixture

The complete synthetic exploration source builds 64 → 128 → 64 → 4 with 16,836 parameters. dataset.synthetic(rows: 65536, inputs: 64, classes: 4, seed: 42) generates finite uniform features and labels from a seeded linear teacher, records a source fingerprint, and supports protected partitions. The source is stored as a bounded Dataset; lazy batches avoid a full source Tensor that would exceed its one million element guard.

The generator admits at most one million rows, 4,096 inputs, 256 classes, eight million feature values and 500 million teacher work units. It requires at least two classes. Generation time is O(Ndc), source space O(Nd + N + dc), for N rows, d inputs and c classes. This is a controlled numerical fixture rather than a real world task accuracy claim. The matched performance protocol fixes ten epochs and 300 global updates, with 65,536 source rows, batch 1,536, SGD, rate 0.05 and seed 42. Protected partition identities and the final source/build contract are recorded before timed comparisons. See the controlled Dense benchmark for its measurement status and scope.

aner parallel neural --device cpu --dtype Float64 --rows 65536 \
    --epochs 10 --batch 1536 --rate 0.05 --output dense-cpu
aner parallel neural --cluster private.neural.cluster --dtype Float32 \
    --rows 65536 --epochs 10 --batch 1536 --rate 0.05 --output dense-group

Output directories must be new. Source exploration and a measured CLI report have distinct timing/observation boundaries. Hardware measurements belong in a separately identified report with recorded source/build/configuration IDs, three or more trials, stated warm ups, held out quality and the fastest selected single worker reference. No speedup or resource usage number is implied by this guide.

Time, space and scaling

Let E be epochs, N training rows, b the fixed global batch, r replicas, P parameters, S the sum of Dense input width × output width products, A the sum of activation widths, d input width and K = E × ceil(N/b) updates. Admission and final/validation evaluation add their own bounded passes.

These bounds describe work and storage, not elapsed time guarantees. Graph traversal, host encoding, device synchronization, finite checks, input staging, parameter broadcast, gradient readback, network startup and the slowest worker affect observed wall time. A group may be slower than its fastest selected device. Unequal batch weights expose deliberate allocation; automatic device cost scheduling is not inferred.

Each replica must fit a complete model and local activations. Two Macs do not become one GPU with pooled unified memory. Canonical host optimizer state can reduce the replicated optimizer requirement, but model/gradient sharding remains a distinct unimplemented algorithm. Tensor owned allocation peaks, provider allocations, process RSS and driver sampled GPU usage have different scopes. Metal shared memory overlaps host RAM; do not add it to RAM as an independent total. Unmeasured transfer, utilization, RSS or bandwidth fields remain unavailable.