Skip to content

Shard Cache Hierarchy

This document explains how Datavo caches shard tars on the consumer side and how to shape that cache for fast IO.

The shard cache hierarchy is a property of the machine running the consumer code. Datavo dataset consumers (SampleStream.from_dataset(...)) read from it transparently. They never need to know which tiers exist, where they live, or how big they are.

Goals

The hierarchy is designed around one boundary:

  • Consumers open a dataset and iterate samples. They do not configure storage.
  • The local IO topology is described in a single YAML file — no tier needs pre-warming.

Specifically:

  • A consumer-side iteration over a Datavo split should hit local disk on the second pass, even when several users share one machine.
  • You can add a shared scratch dir, a per-user SSD, or both by editing one config file. No code change on the consumer side.
  • The cache should fill itself naturally as consumers read. There is no separate prefetch or bulk-import step.

Identity And Cache Key

Datavo workset shards are addressed by content_hash. The hash is a deterministic function of the dataset state and the requested input keys:

  • dataset manifest reference,
  • dataset revision watermark,
  • shard index,
  • requested input keys,
  • the ordered list of sample ids for that shard.

For a given (dataset, shard_index, required_input_keys, revision watermark) the hash is stable across worksets and processes. That makes it safe to use as a global cache key. Two different worksets that ask for the same shard of the same dataset with the same keys hit the same cache file.

Files are stored as <content_hash>.tar at the top level of each tier directory. Tier directories also contain a .staging/ subdirectory for in-flight downloads and a .locks/ subdirectory for per-shard advisory locks.

Tiers

A tier is a backend that can hold shards by content hash. Two kinds of tier ship today; both implement the same small interface so the hierarchy can treat them uniformly.

Folder tiers

A folder tier is a directory on a local filesystem.

ShardCacheTier(
    path="/scratch/datavo/shards",
    max_bytes=1_099_511_627_776,   # 1 TiB
    name="scratch",
)

Folder tier semantics:

  • path is created on first use. ~ and environment variables are expanded.
  • max_bytes is optional. Missing or zero means unbounded.
  • Eviction is LRU based on file mtime. Reads update mtime so popular shards stay resident.
  • Each tier locks per content_hash (POSIX fcntl.flock) so concurrent processes on the same machine do not double-fetch or stomp on installs.
  • The tier is directly readable: the consumer iterates the tar file in place, no copying involved.

Cloud tiers

Three cloud-backed tiers ship today, one per major cloud:

  • AzureBlobShardCacheTier — Azure Blob container
  • S3ShardCacheTier — AWS S3 bucket
  • GcsShardCacheTier — Google Cloud Storage bucket

All three behave the same way from the hierarchy's perspective:

  • Shards are stored at <content_hash>.tar inside the bucket/container. An optional prefix extends that to <prefix>/<content_hash>.tar for sharing one bucket across deployments.
  • The tier is not directly readable. A cloud-tier hit is materialized into a folder tier first; the consumer always reads from local disk. See Cloud topology below.
  • max_bytes is informational. Cloud buckets do not evict locally — eviction is expected to be managed via cloud-side lifecycle policies (Azure blob rules, S3 lifecycle, GCS lifecycle/autoclass). Setting max_bytes still lets the hierarchy skip the tier for shards that exceed the limit.
  • The cloud SDK is imported lazily. A tier is constructable without the SDK installed; the import only fires when you actually exercise the tier. Concrete extras:
  • Azure: pip install 'datavo-sdk[azure]' (azure-storage-blob, plus azure-identity for managed-identity auth).
  • AWS: pip install 'datavo-sdk[aws]' (boto3).
  • GCP: pip install 'datavo-sdk[gcp]' (google-cloud-storage).
  • All three: pip install 'datavo-sdk[all-clouds]'.
AzureBlobShardCacheTier(
    container_url="https://datavoeastus.blob.core.windows.net/shards",
    name="region-eastus",
    # sas_token: SAS string for ad-hoc auth, otherwise DefaultAzureCredential
    # managed_identity_client_id: pin to a specific user-assigned MI
)

S3ShardCacheTier(
    bucket="datavo-cache-us-east-1",
    region="us-east-1",
    name="region-us-east-1",
    # boto3's default credential chain — env vars, EC2 instance profile,
    # ECS task role, EKS service account...
)

GcsShardCacheTier(
    bucket="datavo-cache-us-central1",
    project="datavo-prod",
    name="region-us-central1",
    # Application Default Credentials — workload identity on GKE, service
    # account on GCE, gcloud auth locally...
)

