Row capacity and overflow

56 · Solver and runtime

Constraint rows are stored per world with fixed capacities. row_watermark=True records the most rows each family needed, so you can size free_body_constraint_max and articulation_constraint_max. Rows beyond the capacity are dropped, flagged in solver.constraint_overflow, and reported by check_constraint_capacity().

drag to orbit · scroll to zoom

0.00 s

Cost

Time per step, CUDA graph replay, median of 7 interleaved rounds on one NVIDIA RTX PRO 6000 Blackwell Server Edition; ratios are against the first row. Solver allocation is the GPU memory the solver allocates at construction, before warm-up and graph capture. Scene: the 8-box stack in each world, 8 steps of 1/240 s per frame.

Option	1 world	64 worlds	1024 worlds	solver allocation, 1024 worlds	result in this tutorial
free_body_constraint_max=128	0.84 ms	1.02 ms	1.31 ms	152 MB	drops rows on this stack
free_body_constraint_max=512 (default)	1.01 ms 1.21×	1.21 ms 1.18×	1.65 ms 1.26×	215 MB	fits
free_body_constraint_max=2048	1.02 ms 1.22×	1.21 ms 1.18×	1.68 ms 1.29×	466 MB	fits

What to look at

The nine boxes need about 310 free-body rows at peak, which the default 512 fits.
With 200 rows, contacts are dropped as the boxes land and check_constraint_capacity() raises; here the boxes sink only a few millimetres, but in a larger scene objects fall through.

Key parameter

row_watermark / check_constraint_capacity()

Read constraint_row_watermarks() after a representative run and set the capacity with some margin.

Try changing

Find the smallest capacity with no overflow.
Read solver.constraint_overflow in a kernel as an RL termination signal.

Concept

Row capacity

What it is

Rows live in fixed per-world buffers: 32 rows for articulations and 512 for free bodies by default. Rows that do not fit are dropped and the world is flagged in constraint_overflow, which a kernel can read during a captured run.

In this scene

Nine falling boxes need about 310 free-body rows; a capacity of 200 drops contacts as they land.

Considerations

Cost: Memory grows with capacity × worlds: for 1024 eight-box worlds the solver allocates 152 MB at 128 rows, 215 MB at the default 512 and 467 MB at 2048; time barely changes.
Failure modes: Dropped contacts let objects sink or fall through, sometimes only for a frame. Which contacts keep their rows follows the order of reservation, which by default changes every step; a contact that penetrated while dropped is pushed out when it gets rows back, which adds energy and can make the state diverge. With deterministic=RUN_TO_RUN and CollisionPipeline(deterministic=True) the selection repeats for the same inputs; in a measured folding-chain scene the kept set changed much less between steps and repeated runs matched bitwise. That helps reproduce an overflow but does not fix it.
Use it when: Size from row_watermark over a representative episode with margin, so that nothing overflows; check check_constraint_capacity in tests.

The script

cd tutorials && python 56_row_capacity.py prints each variant's final metric; --record out.json also writes the replay. Every tutorial imports replay.py from the same folder.

56_row_capacity.py
replay.py

# SPDX-FileCopyrightText: Copyright (c) 2026 Skild AI and the Steel contributors
# SPDX-License-Identifier: Apache-2.0

"""Row capacities: measure with row_watermark, catch overflow with check_constraint_capacity."""

import newton
import warp as wp
from newton.solvers import SolverSteel

from replay import Recorder

DT = 1.0 / 240.0
SUBSTEPS = 8


def build():
    builder = newton.ModelBuilder()
    cfg = newton.ModelBuilder.ShapeConfig(mu=0.6)
    builder.add_ground_plane(cfg=cfg)
    for i in range(3):
        for j in range(3):
            box = builder.add_body(
                xform=wp.transform(
                    (0.12 * i - 0.12, 0.12 * j - 0.12, 0.3 + 0.03 * ((i + j) % 3)),
                    wp.quat_rpy(0.1 * i, 0.1 * j, 0.0),
                )
            )
            builder.add_shape_box(box, hx=0.05, hy=0.05, hz=0.05, cfg=cfg)
    return builder.finalize()


