Scaling Data
Lazy tensor shards and distributed loaders keep data execution independent from
the selected SILVA family. A JSON manifest records aligned tensor keys, sample
shapes, dtypes, order, and shard lengths. SILVAShardedTensorDataset then keeps
only one shard cached in each worker process.
Data Flow
from silva_networks import (
SILVAShardedTensorDataset,
make_silva_dataloader,
runtime_for_tier,
write_silva_tensor_shards,
)
manifest = write_silva_tensor_shards(
{"x": inputs, "y": targets},
"data/train",
shard_size=512,
)
dataset = SILVAShardedTensorDataset(manifest)
runtime = runtime_for_tier("workstation")
loader = make_silva_dataloader(dataset, runtime.data_config())
Writing is atomic per shard and for the final manifest. Existing manifests or
shards are rejected unless overwrite=True. For distributed runs, the loader
uses one DistributedSampler; fit_supervised advances its epoch before each
training pass.
Operational Contract
This API surface connects coverage, reproduction, data, and scale configuration to the same SILVA experiment
contract used by the learning pages and notebooks. Its central relation is
\[
F_\theta(z;x)=0,\qquad \widehat F_{\theta,s}(z;x)=0\ \text{uses the same mathematical contract at scale tier }s
\]
| Part |
What must remain inspectable |
| State |
the selected family, constructor contract, runtime tier, and data-loader configuration. |
| Condition |
changing a runtime tier may change numerical budgets and resource use but must not silently change the family equation. |
| Diagnostic |
coverage record, verification level, solver settings, effective batch size, and source-scale metrics. |
| Replacement point |
compact defaults with family-specific modules, official data adapters, and an archived experiment configuration. |
| Scale axes |
solver iterations, tolerance, model width, batch size, precision, workers, process count, and checkpoint interval. |
The relevant method lineage is recorded in the SILVA construction [1] and the selected family's primary references. Those references
define the source mechanisms; this API exposes them through SILVA objects so a
reader can inspect, replace, solve, differentiate, and scale the construction.
Complete Compact Study
Run the complete repository program below from the project root. The page uses
the same file that is exercised by the test suite, so the displayed call is not
an isolated fragment.
"""Inspect one family from public API coverage through executable scale defaults."""
from __future__ import annotations
from silva_networks import (
SILVADataLoaderConfig,
implementation_cases,
runtime_for_tier,
silva_family_guide,
silva_reproduction_spec,
silva_scaling_defaults,
)
family = "fno_deq"
case = next(item for item in implementation_cases() if item.key == "recent_equilibrium_families")
guide = silva_family_guide(family)
reproduction = silva_reproduction_spec(family)
defaults = silva_scaling_defaults(family, tier="smoke")
runtime = runtime_for_tier("smoke")
loader = SILVADataLoaderConfig(batch_size=4, workers=0)
print("family", family)
print("public objects", len(case.public_objects))
print("verification", reproduction.verification_level)
print("benchmark tasks", len(guide.benchmark_tasks))
print("solver", defaults["config"].solver)
print("max iterations", defaults["config"].max_iter)
print("runtime", runtime.device, runtime.mixed_precision)
print("loader", loader.batch_size, loader.workers)
python examples/api_scale_workflow.py
Measured Compact Output
family fno_deq
public objects 12
verification compact-verified
benchmark tasks 2
solver anderson
max iterations 12
runtime auto none
loader 4 0
Interpret the Output
The family resolves through four independent registries: public coverage, source relation, scale guidance, and runtime/data configuration. The compact-verified label describes repository evidence; it does not convert the two listed benchmark tasks into claimed benchmark results.
For a controlled experiment, retain the compact call as a regression case and
change one scale axis at a time. Record the resolved constructor, data source
and split, preprocessing, seed, forward and backward solver settings, task
metric, normalized residual, iteration count, runtime, peak memory, and any
failed convergence case. A larger run becomes evidence only when its own
resolved configuration and outputs are archived; the compact output above is
evidence for the executable mechanism and its stated invariants.
Sharded tensor datasets and distributed data loading for SILVA experiments.
SILVADataLoaderConfig
dataclass
Scale-aware options for a SILVA data loader.
Source code in src/silva_networks/scaling_data.py
| @dataclass(frozen=True)
class SILVADataLoaderConfig:
"""Scale-aware options for a SILVA data loader."""
batch_size: int
workers: int = 0
shuffle: bool = True
pin_memory: bool = False
persistent_workers: bool = False
prefetch_factor: int = 2
drop_last: bool = False
distributed: bool | None = None
seed: int = 0
def __post_init__(self) -> None:
if self.batch_size < 1:
raise ValueError("batch_size must be positive")
if self.workers < 0:
raise ValueError("workers must be nonnegative")
if self.prefetch_factor < 1:
raise ValueError("prefetch_factor must be positive")
if self.persistent_workers and self.workers == 0:
raise ValueError("persistent_workers requires workers > 0")
|
SILVAShardedTensorDataset
Bases: Dataset[dict[str, Tensor]]
Lazy one-shard cache for aligned tensor datasets larger than memory.
Source code in src/silva_networks/scaling_data.py
| class SILVAShardedTensorDataset(Dataset[dict[str, Tensor]]):
"""Lazy one-shard cache for aligned tensor datasets larger than memory."""
def __init__(self, manifest: str | Path):
self.manifest_path = Path(manifest)
metadata = json.loads(self.manifest_path.read_text(encoding="utf-8"))
if metadata.get("format") != "silva-tensor-shards" or metadata.get("version") != 1:
raise ValueError("unsupported SILVA tensor-shard manifest")
self.keys = tuple(metadata.get("keys", ()))
if not self.keys:
raise ValueError("manifest must define tensor keys")
self._shards = tuple(metadata.get("shards", ()))
if not self._shards:
raise ValueError("manifest must contain at least one shard")
lengths = [int(entry["length"]) for entry in self._shards]
if any(length < 1 for length in lengths):
raise ValueError("every shard length must be positive")
self._ends: list[int] = []
running = 0
for length in lengths:
running += length
self._ends.append(running)
if running != int(metadata.get("length", -1)):
raise ValueError("manifest length does not match shard lengths")
self._length = running
self._cached_index: int | None = None
self._cached_payload: dict[str, Tensor] | None = None
def __len__(self) -> int:
return self._length
def __getitem__(self, index: int) -> dict[str, Tensor]:
if index < 0:
index += self._length
if index < 0 or index >= self._length:
raise IndexError(index)
shard_index = bisect.bisect_right(self._ends, index)
start = 0 if shard_index == 0 else self._ends[shard_index - 1]
payload = self._load_shard(shard_index)
local_index = index - start
return {key: payload[key][local_index] for key in self.keys}
def _load_shard(self, shard_index: int) -> dict[str, Tensor]:
if self._cached_index == shard_index and self._cached_payload is not None:
return self._cached_payload
entry = self._shards[shard_index]
path = self.manifest_path.parent / entry["path"]
payload = torch.load(path, map_location="cpu", weights_only=True)
if not isinstance(payload, dict) or set(payload) != set(self.keys):
raise ValueError(f"shard keys do not match manifest: {path}")
expected = int(entry["length"])
for key, value in payload.items():
if not torch.is_tensor(value) or value.dim() == 0 or value.shape[0] != expected:
raise ValueError(f"invalid tensor {key!r} in shard: {path}")
self._cached_index = shard_index
self._cached_payload = payload
return payload
|
make_silva_dataloader
make_silva_dataloader(dataset, config, *, rank=None, world_size=None)
Build an ordinary or distributed data loader from one configuration.
Source code in src/silva_networks/scaling_data.py
| def make_silva_dataloader(
dataset: Dataset[Any],
config: SILVADataLoaderConfig,
*,
rank: int | None = None,
world_size: int | None = None,
) -> DataLoader[Any]:
"""Build an ordinary or distributed data loader from one configuration."""
distributed = (
dist.is_available() and dist.is_initialized()
if config.distributed is None
else config.distributed
)
sampler: DistributedSampler[Any] | None = None
if distributed:
if rank is None or world_size is None:
if not dist.is_available() or not dist.is_initialized():
raise RuntimeError(
"distributed loading requires an initialized process group or rank/world_size"
)
rank = dist.get_rank()
world_size = dist.get_world_size()
sampler = DistributedSampler(
dataset,
num_replicas=world_size,
rank=rank,
shuffle=config.shuffle,
seed=config.seed,
drop_last=config.drop_last,
)
kwargs: dict[str, Any] = {
"batch_size": config.batch_size,
"shuffle": config.shuffle if sampler is None else False,
"sampler": sampler,
"num_workers": config.workers,
"pin_memory": config.pin_memory,
"persistent_workers": config.persistent_workers,
"drop_last": config.drop_last,
}
if config.workers:
kwargs["prefetch_factor"] = config.prefetch_factor
return DataLoader(dataset, **kwargs)
|
write_silva_tensor_shards
write_silva_tensor_shards(tensors, directory, *, shard_size, prefix='shard', overwrite=False)
Write aligned tensors as independently loadable shards plus JSON metadata.
Each tensor must have the same first dimension. The returned manifest can
be opened by :class:SILVAShardedTensorDataset without loading every shard.
Source code in src/silva_networks/scaling_data.py
| def write_silva_tensor_shards(
tensors: Mapping[str, Tensor],
directory: str | Path,
*,
shard_size: int,
prefix: str = "shard",
overwrite: bool = False,
) -> Path:
"""Write aligned tensors as independently loadable shards plus JSON metadata.
Each tensor must have the same first dimension. The returned manifest can
be opened by :class:`SILVAShardedTensorDataset` without loading every shard.
"""
if shard_size < 1:
raise ValueError("shard_size must be positive")
if not tensors:
raise ValueError("tensors cannot be empty")
if not prefix or Path(prefix).name != prefix:
raise ValueError("prefix must be a nonempty filename component")
keys = tuple(tensors)
if any(not key for key in keys):
raise ValueError("tensor keys cannot be empty")
if any(not torch.is_tensor(value) or value.dim() == 0 for value in tensors.values()):
raise ValueError("every value must be a tensor with a sample dimension")
lengths = {int(value.shape[0]) for value in tensors.values()}
if len(lengths) != 1:
raise ValueError("all tensors must share their first dimension")
length = lengths.pop()
if length < 1:
raise ValueError("tensors must contain at least one sample")
destination = Path(directory)
destination.mkdir(parents=True, exist_ok=True)
manifest_path = destination / f"{prefix}-manifest.json"
if manifest_path.exists() and not overwrite:
raise FileExistsError(f"refusing to overwrite existing manifest: {manifest_path}")
shard_plan = [
(
shard_index,
start,
min(start + shard_size, length),
destination / f"{prefix}-{shard_index:05d}.pt",
)
for shard_index, start in enumerate(range(0, length, shard_size))
]
if not overwrite:
existing = next((path for _, _, _, path in shard_plan if path.exists()), None)
if existing is not None:
raise FileExistsError(f"refusing to overwrite existing shard: {existing}")
shard_entries: list[dict[str, Any]] = []
for _shard_index, start, stop, path in shard_plan:
filename = path.name
if path.exists() and not overwrite:
raise FileExistsError(f"refusing to overwrite existing shard: {path}")
payload = {
key: value[start:stop].detach().cpu().contiguous() for key, value in tensors.items()
}
temporary = path.with_suffix(path.suffix + ".tmp")
torch.save(payload, temporary)
temporary.replace(path)
shard_entries.append({"path": filename, "length": stop - start})
metadata = {
"format": "silva-tensor-shards",
"version": 1,
"length": length,
"keys": list(keys),
"tensors": {
key: {
"dtype": str(value.dtype).removeprefix("torch."),
"sample_shape": list(value.shape[1:]),
}
for key, value in tensors.items()
},
"shards": shard_entries,
}
temporary_manifest = manifest_path.with_suffix(".json.tmp")
temporary_manifest.write_text(json.dumps(metadata, indent=2) + "\n", encoding="utf-8")
temporary_manifest.replace(manifest_path)
return manifest_path
|
Where to Go Next