Tiers are independent of each other. Each tier evicts on its own schedule against its own capacity. The hierarchy never coordinates eviction across tiers.

Hierarchy

A ShardCacheHierarchy is an ordered list of tiers, fastest first. The default order is the order they appear in YAML.

The hierarchy enforces one invariant: at least one folder-backed tier must be present. That tier is the "local landing zone" where any remote-tier hit is materialized so the consumer can read a regular file. A hierarchy made up of only remote tiers would have nothing to hand back to the consumer.

Reads

hierarchy.get_or_fetch(content_hash, fetcher) walks tiers fastest to slowest looking for the shard. The first tier that has it serves the read.

If no tier has it, the caller-supplied fetcher writes the tar into a staging file. The fetcher is opaque to the cache. For a Datavo consumer the fetcher issues a workset stream ticket and downloads the shard.

Writes: populate on miss

On a complete miss the staged tar is installed into every tier that can hold it. Each tier respects its own capacity:

  • if a tier's max_bytes is smaller than the shard, the tier is skipped with a warning;
  • if at least one tier accepts the shard, the call succeeds and returns the topmost installed path;
  • if no tier accepts, the call raises.

This is what makes the cache self-warming. You do not need to pre-fill the scratch tier. The first consumer to read a shard puts it everywhere.

Writes: promote on hit

When a slower tier serves a read, the shard is also copied into the faster tiers above it that did not have it. After promotion the call returns the topmost installed path so the caller reads from the fastest available tier even on a first-time slow-tier hit.

Promotion is per-tier subject to capacity:

  • tiers too small for the shard are skipped with a warning;
  • if every faster folder tier rejects, the read is served from the original slower-tier hit (no error) — unless the hit was on a remote tier, in which case the call raises because there is no local landing zone for the consumer.

Promotion happens once per call. After a promote the shard is in the faster tier; subsequent calls find it there directly.

Reads from remote tiers

A hit on a blob tier (or any tier with supports_direct_read = False) is materialized through the topmost folder tier's .staging/ area on its way to the consumer:

  1. The hierarchy downloads the blob into a staging file inside the topmost folder tier.
  2. The staging file is then installed into every faster folder tier that can hold it (the same promote-on-hit path used for slow-folder hits).
  3. The hierarchy returns the path inside whichever folder tier accepted first — the consumer reads from a regular local file.

That single staging step is what couples the local tier and the regional blob tier into a real hierarchy: the consumer doesn't notice the blob exists, and the blob's network cost is paid once per shard per VM.

Hits do not refetch

The fetcher is never called when any tier has the shard, including the promote-on-hit path. There is no revalidation step. A content hash uniquely identifies the bytes; if a tier has content_hash.tar, it is by definition the correct bytes.

Configuration

The hierarchy is resolved at consumer startup. The resolution order is:

  1. An explicit cache_config_path argument to SampleStream.from_dataset(...) or get_default_shard_cache(...).
  2. /etc/datavo/shards.yaml (system-wide).
  3. ~/.config/datavo/shards.yaml (per-user override).
  4. A built-in default: a single user-local tier at $DATAVO_SHARD_CACHE_DIR or ~/.cache/altavo/datavo/shards with 50 GiB LRU capacity.

System-first is intentional. The shared IO topology is a property of the machine, not the user. Per-user overrides exist for laptops and developer workstations.

YAML schema

tiers:
  - name: scratch
    type: folder                  # default; can be omitted
    path: /scratch/datavo/shards
    max_bytes: 1TB
  - name: home
    path: ~/.cache/altavo/datavo/shards
    max_bytes: 50GB
  - name: region-eastus
    type: azure_blob
    container_url: https://datavoeastus.blob.core.windows.net/shards
    # sas_token: "..."            # optional; otherwise DefaultAzureCredential
    # prefix: "team-a"            # optional namespacing inside the container
    # managed_identity_client_id: "..."   # optional; specific UAMI client id
  - name: region-us-east-1
    type: s3
    bucket: datavo-cache-us-east-1
    region: us-east-1
    # prefix: "team-a"
  - name: region-us-central1
    type: gcs
    bucket: datavo-cache-us-central1
    # project: "datavo-prod"      # optional
    # prefix: "team-a"