def simulate(rec, label, **options):
    model = build()
    solver = SolverSteel(model, row_watermark=True, **options)
    pipeline = newton.CollisionPipeline(model)
    contacts = pipeline.contacts()
    state_0, state_1, control = model.state(), model.state(), model.control()

    run = rec.variant(label, model)
    for _ in range(rec.frames(2.5)):
        marks = solver.constraint_row_watermarks()
        run.frame(
            state_0,
            rows_high_water=marks["free_body_raw_high_water"],
            lowest_z=state_0.body_q.numpy()[:, 2].min(),
        )
        for _ in range(SUBSTEPS):
            pipeline.collide(state_0, contacts)
            solver.step(state_0, state_1, control, contacts, DT)
            state_0, state_1 = state_1, state_0
    run.check(solver)
    try:
        solver.check_constraint_capacity()
        print(f"{label}: fits, peak {marks['free_body_raw_high_water']} free-body rows")
    except RuntimeError as error:
        print(f"{label}: {error}")


if __name__ == "__main__":
    rec = Recorder()
    simulate(rec, "free_body_constraint_max=512 (default)")
    simulate(rec, "free_body_constraint_max=200: rows dropped", free_body_constraint_max=200)
    rec.save()

# SPDX-FileCopyrightText: Copyright (c) 2026 Skild AI and the Steel contributors
# SPDX-License-Identifier: Apache-2.0

"""Record Steel tutorial runs as replay JSON for the tutorials page, or print a summary.

Every tutorial imports this module. ``python NN_name.py`` runs the variants and prints
the final metric of each; ``--record out.json`` also writes body poses, shapes and metric
series that the page replays in 3D.
"""

import argparse
import base64
import json
import os
import subprocess
import sys
import time

import newton
import numpy as np
import warp as wp

from source_hash import source_sha256

_RENDERED_TYPES = {
    newton.GeoType.PLANE,
    newton.GeoType.SPHERE,
    newton.GeoType.CAPSULE,
    newton.GeoType.ELLIPSOID,
    newton.GeoType.CYLINDER,
    newton.GeoType.BOX,
    newton.GeoType.MESH,
    newton.GeoType.CONE,
    newton.GeoType.CONVEX_MESH,
}


class Recorder:
    """Collect the variants of one tutorial; ``fps`` is the replay frame rate."""

    def __init__(self, fps: int = 30):
        parser = argparse.ArgumentParser()
        parser.add_argument(
            "--record", metavar="OUT.json", help="write a replay file for the tutorials page"
        )
        parser.add_argument("--device", default=None, help="Warp device, for example cuda:0")
        self.args = parser.parse_args()
        if self.args.device:
            wp.set_device(self.args.device)
        self.fps = fps
        self.script = os.path.abspath(sys.argv[0])
        self.variants: list[Variant] = []

    def frames(self, seconds: float) -> int:
        """Number of replay frames in ``seconds``, including the first."""
        return int(round(seconds * self.fps)) + 1

    def variant(
        self,
        label: str,
        model: newton.Model,
        *,
        ghost: bool = False,
        slot: int | None = None,
        scene: str | None = None,
    ) -> "Variant":
        """Start one run; ``ghost`` overlays run ``slot``, ``scene`` groups runs for charts."""
        variant = Variant(label, model, self.fps, ghost, slot, scene)
        self.variants.append(variant)
        return variant

    def save(self) -> None:
        """Print a summary and, with ``--record``, write the replay file."""
        for variant in self.variants:
            last = variant.metrics[-1] if variant.metrics else {}
            flag = "  (rows dropped)" if variant.overflow else ""
            print(f"{variant.label:<48} {json.dumps(last)}  {variant.wall:.2f} s{flag}")
        if not self.args.record:
            return
        source_sha = source_sha256(self.script)
        doc = {
            "script": os.path.basename(self.script),
            "script_sha256": source_sha,
            "newton_commit": _newton_commit(),
            "newton_version": newton.__version__,
            "warp_version": wp.config.version,
            "gpu": wp.get_device().name,
            "fps": self.fps,
            "variants": [v.to_json() for v in self.variants],
        }
        with open(self.args.record, "w") as f:
            json.dump(doc, f, separators=(",", ":"))
        print(f"wrote {self.args.record}")


