Configure ODS from measured throughput

Configure ODS from measured throughput#

Use embodichain.data_pipeline.configuration to plan bounded data consumption from two rates: accepted trajectory production and active learner demand. The result is a static configuration plan. A consumer applies its cache and sampling settings, then can select ODSFlowController for explicit owner-side pacing/refill or provide its own scheduler.

This method applies across tasks and learners. The Franka integration provides a worked binding; its measurements and reuse bounds are provisional case evidence. See ODS for engine and source ownership and Verify the plan for checking the resulting behavior.

Inputs and units#

Input	Meaning and ownership
`accepted_episodes_per_second`, g	Successfully accepted/published trajectories per active production second, including failed attempts and retry cost. Measure complete production intervals.
`training_chunks_per_second`, t	Completed training selections per active learner second, including sampling, copies, feature conversion, transfer and optimizer completion. Exclude deliberate pacing and evaluation waits.
`ODSReuseProfileCfg`	A versioned task/model/sampler protection policy with recommended minimum/maximum selections and a hard ceiling. Supplied by the caller.
`ODSRuntimeCfg`	Measured batch B, producer concurrency P, first-publication latency, minimum cache diversity, measurement scope and optional storage allowance.

One selection is one returned chunk anchored in a source trajectory. Ten draws from a trajectory in a batch consume ten selections. This is neither ten full trajectory passes nor a count of supervised frames. History/horizon, trajectory length and sampling distribution affect exposure; record them in the profile. If the probe reports completed updates/s, multiply by its batch size B to obtain t. Observation frames/s require an explicit conversion and cannot be substituted.

Measure production and training under the intended concurrent placement. A producer-only benchmark does not establish its capacity while sharing a GPU with the learner. A paced wall rate is not the learner’s active capacity. Repeat measurements after changing concurrency, batch, representation, device placement, display or the pacing schedule.

measure_ods_throughput(production, training) accepts two ODSRateMeasurementCfg records with the same absolute monotonic window, explicit count units and active durations. It returns active and wall rates. It calculates rates from supplied observations; it does not run a probe or verify simultaneous saturation. get_stats() is also an observation interface, not an automatic capacity estimator. Counters and production duration advance at different publication/refill boundaries, so a mid-refill counter ratio can misstate throughput. Prefer completed refill intervals or lifecycle timestamps.

Reuse profile#

The profile supplies integer bounds K_min <= K_max <= K_hard, an explicit qualified flag, an identity/scope and a latency safety factor s >= 1. Throughput cannot establish learning-safe bounds. A new profile may be marked qualified=False; the plan retains that qualification and a diagnostic reason.

Changing hardware changes measured rates. Material changes to the model, trajectory distribution, horizon or sampler require profile revalidation. The public calculator validates fields and ranges, not the scientific truth of a caller’s qualification or scope. It also cannot detect stale measurements from two scalar rates; freshness and workload compatibility belong to the host.

Balance, cache and rate limits#

The calculator keeps B and P at the measured values. Two rates do not predict the scaling curve of other batch sizes or producer counts, nor determine the optimizer/model settings or total updates needed for convergence.

K_balance = t / g
K = clamp(floor(K_balance), K_min, K_max)

L_safe = first_publication_p95_seconds * s
C_base = round_up_to_P(max(2*P, min_cache_episodes, P + ceil(B/K)))
C_desired = round_up_to_P(max(C_base, P + ceil(t*L_safe/K)))

For memory allowance M and measured episode-storage bytes E:

C_memory = P * floor(M / E / P)
C = min(C_desired, C_memory)

Both memory fields must be supplied together. The calculator rejects capacity below C_base. Without a memory allowance it uses C_desired. The allowance covers episode storage, not model, native simulator, SDK cache or allocator memory; reserve these separately.

The actual rates also include the active limit imposed by usable cache quota:

unlocked_budget = (C - P) * K
t_active_cap = min(t, unlocked_budget / L_safe)
t_target = min(t_active_cap, g*K)
u_target = t_target / B
g_target = t_target / K
refill_remaining_chunks = max(B, ceil(t_active_cap * L_safe))
initial_max_updates = floor(C*K / B)
learner_duty_estimate = t_target / t
producer_duty_estimate = g_target / g

The implementation conservatively rounds published floating-point limits down where required. Use returned fields rather than reproducing floating arithmetic in another scheduler. Cache/refill sizing is a latency heuristic: first publication does not guarantee sustained subsequent supply or protection from arbitrary stalls. Validate later publication spacing and actual unlocked windows.

When demand exceeds bounded supply, pace the learner. When supply exceeds demand, request less production while retaining the profile’s planned minimum budget. K_min is not a guarantee that every trajectory receives that exposure: replacement, failures and early stopping may leave allowances unused.

Calculate a plan#

The following complete CPU-only example uses illustrative rates and bounds, not a qualified task profile:

from dataclasses import asdict
from embodichain.data_pipeline.configuration import (
    ODSReuseProfileCfg,
    ODSRuntimeCfg,
    configure_ods,
)