Common fields:

  • tiers (required, non-empty list). Order is fastest first. At least one tier must be folder-backed.
  • tiers[].type (optional). One of folder, azure_blob, s3, gcs. Defaults to folder.
  • tiers[].name (optional). Display name used in logs.
  • tiers[].max_bytes (optional). Integer or string with a SI-style suffix (B, KB, MB, GB, TB, 1024-based). Missing or zero means unbounded.

Folder-tier fields:

  • tiers[].path (required). Local filesystem path. ~ and $VAR are expanded.

Azure-blob-tier fields:

  • tiers[].container_url (required). Azure Blob container URL.
  • tiers[].sas_token (optional). When present, used for SAS-based auth. Otherwise the tier falls back to DefaultAzureCredential.
  • tiers[].prefix (optional). Path-style prefix prepended to each blob name.
  • tiers[].managed_identity_client_id (optional). Passed through to DefaultAzureCredential for pinning to a specific user-assigned managed identity.

S3-tier fields:

  • tiers[].bucket (required). S3 bucket name.
  • tiers[].region (optional). AWS region; boto3 figures it out from the default chain if omitted but explicit is usually clearer.
  • tiers[].prefix (optional). Key prefix prepended to each object name.

GCS-tier fields:

  • tiers[].bucket (required). GCS bucket name.
  • tiers[].project (optional). GCP project id; ADC usually resolves it.
  • tiers[].prefix (optional). Object name prefix.

The legacy root field from the pre-Datavo cache hierarchy is accepted for backward compatibility but ignored, with a warning. The source of truth in the Datavo model is the server itself; the hierarchy only describes local tiers.

For a shared GPU workstation with a fast NVMe and a slow shared NFS scratch share:

# /etc/datavo/shards.yaml
tiers:
  - name: local
    path: /var/cache/datavo/shards
    max_bytes: 200GB
  - name: scratch
    path: /scratch/datavo/shards
    max_bytes: 5TB

The first consumer on this machine to read a shard downloads it once and populates both tiers. Subsequent reads on the same shard hit local. A different user reading the same shard later still hits scratch and gets the shard promoted into local on their first read.

Single-user laptop

The built-in default is usually enough; no YAML required.

Cloud topology

On a cloud cluster of GPU VMs the latency and egress profile of the origin Datavo server is unfavorable to read from on every iteration. The recommended pattern is to put one bucket per region in between the per-VM SSD and the origin: every VM in that region shares the cache, and the cache fills naturally as users in the region consume.

The same shape works on all three clouds, swapping the bucket-side tier:

# /etc/datavo/shards.yaml on each GPU VM in eastus (Azure)
tiers:
  - name: nvme
    path: /mnt/cache/datavo/shards
    max_bytes: 200GB
  - name: region-eastus
    type: azure_blob
    container_url: https://datavoeastus.blob.core.windows.net/shards
    # managed identity picks up the VM's identity; no secrets in this file

# us-east-1 on AWS
tiers:
  - name: nvme
    path: /mnt/cache/datavo/shards
    max_bytes: 200GB
  - name: region-us-east-1
    type: s3
    bucket: datavo-cache-us-east-1
    region: us-east-1

# us-central1 on GCP
tiers:
  - name: nvme
    path: /mnt/cache/datavo/shards
    max_bytes: 200GB
  - name: region-us-central1
    type: gcs
    bucket: datavo-cache-us-central1

What happens (identical across clouds):

  1. First VM in the region opens a dataset. Both tiers miss. The origin server streams the tar to the VM, the hierarchy stages it under /mnt/cache/datavo/shards/.staging/, then installs into nvme and uploads to the region bucket. The shard is now warm for the whole region.
  2. Second VM in the same region opens the same dataset. nvme misses on that VM but the region bucket hits. The hierarchy materializes the object into the VM's nvme.staging/, installs into nvme, and returns the local path. The origin server is never contacted.
  3. A VM in a different region opens the same dataset. That region's bucket misses (different bucket), so the call falls through to the origin server and warms up the new region. Cross-region egress is paid once per region, not once per VM.

For the consumer:

  • Nothing changes in code. SampleStream.from_dataset(...) reads the local YAML, walks the hierarchy, and hands back a local path.

The hierarchy treats all three cloud tiers identically; you can in principle mix providers in one config (e.g. an Azure deployment that also reads from a shared GCS bucket of legacy datasets), though that's an unusual setup.

Why not stream straight from the bucket?

