Solve modes

54 · Solver and runtime

A three-link arm sweeps through 25 boxes. The default serial mode sweeps each world's rows in order in one fused kernel. solve_mode="colored" with the propagation response colors contacts that share no body and solves each color in parallel, and computes articulated responses from the tree factorization instead of per row. Which is faster depends on the scene.

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 3-link arm in 25 boxes in each world.

Option	1 world	64 worlds	256 worlds	solver allocation, 256 worlds	result in this tutorial
serial + immediate	5.57 ms	6.48 ms	7.64 ms	538 MB	stack sag 2.07 mm
colored + propagation	1.46 ms 0.26×	1.94 ms 0.30×	2.65 ms 0.35×	715 MB	stack sag 2.00 mm

What to look at

Both modes push the boxes the same way; the details differ because the pile is chaotic.
Second scene, the 8-box stack at 12 iterations: the two sweeps converge alike (2.07 vs 2.00 mm sag), and the colored sweep is several times faster on the arm scene (see Cost).

Key parameter

solve_mode / articulated_contact_response

"serial" + "immediate" (default) or "colored" + "propagation". propagation_constraint_max sizes the propagation rows.

Try changing

Replicate the scene into 512 worlds and time both modes.
Try propagation_color_order="rotating".

Concept

Serial vs colored solve

What it is

The serial sweep solves each world's rows in order in one fused kernel. The colored sweep groups contacts that share no responding body and solves each group in parallel; with the propagation response, articulated contacts use each body's response from the tree factorization instead of a dense row per contact.

In this scene

Both modes push the 25 boxes; trajectories differ because the pile is chaotic. On the 8-box stack at 12 iterations they converge alike: 2.07 mm (serial) vs 2.00 mm (colored) sag.

Considerations

Cost: On this arm-and-boxes scene, colored+propagation took 1.5 vs 5.6 ms per step in one world and 2.7 vs 7.6 ms in 256 worlds, with 30 to 80% more solver memory.
Interactions: Nesterov needs serial; sleeping needs colored when propagation is on.
Use it when: Contact-rich scenes with articulations; measure, since it depends on the scene.

The script

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

54_colored_propagation.py
replay.py

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

"""Solve modes: the serial sweep vs the colored sweep with the propagation contact response."""

import time

import newton
import numpy as np
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)
    # a 3-link arm turning clockwise into a 5 x 5 grid of boxes
    base = builder.add_link(xform=wp.transform((0.0, -0.45, 0.06), wp.quat_identity()))
    builder.add_shape_box(base, hx=0.04, hy=0.04, hz=0.04, cfg=cfg)
    joints = [
        builder.add_joint_revolute(
            -1,
            base,
            axis=newton.Axis.Z,
            parent_xform=wp.transform((0.0, -0.45, 0.06), wp.quat_identity()),
            target_kd=50.0,
            target_qd=-0.8,
        )
    ]
    parent = base
    for i in range(3):
        link = builder.add_link(
            xform=wp.transform((0.0, -0.45 + 0.12 + 0.24 * i, 0.06), wp.quat_identity())
        )
        builder.add_shape_box(link, hx=0.02, hy=0.11, hz=0.03, cfg=cfg)
        anchor = (0.0, 0.0, 0.0) if i == 0 else (0.0, 0.12, 0.0)
        joints.append(
            builder.add_joint_revolute(
                parent,
                link,
                axis=newton.Axis.Z,
                parent_xform=wp.transform(anchor, wp.quat_identity()),
                child_xform=wp.transform((0.0, -0.12, 0.0), wp.quat_identity()),
                target_ke=60.0,
                target_kd=2.0,
            )
        )
        parent = link
    builder.add_articulation(joints)
    for i in range(5):
        for j in range(5):
            box = builder.add_body(
                xform=wp.transform((0.13 * i + 0.1, 0.13 * j - 0.36, 0.04), wp.quat_identity())
            )
            builder.add_shape_box(
                box,
                hx=0.04,
                hy=0.04,
                hz=0.04,
                cfg=newton.ModelBuilder.ShapeConfig(density=300.0, mu=0.6),
            )
    return builder.finalize()


def build_stack(n=8, h=0.05):
    builder = newton.ModelBuilder()
    builder.add_ground_plane(cfg=newton.ModelBuilder.ShapeConfig(mu=0.6))
    for i in range(n):
        box = builder.add_body(
            xform=wp.transform((0.0, 0.0, h + 2 * h * i + 0.0002 * i), wp.quat_identity())
        )
        builder.add_shape_box(box, hx=h, hy=h, hz=h, cfg=newton.ModelBuilder.ShapeConfig(mu=0.6))
    return builder.finalize()


def simulate_stack(rec, label, **options):
    """The same sag check as the iterations tutorial, to compare convergence of the two sweeps."""
    model = build_stack()
    solver = SolverSteel(model, **options)
    pipeline = newton.CollisionPipeline(model)
    contacts = pipeline.contacts()
    state_0, state_1, control = model.state(), model.state(), model.control()

    run = rec.variant(label, model, scene="stack")
    for _ in range(rec.frames(4.0)):
        run.frame(state_0, sag_mm=1000.0 * (0.75 - state_0.body_q.numpy()[-1, 2]))
        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)


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

    run = rec.variant(label, model, scene="arm")
    elapsed = 0.0
    for frame in range(rec.frames(5.0)):
        moved = np.linalg.norm(
            state_0.body_q.numpy()[4:, :2] - model.body_q.numpy()[4:, :2], axis=1
        )
        run.frame(state_0, mean_push_mm=1000.0 * moved.mean())
        wp.synchronize()
        start = time.perf_counter()
        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
        wp.synchronize()
        if frame > 0:  # the first frame includes kernel compilation
            elapsed += time.perf_counter() - start
    run.check(solver)
    print(f"{label}: {1000.0 * elapsed / ((rec.frames(5.0) - 1) * SUBSTEPS):.3f} ms per eager step")


if __name__ == "__main__":
    rec = Recorder()
    simulate(rec, 'solve_mode="serial" (default)')
    simulate(
        rec,
        'solve_mode="colored" + "propagation"',
        solve_mode="colored",
        articulated_contact_response="propagation",
    )
    simulate_stack(rec, 'stack, "serial"')
    simulate_stack(
        rec, 'stack, "colored"', solve_mode="colored", articulated_contact_response="propagation"
    )
    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.