profile = ODSReuseProfileCfg(
    profile_id="illustrative-v1",
    scope="Illustrative numeric chunk workload; requires learning validation",
    qualified=False,
    min_chunks_per_episode=200,
    max_chunks_per_episode=600,
    hard_max_chunks_per_episode=1000,
)
runtime = ODSRuntimeCfg(
    batch_size=64,
    producer_envs=4,
    first_publication_p95_seconds=2.0,
    min_cache_episodes=8,
    measurement_scope="Illustrative concurrent rates, not a hardware result",
)
plan = configure_ods(1.0, 1000.0, profile=profile, runtime=runtime)
assert plan.episode_sample_budget == 600
print(asdict(plan))

ODSConfiguration includes assigned budget and hard limit, cache size/storage, initial credits/update ceiling, active/average rate targets, refill threshold, duty estimates, profile qualification, diagnostic reasons and stopping rule. Invalid/non-positive rates, invalid profiles and infeasible memory raise an error; they do not silently remove protection.

Apply the plan#

Plan field	Runtime owner
`buffer_size`, `batch_size`, `producer_envs`	Host constructs the cache/consumer at these measured settings.
`episode_sample_budget`, `episode_sample_hard_limit`	Pass to `OnlineDataEngineCfg`; engine samplers enforce per-source delivery under the producer lock. Resident sources have their own optional ledger.
`active_chunks_per_second_cap`, `training_chunks_per_second_target`	`ODSFlowController.prepare_batch()` spaces batch starts at the smaller rate. The host calls it before sampling/training; actual completed throughput still needs measurement.
`refill_remaining_chunks`	With a plan, the controller compares coarse unlocked quota with this threshold and schedules manual refill.
`publication_episodes_per_second_target`	Production target to compare with observed publication. The controller schedules whole-cache refills by quota, without a separate per-episode time gate.

For example, after supplying the task’s validated gym_config and state width:

from embodichain.data_pipeline.datasets import OnlineDataSource, SequenceSpecCfg
from embodichain.data_pipeline.engine import OnlineDataEngine, OnlineDataEngineCfg
from embodichain.data_pipeline.runtime import ODSFlowController, ODSFlowControllerCfg

engine_cfg = OnlineDataEngineCfg(
    gym_config={**your_gym_config, "num_envs": plan.producer_envs},
    state_dim=your_state_width,
    action_dim=your_action_width,
    buffer_size=plan.buffer_size,
    buffer_device="cpu",
    refill_policy="manual",
    episode_sample_budget=plan.episode_sample_budget,
    episode_sample_hard_limit=plan.episode_sample_hard_limit,
    include_provenance=True,
)

# The application supplies the truthful schema, feature transform and train_step.
with OnlineDataEngine(engine_cfg) as engine:
    source = OnlineDataSource(
        engine, schema=your_schema, observation_transform=your_feature_transform
    )
    flow = ODSFlowController(
        engine,
        ODSFlowControllerCfg(max_generated_episodes=3 * plan.buffer_size),
        plan=plan,
    )
    sequence = SequenceSpecCfg(history_length=2, prediction_horizon=64)
    for _ in range(1000):
        if not flow.prepare_batch(plan.batch_size):
            break  # Final supply cannot fund another complete batch.
        batch = flow.sample_context(source, plan.batch_size, sequence)
        train_step(batch)

The context manager starts and stops the engine. The controller requires manual refill and runs only in its creating process/thread; it creates no background scheduler or DataLoader worker. Its sample_context() accepts only an OnlineDataSource wrapping the controlled engine. Sample resident sources directly through their own sample_context(); do not pass them to the controller. It validates the engine’s cache, producer count and assigned/hard quotas against the plan; prepare_batch() validates batch size. With either a plan or a production limit, engine.cfg.gym_config must declare a positive integer num_envs that divides cache size, so each fill produces exactly one cache. Unplanned, unlimited legacy engines do not acquire this restriction. The host retains responsibility for workload and semantic schema compatibility. Constructing a controller does not apply engine settings.

With a plan, prepare_batch() schedules refill, waits for enough coarse unlocked quota and spaces batch starts. It returns False when no further whole-cache refill fits the production limit and remaining quota cannot fund a full batch. Slow updates or evaluation pauses do not accumulate catch-up credits. Call it before each planned batch: sample_context() alone does not apply pacing.

max_generated_episodes includes initial fill and must be at least cache size. It limits requests made through this controller to complete cache refills; unused partial-cache capacity remains unproduced. An already pending refill is adopted without another engine request or a fabricated request timestamp. Calls made directly to the engine bypass this controller policy.

Without plan, prepare_batch() only polls health/events. Refill stays explicit through flow.request_refill(), which returns a target generation or None when no whole cache fits. This supports caller-defined round schedules without automatically changing their data distribution.

The planner’s remaining-quota threshold is not engine refill_threshold. The engine’s default sample_threshold policy counts delivered samples against refill_threshold * buffer_size after successful calls. Quota exhaustion does not itself request refill. Use manual scheduling for the plan above.

Bounded waits and callbacks#