The hierarchy could in principle read the tar straight out of the bucket over the network, skipping the local copy. The current design instead materializes the object into the local tier on first read because:

  • Iterating a Datavo dataset is usually a long-running training loop that reads each shard several times across epochs; the local copy pays for itself after one re-read.
  • Local mtime-driven LRU then becomes the per-VM working-set policy without any bucket-side coordination.
  • The consumer's read path stays a single tarfile.open(local_path) call; there is no separate "remote shard" iteration mode.

If a future workload changes that tradeoff (pure shuffle-once, very large shards) a streaming read mode can be added as an opt-in flag without touching the rest of the design.

Concurrency

Folder-tier installs are guarded by per-content-hash advisory file locks (tier.path/.locks/<content_hash>.lock). Two processes on the same machine that race on the same shard:

  • both stage their own download into the same tier's .staging/,
  • one acquires the lock first, finishes its rename, releases the lock,
  • the second acquires the lock, observes the file already exists, discards its own staged copy, and returns the existing path.

The lock is per-shard, not per-tier, so concurrent fetches of different shards do not block each other. The hierarchy itself holds no locks; the locking lives entirely inside each tier.

Cloud-tier installs are race-free at the storage layer: uploads of the same content hash from two VMs may both succeed, but they write the same bytes (same content hash means same payload), so the resulting object is correct either way. The cost is one redundant upload, which is negligible compared to the network fetch from the origin server that both VMs would otherwise do.

Operational Considerations

  • Disk usage. Each tier independently respects its max_bytes. If you set the top tier too small, large shards are skipped there and shared tiers do the work. This is intentional: a too-tight top tier does not break reads, it just degrades them.
  • Filesystem requirements. Tier directories must support os.rename atomically within themselves (any POSIX filesystem does). Cross-tier installs use shutil.copyfile and tolerate cross-filesystem layouts.
  • Cleanup. Stale entries in .staging/ may accumulate if a process is killed mid-download. They are safe to remove manually.
  • Cache poisoning. Files in a tier are trusted because they are addressed by content hash. Anyone who manually drops tars into a tier directory must name them with the correct hash.
  • Observability. The hierarchy logs skipped tiers, promotion attempts, and install failures at WARNING. There is no per-tier metrics emission today; capacity and hit rates are surfaced via the standard logging stream.

Programmatic Access

The hierarchy and its tiers are public SDK surface:

from datavo_sdk import (
    AzureBlobShardCacheTier,
    SampleStream,
    GcsShardCacheTier,
    S3ShardCacheTier,
    ShardCacheHierarchy,
    ShardCacheTier,
    get_default_shard_cache,
)

# Default consumer path.
ds = SampleStream.from_dataset("voice_demo_v1/train", keys=["audio.wav"])

# Inspect or override:
cache = get_default_shard_cache()
for tier in cache.tiers:
    print(tier.name, tier.max_bytes, tier.supports_direct_read)

# Custom on-prem topology, e.g. for tests.
custom = ShardCacheHierarchy(
    [
        ShardCacheTier("/tmp/fast", max_bytes=10 * 1024**3, name="fast"),
        ShardCacheTier("/tmp/slow", name="slow"),
    ]
)
ds = SampleStream.from_dataset("voice_demo_v1/train", keys=["audio.wav"], cache=custom)

# Cloud topology — pick the tier class that matches the region's cloud.
cloud_azure = ShardCacheHierarchy(
    [
        ShardCacheTier("/mnt/cache/datavo/shards", max_bytes=200 * 1024**3, name="nvme"),
        AzureBlobShardCacheTier(
            container_url="https://datavoeastus.blob.core.windows.net/shards",
            name="region-eastus",
        ),
    ]
)

cloud_aws = ShardCacheHierarchy(
    [
        ShardCacheTier("/mnt/cache/datavo/shards", max_bytes=200 * 1024**3, name="nvme"),
        S3ShardCacheTier(
            bucket="datavo-cache-us-east-1",
            region="us-east-1",
            name="region-us-east-1",
        ),
    ]
)

cloud_gcp = ShardCacheHierarchy(
    [
        ShardCacheTier("/mnt/cache/datavo/shards", max_bytes=200 * 1024**3, name="nvme"),
        GcsShardCacheTier(
            bucket="datavo-cache-us-central1",
            name="region-us-central1",
        ),
    ]
)

The same ShardCacheHierarchy can be reused across multiple SampleStream instances. It carries no per-dataset state.

Importing a cloud tier class does not require the cloud SDK; the SDK is imported lazily the first time the tier hits the wire. That keeps the cost of from datavo_sdk import S3ShardCacheTier to zero when boto3 is not installed.