Sleeping islands

50 · Solver and runtime

Four towers settle and fall asleep (shown in blue) once their bodies stay below 0.05 m/s and 0.15 rad/s for 0.5 s. Islands are rebuilt every step from contacts, and sleeping islands skip their rows and their dynamics. A ball pushed at 7 s wakes only the tower it hits. A second scene runs 64 worlds of the same towers, with the default collision pipeline and with CollisionPipeline(deterministic=True); the replay shows world 0 and the chart counts every world.

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: four 5-box towers per world, timed after 4 s of settling.

Option	1 world	64 worlds	1024 worlds	solver allocation, 1024 worlds	result in this tutorial
enable_sleeping=False	1.98 ms	2.35 ms	3.34 ms	702 MB
enable_sleeping=True	0.77 ms 0.39×	2.45 ms 1.04×	3.43 ms 1.03×	709 MB	1 world: all asleep at 2.1 s; 64 and 1024 worlds: most stay awake with this default pipeline

What to look at

One world: the towers settle, and by about 2.1 s every body is asleep.
At 7 s the ball wakes and topples one tower; the other three stay asleep.
64 worlds, default pipeline: only 7% of the bodies are asleep after 6 s. The pipeline can write the same contacts in a different order on each call, and sleeping rules out warm start, so resting boxes keep a jitter whose fast tail reaches the 0.15 rad/s threshold.
64 worlds, CollisionPipeline(deterministic=True): with sorted contacts the jitter was lower in these runs, and every body is asleep by 1.0 s.

Key parameter

enable_sleeping

Experimental. sleep_linear_threshold, sleep_angular_threshold and sleep_quiet_time set when an island sleeps; the body_awake observable reports it.

Try changing

Set sleep_quiet_time=2.0.
Pass sleep_skip_constraints=False to keep the rows of sleeping islands.
Keep the default pipeline in the 64-world scene and set sleep_angular_threshold=0.3.

Concept

Sleeping islands

What it is

Bodies in contact are joined into islands every step. A supported island whose bodies stay below 0.05 m/s and 0.15 rad/s for 0.5 s falls asleep: its state is frozen and its rows and dynamics are skipped until a wake event such as a force, a contact with a moving body or lost support.

In this scene

All 21 bodies are asleep at 2.1 s; the ball pushed at 7 s wakes one tower plus itself. In 64 worlds, 7% of the bodies sleep after 6 s with the default pipeline and all of them by 1.0 s with the deterministic one.

Considerations

Cost: In one world sleeping cut the step from 2.0 to 0.77 ms in this table; the saving depends on how many islands are asleep in the timed window, which varies between runs. The table's 64 and 1024 worlds use the default pipeline and show no gain, because most bodies stay awake. In a separate measurement on the same GPU with CollisionPipeline(deterministic=True), every island slept and a 1024-world solver step took 1.09 ms instead of 3.49 ms.
Failure modes: With the default pipeline, whether islands sleep varies between runs and does not change monotonically with the world count: in separate runs 8 worlds mostly slept, while 16 to 128 worlds left most bodies awake. Nothing in this scene couples the worlds. Check body_awake in batched setups.
Interactions: Not with warm start, velocity iterations or adhesion; driven and sprung articulations stay awake; contact_f reads zero under sleeping bodies.
Use it when: Large mostly-resting scenes, with CollisionPipeline(deterministic=True) or a larger sleep_angular_threshold in batched setups; measure on your own setup.

The script

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

50_sleeping.py
replay.py

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

"""Sleeping islands: settled towers fall asleep and skip their work until something wakes them."""

from types import SimpleNamespace

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
KICK_AT = 7.0  # s
WORLDS = 64
CFG = newton.ModelBuilder.ShapeConfig(mu=0.7)


def add_towers(builder):
    for k in range(4):
        for i in range(5):
            box = builder.add_body(
                xform=wp.transform((0.5 * k - 0.75, 0.0, 0.05 + 0.1005 * i), wp.quat_identity())
            )
            builder.add_shape_box(box, hx=0.05, hy=0.05, hz=0.05, cfg=CFG)


def build():
    builder = newton.ModelBuilder()
    builder.add_ground_plane(cfg=CFG)
    add_towers(builder)
    ball = builder.add_body(xform=wp.transform((-0.25, -1.6, 0.08), wp.quat_identity()))
    builder.add_shape_sphere(
        ball, radius=0.08, cfg=newton.ModelBuilder.ShapeConfig(density=3000.0, mu=0.3)
    )
    return builder.finalize(), ball


def simulate(rec, label):
    model, ball = build()
    solver = SolverSteel(model, enable_sleeping=True)  # thresholds: 0.05 m/s, 0.15 rad/s for 0.5 s
    pipeline = newton.CollisionPipeline(model)
    contacts = pipeline.contacts()
    observed = solver.observables({"body_awake"})
    state_0, state_1, control = model.state(), model.state(), model.control()
    kick = np.zeros((model.body_count, 6), dtype=np.float32)
    # one step of this force gives the ball 4 m/s toward the second tower
    kick[ball, 1] = model.body_mass.numpy()[ball] * 4.0 / DT

    run = rec.variant(label, model, scene="one")
    for frame in range(rec.frames(10.0)):
        awake = observed.body_awake
        run.frame(state_0, awake=awake, awake_bodies=awake.numpy().sum())
        for step in range(SUBSTEPS):
            state_0.clear_forces()
            if frame == int(KICK_AT * rec.fps) and step == 0:
                state_0.body_f.assign(kick)  # a nonzero force also wakes the ball's island
            pipeline.collide(state_0, contacts)
            solver.step(state_0, state_1, control, contacts, DT, observables=observed)
            state_0, state_1 = state_1, state_0
    run.check(solver)


def simulate_worlds(rec, deterministic):
    """WORLDS copies of the towers; the replay shows world 0, the chart counts every world."""
    towers = newton.ModelBuilder()
    add_towers(towers)
    builder = newton.ModelBuilder()
    builder.add_ground_plane(cfg=CFG)
    builder.replicate(towers, WORLDS)
    model = builder.finalize()
    shown = newton.ModelBuilder()
    shown.add_ground_plane(cfg=CFG)
    add_towers(shown)
    shown = shown.finalize()

    solver = SolverSteel(model, enable_sleeping=True)
    # the default pipeline can order contacts differently on each collide; this one sorts them
    pipeline = newton.CollisionPipeline(model, deterministic=deterministic)
    contacts = pipeline.contacts()
    observed = solver.observables({"body_awake"})
    state_0, state_1, control = model.state(), model.state(), model.control()

    pipe = "CollisionPipeline(deterministic=True)" if deterministic else "default pipeline"
    run = rec.variant(f"{WORLDS} worlds, {pipe}", shown, scene="worlds")
    for _ in range(rec.frames(6.0)):
        awake = observed.body_awake
        asleep_pct = 100.0 * (1.0 - awake.numpy().mean())
        first = SimpleNamespace(body_q=state_0.body_q[: shown.body_count])
        run.frame(first, awake=awake[: shown.body_count], asleep_pct=asleep_pct)
        for _ in range(SUBSTEPS):
            pipeline.collide(state_0, contacts)
            solver.step(state_0, state_1, control, contacts, DT, observables=observed)
            state_0, state_1 = state_1, state_0
    run.check(solver)


if __name__ == "__main__":
    rec = Recorder()
    simulate(rec, "enable_sleeping=True")
    for deterministic in [False, True]:
        simulate_worlds(rec, deterministic)
    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.