ODSFlowControllerCfg defaults to 30 seconds for data/quota waits, 1800 seconds for a complete generation and 0.02 seconds between polling probes. A transient live DataUnavailableError or DataBudgetExhaustedError is retried only within the configured data deadline. Worker failures, invalid inputs and caller cancellation propagate. Callbacks run synchronously; an interval observer failure preserves an already active error with a diagnostic note. Resident sources are sampled directly; Source and Dataset APIs retain their immediate error semantics.

flow.wait_for_generation(generation, deadline=...) waits for the whole fill, not its first publication. Its optional absolute time.monotonic() deadline can shorten the configured generation timeout. This lets a collector count time spent writing a previous generation against the same request deadline.

All waiting/pacing loops poll caller progress, drain lifecycle events and check engine health. Optional callbacks let the application retain its own accounting:

Callback	Meaning
`progress()`	Check application deadlines or cancellation; raise to interrupt.
`on_events(events)`	Receive drained publication/eviction records. Events are drained even without an observer; install one to retain accounting.
`on_refill(generation, requested_at)`	Observe new requests made by this controller, once per generation, in absolute host monotonic seconds. Adopting an already pending generation emits no request callback.
`on_interval(kind, started, ended)`	Receive `data_wait`, `quota_wait` and `usage_throttle` intervals in absolute host monotonic seconds, including zero-duration waits.

Convert callback times explicitly when an application reports seconds since launch; do not mix them with Unix acceptance timestamps. The application owns completed optimizer counts, telemetry and shutdown. flow.drain_events() can also drain final records after stop/failure. An enabled lifecycle log still needs capacity for initial fill before start() returns. The controller does not snapshot or restore cache, quotas or pacing state.

For frequent probes, engine.check_health() checks readiness/liveness without sampling or refill. get_flow_status() returns compact detached generation, accepted-episode and optional quota counters without constructing per-row audit records or copying trajectory payloads. Producer and quota counters use separate lock snapshots. Unlocked quota requires a valid continuity-qualified anchor and excludes the write window; sequence/mode/generation filters may further restrict sampling. Use get_sampling_budget_snapshot() for the full retained-source audit rather than in every pacing poll.

Strict budget and consumer accounting#

For an immutable (run_id, episode_id), each requested batch’s selection, copy and debit share the producer lock. Within-batch duplicate draws count separately. The assigned K is a delivery allowance; the hard limit bounds its configuration. Current budgets are fixed for an engine/source lifetime: the planner does not implement online adaptation or replenish exhausted identities.

Failed optimization, transforms or prefetched data do not refund delivered chunks. Track optimizer-completed chunks separately. Live replacement retires unused credits; it cannot move them to another identity. A new live run creates new identities. Replaying existing identities requires restored usage accounting; the demo rejects protected offline resume without that state.

The initial update count is a ceiling, not a convergence recommendation. After production stops, the host may consume only remaining quotas and must stop when a requested full batch cannot be funded. Flow control applies the given profile and plan; it does not derive a new safe K or convergence budget.

Verify the plan#

Observe several complete refills under the intended concurrent workload and compare the resolved plan with these measurements:

Check	What to record
Supply and consumption	Accepted episodes/s, completed chunks/s, and first/full-publication latency. Keep active service rates separate from wall rates that include pacing and waits.
Protected reuse	Per-source delivered chunks versus assigned K, completed optimizer chunks, generation shares and unused allowance retired on replacement. Retain source history in the application because engine snapshots contain only current rows.
Frame exposure	Unique anchors and valid supervised-frame appearances through `loss_mask`; a chunk selection is not a whole-episode pass.
Waiting and resources	Data/quota waits, intentional pacing, actual cache storage and total CPU/GPU memory. The cache allowance covers episode payloads only.

Use one monotonic time domain and retain counts with their interval boundaries. Sampled acceptance timestamps use Unix time and require explicit alignment before comparison with monotonic publication events. Cross-check lifetime deliveries against application records; prefetched or failed consumer work still spends delivery quota.

Measure the overhead of accounting and resource sampling in the complete consumer. Record task quality separately: rate matching and bounded reuse do not establish convergence or an efficiency advantage. The BC example’s comparison protocol defines experiment arms, checkpoint evaluation and cold-start/data-ready costs.

Recalibration and diagnosis#

Observation	Check before changing configuration
Learner waits at zero quota	Separate intentional protection from missing production. Measure joint supply before changing P; keep profile bounds.
Quota remains but sampling fails	Check write locks, continuity, sequence length and segment/generation filters. Positive credits alone do not prove an eligible context exists.
New trajectories retire unused	Inspect residence, cache replacement and refill timing; more production may increase waste.
Old/first-generation data dominates	Inspect startup allowance, gaps, source age and post-production training. The average target alone does not constrain every tail.
Quality regresses at healthy throughput	Audit exposure and development rollouts; rates cannot determine convergence or diagnose all policy failures.
More producers reduce throughput	Re-measure learner capacity and CPU/NUMA/GPU contention under joint load.

Record the resolved plan and inputs, actual accepted/delivered/completed counts, generation shares, retired allowances, latency observations, waits and resource scope. Compare predictions with measured rates after changing the schedule.