class Variant:
    """One recorded run: a pose per frame for every body, plus named metric values."""

    def __init__(self, label, model, fps, ghost, slot, scene):
        self.label, self.model, self.fps, self.ghost, self.slot = label, model, fps, ghost, slot
        self.scene = scene
        self.poses: list[np.ndarray] = []
        self.metrics: list[dict] = []
        self.awake: list[list[int]] = []
        self.particles: list[np.ndarray] = []
        self.overflow = False
        self.start = time.perf_counter()
        self.wall = 0.0

    def frame(self, state: newton.State, awake=None, particles=None, **metrics) -> None:
        """Record ``state.body_q``, optional awake flags, particle positions and metric values."""
        self.poses.append(state.body_q.numpy().astype(np.float32).copy())
        if particles is not None:
            self.particles.append(particles.numpy().astype(np.float32).copy())
        self.metrics.append({k: _round(v) for k, v in metrics.items()})
        if awake is not None:
            self.awake.append([int(x) for x in awake.numpy()])
        self.wall = time.perf_counter() - self.start

    def check(self, solver) -> None:
        """Note whether the solver dropped rows; call once after the run."""
        try:
            solver.check_constraint_capacity()
        except RuntimeError:
            self.overflow = True

    def to_json(self) -> dict:
        poses = (
            np.stack(self.poses)
            if self.poses
            else np.zeros((0, self.model.body_count, 7), np.float32)
        )
        return {
            "label": self.label,
            "fps": self.fps,
            "ghost": self.ghost,
            "slot": self.slot,
            "scene": self.scene,
            "bodies": int(self.model.body_count),
            "frames": int(poses.shape[0]),
            "poses": base64.b64encode(poses.astype(np.float32).tobytes()).decode(),
            "shapes": _shape_table(self.model),
            "metric": self.metrics,
            "awake": self.awake or None,
            "particles": self._particles_json(),
            "wall_s": round(self.wall, 3),
            "overflow": self.overflow,
        }

    def _particles_json(self):
        if not self.particles:
            return None
        frames = np.stack(self.particles)
        tris = self.model.tri_indices
        radius = self.model.particle_radius.numpy()
        return {
            "count": int(frames.shape[1]),
            "frames": base64.b64encode(frames.astype(np.float32).tobytes()).decode(),
            "radius": round(float(radius.mean()), 5) if radius.size else 0.0,
            "tris": [int(v) for v in tris.numpy().reshape(-1)]
            if tris is not None and tris.shape[0]
            else None,
        }


def _shape_table(model):
    types = model.shape_type.numpy()
    scales = model.shape_scale.numpy()
    xforms = model.shape_transform.numpy()
    bodies = model.shape_body.numpy()
    flags = model.shape_flags.numpy()
    shapes = []
    for i, t in enumerate(types):
        if int(t) not in _RENDERED_TYPES or not flags[i] & int(newton.ShapeFlags.VISIBLE):
            continue
        shape = {
            "b": int(bodies[i]),
            "t": int(t),
            "s": [round(float(v), 5) for v in scales[i]],
            "x": [round(float(v), 6) for v in xforms[i]],
        }
        if int(t) in (newton.GeoType.MESH, newton.GeoType.CONVEX_MESH):
            mesh = model.shape_source[i]
            shape["v"] = [round(float(v), 5) for v in np.asarray(mesh.vertices).reshape(-1)]
            shape["i"] = [int(v) for v in np.asarray(mesh.indices).reshape(-1)]
        shapes.append(shape)
    return shapes


def _round(value):
    value = float(value)
    return round(value, 5) if np.isfinite(value) else None


def _newton_commit():
    if os.environ.get("NEWTON_COMMIT"):
        return os.environ["NEWTON_COMMIT"]
    try:
        root = os.path.dirname(os.path.dirname(newton.__file__))
        return subprocess.check_output(["git", "-C", root, "rev-parse", "HEAD"], text=True).strip()
    except (OSError, subprocess.CalledProcessError):
        return None

Recorded with SolverSteel on Newton 1.9.0.dev0, Warp 1.19.0, NVIDIA RTX PRO 6000 Blackwell Server Edition. Replays read the state back every frame, so they are not a speed benchmark. Script · page text

Steel docs and tutorials are licensed under Apache-2.0. Third-party notices. "Skild", "Skild AI" and the Skild logo are trademarks of Skild AI and are not licensed under Apache-2.0.