Deep memory — memray, folded to your source lines
+
-
-
Memory over time
- {samples && samples.length > 0 ? (
- No memory samples yet.
- )}
+ {hotLine && (
+
+ Hottest allocation
+
+ {hotLine.filename.split("/").slice(-1)[0]}:{hotLine.lineno} {hotLine.function}
+
+ {bytes(hotLine.hwm_bytes)}
+
+
+ Deep memory →
+
+ )}
+
+
+ Memory over time ·{" "}
+
+ open the Timeline to zoom & inspect spikes →
+
+
+ {samples && samples.length > 0 ? (
+ No memory samples yet.
+ )}
+
+
Tasks by layer{duration ? ` · ${duration.toFixed(1)}s total` : ""}
diff --git a/web/app/runs/[id]/stream/page.tsx b/web/app/runs/[id]/stream/page.tsx
new file mode 100644
index 0000000..051b115
--- /dev/null
+++ b/web/app/runs/[id]/stream/page.tsx
@@ -0,0 +1,12 @@
+"use client";
+
+import { TaskStream } from "@/components/TaskStream";
+
+export default function StreamPage() {
+ return (
+ <>
+ () {
- const ref = useRef(null);
- const [w, setW] = useState(960);
- useEffect(() => {
- if (!ref.current) return;
- const ro = new ResizeObserver(([e]) => setW(e.contentRect.width));
- ro.observe(ref.current);
- return () => ro.disconnect();
- }, []);
- return [ref, w] as const;
-}
-
-function fmtTime(s: number): string {
- return s < 1 ? `${(s * 1000).toFixed(0)}ms` : `${s.toFixed(s < 10 ? 1 : 0)}s`;
-}
-
-export function AlignedTimeline({ runId }: { runId: string }) {
- const { data: samples } = useTimeline(runId);
- const { data: spans } = useSpans(runId);
- const [wrapRef, width] = useWidth();
- const colorOf = useMemo(() => layerColorMap(), []);
-
- const model = useMemo(() => {
- const S: Sample[] = samples ?? [];
- const T: TaskSpan[] = spans ?? [];
- const times = [
- ...S.map((s) => s.timestamp),
- ...T.map((t) => t.start),
- ...T.map((t) => t.end),
- ];
- if (times.length === 0) return null;
- const t0 = Math.min(...times);
- const t1 = Math.max(...times);
- const dur = Math.max(0.001, t1 - t0);
- const workers = [...new Set([...S.map((s) => s.worker), ...T.map((t) => t.worker)])].sort();
- const peak = Math.max(1, ...S.map((s) => s.rss_bytes));
- const layers = [...new Set(T.map((t) => baseName(t.layer)))];
- return { S, T, t0, dur, workers, peak, layers };
- }, [samples, spans]);
-
- if (!model)
- return ;
+}
+
+interface Cell {
+ node: FNode;
+ x: number; // 0..1 within the rendered root
+ w: number;
+ depth: number;
+}
+
+function shortFile(path: string): string {
+ return path.split("/").slice(-1)[0] ?? path;
+}
+function isLib(path: string): boolean {
+ return /site-packages|dist-packages|\/(numpy|dask|distributed|pandas|xarray|zarr)\//.test(path);
+}
+
+function buildTree(stacks: FlameStack[]): FNode {
+ const root: FNode = { key: "root", frame: null, bytes: 0, children: new Map() };
+ for (const s of stacks) {
+ root.bytes += s.hwm_bytes;
+ let node = root;
+ for (const f of s.frames) {
+ const k = `${f.function}|${f.filename}|${f.lineno}`;
+ let c = node.children.get(k);
+ if (!c) {
+ c = { key: `${node.key}/${k}`, frame: f, bytes: 0, children: new Map() };
+ node.children.set(k, c);
+ }
+ c.bytes += s.hwm_bytes;
+ node = c;
+ }
+ }
+ return root;
+}
+
+function layout(rootNode: FNode): { cells: Cell[]; maxDepth: number } {
+ const cells: Cell[] = [];
+ let maxDepth = 0;
+ const place = (node: FNode, x: number, w: number, depth: number) => {
+ if (w < 0.0015) return; // too thin to see or interact with
+ cells.push({ node, x, w, depth });
+ maxDepth = Math.max(maxDepth, depth);
+ const kids = [...node.children.values()].sort((a, b) => b.bytes - a.bytes);
+ let cx = x;
+ for (const c of kids) {
+ const cw = w * (c.bytes / node.bytes);
+ place(c, cx, cw, depth + 1);
+ cx += cw;
+ }
+ };
+ place(rootNode, 0, 1, 0);
+ return { cells, maxDepth };
+}
+
+function findByKey(root: FNode, key: string): FNode | null {
+ if (root.key === key) return root;
+ for (const c of root.children.values()) {
+ const hit = findByKey(c, key);
+ if (hit) return hit;
+ }
+ return null;
+}
+
+export function Flamegraph({ runId }: { runId: string }) {
+ const [worker, setWorker] = useState(null);
+ const [focusKey, setFocusKey] = useState("root");
+ const [selected, setSelected] = useState(null);
+ const [hover, setHover] = useState<{ x: number; y: number; node: FNode } | null>(null);
+ const { data, mutate } = useFlamegraph(runId, worker);
+ const { deepNonce } = useLive();
+ const colorOf = useMemo(() => layerColorMap(), []);
+
+ useEffect(() => {
+ if (deepNonce > 0) mutate();
+ }, [deepNonce, mutate]);
+
+ const tree = useMemo(() => buildTree(data?.stacks ?? []), [data]);
+ const focus = useMemo(() => findByKey(tree, focusKey) ?? tree, [tree, focusKey]);
+ const { cells, maxDepth } = useMemo(() => layout(focus), [focus]);
+
+ const total = tree.bytes || 1;
+ const workers = data?.workers ?? [];
+ const hasStacks = (data?.stacks?.length ?? 0) > 0;
+
+ return (
+ (null);
+ const canvasRef = useRef(null);
+ const [size, setSize] = useState({ w: 900, h: 620 });
+ const view = useRef({ scale: 1, tx: 0, ty: 0 });
+ const drag = useRef<{ x: number; y: number; moved: boolean } | null>(null);
+ const marquee = useRef<{ x0: number; y0: number } | null>(null);
+ const [box, setBox] = useState<{ x: number; y: number; w: number; h: number } | null>(null);
+ const [hover, setHover] = useState<{ x: number; y: number; node: CNode } | null>(null);
+ const [, force] = useState(0);
+ const redraw = () => force((n) => n + 1);
+
+ const layout = useMemo(() => {
+ const g = new dagre.graphlib.Graph();
+ g.setGraph({ rankdir: "TB", nodesep: 8, ranksep: 24 });
+ g.setDefaultEdgeLabel(() => ({}));
+ const ids = new Set(nodes.map((n) => n.id));
+ nodes.forEach((n) => g.setNode(n.id, { width: NW, height: NH }));
+ const drawnEdges: [string, string][] = [];
+ for (const [s, t] of edges) {
+ if (ids.has(s) && ids.has(t)) {
+ g.setEdge(s, t);
+ drawnEdges.push([s, t]);
+ }
+ }
+ dagre.layout(g);
+ const placed: Placed[] = nodes.map((n) => {
+ const p = g.node(n.id);
+ return { id: n.id, x: p.x, y: p.y, w: NW, h: NH, data: n };
+ });
+ const byId = new Map(placed.map((p) => [p.id, p]));
+ const gw = (g.graph().width ?? 1000) as number;
+ const gh = (g.graph().height ?? 1000) as number;
+ return { placed, byId, drawnEdges, gw, gh };
+ }, [nodes, edges]);
+
+ // Fit to view whenever the graph changes.
+ useEffect(() => {
+ const { gw, gh } = layout;
+ const scale = Math.min(size.w / (gw + 40), size.h / (gh + 40), 1.5);
+ view.current = {
+ scale,
+ tx: (size.w - gw * scale) / 2,
+ ty: 20,
+ };
+ redraw();
+ }, [layout, size]);
+
+ useEffect(() => {
+ const el = wrapRef.current;
+ if (!el) return;
+ const ro = new ResizeObserver(() =>
+ setSize({ w: el.clientWidth, h: el.clientHeight }),
+ );
+ ro.observe(el);
+ setSize({ w: el.clientWidth, h: el.clientHeight });
+ return () => ro.disconnect();
+ }, []);
+
+ useEffect(() => {
+ const canvas = canvasRef.current;
+ if (!canvas) return;
+ const dpr = window.devicePixelRatio || 1;
+ canvas.width = size.w * dpr;
+ canvas.height = size.h * dpr;
+ const ctx = canvas.getContext("2d");
+ if (!ctx) return;
+ ctx.setTransform(dpr, 0, 0, dpr, 0, 0);
+ ctx.clearRect(0, 0, size.w, size.h);
+ const { scale, tx, ty } = view.current;
+ ctx.save();
+ ctx.translate(tx, ty);
+ ctx.scale(scale, scale);
+
+ // edges
+ ctx.strokeStyle = "#c9ccd2";
+ ctx.lineWidth = 0.5 / scale;
+ ctx.beginPath();
+ for (const [s, t] of layout.drawnEdges) {
+ const a = layout.byId.get(s);
+ const b = layout.byId.get(t);
+ if (!a || !b) continue;
+ ctx.moveTo(a.x, a.y + a.h / 2);
+ ctx.lineTo(b.x, b.y - b.h / 2);
+ }
+ ctx.stroke();
+
+ // nodes — cull to the visible world rect so deep zoom stays fast
+ const showLabels = scale > LABEL_SCALE;
+ const vx0 = -tx / scale;
+ const vy0 = -ty / scale;
+ const vx1 = (size.w - tx) / scale;
+ const vy1 = (size.h - ty) / scale;
+ const fontPx = Math.min(9, Math.max(5, 8));
+ for (const p of layout.placed) {
+ if (p.x < vx0 - p.w || p.x > vx1 + p.w || p.y < vy0 - p.h || p.y > vy1 + p.h) continue;
+ ctx.fillStyle = p.data.color;
+ ctx.fillRect(p.x - p.w / 2, p.y - p.h / 2, p.w, p.h);
+ const isSel = p.id === selected;
+ const isHover = hover?.node.id === p.id;
+ if (p.data.hot || isSel || isHover) {
+ ctx.strokeStyle = isSel ? "#1b1d21" : isHover ? "#e8590c" : "#d43b3b";
+ ctx.lineWidth = (isSel || isHover ? 2.5 : 2) / scale;
+ ctx.strokeRect(p.x - p.w / 2, p.y - p.h / 2, p.w, p.h);
+ }
+ if (showLabels) {
+ ctx.fillStyle = "#1b1d21";
+ ctx.font = `${fontPx}px ui-monospace, monospace`;
+ ctx.textBaseline = "middle";
+ ctx.save();
+ ctx.beginPath();
+ ctx.rect(p.x - p.w / 2, p.y - p.h / 2, p.w, p.h);
+ ctx.clip();
+ ctx.fillText(p.data.label, p.x - p.w / 2 + 2, p.y);
+ ctx.restore();
+ }
+ }
+
+ // highlight the active node's edges so its connections are unmistakable
+ const active = hover?.node.id ?? selected;
+ if (active) {
+ ctx.strokeStyle = "#e8590c";
+ ctx.lineWidth = 1.6 / scale;
+ ctx.beginPath();
+ for (const [s, t] of layout.drawnEdges) {
+ if (s !== active && t !== active) continue;
+ const a = layout.byId.get(s);
+ const b = layout.byId.get(t);
+ if (!a || !b) continue;
+ ctx.moveTo(a.x, a.y + a.h / 2);
+ ctx.lineTo(b.x, b.y - b.h / 2);
+ }
+ ctx.stroke();
+ }
+ ctx.restore();
+ });
+
+ const toWorld = (cx: number, cy: number) => {
+ const { scale, tx, ty } = view.current;
+ return { x: (cx - tx) / scale, y: (cy - ty) / scale };
+ };
+
+ const onWheel = (e: React.WheelEvent) => {
+ e.preventDefault();
+ const rect = canvasRef.current!.getBoundingClientRect();
+ const cx = e.clientX - rect.left;
+ const cy = e.clientY - rect.top;
+ const before = toWorld(cx, cy);
+ // finer step for precise control; trackpads send small deltas → scale them
+ const step = Math.min(0.25, Math.abs(e.deltaY) / 400 + 0.06);
+ const factor = e.deltaY < 0 ? 1 + step : 1 / (1 + step);
+ const v = view.current;
+ v.scale = Math.max(MIN_SCALE, Math.min(MAX_SCALE, v.scale * factor));
+ v.tx = cx - before.x * v.scale;
+ v.ty = cy - before.y * v.scale;
+ redraw();
+ };
+
+ const nodeAt = (cx: number, cy: number): Placed | null => {
+ const w = toWorld(cx, cy);
+ for (const p of layout.placed) {
+ if (Math.abs(w.x - p.x) <= p.w / 2 + 1 && Math.abs(w.y - p.y) <= p.h / 2 + 1) return p;
+ }
+ return null;
+ };
+
+ const canvasXY = (e: React.MouseEvent) => {
+ const rect = canvasRef.current!.getBoundingClientRect();
+ return { cx: e.clientX - rect.left, cy: e.clientY - rect.top };
+ };
+
+ const onDown = (e: React.MouseEvent) => {
+ // Shift-drag draws a box to zoom into that region; plain drag pans.
+ if (e.shiftKey) {
+ const { cx, cy } = canvasXY(e);
+ marquee.current = { x0: cx, y0: cy };
+ setBox({ x: cx, y: cy, w: 0, h: 0 });
+ return;
+ }
+ drag.current = { x: e.clientX, y: e.clientY, moved: false };
+ };
+ const onMoveDrag = (e: React.MouseEvent) => {
+ if (marquee.current) {
+ const { cx, cy } = canvasXY(e);
+ const { x0, y0 } = marquee.current;
+ setBox({ x: Math.min(x0, cx), y: Math.min(y0, cy), w: Math.abs(cx - x0), h: Math.abs(cy - y0) });
+ return;
+ }
+ if (drag.current) {
+ const dx = e.clientX - drag.current.x;
+ const dy = e.clientY - drag.current.y;
+ if (Math.abs(dx) + Math.abs(dy) > 2) drag.current.moved = true;
+ view.current.tx += dx;
+ view.current.ty += dy;
+ drag.current.x = e.clientX;
+ drag.current.y = e.clientY;
+ if (hover) setHover(null);
+ redraw();
+ return;
+ }
+ // hover hit-test for the tooltip + highlight
+ const rect = canvasRef.current!.getBoundingClientRect();
+ const cx = e.clientX - rect.left;
+ const cy = e.clientY - rect.top;
+ const p = nodeAt(cx, cy);
+ if (p) setHover({ x: cx, y: cy, node: p.data });
+ else if (hover) setHover(null);
+ };
+ const onUp = (e: React.MouseEvent) => {
+ if (marquee.current) {
+ marquee.current = null;
+ const m = box;
+ setBox(null);
+ if (m && m.w > 8 && m.h > 8) {
+ const w0 = toWorld(m.x, m.y);
+ const w1 = toWorld(m.x + m.w, m.y + m.h);
+ const worldW = Math.max(1, w1.x - w0.x);
+ const worldH = Math.max(1, w1.y - w0.y);
+ const scale = Math.max(
+ MIN_SCALE,
+ Math.min(MAX_SCALE, Math.min(size.w / worldW, size.h / worldH)),
+ );
+ const cxw = (w0.x + w1.x) / 2;
+ const cyw = (w0.y + w1.y) / 2;
+ view.current = { scale, tx: size.w / 2 - cxw * scale, ty: size.h / 2 - cyw * scale };
+ redraw();
+ }
+ return;
+ }
+ const d = drag.current;
+ drag.current = null;
+ if (d && !d.moved) {
+ const { cx, cy } = canvasXY(e);
+ const p = nodeAt(cx, cy);
+ if (p) onSelect(p.id);
+ }
+ };
+
+ const zoomBy = (factor: number) => {
+ const v = view.current;
+ const cx = size.w / 2;
+ const cy = size.h / 2;
+ const before = toWorld(cx, cy);
+ v.scale = Math.max(MIN_SCALE, Math.min(MAX_SCALE, v.scale * factor));
+ v.tx = cx - before.x * v.scale;
+ v.ty = cy - before.y * v.scale;
+ redraw();
+ };
+ const fit = () => {
+ const { gw, gh } = layout;
+ const scale = Math.min(size.w / (gw + 40), size.h / (gh + 40), 1.5);
+ view.current = { scale, tx: (size.w - gw * scale) / 2, ty: 20 };
+ redraw();
+ };
+
+ return (
+ ;
+}
+
+// Deep-memory allocations over time, stacked by task layer — canvas so it has
+// the same scroll-to-zoom / drag-to-pan / shift-box-zoom as the other charts.
+export function LayerMemoryChart({ runId }: { runId: string }) {
+ const { data, mutate } = useAllocTimeline(runId);
+ const { deepNonce } = useLive();
+ const colorOf = useMemo(() => layerColorMap(), []);
+ const wrapRef = useRef(null);
+ const canvasRef = useRef(null);
+ const [width, setWidth] = useState(900);
+ const [hover, setHover] = useState<{ x: number; y: number; t: number } | null>(null);
+
+ useEffect(() => {
+ if (deepNonce > 0) mutate();
+ }, [deepNonce, mutate]);
+
+ const { buckets, layers, t0, domainHi, yMax } = useMemo(() => {
+ const raw = (data ?? []).map((r) => ({ ...r, bytes: Number(r.bytes) }));
+ if (raw.length === 0)
+ return { buckets: [] as Bucket[], layers: [] as string[], t0: 0, domainHi: 1, yMax: 1 };
+ const t0 = Math.min(...raw.map((r) => r.ts));
+ const peak = new Map();
+ for (const r of raw) peak.set(r.layer, Math.max(peak.get(r.layer) ?? 0, r.bytes));
+ const layers = [...peak.entries()]
+ .sort((a, b) => b[1] - a[1])
+ .map(([l]) => l)
+ .slice(0, MAX_LAYERS);
+ const keep = new Set(layers);
+ const hasOther = raw.some((r) => !keep.has(r.layer));
+ const byT = new Map>();
+ for (const r of raw) {
+ const t = Math.round((r.ts - t0) * 2) / 2;
+ const row = byT.get(t) ?? {};
+ const key = keep.has(r.layer) ? r.layer : "(other)";
+ row[key] = (row[key] ?? 0) + r.bytes;
+ byT.set(t, row);
+ }
+ if (hasOther) layers.push("(other)");
+ const buckets: Bucket[] = [...byT.entries()]
+ .map(([t, vals]) => ({ t, vals }))
+ .sort((a, b) => a.t - b.t);
+ let yMax = 1;
+ let domainHi = 1;
+ for (const b of buckets) {
+ const tot = layers.reduce((a, l) => a + (b.vals[l] ?? 0), 0);
+ if (tot > yMax) yMax = tot;
+ if (b.t > domainHi) domainHi = b.t;
+ }
+ return { buckets, layers, t0, domainHi, yMax: yMax * 1.05 };
+ }, [data]);
+
+ const view = useRef({ lo: 0, hi: domainHi });
+ const [, force] = useState(0);
+ const redraw = () => force((n) => n + 1);
+ const drag = useRef<{ x: number; lo: number; hi: number } | null>(null);
+ const marquee = useRef<{ x0: number } | null>(null);
+ const [box, setBox] = useState<{ x: number; w: number } | null>(null);
+
+ useEffect(() => {
+ view.current = { lo: 0, hi: domainHi };
+ redraw();
+ }, [domainHi]);
+
+ useEffect(() => {
+ const el = wrapRef.current;
+ if (!el) return;
+ const ro = new ResizeObserver(() => setWidth(el.clientWidth));
+ ro.observe(el);
+ setWidth(el.clientWidth);
+ return () => ro.disconnect();
+ }, []);
+
+ const plotL = PAD_L;
+ const plotW = Math.max(1, width - PAD_L - PAD_R);
+ const plotT = PAD_T;
+ const plotH = Math.max(1, HEIGHT - PAD_T - AXIS_H);
+ const xOf = (t: number) =>
+ plotL + ((t - view.current.lo) / (view.current.hi - view.current.lo)) * plotW;
+ const tOf = (x: number) =>
+ view.current.lo + ((x - plotL) / plotW) * (view.current.hi - view.current.lo);
+ const yOf = (v: number) => plotT + (1 - v / yMax) * plotH;
+
+ useEffect(() => {
+ const canvas = canvasRef.current;
+ if (!canvas) return;
+ const dpr = window.devicePixelRatio || 1;
+ canvas.width = width * dpr;
+ canvas.height = HEIGHT * dpr;
+ const ctx = canvas.getContext("2d");
+ if (!ctx) return;
+ ctx.setTransform(dpr, 0, 0, dpr, 0, 0);
+ ctx.clearRect(0, 0, width, HEIGHT);
+ const { lo, hi } = view.current;
+
+ // y grid
+ ctx.strokeStyle = "#eef0f3";
+ ctx.fillStyle = "#9298a2";
+ ctx.font = "10px ui-monospace, monospace";
+ ctx.textBaseline = "middle";
+ for (let i = 0; i <= 4; i++) {
+ const v = (yMax * i) / 4;
+ const y = yOf(v);
+ ctx.beginPath();
+ ctx.moveTo(plotL, y);
+ ctx.lineTo(width - PAD_R, y);
+ ctx.stroke();
+ ctx.fillText(bytes(v), 6, y);
+ }
+
+ const vis = buckets.filter((b) => b.t >= lo - 1 && b.t <= hi + 1);
+ if (vis.length >= 2) {
+ ctx.save();
+ ctx.beginPath();
+ ctx.rect(plotL, plotT, plotW, plotH);
+ ctx.clip();
+ // stack from bottom: keep a running cumulative per bucket
+ const cum = new Array(vis.length).fill(0);
+ for (let li = layers.length - 1; li >= 0; li--) {
+ const layer = layers[li];
+ ctx.beginPath();
+ // top edge L->R
+ for (let i = 0; i < vis.length; i++) {
+ const top = cum[i] + (vis[i].vals[layer] ?? 0);
+ const x = xOf(vis[i].t);
+ const y = yOf(top);
+ if (i === 0) ctx.moveTo(x, y);
+ else ctx.lineTo(x, y);
+ }
+ // bottom edge R->L
+ for (let i = vis.length - 1; i >= 0; i--) {
+ ctx.lineTo(xOf(vis[i].t), yOf(cum[i]));
+ }
+ ctx.closePath();
+ ctx.fillStyle = colorOf(layer);
+ ctx.globalAlpha = 0.78;
+ ctx.fill();
+ ctx.globalAlpha = 1;
+ for (let i = 0; i < vis.length; i++) cum[i] += vis[i].vals[layer] ?? 0;
+ }
+ // hover crosshair
+ if (hover) {
+ ctx.strokeStyle = "#8a8f98";
+ ctx.beginPath();
+ ctx.moveTo(hover.x, plotT);
+ ctx.lineTo(hover.x, plotT + plotH);
+ ctx.stroke();
+ }
+ ctx.restore();
+ }
+
+ // x axis
+ ctx.fillStyle = "#8a8f98";
+ ctx.textBaseline = "top";
+ for (let i = 0; i <= 8; i++) {
+ const t = lo + ((hi - lo) * i) / 8;
+ ctx.fillText(`${t.toFixed(1)}s`, Math.min(xOf(t), width - 28), plotT + plotH + 4);
+ }
+ });
+
+ const canvasXY = (e: React.MouseEvent) => {
+ const r = canvasRef.current!.getBoundingClientRect();
+ return { cx: e.clientX - r.left, cy: e.clientY - r.top };
+ };
+ const onWheel = (e: React.WheelEvent) => {
+ e.preventDefault();
+ const { cx } = canvasXY(e);
+ if (cx < plotL) return;
+ const { lo, hi } = view.current;
+ const tc = tOf(cx);
+ const step = Math.min(0.25, Math.abs(e.deltaY) / 400 + 0.06);
+ const factor = e.deltaY < 0 ? 1 - step : 1 + step;
+ const nlo = Math.max(0, tc - (tc - lo) * factor);
+ const nhi = Math.min(domainHi, tc + (hi - tc) * factor);
+ if (nhi - nlo > 0.02) {
+ view.current = { lo: nlo, hi: nhi };
+ redraw();
+ }
+ };
+ const onDown = (e: React.MouseEvent) => {
+ const { cx } = canvasXY(e);
+ if (cx < plotL) return;
+ if (e.shiftKey) {
+ marquee.current = { x0: cx };
+ setBox({ x: cx, w: 0 });
+ return;
+ }
+ drag.current = { x: cx, lo: view.current.lo, hi: view.current.hi };
+ };
+ const onMove = (e: React.MouseEvent) => {
+ const { cx, cy } = canvasXY(e);
+ if (marquee.current) {
+ setBox({ x: Math.min(marquee.current.x0, cx), w: Math.abs(cx - marquee.current.x0) });
+ return;
+ }
+ if (drag.current) {
+ const { lo, hi } = drag.current;
+ const dt = ((cx - drag.current.x) / plotW) * (hi - lo);
+ let nlo = lo - dt;
+ let nhi = hi - dt;
+ const span = hi - lo;
+ if (nlo < 0) {
+ nlo = 0;
+ nhi = span;
+ }
+ if (nhi > domainHi) {
+ nhi = domainHi;
+ nlo = domainHi - span;
+ }
+ view.current = { lo: nlo, hi: nhi };
+ setHover(null);
+ redraw();
+ return;
+ }
+ if (cx >= plotL && cx <= width - PAD_R) setHover({ x: cx, y: cy, t: tOf(cx) });
+ else setHover(null);
+ };
+ const onUp = () => {
+ if (marquee.current) {
+ const m = box;
+ marquee.current = null;
+ setBox(null);
+ if (m && m.w > 6) {
+ const nlo = tOf(m.x);
+ const nhi = tOf(m.x + m.w);
+ if (nhi - nlo > 0.02) {
+ view.current = { lo: nlo, hi: nhi };
+ redraw();
+ }
+ }
+ return;
+ }
+ drag.current = null;
+ };
+ const reset = () => {
+ view.current = { lo: 0, hi: domainHi };
+ redraw();
+ };
+
+ if (!data || buckets.length === 0)
+ return (
+ [], workers: [] as string[] };
- const t0 = Math.min(...samples.map((s) => s.timestamp));
- const workers = [...new Set(samples.map((s) => s.worker))];
- const byBucket = new Map>();
- for (const s of samples) {
- const t = Math.round((s.timestamp - t0) * 4) / 4;
- const row = byBucket.get(t) ?? { t };
- row[s.worker] = s.rss_bytes;
- byBucket.set(t, row);
- }
- const rows = [...byBucket.values()].sort((a, b) => a.t - b.t);
- return { rows, workers };
+function shortWorker(w: string) {
+ return w.replace(/^tcp:\/\//, "");
}
-export function MemoryChart({ samples }: { samples: Sample[] }) {
- const { rows, workers } = toRows(samples);
- const peak = Math.max(1, ...samples.map((s) => s.rss_bytes));
+interface Series {
+ worker: string;
+ color: string;
+ pts: { t: number; rss: number }[];
+}
+
+// Memory over time on a canvas, with the same navigation as the graph / task
+// stream: scroll to zoom the time axis, drag to pan, shift-drag to box-zoom.
+// Click a point to pin an instant (the parent's spike inspector).
+export function MemoryChart({
+ samples,
+ deaths = [],
+ selectedTime = null,
+ onSelect,
+ height = 400,
+}: {
+ samples: Sample[];
+ deaths?: DeathEvent[];
+ selectedTime?: number | null;
+ onSelect?: (absTime: number) => void;
+ height?: number;
+}) {
+ const wrapRef = useRef(null);
+ const canvasRef = useRef(null);
+ const [width, setWidth] = useState(900);
+ const [hover, setHover] = useState<{ x: number; y: number; t: number } | null>(null);
+
+ const { t0, series, peak, domainHi } = useMemo(() => {
+ if (samples.length === 0) return { t0: 0, series: [] as Series[], peak: 1, domainHi: 1 };
+ const t0 = Math.min(...samples.map((s) => s.timestamp));
+ const byW = new Map();
+ let peak = 1;
+ let domainHi = 1;
+ for (const s of samples) {
+ if (!s.worker) continue;
+ const t = s.timestamp - t0;
+ (byW.get(s.worker) ?? byW.set(s.worker, []).get(s.worker)!).push({ t, rss: s.rss_bytes });
+ if (s.rss_bytes > peak) peak = s.rss_bytes;
+ if (t > domainHi) domainHi = t;
+ }
+ const series: Series[] = [...byW.entries()].map(([worker, pts], i) => ({
+ worker,
+ color: PALETTE[i % PALETTE.length],
+ pts: pts.sort((a, b) => a.t - b.t),
+ }));
+ return { t0, series, peak, domainHi };
+ }, [samples]);
+
+ const view = useRef({ lo: 0, hi: domainHi });
+ const [, force] = useState(0);
+ const redraw = () => force((n) => n + 1);
+ const drag = useRef<{ x: number; lo: number; hi: number; moved: boolean } | null>(null);
+ const marquee = useRef<{ x0: number } | null>(null);
+ const [box, setBox] = useState<{ x: number; w: number } | null>(null);
+
+ useEffect(() => {
+ view.current = { lo: 0, hi: domainHi };
+ redraw();
+ }, [domainHi]);
+
+ useEffect(() => {
+ const el = wrapRef.current;
+ if (!el) return;
+ const ro = new ResizeObserver(() => setWidth(el.clientWidth));
+ ro.observe(el);
+ setWidth(el.clientWidth);
+ return () => ro.disconnect();
+ }, []);
+
+ const plotL = PAD_L;
+ const plotW = Math.max(1, width - PAD_L - PAD_R);
+ const plotT = PAD_T;
+ const plotH = Math.max(1, height - PAD_T - AXIS_H);
+ const yMax = peak * 1.1;
+ const xOf = (t: number) => plotL + ((t - view.current.lo) / (view.current.hi - view.current.lo)) * plotW;
+ const tOf = (x: number) => view.current.lo + ((x - plotL) / plotW) * (view.current.hi - view.current.lo);
+ const yOf = (rss: number) => plotT + (1 - rss / yMax) * plotH;
+
+ useEffect(() => {
+ const canvas = canvasRef.current;
+ if (!canvas) return;
+ const dpr = window.devicePixelRatio || 1;
+ canvas.width = width * dpr;
+ canvas.height = height * dpr;
+ const ctx = canvas.getContext("2d");
+ if (!ctx) return;
+ ctx.setTransform(dpr, 0, 0, dpr, 0, 0);
+ ctx.clearRect(0, 0, width, height);
+ const { lo, hi } = view.current;
+
+ // y grid + labels
+ ctx.strokeStyle = "#eef0f3";
+ ctx.fillStyle = "#9298a2";
+ ctx.font = "10px ui-monospace, monospace";
+ ctx.textBaseline = "middle";
+ for (let i = 0; i <= 4; i++) {
+ const v = (yMax * i) / 4;
+ const y = yOf(v);
+ ctx.beginPath();
+ ctx.moveTo(plotL, y);
+ ctx.lineTo(width - PAD_R, y);
+ ctx.stroke();
+ ctx.fillText(bytes(v), 6, y);
+ }
+
+ // clip to plot for the lines
+ ctx.save();
+ ctx.beginPath();
+ ctx.rect(plotL, plotT, plotW, plotH);
+ ctx.clip();
+ for (const s of series) {
+ ctx.strokeStyle = s.color;
+ ctx.lineWidth = 1.6;
+ ctx.beginPath();
+ let started = false;
+ for (const p of s.pts) {
+ if (p.t < lo - 1 || p.t > hi + 1) {
+ started = false;
+ continue;
+ }
+ const x = xOf(p.t);
+ const y = yOf(p.rss);
+ if (!started) {
+ ctx.moveTo(x, y);
+ started = true;
+ } else ctx.lineTo(x, y);
+ }
+ ctx.stroke();
+ }
+ // death markers
+ for (const d of deaths) {
+ const t = d.timestamp - t0;
+ if (t < lo || t > hi) continue;
+ const x = xOf(t);
+ ctx.strokeStyle = "#d43b3b";
+ ctx.setLineDash([3, 3]);
+ ctx.beginPath();
+ ctx.moveTo(x, plotT);
+ ctx.lineTo(x, plotT + plotH);
+ ctx.stroke();
+ ctx.setLineDash([]);
+ }
+ // selected time
+ if (selectedTime != null) {
+ const t = selectedTime - t0;
+ if (t >= lo && t <= hi) {
+ ctx.strokeStyle = "#e8590c";
+ ctx.lineWidth = 1.5;
+ ctx.beginPath();
+ ctx.moveTo(xOf(t), plotT);
+ ctx.lineTo(xOf(t), plotT + plotH);
+ ctx.stroke();
+ }
+ }
+ // hover crosshair + a dot on each line at the cursor time
+ if (hover) {
+ ctx.strokeStyle = "#c9ccd2";
+ ctx.beginPath();
+ ctx.moveTo(hover.x, plotT);
+ ctx.lineTo(hover.x, plotT + plotH);
+ ctx.stroke();
+ for (const s of series) {
+ let best: { t: number; rss: number } | null = null;
+ for (const p of s.pts)
+ if (!best || Math.abs(p.t - hover.t) < Math.abs(best.t - hover.t)) best = p;
+ if (best && best.t >= lo && best.t <= hi) {
+ const x = xOf(best.t);
+ const y = yOf(best.rss);
+ ctx.fillStyle = s.color;
+ ctx.beginPath();
+ ctx.arc(x, y, 3.5, 0, Math.PI * 2);
+ ctx.fill();
+ ctx.strokeStyle = "#fff";
+ ctx.lineWidth = 1.2;
+ ctx.stroke();
+ }
+ }
+ }
+ ctx.restore();
+
+ // death labels (outside clip, at top)
+ ctx.fillStyle = "#d43b3b";
+ ctx.textBaseline = "alphabetic";
+ for (const d of deaths) {
+ const t = d.timestamp - t0;
+ if (t < lo || t > hi) continue;
+ ctx.fillText("death", Math.min(xOf(t) + 3, width - 34), plotT - 8);
+ }
+
+ // x axis
+ ctx.fillStyle = "#8a8f98";
+ ctx.textBaseline = "top";
+ for (let i = 0; i <= 8; i++) {
+ const t = lo + ((hi - lo) * i) / 8;
+ ctx.fillText(`${t.toFixed(1)}s`, Math.min(xOf(t), width - 28), plotT + plotH + 4);
+ }
+ });
+
+ const canvasXY = (e: React.MouseEvent) => {
+ const r = canvasRef.current!.getBoundingClientRect();
+ return { cx: e.clientX - r.left, cy: e.clientY - r.top };
+ };
+
+ const onWheel = (e: React.WheelEvent) => {
+ e.preventDefault();
+ const { cx } = canvasXY(e);
+ if (cx < plotL) return;
+ const { lo, hi } = view.current;
+ const tc = tOf(cx);
+ const step = Math.min(0.25, Math.abs(e.deltaY) / 400 + 0.06);
+ const factor = e.deltaY < 0 ? 1 - step : 1 + step;
+ let nlo = tc - (tc - lo) * factor;
+ let nhi = tc + (hi - tc) * factor;
+ nlo = Math.max(0, nlo);
+ nhi = Math.min(domainHi, nhi);
+ if (nhi - nlo > 0.02) {
+ view.current = { lo: nlo, hi: nhi };
+ redraw();
+ }
+ };
+
+ const onDown = (e: React.MouseEvent) => {
+ const { cx } = canvasXY(e);
+ if (cx < plotL) return;
+ if (e.shiftKey) {
+ marquee.current = { x0: cx };
+ setBox({ x: cx, w: 0 });
+ return;
+ }
+ drag.current = { x: cx, lo: view.current.lo, hi: view.current.hi, moved: false };
+ };
+ const onMove = (e: React.MouseEvent) => {
+ const { cx, cy } = canvasXY(e);
+ if (marquee.current) {
+ setBox({ x: Math.min(marquee.current.x0, cx), w: Math.abs(cx - marquee.current.x0) });
+ return;
+ }
+ if (drag.current) {
+ const dx = cx - drag.current.x;
+ if (Math.abs(dx) > 2) drag.current.moved = true;
+ const { lo, hi } = drag.current;
+ const dt = (dx / plotW) * (hi - lo);
+ let nlo = lo - dt;
+ let nhi = hi - dt;
+ const span = hi - lo;
+ if (nlo < 0) {
+ nlo = 0;
+ nhi = span;
+ }
+ if (nhi > domainHi) {
+ nhi = domainHi;
+ nlo = domainHi - span;
+ }
+ view.current = { lo: nlo, hi: nhi };
+ setHover(null);
+ redraw();
+ return;
+ }
+ if (cx >= plotL && cx <= width - PAD_R) setHover({ x: cx, y: cy, t: tOf(cx) });
+ else setHover(null);
+ };
+ const onUp = () => {
+ if (marquee.current) {
+ const m = box;
+ marquee.current = null;
+ setBox(null);
+ if (m && m.w > 6) {
+ const nlo = tOf(m.x);
+ const nhi = tOf(m.x + m.w);
+ if (nhi - nlo > 0.02) {
+ view.current = { lo: nlo, hi: nhi };
+ redraw();
+ }
+ }
+ return;
+ }
+ const d = drag.current;
+ drag.current = null;
+ if (d && !d.moved && onSelect) onSelect(t0 + tOf(d.x));
+ };
+ const reset = () => {
+ view.current = { lo: 0, hi: domainHi };
+ redraw();
+ };
+
+ if (samples.length === 0) return
-
- `${t}s`}
- stroke="#e6e8ec"
- />
- bytes(v)}
- domain={[0, Math.ceil(peak * 1.1)]}
- stroke="#e6e8ec"
- width={64}
- />
- `t = ${t}s`}
- formatter={(v: number, name: string) => [bytes(v), name.replace("tcp://", "")]}
- />
- {workers.map((w, i) => (
-
-
);
}
diff --git a/web/components/MemoryDeep.tsx b/web/components/MemoryDeep.tsx
new file mode 100644
index 0000000..6cc78ce
--- /dev/null
+++ b/web/components/MemoryDeep.tsx
@@ -0,0 +1,114 @@
+"use client";
+
+import { useEffect } from "react";
+import { useAllocSites, useTaskMemory } from "@/lib/api";
+import { useLive } from "@/lib/live";
+import { bytes, shortKey } from "@/lib/format";
+import { baseName } from "@/lib/colors";
+import { Flamegraph } from "./Flamegraph";
+
+// The 10x view: memray-derived, per-source-line memory. Answers "which line
+// allocated the array that's filling memory", not just "how much is resident".
+export function MemoryDeep({ runId }: { runId: string }) {
+ const { data: sites, mutate: mSites } = useAllocSites(runId);
+ const { data: tasks, mutate: mTasks } = useTaskMemory(runId);
+ const { deepNonce } = useLive();
+
+ // Pull fresh aggregates the moment a new epoch's deep data streams in.
+ useEffect(() => {
+ if (deepNonce > 0) {
+ mSites();
+ mTasks();
+ }
+ }, [deepNonce, mSites, mTasks]);
+
+ const rows = sites ?? [];
+ const taskRows = tasks ?? [];
+
+ if (rows.length === 0 && taskRows.length === 0)
+ return (
+ (null);
+
+ if (samples.length === 0) return ;
+ const connect = () => {
+ ws = new WebSocket(`${base}/ws/runs`);
+ ws.onmessage = () => mutate();
+ ws.onclose = () => {
+ if (!closed) retry = setTimeout(connect, 1500);
+ };
+ ws.onerror = () => ws?.close();
+ };
+ connect();
+ return () => {
+ closed = true;
+ clearTimeout(retry);
+ ws?.close();
+ };
+ }, [mutate]);
+
return (
<>
Task stream · live
+ Memory & task stream — time flows top to bottom
- Worker memory over time
+
+ Allocations by task layer over time
+
+ Workers · live
+ No timeline yet — samples and task spans will appear here.
;
-
- const { S, T, t0, dur, workers, peak, layers } = model;
- const plotH = Math.min(2400, Math.max(460, dur * 44));
- const H = plotH + PAD_T + PAD_B;
- const streamW = Math.max(120, width - STREAM_X0 - 8);
- const laneW = streamW / Math.max(1, workers.length);
- const y = (t: number) => PAD_T + ((t - t0) / dur) * plotH;
- const memX = (rss: number) => MEM_X0 + (rss / peak) * MEM_W;
- const laneIdx = new Map(workers.map((w, i) => [w, i]));
-
- // time gridlines
- const ticks = 6;
- const gridlines = Array.from({ length: ticks + 1 }, (_, i) => (i / ticks) * dur);
-
- return (
-
-
- );
-}
diff --git a/web/components/AppShell.tsx b/web/components/AppShell.tsx
new file mode 100644
index 0000000..3e4490b
--- /dev/null
+++ b/web/components/AppShell.tsx
@@ -0,0 +1,79 @@
+"use client";
+
+import Link from "next/link";
+import { useEffect, useState } from "react";
+import { Sidebar } from "@/components/Sidebar";
+
+// Array cells stacked into a flamegraph silhouette — arrays + flamegraphs, in
+// Dask's warm palette. Mirrors app/icon.svg.
+const LOGO_COLS = [2, 4, 3, 5, 4];
+const LOGO_RAMP = ["#D6402E", "#E8552D", "#F2762E", "#FDA33E", "#FFC83D"];
+function Logo() {
+ const cells: React.ReactNode[] = [];
+ LOGO_COLS.forEach((h, c) => {
+ for (let r = 0; r < h; r++) {
+ cells.push(
+
- {layers.slice(0, 12).map((l) => (
-
-
- {l}
-
- ))}
-
-
-
-
-
- {T.length} task spans across {workers.length} worker{workers.length !== 1 ? "s" : ""} · time
- runs top → bottom
-
-
+
+
+
+
+
+ );
+}
diff --git a/web/components/Flamegraph.tsx b/web/components/Flamegraph.tsx
new file mode 100644
index 0000000..076cc1d
--- /dev/null
+++ b/web/components/Flamegraph.tsx
@@ -0,0 +1,242 @@
+"use client";
+
+import { useMemo, useState } from "react";
+import { useFlamegraph } from "@/lib/api";
+import { useLive } from "@/lib/live";
+import { useEffect } from "react";
+import { bytes } from "@/lib/format";
+import { layerColorMap } from "@/lib/colors";
+import type { FlameFrame, FlameStack } from "@/lib/types";
+
+const ROW = 20; // px per stack depth
+
+interface FNode {
+ key: string;
+ frame: FlameFrame | null;
+ bytes: number;
+ children: Map
+
+
+ {children}
+
+
+ );
+}
+
+function shortWorker(w: string): string {
+ return w.replace(/^tcp:\/\//, "");
+}
diff --git a/web/components/GraphCanvas.tsx b/web/components/GraphCanvas.tsx
new file mode 100644
index 0000000..b9b5fb4
--- /dev/null
+++ b/web/components/GraphCanvas.tsx
@@ -0,0 +1,350 @@
+"use client";
+
+import dagre from "@dagrejs/dagre";
+import { useEffect, useMemo, useRef, useState } from "react";
+
+export interface CNode {
+ id: string;
+ label: string;
+ layer: string;
+ color: string;
+ hot: boolean;
+}
+
+interface Placed {
+ id: string;
+ x: number;
+ y: number;
+ w: number;
+ h: number;
+ data: CNode;
+}
+
+const NW = 26;
+const NH = 14;
+const MAX_SCALE = 40; // deep zoom for reading individual tasks
+const MIN_SCALE = 0.04;
+const LABEL_SCALE = 1.4; // show node labels once zoomed in past this
+
+// A canvas DAG for large task graphs — the real connected graph, not a
+// layer-level summary. dagre gives a top-to-bottom layered layout; we draw it
+// on a canvas with pan (drag) and zoom (wheel) so thousands of nodes stay
+// responsive where a DOM/SVG renderer would choke. Click hit-tests to select.
+export function GraphCanvas({
+ nodes,
+ edges,
+ onSelect,
+ selected,
+}: {
+ nodes: CNode[];
+ edges: [string, string][];
+ onSelect: (id: string) => void;
+ selected: string | null;
+}) {
+ const wrapRef = useRef
+
+ Allocation flamegraph {worker ? `· ${shortWorker(worker)}` : "· all workers"} · peak{" "}
+ {bytes(focus.bytes)}
+
+
+
+ {!hasStacks && (
+
+ {focusKey !== "root" && (
+
+ )}
+
+
+
+ {data
+ ? worker
+ ? "No call-stack data for this worker."
+ : "No call-stack data yet — needs the memray engine (deep=True) and one epoch."
+ : "Loading…"}
+
+ )}
+
+
+
+ setHover(null)}
+ >
+ {cells.map((c) => {
+ const f = c.node.frame;
+ const label = f ? `${f.function}` : "all";
+ const lib = f ? isLib(f.filename) : false;
+ const widthPct = c.w * 100;
+ return (
+
+
+ {selected?.frame && (
+ {
+ const r = e.currentTarget.parentElement!.getBoundingClientRect();
+ setHover({ x: e.clientX - r.left, y: e.clientY - r.top, node: c.node });
+ }}
+ onClick={() => {
+ setSelected(c.node);
+ if (c.node.children.size > 0) setFocusKey(c.node.key);
+ }}
+ >
+ {widthPct > 4 ? label : ""}
+
+ );
+ })}
+ {hover && (
+
+ {hover.node.frame ? (
+ <>
+
+ )}
+ {hover.node.frame.function}
+
+ {shortFile(hover.node.frame.filename)}:{hover.node.frame.lineno}
+
+
+ ) : (
+ all allocations
+ )}
+
+ {bytes(hover.node.bytes)} · {((hover.node.bytes / total) * 100).toFixed(1)}%
+
+
+
+ )}
+ {selected.frame.function}
+
+ {selected.frame.filename}:{selected.frame.lineno}
+
+
+ peak held
+ {bytes(selected.bytes)}
+
+
+ share of total
+ {((selected.bytes / total) * 100).toFixed(1)}%
+
+
+ in this library?
+ {isLib(selected.frame.filename) ? "yes (framework)" : "your code"}
+
+
+
+ Each bar is a call frame; width is peak bytes held below it. Click to zoom, again for
+ detail — the memray tree read, on your own code.
+
+
+
+ );
+}
diff --git a/web/components/LayerMemoryChart.tsx b/web/components/LayerMemoryChart.tsx
new file mode 100644
index 0000000..2c18065
--- /dev/null
+++ b/web/components/LayerMemoryChart.tsx
@@ -0,0 +1,336 @@
+"use client";
+
+import { useEffect, useMemo, useRef, useState } from "react";
+import { useAllocTimeline } from "@/lib/api";
+import { useLive } from "@/lib/live";
+import { baseName, layerColorMap } from "@/lib/colors";
+import { bytes } from "@/lib/format";
+
+const PAD_L = 66;
+const PAD_R = 12;
+const PAD_T = 10;
+const AXIS_H = 20;
+const HEIGHT = 360;
+const MAX_LAYERS = 12;
+
+interface Bucket {
+ t: number;
+ vals: Record
+ No per-layer allocation data yet — needs the memray engine (
+ );
+
+ const zoomed = view.current.lo > 1e-6 || view.current.hi < domainHi - 1e-6;
+ // nearest bucket for the hover tooltip
+ let nb: Bucket | null = null;
+ if (hover) for (const b of buckets) if (!nb || Math.abs(b.t - hover.t) < Math.abs(nb.t - hover.t)) nb = b;
+ const hoverRows = nb
+ ? layers
+ .map((l) => ({ l, v: nb!.vals[l] ?? 0 }))
+ .filter((r) => r.v > 0)
+ .sort((a, b) => b.v - a.v)
+ : [];
+
+ return (
+ deep=True).
+
+
+ );
+}
diff --git a/web/components/MemoryChart.tsx b/web/components/MemoryChart.tsx
index 7b58571..b56fc5d 100644
--- a/web/components/MemoryChart.tsx
+++ b/web/components/MemoryChart.tsx
@@ -1,90 +1,369 @@
"use client";
-import {
- CartesianGrid,
- Line,
- LineChart,
- ResponsiveContainer,
- Tooltip,
- XAxis,
- YAxis,
-} from "recharts";
+import { useEffect, useMemo, useRef, useState } from "react";
import { bytes } from "@/lib/format";
-import type { Sample } from "@/lib/types";
+import type { DeathEvent, Sample } from "@/lib/types";
const PALETTE = ["#4f46e5", "#0891b2", "#16a34a", "#d97706", "#db2777", "#7c3aed"];
+const PAD_L = 66;
+const PAD_R = 12;
+const PAD_T = 22; // room for the "death" labels
+const AXIS_H = 20;
-function toRows(samples: Sample[]) {
- if (samples.length === 0) return { rows: [] as Record
+ {layers.map((l) => (
+
+
+ {baseName(l)}
+
+ ))}
+ · scroll to zoom · drag to pan · shift-drag box-zoom
+
+
+
+ No memory samples yet.
;
+
+ const zoomed = view.current.lo > 1e-6 || view.current.hi < domainHi - 1e-6;
+ const hoverVals =
+ hover &&
+ series.map((s) => {
+ // nearest point to hovered time
+ let best: { t: number; rss: number } | null = null;
+ for (const p of s.pts) if (!best || Math.abs(p.t - hover.t) < Math.abs(best.t - hover.t)) best = p;
+ return { worker: s.worker, color: s.color, rss: best?.rss ?? 0 };
+ });
return (
- {workers.map((w, i) => (
-
-
- {w.replace("tcp://", "")}
+ {series.map((s) => (
+
+
+ {shortWorker(s.worker)}
))}
+ {onSelect && (
+ · scroll to zoom · drag to pan · shift-drag box-zoom · click to inspect
+ )}
+
+
+
-
+ No deep memory data. Enable the memray engine for this run:{" "}
+
+ );
+
+ const topLine = rows[0]?.hwm_bytes ?? 1;
+ const topTask = taskRows[0]?.peak_rss_delta ?? 1;
+
+ return (
+ <>
+ register(client, url, deep=True) or{" "}
+ LocalProfiler(..., deep=True).
+
+
+
+ );
+}
diff --git a/web/components/MemoryExplorer.tsx b/web/components/MemoryExplorer.tsx
new file mode 100644
index 0000000..5391510
--- /dev/null
+++ b/web/components/MemoryExplorer.tsx
@@ -0,0 +1,39 @@
+"use client";
+
+import { useState } from "react";
+import { useLive } from "@/lib/live";
+import { MemoryChart } from "./MemoryChart";
+import { SpikeInspector } from "./SpikeInspector";
+
+// The full-page memory explorer: a large, zoomable memory-over-time chart with a
+// roomy spike inspector alongside. Click any point to pin an instant and see
+// what was running and allocating then.
+export function MemoryExplorer({ runId }: { runId: string }) {
+ const { samples, spans, deaths } = useLive();
+ const [pinned, setPinned] = useState
+
+
+
+
+
+ Peak memory by source line
+ | Source line | +Peak | +Allocs | +
|---|---|---|
|
+
+ {s.filename.split("/").slice(-1)[0]}:{s.lineno}
+
+ {s.function}
+ |
+
+
+
+ {bytes(s.hwm_bytes)}
+
+ |
+ {s.n_allocations.toLocaleString()} | +
+
+
+
+
+ Peak memory by task
+ | Task | +Peak Δ RSS | +Top line | +
|---|---|---|
|
+ {shortKey(t.key)}
+ {baseName(t.layer)}
+ |
+
+
+
+ {bytes(t.peak_rss_delta)}
+
+ |
+ + {top ? `${top.filename.split("/").slice(-1)[0]}:${top.lineno}` : "—"} + | +
No memory samples yet.
;
+
+ return (
+
+
+ );
+}
diff --git a/web/components/PostMortem.tsx b/web/components/PostMortem.tsx
index 64264ea..57745b6 100644
--- a/web/components/PostMortem.tsx
+++ b/web/components/PostMortem.tsx
@@ -1,8 +1,9 @@
"use client";
-import { useDeaths, useGraph } from "@/lib/api";
+import { useGraph } from "@/lib/api";
+import { useLive } from "@/lib/live";
import { bytes, layerToken, shortKey } from "@/lib/format";
-import type { ChunkMeta, GraphLayer } from "@/lib/types";
+import type { AllocationSite, ChunkMeta, GraphLayer } from "@/lib/types";
import { CodeLine } from "./CodeLine";
function sourceFor(key: string, layers: GraphLayer[]): GraphLayer | undefined {
@@ -23,14 +24,31 @@ function Chunks({ chunks }: { chunks: ChunkMeta[] }) {
);
}
+// The deep-memory cause: the exact source lines at the high-water mark when the
+// worker died — the "line 42 allocated 12.8 GB" answer above the chunk view.
+function Sites({ sites }: { sites: AllocationSite[] }) {
+ if (sites.length === 0) return null;
+ return (
+
+
+
+ setPinned(null)}
+ />
+
+
+
+ );
+}
+
export function PostMortem({ runId }: { runId: string }) {
- const { data: deaths, isLoading } = useDeaths(runId);
+ const { deaths } = useLive();
const { data: graph } = useGraph(runId);
const layers = graph?.layers ?? [];
- if (isLoading) return At the high-water mark:
+ {sites.slice(0, 5).map((s, i) => (
+
+
+ {s.filename.split("/").slice(-1)[0]}:{s.lineno} {s.function}
+
+ {bytes(s.hwm_bytes)}
+
+ ))}
+ Loading…
;
-
- const relevant = (deaths ?? []).filter((d) => d.suspect_keys.length > 0);
+ const relevant = deaths.filter((d) => d.suspect_keys.length > 0);
if (relevant.length === 0)
return (
@@ -55,6 +73,7 @@ export function PostMortem({ runId }: { runId: string }) {
{d.worker}
{d.reason}
+
@@ -37,6 +63,11 @@ export function Sidebar() {
(
+ time != null
+ ? `/api/runs/${runId}/alloc-sites?start=${time - WINDOW}&end=${time + WINDOW}`
+ : null,
+ fetcher,
+ );
+
+ const t0 = useMemo(
+ () => (samples.length ? Math.min(...samples.map((s) => s.timestamp)) : 0),
+ [samples],
+ );
+
+ const info = useMemo(() => {
+ if (time == null) return null;
+ // RSS per worker at the instant (nearest sample within 2s)
+ const perWorker = new Map();
+ const nearest = new Map();
+ for (const s of samples) {
+ if (!s.worker) continue;
+ const dt = Math.abs(s.timestamp - time);
+ const cur = nearest.get(s.worker);
+ if (!cur || dt < cur.dt) nearest.set(s.worker, { dt, rss: s.rss_bytes });
+ }
+ for (const [w, v] of nearest) if (v.dt < 2) perWorker.set(w, v.rss);
+ const totalRss = [...perWorker.values()].reduce((a, b) => a + b, 0);
+
+ // tasks whose span covers this instant → "where we were in the graph"
+ const running = spans
+ .filter((s) => s.start <= time && s.end >= time)
+ .sort((a, b) => b.end - b.start - (a.end - a.start));
+
+ return { perWorker, totalRss, running };
+ }, [time, samples, spans]);
+
+ if (time == null)
+ return (
+ , colorOf: (l: string) => string) {
- const taskLevel = graph.nodes.length > 0 && graph.nodes.length <= TASK_LIMIT && !graph.truncated;
+function build(
+ graph: GraphData,
+ hot: Set,
+ colorOf: (l: string) => string,
+ allowTaskLevel = true,
+) {
+ const taskLevel =
+ allowTaskLevel &&
+ graph.nodes.length > 0 &&
+ graph.nodes.length <= TASK_LIMIT &&
+ !graph.truncated;
if (taskLevel) {
const raw = graph.nodes.map((n) => ({
id: n.key,
@@ -160,23 +172,50 @@ function SourcePanel({
}
export function TaskGraph({ runId }: { runId: string }) {
- const { data: graph, isLoading } = useGraph(runId);
- const { data: deaths } = useDeaths(runId);
+ const { data: graph, isLoading, mutate } = useGraph(runId);
+ const { deaths, graphNonce } = useLive();
const [selected, setSelected] = useState(null);
+ // For big graphs the readable default is the layer-level DAG; drilling into
+ // the full task canvas is opt-in (thousands of nodes is a hairball otherwise).
+ const [showTasks, setShowTasks] = useState(false);
+
+ // The graph is uploaded once the collection is known; refetch when the live
+ // stream signals a (re)upload landed.
+ useEffect(() => {
+ if (graphNonce > 0) mutate();
+ }, [graphNonce, mutate]);
const hot = useMemo(
- () => new Set((deaths ?? []).flatMap((d) => d.suspect_keys.map(layerToken))),
+ () => new Set(deaths.flatMap((d) => d.suspect_keys.map(layerToken))),
[deaths],
);
const colorOf = useMemo(() => layerColorMap(), []);
+ const hasTasks = !!graph && graph.nodes.length > 0 && !graph.truncated;
+ const large = !!graph && graph.nodes.length > TASK_LIMIT && !graph.truncated;
+ // Show the full task canvas only for small graphs, or when the user opts in.
+ const useCanvas = hasTasks && (large ? showTasks : false);
+ const useTaskReactFlow = hasTasks && !large; // small graph → interactive task view
+
+ const canvasNodes = useMemo(() => {
+ if (!graph || !useCanvas) return [];
+ return graph.nodes.map((n) => ({
+ id: n.key,
+ label: shortKey(n.key),
+ layer: n.layer,
+ color: colorOf(n.layer),
+ hot: [...hot].some((t) => n.key.includes(t) || layerToken(n.key).startsWith(t)),
+ }));
+ }, [graph, useCanvas, hot, colorOf]);
+
+ // react-flow path: the interactive task view (small graphs) or the layer DAG.
const { nodes, edges, taskLevel } = useMemo(
() =>
- graph
- ? build(graph, hot, colorOf)
+ graph && !useCanvas
+ ? build(graph, hot, colorOf, useTaskReactFlow)
: { nodes: [] as Node[], edges: [] as Edge[], taskLevel: false },
- [graph, hot, colorOf],
+ [graph, useCanvas, useTaskReactFlow, hot, colorOf],
);
if (isLoading) return
{ago(r.created_at)} · {r.counts.workers ?? 0}w · {r.counts.samples ?? 0}s
+ {r.origin || r.origin_ip ? (
+
+ {r.origin || r.origin_ip}
+
+ ) : null}
))
)}
diff --git a/web/components/SpikeInspector.tsx b/web/components/SpikeInspector.tsx
new file mode 100644
index 0000000..91b007e
--- /dev/null
+++ b/web/components/SpikeInspector.tsx
@@ -0,0 +1,147 @@
+"use client";
+
+import { useMemo } from "react";
+import useSWR from "swr";
+import { useGraph } from "@/lib/api";
+import { baseName } from "@/lib/colors";
+import { bytes, shortKey } from "@/lib/format";
+import type { AllocSiteRow, GraphLayer, Sample, TaskSpan } from "@/lib/types";
+import { CodeLine } from "./CodeLine";
+
+const WINDOW = 3; // seconds around the click to attribute allocations to
+
+const fetcher = (u: string) => fetch(u).then((r) => r.json());
+
+function sourceForLayer(layer: string, layers: GraphLayer[]): GraphLayer | undefined {
+ return layers.find((l) => layer === l.layer || layer.startsWith(l.layer) || l.layer.startsWith(layer));
+}
+
+// Given a pinned instant, answer "what was happening here": which tasks were
+// running (where we were in the graph), what they cost in memory, and which
+// source lines were allocating in that window — the "what caused this spike".
+export function SpikeInspector({
+ runId,
+ time,
+ samples,
+ spans,
+ onClose,
+}: {
+ runId: string;
+ time: number | null;
+ samples: Sample[];
+ spans: TaskSpan[];
+ onClose: () => void;
+}) {
+ const { data: graph } = useGraph(runId);
+ const { data: sites } = useSWR
+ Click any point on the memory chart to see what was running and allocating at that instant.
+
+ );
+
+ const layers = graph?.layers ?? [];
+ const rel = (time - t0).toFixed(1);
+
+ return (
+
+
+ );
+}
diff --git a/web/components/TaskGraph.tsx b/web/components/TaskGraph.tsx
index 7dc8c02..8db11a8 100644
--- a/web/components/TaskGraph.tsx
+++ b/web/components/TaskGraph.tsx
@@ -12,17 +12,20 @@ import {
Position,
ReactFlow,
} from "@xyflow/react";
-import { useMemo, useState } from "react";
-import { useChunks, useDeaths, useGraph } from "@/lib/api";
+import { useEffect, useMemo, useState } from "react";
+import { useChunks, useGraph } from "@/lib/api";
+import { useLive } from "@/lib/live";
import { baseName, layerColorMap } from "@/lib/colors";
import { bytes, layerToken, shortKey } from "@/lib/format";
import type { GraphData, GraphLayer } from "@/lib/types";
import { CodeLine } from "./CodeLine";
+import { GraphCanvas, type CNode } from "./GraphCanvas";
const NODE_W = 168;
const NODE_H = 38;
-// Above this many task nodes we render the layer-level graph instead — a few
-// hundred nodes is the sweet spot for a readable, responsive DAG.
+// Above this many task nodes we switch from the interactive react-flow view to
+// the canvas DAG renderer, which draws the *whole* graph (thousands of nodes)
+// rather than collapsing to a layer summary.
const TASK_LIMIT = 400;
type NodeData = { label: string; layer: string; color: string; hot: boolean };
@@ -69,8 +72,17 @@ function laidOut(
return { nodes, edges };
}
-function build(graph: GraphData, hot: Set
+
+ Memory at +{rel}s · {info ? bytes(info.totalRss) : "—"} total RSS
+
+
+
+
+
+
+
+ {sites.slice(0, 8).map((s, i) => (
+
+ ) : (
+
+ Running here ({info?.running.length ?? 0})
+ {info && info.running.length > 0 ? (
+
+ {info.running.slice(0, 8).map((s, i) => {
+ const src = sourceForLayer(s.layer, layers);
+ return (
+
+ ) : (
+
+
+ );
+ })}
+
+ {shortKey(s.key)}
+ {s.worker.replace(/^tcp:\/\//, "")}
+
+ {baseName(s.layer)}
+ {src ? (
+ <>
+
+ {src.filename}:{src.lineno}
+
+ No task span covers this instant.
+ )}
+
+
+ Allocating in this window (±{WINDOW}s)
+
+ {sites && sites.length > 0 ? (
+ | + {s.filename.split("/").slice(-1)[0]}:{s.lineno} + {s.function} + | ++ {bytes(s.hwm_bytes)} + | +
+ No deep allocation data here — enable
+ )}
+ deep=True for line-level cause.
+ Loading…
;
@@ -191,40 +230,68 @@ export function TaskGraph({ runId }: { runId: string }) {
const layerOf = new Map(graph.nodes.map((n) => [n.key, n.layer]));
const sourceOf = new Map(graph.layers.map((l) => [l.layer, l]));
- const selLayer = selected ? (taskLevel ? layerOf.get(selected) ?? "" : selected) : "";
+ const selTaskLevel = useCanvas || taskLevel;
+ const selLayer = selected ? (selTaskLevel ? layerOf.get(selected) ?? "" : selected) : "";
+
+ const note = useCanvas
+ ? `${graph.task_count} tasks · full task graph`
+ : taskLevel
+ ? `${graph.task_count} tasks · task-level graph`
+ : `${graph.task_count} tasks · grouped by layer (${
+ graph.layers.length || new Set(graph.nodes.map((n) => n.layer)).size
+ } layers)`;
return (
<>
- {graph.task_count} tasks ·{" "}
- {taskLevel ? "task-level graph" : "layer-level graph (too large for task view)"} · click a
- node for its source
+ {note} · click a node for its source
+ {large && (
+
+ )}
- setSelected(null)}
/>
)}
diff --git a/web/components/TaskStream.tsx b/web/components/TaskStream.tsx
new file mode 100644
index 0000000..ec3a370
--- /dev/null
+++ b/web/components/TaskStream.tsx
@@ -0,0 +1,327 @@
+"use client";
+
+import { useEffect, useMemo, useRef, useState } from "react";
+import { useLive } from "@/lib/live";
+import { baseName, layerColorMap } from "@/lib/colors";
+import { shortKey } from "@/lib/format";
+import type { TaskSpan } from "@/lib/types";
+
+const LANE_H = 18;
+const LANE_GAP = 3;
+const SECTION_GAP = 16;
+const GUTTER = 96; // left label column (fixed, not zoomed)
+const PAD_R = 10;
+const PAD_T = 8;
+const AXIS_H = 18;
+
+type Hover = { x: number; y: number; span: TaskSpan } | null;
+
+// Greedy interval packing: assign each span to the first lane whose last task
+// has finished — the compact "global" activity view across all workers.
+function packLanes(spans: TaskSpan[]): { span: TaskSpan; lane: number }[] {
+ const sorted = [...spans].sort((a, b) => a.start - b.start);
+ const laneEnds: number[] = [];
+ const out: { span: TaskSpan; lane: number }[] = [];
+ for (const s of sorted) {
+ let lane = laneEnds.findIndex((end) => end <= s.start);
+ if (lane === -1) {
+ lane = laneEnds.length;
+ laneEnds.push(s.end);
+ } else {
+ laneEnds[lane] = s.end;
+ }
+ out.push({ span: s, lane });
+ }
+ return out;
+}
+
+export function TaskStream() {
+ const { spans } = useLive();
+ const wrapRef = useRef(null);
+ const canvasRef = useRef(null);
+ const [width, setWidth] = useState(900);
+ const [hover, setHover] = useState(null);
+ const colorOf = useMemo(() => layerColorMap(), []);
+
+ const domain = useMemo(() => {
+ if (spans.length === 0) return { lo: 0, hi: 1 };
+ let lo = Infinity;
+ let hi = -Infinity;
+ for (const s of spans) {
+ if (s.start < lo) lo = s.start;
+ if (s.end > hi) hi = s.end;
+ }
+ return { lo, hi: hi <= lo ? lo + 1 : hi };
+ }, [spans]);
+
+ // the visible time window (zoom/pan state)
+ const view = useRef({ lo: domain.lo, hi: domain.hi });
+ const [, force] = useState(0);
+ const redraw = () => force((n) => n + 1);
+ const drag = useRef<{ x: number; lo: number; hi: number } | null>(null);
+ const marquee = useRef<{ x0: number } | null>(null);
+ const [box, setBox] = useState<{ x: number; w: number } | null>(null);
+
+ // reset the window when the run/domain changes
+ useEffect(() => {
+ view.current = { lo: domain.lo, hi: domain.hi };
+ redraw();
+ }, [domain.lo, domain.hi]);
+
+ const workers = useMemo(() => Array.from(new Set(spans.map((s) => s.worker))).sort(), [spans]);
+ const global = useMemo(() => packLanes(spans), [spans]);
+ const globalLanes = useMemo(() => Math.max(1, ...global.map((g) => g.lane + 1)), [global]);
+ const laneOf = useMemo(() => {
+ const m = new Map();
+ workers.forEach((w, i) => m.set(w, i));
+ return m;
+ }, [workers]);
+
+ const globalH = globalLanes * (LANE_H + LANE_GAP);
+ const workerH = workers.length * (LANE_H + LANE_GAP);
+ const globalTop = PAD_T + 14; // room for a section label
+ const workerTop = globalTop + globalH + SECTION_GAP + 14;
+ const height = workerTop + workerH + AXIS_H;
+
+ useEffect(() => {
+ const el = wrapRef.current;
+ if (!el) return;
+ const ro = new ResizeObserver(() => setWidth(el.clientWidth));
+ ro.observe(el);
+ setWidth(el.clientWidth);
+ return () => ro.disconnect();
+ }, []);
+
+ const plotL = GUTTER;
+ const plotW = Math.max(1, width - GUTTER - PAD_R);
+ const xOf = (t: number) => {
+ const { lo, hi } = view.current;
+ return plotL + ((t - lo) / (hi - lo)) * plotW;
+ };
+ const tOf = (x: number) => {
+ const { lo, hi } = view.current;
+ return lo + ((x - plotL) / plotW) * (hi - lo);
+ };
+
+ useEffect(() => {
+ const canvas = canvasRef.current;
+ if (!canvas) return;
+ const dpr = window.devicePixelRatio || 1;
+ canvas.width = width * dpr;
+ canvas.height = height * dpr;
+ const ctx = canvas.getContext("2d");
+ if (!ctx) return;
+ ctx.setTransform(dpr, 0, 0, dpr, 0, 0);
+ ctx.clearRect(0, 0, width, height);
+ const { lo, hi } = view.current;
+
+ // section labels
+ ctx.fillStyle = "#62666d";
+ ctx.font = "600 11px ui-monospace, monospace";
+ ctx.textBaseline = "alphabetic";
+ ctx.fillText("ALL TASKS", 6, globalTop - 3);
+ ctx.fillText("PER WORKER", 6, workerTop - 3);
+
+ const drawRect = (s: TaskSpan, y: number) => {
+ if (s.end < lo || s.start > hi) return;
+ const x0 = Math.max(plotL, xOf(s.start));
+ const x1 = Math.min(width - PAD_R, xOf(s.end));
+ if (x1 <= x0) return;
+ ctx.fillStyle = colorOf(s.layer);
+ ctx.fillRect(x0, y, Math.max(1, x1 - x0), LANE_H);
+ };
+
+ // global lane backgrounds + rects
+ ctx.fillStyle = "#f4f5f7";
+ for (let i = 0; i < globalLanes; i++)
+ ctx.fillRect(plotL, globalTop + i * (LANE_H + LANE_GAP), plotW, LANE_H);
+ for (const g of global) drawRect(g.span, globalTop + g.lane * (LANE_H + LANE_GAP));
+
+ // per-worker lanes
+ ctx.fillStyle = "#f7f8fa";
+ workers.forEach((_, i) =>
+ ctx.fillRect(plotL, workerTop + i * (LANE_H + LANE_GAP), plotW, LANE_H),
+ );
+ ctx.fillStyle = "#62666d";
+ ctx.font = "10px ui-monospace, monospace";
+ workers.forEach((w, i) => {
+ const y = workerTop + i * (LANE_H + LANE_GAP);
+ ctx.fillStyle = "#8a8f98";
+ ctx.fillText(w.replace(/^tcp:\/\//, "").slice(0, 15), 6, y + LANE_H - 5);
+ });
+ for (const s of spans) {
+ const lane = laneOf.get(s.worker) ?? 0;
+ drawRect(s, workerTop + lane * (LANE_H + LANE_GAP));
+ }
+
+ // axis
+ ctx.fillStyle = "#8a8f98";
+ ctx.font = "10px ui-monospace, monospace";
+ ctx.textBaseline = "top";
+ const ticks = 8;
+ for (let i = 0; i <= ticks; i++) {
+ const t = lo + ((hi - lo) * i) / ticks;
+ const x = xOf(t);
+ if (x < plotL - 1) continue;
+ ctx.fillText(`${(t - domain.lo).toFixed(1)}s`, Math.min(x, width - 28), height - AXIS_H + 2);
+ }
+ });
+
+ const canvasXY = (e: React.MouseEvent) => {
+ const r = canvasRef.current!.getBoundingClientRect();
+ return { cx: e.clientX - r.left, cy: e.clientY - r.top };
+ };
+
+ const onWheel = (e: React.WheelEvent) => {
+ e.preventDefault();
+ const { cx } = canvasXY(e);
+ if (cx < plotL) return;
+ const { lo, hi } = view.current;
+ const tc = tOf(cx);
+ const step = Math.min(0.25, Math.abs(e.deltaY) / 400 + 0.06);
+ const factor = e.deltaY < 0 ? 1 - step : 1 + step; // in = shrink window
+ let nlo = tc - (tc - lo) * factor;
+ let nhi = tc + (hi - tc) * factor;
+ // clamp to domain and a min window
+ nlo = Math.max(domain.lo, nlo);
+ nhi = Math.min(domain.hi, nhi);
+ if (nhi - nlo > 0.02) {
+ view.current = { lo: nlo, hi: nhi };
+ redraw();
+ }
+ };
+
+ const onDown = (e: React.MouseEvent) => {
+ const { cx } = canvasXY(e);
+ if (cx < plotL) return;
+ if (e.shiftKey) {
+ marquee.current = { x0: cx };
+ setBox({ x: cx, w: 0 });
+ return;
+ }
+ drag.current = { x: cx, lo: view.current.lo, hi: view.current.hi };
+ };
+ const onMove = (e: React.MouseEvent) => {
+ const { cx, cy } = canvasXY(e);
+ if (marquee.current) {
+ setBox({ x: Math.min(marquee.current.x0, cx), w: Math.abs(cx - marquee.current.x0) });
+ return;
+ }
+ if (drag.current) {
+ const { lo, hi } = drag.current;
+ const dt = ((cx - drag.current.x) / plotW) * (hi - lo);
+ let nlo = lo - dt;
+ let nhi = hi - dt;
+ const span = hi - lo;
+ if (nlo < domain.lo) {
+ nlo = domain.lo;
+ nhi = domain.lo + span;
+ }
+ if (nhi > domain.hi) {
+ nhi = domain.hi;
+ nlo = domain.hi - span;
+ }
+ view.current = { lo: nlo, hi: nhi };
+ setHover(null);
+ redraw();
+ return;
+ }
+ // hover hit-test
+ const t = tOf(cx);
+ const yrow = (top: number, n: number) => {
+ const i = Math.floor((cy - top) / (LANE_H + LANE_GAP));
+ return i >= 0 && i < n ? i : -1;
+ };
+ let found: TaskSpan | null = null;
+ const gi = yrow(globalTop, globalLanes);
+ if (gi >= 0) {
+ for (const g of global)
+ if (g.lane === gi && g.span.start <= t && g.span.end >= t) found = g.span;
+ } else {
+ const wi = yrow(workerTop, workers.length);
+ if (wi >= 0) {
+ const w = workers[wi];
+ for (const s of spans)
+ if (s.worker === w && s.start <= t && s.end >= t) found = s;
+ }
+ }
+ setHover(found ? { x: cx, y: cy, span: found } : null);
+ };
+ const onUp = () => {
+ if (marquee.current) {
+ const m = box;
+ marquee.current = null;
+ setBox(null);
+ if (m && m.w > 6) {
+ const nlo = tOf(m.x);
+ const nhi = tOf(m.x + m.w);
+ if (nhi - nlo > 0.02) {
+ view.current = { lo: nlo, hi: nhi };
+ redraw();
+ }
+ }
+ return;
+ }
+ drag.current = null;
+ };
+ const reset = () => {
+ view.current = { lo: domain.lo, hi: domain.hi };
+ redraw();
+ };
+
+ if (spans.length === 0) return (
+ () => Object.values(statuses).sort((a, b) => a.worker.localeCompare(b.worker)),
+ [statuses],
+ );
+
+ if (rows.length === 0) return
+
+
+
+ );
+}
diff --git a/web/lib/api.ts b/web/lib/api.ts
index 9d08c1c..6a485fb 100644
--- a/web/lib/api.ts
+++ b/web/lib/api.ts
@@ -2,13 +2,18 @@
import useSWR from "swr";
import type {
+ AllocSiteRow,
+ AllocTimelineRow,
ChunkMeta,
DeathEvent,
+ FlameData,
GraphData,
LayerStat,
RunInfo,
Sample,
+ TaskMemoryRow,
TaskSpan,
+ WorkerStatus,
} from "./types";
const fetcher = async (url: string) => {
@@ -48,6 +53,31 @@ export function useLayerStats(id: string) {
return useSWR(`/api/runs/${id}/layer-stats`, fetcher, LIVE);
}
+export function useWorkers(id: string) {
+ return useSWR(`/api/runs/${id}/workers`, fetcher, LIVE);
+}
+
+export function useAllocSites(id: string) {
+ return useSWR(`/api/runs/${id}/alloc-sites`, fetcher, LIVE);
+}
+
+export function useTaskMemory(id: string) {
+ return useSWR(`/api/runs/${id}/task-memory`, fetcher, LIVE);
+}
+
+export function useAllocTimeline(id: string) {
+ return useSWR(`/api/runs/${id}/alloc-timeline`, fetcher, LIVE);
+}
+
+export function useFlamegraph(id: string, worker: string | null) {
+ const q = worker ? `?worker=${encodeURIComponent(worker)}` : "";
+ // keepPreviousData so switching worker doesn't blank the panel mid-fetch.
+ return useSWR(`/api/runs/${id}/flamegraph${q}`, fetcher, {
+ ...LIVE,
+ keepPreviousData: true,
+ });
+}
+
export function useChunks(id: string, key: string | null) {
return useSWR(
key ? `/api/runs/${id}/chunks/${encodeURIComponent(key)}` : null,
diff --git a/web/lib/live.tsx b/web/lib/live.tsx
new file mode 100644
index 0000000..3023463
--- /dev/null
+++ b/web/lib/live.tsx
@@ -0,0 +1,192 @@
+"use client";
+
+// Real-time run state. One WebSocket per open run streams the collector's
+// ingest as it happens; we seed from REST once, then apply each frame as a
+// delta into in-memory rings. Components read the live state through useLive()
+// instead of polling — the memory chart, Workers table and task stream all
+// update the instant a worker flushes.
+
+import {
+ createContext,
+ useContext,
+ useEffect,
+ useMemo,
+ useRef,
+ useState,
+} from "react";
+import type { DeathEvent, Sample, TaskSpan, WorkerStatus } from "./types";
+
+const SAMPLE_CAP = 6000; // ~ tens of minutes at 0.2s across a few workers
+const SPAN_CAP = 20000;
+
+type Frame =
+ | {
+ type: "batch";
+ worker: string;
+ samples: Sample[];
+ spans: TaskSpan[];
+ statuses: WorkerStatus[];
+ epochs: unknown[];
+ task_memory: unknown[];
+ }
+ | { type: "death"; data: DeathEvent }
+ | { type: "graph" };
+
+export interface LiveState {
+ connected: boolean;
+ // Wall-clock (ms) of the last data frame received. A run is only "live" if
+ // data is actually still arriving — a finished run has an open socket but no
+ // recent frames, so the UI shows it idle rather than falsely "live".
+ lastFrameAt: number;
+ samples: Sample[];
+ statuses: Record;
+ spans: TaskSpan[];
+ deaths: DeathEvent[];
+ // Bumps when the deep-memory stream or graph advances, so views backed by
+ // aggregate REST endpoints (alloc-sites, task-memory, graph) can revalidate.
+ deepNonce: number;
+ graphNonce: number;
+}
+
+const empty: LiveState = {
+ connected: false,
+ lastFrameAt: 0,
+ samples: [],
+ statuses: {},
+ spans: [],
+ deaths: [],
+ deepNonce: 0,
+ graphNonce: 0,
+};
+
+const LiveContext = createContext(empty);
+
+export function useLive(): LiveState {
+ return useContext(LiveContext);
+}
+
+export function collectorWsBase(): string {
+ const env = process.env.NEXT_PUBLIC_COLLECTOR_WS;
+ if (env) return env.replace(/\/$/, "");
+ // Default: the collector's published port on the same host the dashboard is
+ // served from (docker-compose maps 8765:8765; dev runs it on localhost).
+ if (typeof window !== "undefined") {
+ const proto = window.location.protocol === "https:" ? "wss" : "ws";
+ return `${proto}://${window.location.hostname}:8765`;
+ }
+ return "";
+}
+
+function collectorWsUrl(runId: string): string {
+ const base = collectorWsBase();
+ return base ? `${base}/ws/runs/${runId}` : "";
+}
+
+export function LiveProvider({
+ runId,
+ children,
+}: {
+ runId: string;
+ children: React.ReactNode;
+}) {
+ const [state, setState] = useState(empty);
+ const ref = useRef(empty);
+ const set = (next: LiveState) => {
+ ref.current = next;
+ setState(next);
+ };
+
+ // Seed from REST once so the views aren't empty before the first frame.
+ useEffect(() => {
+ let cancelled = false;
+ async function seed() {
+ try {
+ const [timeline, workers, spans, deaths] = await Promise.all([
+ fetch(`/api/runs/${runId}/timeline`).then((r) => r.json()),
+ fetch(`/api/runs/${runId}/workers`).then((r) => r.json()),
+ fetch(`/api/runs/${runId}/spans`).then((r) => r.json()),
+ fetch(`/api/runs/${runId}/deaths`).then((r) => r.json()),
+ ]);
+ if (cancelled) return;
+ const statuses: Record = {};
+ for (const w of workers as WorkerStatus[]) statuses[w.worker] = w;
+ set({
+ ...ref.current,
+ samples: (timeline as Sample[]).slice().sort((a, b) => a.timestamp - b.timestamp),
+ statuses,
+ spans: spans as TaskSpan[],
+ deaths: deaths as DeathEvent[],
+ });
+ } catch {
+ /* collector not up yet; the WS will fill in */
+ }
+ }
+ seed();
+ return () => {
+ cancelled = true;
+ };
+ }, [runId]);
+
+ // Live WebSocket with auto-reconnect.
+ useEffect(() => {
+ const url = collectorWsUrl(runId);
+ if (!url) return;
+ let ws: WebSocket | null = null;
+ let closed = false;
+ let retry: ReturnType;
+
+ const connect = () => {
+ ws = new WebSocket(url);
+ ws.onopen = () => set({ ...ref.current, connected: true });
+ ws.onclose = () => {
+ set({ ...ref.current, connected: false });
+ if (!closed) retry = setTimeout(connect, 1500);
+ };
+ ws.onerror = () => ws?.close();
+ ws.onmessage = (ev) => {
+ let frame: Frame;
+ try {
+ frame = JSON.parse(ev.data);
+ } catch {
+ return;
+ }
+ apply(frame);
+ };
+ };
+
+ const apply = (frame: Frame) => {
+ const cur = { ...ref.current, lastFrameAt: Date.now() };
+ if (frame.type === "batch") {
+ const statuses = { ...cur.statuses };
+ for (const s of frame.statuses) statuses[s.worker] = s;
+ // A batch's MemorySamples don't carry the worker (it's on the envelope);
+ // tag them so the memory chart can key lines by worker.
+ const tagged = frame.samples.map((s) => ({ ...s, worker: frame.worker }));
+ const samples = cur.samples.concat(tagged);
+ const spans = cur.spans.concat(frame.spans);
+ const deep = frame.epochs.length || frame.task_memory.length;
+ set({
+ ...cur,
+ statuses,
+ samples: samples.length > SAMPLE_CAP ? samples.slice(-SAMPLE_CAP) : samples,
+ spans: spans.length > SPAN_CAP ? spans.slice(-SPAN_CAP) : spans,
+ deepNonce: deep ? cur.deepNonce + 1 : cur.deepNonce,
+ });
+ } else if (frame.type === "death") {
+ set({ ...cur, deaths: [frame.data, ...cur.deaths] });
+ } else if (frame.type === "graph") {
+ set({ ...cur, graphNonce: cur.graphNonce + 1 });
+ }
+ };
+
+ connect();
+ return () => {
+ closed = true;
+ clearTimeout(retry);
+ ws?.close();
+ };
+ }, [runId]);
+
+ const value = useMemo(() => state, [state]);
+ return {children} ;
+}
diff --git a/web/lib/types.ts b/web/lib/types.ts
index c87fac1..c21f406 100644
--- a/web/lib/types.ts
+++ b/web/lib/types.ts
@@ -4,6 +4,8 @@ export interface RunInfo {
id: string;
name: string;
created_at: number;
+ origin?: string;
+ origin_ip?: string;
counts: { samples?: number; deaths?: number; workers?: number };
}
@@ -14,15 +16,79 @@ export interface ChunkMeta {
nbytes: number;
}
+export interface AllocationSite {
+ filename: string;
+ lineno: number;
+ function: string;
+ hwm_bytes: number;
+ n_allocations: number;
+ task_key: string;
+ layer: string;
+}
+
export interface DeathEvent {
timestamp: number;
worker: string;
suspect_keys: string[];
suspect_chunks: ChunkMeta[];
+ suspect_sites: AllocationSite[];
suspected_oom: boolean;
reason: string;
}
+export interface WorkerStatus {
+ worker: string;
+ timestamp: number;
+ rss_bytes: number;
+ managed_bytes: number;
+ memory_limit: number;
+ cpu: number;
+ nthreads: number;
+ executing: number;
+ ready: number;
+}
+
+// Aggregated per-source-line deep memory (peak bytes across epochs).
+export interface AllocSiteRow {
+ filename: string;
+ lineno: number;
+ function: string;
+ hwm_bytes: number;
+ n_allocations: number;
+ layers: string[];
+}
+
+export interface TaskMemoryRow {
+ key: string;
+ layer: string;
+ worker: string;
+ peak_rss_delta: number;
+ top_sites: AllocationSite[];
+}
+
+export interface FlameFrame {
+ function: string;
+ filename: string;
+ lineno: number;
+}
+
+export interface FlameStack {
+ frames: FlameFrame[];
+ hwm_bytes: number;
+ n_allocations: number;
+}
+
+export interface FlameData {
+ workers: string[];
+ stacks: FlameStack[];
+}
+
+export interface AllocTimelineRow {
+ ts: number;
+ layer: string;
+ bytes: number;
+}
+
export interface Sample {
worker: string;
timestamp: number;
diff --git a/web/public/logo.svg b/web/public/logo.svg
new file mode 100644
index 0000000..9b8795f
--- /dev/null
+++ b/web/public/logo.svg
@@ -0,0 +1,22 @@
+
From 0a7448fe846dd3b14a38b6c0b9e62d8c6d52c908 Mon Sep 17 00:00:00 2001
From: polymood <36277904+polymood@users.noreply.github.com>
Date: Fri, 3 Jul 2026 16:49:52 +0200
Subject: [PATCH 3/4] feat: examples across Dask + packaging, docs, CI
- examples covering the breadth of Dask: distributed/deep OOM demos, a
minutes-long big pipeline, a self-limiting crash, and one per collection type
(dask.delayed, dask.dataframe, dask.bag, xarray on Zarr and NetCDF).
- Packaging: PyPI metadata (authors, urls, classifiers), MIT LICENSE.
- CI workflow: ruff + mypy + pytest + package build + web build.
- README rewritten; examples/README expanded.
---
.github/workflows/ci.yml | 39 ++++
LICENSE | 21 +++
README.md | 333 ++++++++++++++++-------------------
examples/README.md | 24 +++
examples/big_pipeline_oom.py | 192 ++++++++++++++++++++
examples/dask_bag.py | 55 ++++++
examples/dask_dataframe.py | 62 +++++++
examples/dask_delayed.py | 77 ++++++++
examples/deep_oom.py | 72 ++++++++
examples/distributed_oom.py | 4 +-
examples/threaded_big.py | 86 +++++++++
examples/threaded_crash.py | 93 ++++++++++
examples/xarray_netcdf.py | 63 +++++++
examples/xarray_zarr.py | 66 +++++++
14 files changed, 1003 insertions(+), 184 deletions(-)
create mode 100644 .github/workflows/ci.yml
create mode 100644 LICENSE
create mode 100644 examples/big_pipeline_oom.py
create mode 100644 examples/dask_bag.py
create mode 100644 examples/dask_dataframe.py
create mode 100644 examples/dask_delayed.py
create mode 100644 examples/deep_oom.py
create mode 100644 examples/threaded_big.py
create mode 100644 examples/threaded_crash.py
create mode 100644 examples/xarray_netcdf.py
create mode 100644 examples/xarray_zarr.py
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
new file mode 100644
index 0000000..4ecfea8
--- /dev/null
+++ b/.github/workflows/ci.yml
@@ -0,0 +1,39 @@
+name: CI
+
+on:
+ push:
+ branches: [main]
+ pull_request:
+
+jobs:
+ python:
+ runs-on: ubuntu-latest
+ steps:
+ - uses: actions/checkout@v4
+ - name: Install uv
+ uses: astral-sh/setup-uv@v5
+ - name: Sync (dev + collector + deep)
+ run: uv sync --group dev --extra collector --extra deep
+ - name: Lint
+ run: uv run ruff check .
+ - name: Format check
+ run: uv run ruff format --check .
+ - name: Type check
+ run: uv run mypy src/
+ - name: Tests (unit)
+ run: uv run pytest -q -m "not integration"
+ - name: Build package
+ run: uv build
+
+ web:
+ runs-on: ubuntu-latest
+ defaults:
+ run:
+ working-directory: web
+ steps:
+ - uses: actions/checkout@v4
+ - uses: actions/setup-node@v4
+ with:
+ node-version: "20"
+ - run: npm ci
+ - run: npm run build
diff --git a/LICENSE b/LICENSE
new file mode 100644
index 0000000..fab0d2a
--- /dev/null
+++ b/LICENSE
@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 polymood
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
diff --git a/README.md b/README.md
index 893bd4b..efc0c88 100644
--- a/README.md
+++ b/README.md
@@ -1,116 +1,111 @@
-# DaskGenie
-
-A code-to-graph-to-memory profiler for Dask: when a worker dies from an OOM,
-know exactly which chunk killed it and which line of your code produced it.
-
-Built incrementally, each stage proven before the next is built. See
-**Status** below for what's real today.
-
-## Status
-
-- [x] **GraphCapture.** Maps Dask task-graph layers back to the user source
- line that built them. No UI. This is the de-risking step: prove source
- attribution works before building anything on top of it.
-- [x] **WorkerPlugin + Collector.** Per-task RSS / managed-memory sampling
- with input chunk metadata, batched to a FastAPI collector that stores to
- SQLite, exposes Prometheus `/metrics`, and serves a query API.
-- [x] **SchedulerPlugin — the v1 success criterion.** Tracks which tasks are
- in flight on which worker and, on a worker death, records the suspect
- tasks; the collector joins their chunk metadata so `curl /api/deaths`
- answers *"which chunk killed this worker, and what code produced it."*
-- [x] **Dashboard (Next.js) + Docker.** A standalone dashboard (light theme)
- over the collector API: runs list, per-run overview, post-mortem with
- syntax-highlighted source, memory timeline, and the task-graph DAG.
- `docker compose up` runs collector + dashboard, persisted on a volume.
-- [x] **Any scheduler.** `LocalProfiler` (Dask callback API) profiles memory +
- per-task chunks on the threaded/synchronous/processes schedulers, not
- just `dask.distributed`. Runnable `examples/`.
-- [ ] Next: aligned execution-ordered view, memory flamegraphs, on-demand
- single-task memray, TimescaleDB backend.
-
-## The dashboard (always-on, via Docker)
-
-Two services — the **collector** (JSON API + Prometheus `/metrics` + SQLite,
-port 8765) and the **Next.js dashboard** (port 3000) — both persisted and
-restart-on-failure:
+:8765`). |
+
+The `register()` and `LocalProfiler` calls take `deep=`, `sample_interval=`,
+`flush_interval=`, and `deep_epoch_seconds=` to trade overhead for resolution.
+
+## Notes and limitations
+
+- **Source attribution** is strongest for `dask.array` and `dask.delayed`. For
+ `dask.dataframe` (which builds graphs through dask-expr) and xarray, the heavy
+ work runs inside library/C code, so per-line allocations fold to framework
+ frames — the graph, memory, per-layer and flamegraph views still work, but
+ layer→line mapping is sparse there.
+- **memray** is Linux/macOS + CPython only, and a single tracker runs per
+ process; deep mode is opt-in and costs roughly 1.5–2× runtime.
+- On a **hard OOM kill** the worker's in-flight memray epoch can die before it
+ flushes, so a specific post-mortem's line attribution is best-effort — the
+ Memory tab remains the reliable place to see the culprit line.
+- The **OOM label** is a heuristic, not a certainty (see the post-mortem note).
+- **TimescaleDB** is the default store in Docker; SQLite is the zero-setup
+ backend used locally and by the test suite.
+
+## Examples
+
+Runnable scripts covering the breadth of Dask live in
+[`examples/`](./examples) — distributed OOMs, the deep-memory demo, a big
+minutes-long pipeline, a self-limiting crash, and one per collection type
+(`dask.delayed`, `dask.dataframe`, `dask.bag`, xarray on Zarr and NetCDF). See
+[`examples/README.md`](./examples/README.md).
## Development
```bash
-uv sync --group dev
+uv sync --group dev --extra collector --extra deep
uv run pytest
uv run ruff check .
uv run ruff format .
uv run mypy src/
+
+# collector + dashboard separately (dashboard proxies /api to the collector)
+uv run python -m daskgenie.collector --port 8765 # terminal 1
+cd web && npm install && npm run dev # terminal 2 → :3000
```
## How source attribution works
`HighLevelGraph.from_collections` is the one classmethod almost every Dask
-collection operation (array, dataframe, and by extension xarray, since it
-wraps dask arrays) calls to register a new named graph layer. `track()`
-patches it for the duration of the `with` block: each time a new layer name
-appears, it walks the call stack outward until it finds the first frame that
-isn't inside `dask`/`distributed`/`xarray`/`numpy`/`zarr`/`daskgenie` or a
-`site-packages` install — that's the user's call site. The library-path
-filter is configurable via `track(extra_library_paths=[...])` for teams with
-their own internal wrapper libraries.
+collection operation calls to register a new named graph layer. `track()`
+patches it for the duration of the `with` block: each time a new layer appears,
+it walks the call stack outward until it finds the first frame that isn't inside
+`dask`/`distributed`/`xarray`/`numpy`/`zarr`/`daskgenie` or a `site-packages`
+install — that's your call site. The deep memory engine reuses the same
+library-path filter to fold memray stacks to the first user frame. The
+library-path filter is configurable via `track(extra_library_paths=[...])`.
+
+## License
+
+MIT — see [LICENSE](./LICENSE).
diff --git a/examples/README.md b/examples/README.md
index 90d395c..4c604b2 100644
--- a/examples/README.md
+++ b/examples/README.md
@@ -5,6 +5,15 @@
| `graph_source_map.py` | nothing | Source attribution only — layer → the line that built it. |
| `local_scheduler.py` | running collector | Memory + chunks + graph for a **local** scheduler (threaded/synchronous/processes). |
| `distributed_oom.py` | running collector | A real worker OOM and its post-mortem on `dask.distributed`. |
+| `deep_oom.py` | running collector + `deep` extra | Deep memory (memray): the **source line** that allocated the OOM array, live Workers/Task-stream/Memory tabs. |
+| `big_pipeline_oom.py` | running collector + `deep` extra | A realistic **minutes-long** multi-stage pipeline (FFT + matmul), a **large task graph** (~4.5k nodes), streamed live, ending in a real worker OOM the Memory tab attributes to `upsample_block`. |
+| `threaded_big.py` | running collector + `deep` extra | **Threaded scheduler** (no workers): a pipeline that *hoards* every result in memory instead of spilling to Zarr — the RSS staircase + the accumulation line, the "should have gone to disk" signature. |
+| `threaded_crash.py` | running collector + `deep` extra | Threaded scheduler that **actually crashes** (self-limiting `MemoryError`): the memory curve climbing to the wall and the line that got it there. |
+| `dask_delayed.py` | running collector + `examples` + `deep` extras | **dask.delayed** custom ETL DAG on a distributed cluster — per-line memory on `transform_shard`. |
+| `dask_dataframe.py` | running collector + `examples` + `deep` extras | **dask.dataframe** groupby-aggregation + shuffle (`set_index` rolling) on a synthetic timeseries. |
+| `dask_bag.py` | running collector + `examples` + `deep` extras | **dask.bag** record/JSON aggregation on the **processes** scheduler (per-process memory). |
+| `xarray_zarr.py` | running collector + `examples` + `deep` extras | **xarray on Zarr**: a chunked cube streamed from a Zarr store → climatology + anomaly (bounded memory, the read-don't-hoard pattern). |
+| `xarray_netcdf.py` | running collector + `examples` + `deep` extras | **xarray on NetCDF** (HDF5/h5netcdf): open with Dask chunks, rolling-mean detrend + reduce. |
## Setup
@@ -20,8 +29,23 @@ Then run an example and open the dashboard at http://localhost:3000:
```bash
uv run --extra demo python examples/local_scheduler.py
uv run --extra demo python examples/distributed_oom.py
+uv run --extra demo --extra deep python examples/deep_oom.py
+uv run --extra demo --extra deep python examples/big_pipeline_oom.py # ~2-4 min, big graph, real OOM
+uv run --extra demo --extra deep python examples/threaded_big.py # threaded, memory-hoard demo
+
+# Across Dask's collections (need the `examples` extra: xarray, zarr, pandas, ...)
+uv run --extra examples --extra deep python examples/dask_delayed.py
+uv run --extra examples --extra deep python examples/dask_dataframe.py
+uv run --extra examples --extra deep python examples/dask_bag.py
+uv run --extra examples --extra deep python examples/xarray_zarr.py
+uv run --extra examples --extra deep python examples/xarray_netcdf.py
```
+> Source-line attribution is strongest for `dask.array` and `dask.delayed`. For
+> `dask.dataframe` (dask-expr) and xarray the heavy work runs inside library C
+> code, so allocations fold to framework frames — the **flamegraph** still shows
+> the full call tree, and the graph/memory views work throughout.
+
(No Docker? Run the collector directly with
`uv run python -m daskgenie.collector --port 8765` and the dashboard dev server
from `web/` — see the top-level README.)
diff --git a/examples/big_pipeline_oom.py b/examples/big_pipeline_oom.py
new file mode 100644
index 0000000..ca75499
--- /dev/null
+++ b/examples/big_pipeline_oom.py
@@ -0,0 +1,192 @@
+"""A realistic, minutes-long Dask pipeline that builds a large task graph, does
+genuine heavy per-block compute, and then dies from a real out-of-memory on a
+worker — the full DaskGenie story end to end.
+
+It mimics an image/geospatial processing pipeline over a big float32 field:
+
+ 1. ingest random scene, chunked (big base graph)
+ 2. normalise elementwise (x - mean) / std (per-block)
+ 3. smooth overlapping stencil (map_overlap) (halo exchange → edges)
+ 4. spectral per-block FFT magnitude (CPU heavy)
+ 5. refine ×N matmul-based diffusion + rechunk (burns minutes, huge graph)
+ 6. upsample naive super-resolution (np.kron) (THE OOM — one block
+ explodes to a multi-GB float64 array, past the worker limit)
+
+Stages 1-5 run for a couple of minutes and stream live to the dashboard
+(Workers / Task stream / Memory update in real time). Stage 6 then OOM-kills a
+worker on a single, nameable line, so the Post-mortem and the deep Memory tab
+point straight at ``big_pipeline_oom.py: upsample_block``.
+
+Start the stack first (``docker compose up -d``), then:
+
+ uv run --extra demo --extra deep python examples/big_pipeline_oom.py
+
+Tunables (env): DG_N (field size), DG_CHUNK, DG_ITERS, DG_WORKERS, DG_MEMLIMIT.
+Defaults run ~2-4 min and reliably OOM with a 1900MB worker limit.
+"""
+
+from __future__ import annotations
+
+import os
+import time
+
+import numpy as np
+from distributed import Client, LocalCluster, wait
+
+import daskgenie as dg
+import daskgenie.client as genie
+
+COLLECTOR = os.environ.get("DG_COLLECTOR", "http://localhost:8765")
+N = int(os.environ.get("DG_N", "14000"))
+CHUNK = int(os.environ.get("DG_CHUNK", "1400"))
+ITERS = int(os.environ.get("DG_ITERS", "3"))
+REPEAT = int(os.environ.get("DG_REPEAT", "8")) # inner matmuls per diffuse — runtime knob
+WORKERS = int(os.environ.get("DG_WORKERS", "4"))
+# Headroom so the refine loop runs clean; the upsample is the intended OOM.
+MEMLIMIT = os.environ.get("DG_MEMLIMIT", "2600MB")
+UPSAMPLE = int(os.environ.get("DG_UPSAMPLE", "11")) # Kron factor → ~1.9 GB per tile
+
+
+# -- per-block compute kernels (real work, each a nameable source line) --------
+
+
+def normalise_block(block: np.ndarray) -> np.ndarray:
+ # standardise the tile; cheap, elementwise
+ mu = block.mean()
+ sd = block.std() + 1e-6
+ return ((block - mu) / sd).astype(np.float32)
+
+
+def spectral_block(block: np.ndarray) -> np.ndarray:
+ # 2-D FFT magnitude — genuinely CPU-heavy, keeps workers busy
+ spec = np.fft.rfft2(block)
+ mag = np.abs(spec).astype(np.float32)
+ # pad back to the block's shape so the graph stays rectangular
+ out = np.zeros_like(block)
+ out[:, : mag.shape[1]] = mag[:, : out.shape[1]]
+ return out
+
+
+def diffuse_block(block: np.ndarray) -> np.ndarray:
+ # matmul-based smoothing: an (n x n) gaussian kernel applied REPEAT times.
+ # O(n^3) per pass — the main time sink of the refine loop. Single-sided so it
+ # stays valid for non-square edge tiles after a rechunk; REPEAT (not more
+ # tasks) is the runtime knob, keeping the graph a readable size.
+ n = block.shape[0]
+ kernel = np.exp(-((np.arange(n)[:, None] - np.arange(n)[None, :]) ** 2) / (2 * 64.0))
+ kernel = (kernel / kernel.sum(axis=1, keepdims=True)).astype(np.float32)
+ out = block
+ for _ in range(REPEAT):
+ out = (kernel @ out).astype(np.float32)
+ return out
+
+
+def upsample_block(block: np.ndarray) -> np.ndarray:
+ # Naive super-resolution: Kronecker-upsample each tile. For a 1400x1400 tile
+ # at factor 11 this materialises a 15400x15400 float64 array (~1.9 GB) on ONE
+ # task — the careless allocation the deep Memory tab and post-mortem name.
+ hires = np.kron(block.astype(np.float64), np.ones((UPSAMPLE, UPSAMPLE))) # ~1.9 GB
+ # Hold + touch the array for ~1.2s of real work so a deep-memory epoch closes
+ # and flushes this multi-GB high-water mark *before* the fatal second buffer
+ # below tips the worker over its limit (a hard OOM kill is faster than an
+ # epoch, so without this the killing line dies unrecorded).
+ t0 = time.time()
+ acc = 0.0
+ while time.time() - t0 < 1.2:
+ acc += float(hires.sum())
+ # now blow past the limit: a second equally-huge buffer forces the OOM
+ killer = np.kron(block.astype(np.float64), np.ones((UPSAMPLE, UPSAMPLE))) # +1.9 GB → OOM
+ return ((hires + killer)[::UPSAMPLE, ::UPSAMPLE] + acc * 0.0).astype(np.float32)
+
+
+def main() -> None:
+ import dask.array as da
+
+ cluster = LocalCluster(
+ n_workers=WORKERS,
+ threads_per_worker=1,
+ processes=True,
+ memory_limit=MEMLIMIT,
+ dashboard_address=":0",
+ )
+ client = Client(cluster)
+ print(f"cluster: {WORKERS} workers x {MEMLIMIT}; field {N}x{N} float32, chunk {CHUNK}")
+
+ try:
+ run_id = genie.register(
+ client,
+ COLLECTOR,
+ run_name=f"big pipeline {N}x{N} (OOM)",
+ flush_interval=0.25,
+ deep=True,
+ deep_epoch_seconds=1.0,
+ )
+ print(f"run_id = {run_id} — open the dashboard now and watch it live")
+
+ with dg.track() as source_map:
+ # 1. ingest — a big chunked float32 field
+ x = da.random.random((N, N), chunks=(CHUNK, CHUNK)).astype(np.float32)
+
+ # 2. normalise (elementwise, per-block)
+ x = x.map_blocks(normalise_block, dtype=np.float32)
+
+ # 3. smooth — overlapping stencil, adds halo-exchange edges to the graph
+ x = x.map_overlap(
+ lambda b: (b + np.roll(b, 1, 0) + np.roll(b, -1, 0)) / 3.0,
+ depth=1,
+ boundary="reflect",
+ dtype=np.float32,
+ )
+
+ # 4. spectral features (CPU heavy)
+ x = x.map_blocks(spectral_block, dtype=np.float32)
+
+ # 5. refine loop — matmul diffusion + a rechunk each pass. This is
+ # where the minutes go and the graph balloons.
+ for i in range(ITERS):
+ x = x.map_blocks(diffuse_block, dtype=np.float32)
+ # alternate the chunking so successive passes must reshuffle
+ new_chunk = CHUNK if i % 2 == 0 else CHUNK + 500
+ x = x.rechunk((new_chunk, new_chunk))
+ x = (x - x.mean()) / (x.std() + 1e-6)
+
+ # Upload the FULL refined-field graph (stages 1-5, thousands of tasks)
+ # so the Graph tab shows the real multi-stage DAG — not just the small
+ # post-persist tail. The upsample line is added to the source map below
+ # so the Memory/Post-mortem tabs can still name it.
+ with dg.track() as source_map2:
+ # 6. the OOM: naive super-resolution upsample
+ result = x.map_blocks(upsample_block, dtype=np.float32).sum()
+ source_map.update(source_map2)
+ dg.upload_graph(COLLECTOR, run_id, source_map, collection=x)
+
+ # Persist the refined field so stages 1-5 actually run (and stream) before
+ # the OOM — this is the "couple of minutes of real work" part.
+ print("running stages 1-5 (this takes a few minutes)...")
+ t0 = time.time()
+ x = client.persist(x)
+ wait(x)
+ print(f"refined field ready in {time.time() - t0:.0f}s; now the upsample OOM...")
+
+ # rebuild the OOM step on the now-persisted field
+ result = x.map_blocks(upsample_block, dtype=np.float32).sum()
+
+ try:
+ client.compute(result).result(timeout=180)
+ print("no OOM — try a smaller DG_MEMLIMIT or larger DG_CHUNK")
+ except Exception as exc: # noqa: BLE001 - expected KilledWorker
+ print(f"worker OOM-killed as expected: {type(exc).__name__}")
+
+ print(
+ f"\ndone — run {run_id!r}:\n"
+ " • Memory tab → upsample_block line at multi-GB peak\n"
+ " • Post-mortem → the suspect task, chunk, and (timing permitting) the line\n"
+ " • Graph tab → the full multi-thousand-node task DAG"
+ )
+ finally:
+ client.close()
+ cluster.close()
+
+
+if __name__ == "__main__":
+ main()
diff --git a/examples/dask_bag.py b/examples/dask_bag.py
new file mode 100644
index 0000000..6811ddf
--- /dev/null
+++ b/examples/dask_bag.py
@@ -0,0 +1,55 @@
+"""Profile a **dask.bag** pipeline (processes scheduler).
+
+Bags are Dask's tool for messy, record-oriented data (JSON logs, text). This
+simulates a stream of event records, parses/filters/aggregates them (a word/tag
+frequency count via ``foldby``), and profiles it on the multiprocessing
+scheduler — so DaskGenie shows per-process memory, not just threads.
+
+Needs the ``examples`` + ``deep`` extras. Start the stack, then:
+
+ uv run --extra examples --extra deep python examples/dask_bag.py
+"""
+
+from __future__ import annotations
+
+import os
+import random
+
+import daskgenie as dg
+
+COLLECTOR = os.environ.get("DG_COLLECTOR", "http://localhost:8765")
+N = int(os.environ.get("DG_N", "4000000"))
+PARTS = int(os.environ.get("DG_PARTS", "32"))
+
+TAGS = ["error", "warn", "info", "debug", "trace", "fatal", "audit", "metric"]
+
+
+def make_record(i: int) -> dict:
+ rng = random.Random(i)
+ return {"id": i, "tag": rng.choice(TAGS), "value": rng.random(), "ok": rng.random() > 0.2}
+
+
+def main() -> None:
+ import dask.bag as db
+
+ with dg.track() as source_map:
+ bag = db.from_sequence(range(N), npartitions=PARTS).map(make_record)
+ kept = bag.filter(lambda r: r["ok"])
+ # frequency of each tag among kept records
+ counts = kept.foldby("tag", lambda acc, _r: acc + 1, 0, lambda a, b: a + b)
+ result = counts
+
+ with dg.LocalProfiler(
+ COLLECTOR,
+ run_name="dask.bag log aggregation",
+ source_map=source_map,
+ deep=True,
+ deep_epoch_seconds=2.0,
+ ) as prof:
+ print(f"run {prof.run_id}: folding {N:,} records over {PARTS} partitions...")
+ out = dict(result.compute(scheduler="processes", num_workers=4))
+ print(f"done — tag counts: {out}\nopen the dashboard run {prof.run_id!r}")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/examples/dask_dataframe.py b/examples/dask_dataframe.py
new file mode 100644
index 0000000..8d0a4a2
--- /dev/null
+++ b/examples/dask_dataframe.py
@@ -0,0 +1,62 @@
+"""Profile a **dask.dataframe** pipeline on a distributed cluster.
+
+Uses Dask's built-in synthetic timeseries (no external data), then does the
+memory-heavy things dataframes are known for: a groupby-aggregation and a
+shuffle-y ``set_index``. DaskGenie shows the partition memory over time and the
+per-layer allocation timeline (groupby vs shuffle).
+
+Start the stack (``docker compose up -d``), then:
+
+ uv run --extra examples --extra deep python examples/dask_dataframe.py
+"""
+
+from __future__ import annotations
+
+import os
+
+from distributed import Client, LocalCluster
+
+import daskgenie as dg
+import daskgenie.client as genie
+
+COLLECTOR = os.environ.get("DG_COLLECTOR", "http://localhost:8765")
+DAYS = int(os.environ.get("DG_DAYS", "120")) # ~ size of the synthetic frame
+
+
+def main() -> None:
+ import dask.dataframe as dd
+ import dask.datasets
+
+ cluster = LocalCluster(
+ n_workers=4, threads_per_worker=2, processes=True, dashboard_address=":0"
+ )
+ client = Client(cluster)
+ try:
+ run_id = genie.register(client, COLLECTOR, run_name="dask.dataframe groupby", deep=True)
+
+ with dg.track() as source_map:
+ df = dask.datasets.timeseries(
+ start="2020-01-01",
+ end=f"2020-{1 + DAYS // 30:02d}-01",
+ freq="100ms",
+ partition_freq="1d",
+ dtypes={"x": float, "y": float, "id": int},
+ )
+ df = df.assign(z=(df.x * df.y).abs())
+ # groupby-aggregation (hash aggregation memory)
+ agg = df.groupby("id").agg({"x": "mean", "y": "std", "z": "sum"})
+ # a rolling window over a set_index (shuffle) branch
+ rolled = df.set_index("id").z.rolling(50).mean()
+ result = agg.z.sum() + dd.to_numeric(rolled.fillna(0)).sum()
+ dg.upload_graph(COLLECTOR, run_id, source_map, collection=result)
+
+ print(f"run {run_id}: computing groupby + rolling over the frame...")
+ val = result.compute()
+ print(f"done — result={float(val):.3e}; open the dashboard run {run_id!r}")
+ finally:
+ client.close()
+ cluster.close()
+
+
+if __name__ == "__main__":
+ main()
diff --git a/examples/dask_delayed.py b/examples/dask_delayed.py
new file mode 100644
index 0000000..5b7f5dd
--- /dev/null
+++ b/examples/dask_delayed.py
@@ -0,0 +1,77 @@
+"""Profile a **dask.delayed** pipeline on a distributed cluster.
+
+``@delayed`` builds a task graph out of ordinary Python functions — the classic
+"custom ETL DAG" use of Dask. DaskGenie captures the graph, per-task memory, and
+(with ``deep=True``) the source line behind each allocation, exactly as it does
+for arrays.
+
+Start the stack (``docker compose up -d``), then:
+
+ uv run --extra examples --extra deep python examples/dask_delayed.py
+
+Open the run: the **Graph** tab shows the delayed DAG (load → transform →
+combine), **Memory** attributes bytes to ``transform_shard``.
+"""
+
+from __future__ import annotations
+
+import os
+
+import numpy as np
+from dask import delayed
+from distributed import Client, LocalCluster
+
+import daskgenie as dg
+import daskgenie.client as genie
+
+COLLECTOR = os.environ.get("DG_COLLECTOR", "http://localhost:8765")
+SHARDS = int(os.environ.get("DG_SHARDS", "24"))
+SIZE = int(os.environ.get("DG_SIZE", "2000"))
+
+
+@delayed
+def load_shard(i: int) -> np.ndarray:
+ # stand-in for reading shard i off disk / object store
+ rng = np.random.default_rng(i)
+ return rng.random((SIZE, SIZE), dtype=np.float64)
+
+
+@delayed
+def transform_shard(a: np.ndarray) -> np.ndarray:
+ # per-shard feature transform — where the memory goes
+ return np.tanh(a @ a.T) / (a.std() + 1e-9)
+
+
+@delayed
+def reduce_shard(a: np.ndarray) -> float:
+ return float(a.sum())
+
+
+@delayed
+def combine(parts: list[float]) -> float:
+ return float(np.sum(parts))
+
+
+def main() -> None:
+ cluster = LocalCluster(
+ n_workers=4, threads_per_worker=1, processes=True, dashboard_address=":0"
+ )
+ client = Client(cluster)
+ try:
+ run_id = genie.register(client, COLLECTOR, run_name="dask.delayed ETL", deep=True)
+
+ with dg.track() as source_map:
+ parts = [reduce_shard(transform_shard(load_shard(i))) for i in range(SHARDS)]
+ total = combine(parts)
+ dg.upload_graph(COLLECTOR, run_id, source_map, collection=total)
+
+ print(f"run {run_id}: computing {SHARDS} delayed shards...")
+ result = total.compute()
+ print(f"done — result={result:.3e}; open the dashboard run {run_id!r}")
+ finally:
+ client.close()
+ cluster.close()
+
+
+if __name__ == "__main__":
+ main()
diff --git a/examples/deep_oom.py b/examples/deep_oom.py
new file mode 100644
index 0000000..8174750
--- /dev/null
+++ b/examples/deep_oom.py
@@ -0,0 +1,72 @@
+"""Deep memory profiling end to end: run a pipeline that allocates a huge array
+on a line we can name, with the memray engine on, then read the *source line*
+that dominated memory — not just "a chunk was resident".
+
+Start the stack first (``docker compose up -d``), then:
+
+ uv run --extra demo --extra deep python examples/deep_oom.py
+
+Open http://localhost:3000, pick the run, and watch it live:
+
+- **Workers** / **Task stream** update in real time as the job runs.
+- **Memory** shows the allocation flamegraph and the per-source-line table —
+ ``deep_oom.py: build_monster ~2 GB`` at the top.
+- **Post-mortem** (if the worker is OOM-killed) names that allocation line as
+ the cause, above the input-chunk view.
+"""
+
+from __future__ import annotations
+
+import numpy as np
+from distributed import Client, LocalCluster, wait
+
+import daskgenie as dg
+import daskgenie.client as genie
+
+COLLECTOR = "http://localhost:8765"
+
+
+def build_monster(block: np.ndarray) -> float:
+ # This is the line that eats the memory — the deep engine attributes the
+ # high-water mark straight back here (file:line), then to this task/layer.
+ monster = np.ones((16000, 16000), dtype="float64") # ~2 GB, on THIS line
+ return float(np.asarray(block).sum() + monster.sum())
+
+
+def main() -> None:
+ import dask.array as da
+
+ cluster = LocalCluster(n_workers=2, threads_per_worker=1, processes=True, memory_limit="1500MB")
+ client = Client(cluster)
+ try:
+ # deep=True installs the memray engine on each worker (needs the `deep`
+ # extra). deep_epoch_seconds bounds how quickly per-line data streams in.
+ run_id = genie.register(
+ client,
+ COLLECTOR,
+ run_name="deep OOM (memray)",
+ flush_interval=0.25,
+ deep=True,
+ deep_epoch_seconds=2.0,
+ )
+
+ with dg.track() as source_map:
+ x = client.persist(da.ones((8000, 8000), chunks=(4000, 4000)))
+ wait(x)
+ result = x.map_blocks(build_monster, dtype="float64").sum()
+ dg.upload_graph(COLLECTOR, run_id, source_map, collection=result)
+
+ print(f"running with deep memory profiling — run {run_id!r}")
+ try:
+ client.compute(result).result(timeout=90)
+ except Exception as exc: # noqa: BLE001 - expected KilledWorker on OOM
+ print(f"worker died as expected: {type(exc).__name__}")
+
+ print(f"done — open the dashboard, run {run_id!r}, Memory + Post-mortem tabs")
+ finally:
+ client.close()
+ cluster.close()
+
+
+if __name__ == "__main__":
+ main()
diff --git a/examples/distributed_oom.py b/examples/distributed_oom.py
index fc5c87b..027d44d 100644
--- a/examples/distributed_oom.py
+++ b/examples/distributed_oom.py
@@ -31,9 +31,7 @@ def blowup(block: np.ndarray) -> float:
def main() -> None:
import dask.array as da
- cluster = LocalCluster(
- n_workers=2, threads_per_worker=1, processes=True, memory_limit="1500MB"
- )
+ cluster = LocalCluster(n_workers=2, threads_per_worker=1, processes=True, memory_limit="1500MB")
client = Client(cluster)
try:
run_id = genie.register(client, COLLECTOR, run_name="rechunk OOM", flush_interval=0.1)
diff --git a/examples/threaded_big.py b/examples/threaded_big.py
new file mode 100644
index 0000000..58ccb86
--- /dev/null
+++ b/examples/threaded_big.py
@@ -0,0 +1,86 @@
+"""A big **threaded-scheduler** pipeline that shows the other failure mode: not a
+single monster allocation, but *holding too much in memory that should have been
+spilled to disk (Zarr)*.
+
+There are no distributed workers here — everything runs in one process on
+``scheduler="threads"``, profiled by ``LocalProfiler`` (the Dask callback API).
+The pipeline processes a stack of large scenes and — the bug — keeps **every**
+processed scene resident in a Python list instead of streaming each to a Zarr
+store. Memory climbs step by step; DaskGenie's deep Memory tab attributes the
+growth to the exact accumulation line, and the memory-over-time chart shows the
+staircase that a to-zarr write would have kept flat.
+
+Run (start the stack first with ``docker compose up -d``):
+
+ uv run --extra demo --extra deep python examples/threaded_big.py
+
+Tunables (env): DG_N (scene size), DG_CHUNK, DG_STEPS (how many scenes to hoard),
+DG_THREADS. Defaults hold ~4-5 GB and run ~2 min. Lower DG_STEPS if your machine
+has less RAM — the point is the *shape* of the memory curve, not a hard crash.
+"""
+
+from __future__ import annotations
+
+import os
+import time
+
+import numpy as np
+
+import daskgenie as dg
+
+COLLECTOR = os.environ.get("DG_COLLECTOR", "http://localhost:8765")
+N = int(os.environ.get("DG_N", "8000"))
+CHUNK = int(os.environ.get("DG_CHUNK", "1000"))
+STEPS = int(os.environ.get("DG_STEPS", "18"))
+THREADS = int(os.environ.get("DG_THREADS", "4"))
+
+
+def process_scene(block: np.ndarray) -> np.ndarray:
+ # genuine per-tile work: normalise, FFT magnitude, a matmul smooth
+ n = block.shape[0]
+ b = (block - block.mean()) / (block.std() + 1e-6)
+ spec = np.abs(np.fft.rfft2(b)).astype(np.float32)
+ out = np.zeros_like(b, dtype=np.float32)
+ out[:, : spec.shape[1]] = spec[:, : out.shape[1]]
+ kernel = np.exp(-((np.arange(n)[:, None] - np.arange(n)[None, :]) ** 2) / (2 * 64.0))
+ kernel = (kernel / kernel.sum(axis=1, keepdims=True)).astype(np.float32)
+ return (kernel @ out).astype(np.float32)
+
+
+def main() -> None:
+ import dask.array as da
+
+ print(f"threaded scheduler: {THREADS} threads; {STEPS} scenes of {N}x{N} float32")
+ print("BUG on purpose: every processed scene is kept resident instead of spilled to Zarr\n")
+
+ hoarded: list[np.ndarray] = [] # <-- the leak: everything stays in memory
+
+ with dg.LocalProfiler(
+ COLLECTOR,
+ run_name=f"threaded hoard {STEPS}x{N}",
+ deep=True,
+ deep_epoch_seconds=1.5,
+ ) as prof:
+ print(f"run_id = {prof.run_id} — open the dashboard and watch memory climb\n")
+ for step in range(STEPS):
+ with dg.track():
+ scene = da.random.random((N, N), chunks=(CHUNK, CHUNK)).astype(np.float32)
+ processed = scene.map_blocks(process_scene, dtype=np.float32)
+ # RIGHT way (kept for contrast): stream straight to disk, stay flat
+ # processed.to_zarr(f"/tmp/scenes/{step}.zarr", overwrite=True)
+ # WRONG way (what this script does): materialise and hoard it
+ result = processed.compute(scheduler="threads", num_workers=THREADS)
+ hoarded.append(result) # the line that holds ~256 MB per step forever
+ held_gb = sum(a.nbytes for a in hoarded) / 1e9
+ print(f" step {step + 1:2d}/{STEPS} holding {held_gb:5.2f} GB resident")
+ time.sleep(0.2)
+
+ print(
+ f"\ndone — run {prof.run_id!r}. Memory tab: the hoard line at the top of the\n"
+ "per-source-line table; Overview: a rising staircase instead of a flat line —\n"
+ "the signature of data that should have gone to a Zarr store, not a Python list."
+ )
+
+
+if __name__ == "__main__":
+ main()
diff --git a/examples/threaded_crash.py b/examples/threaded_crash.py
new file mode 100644
index 0000000..2d9c379
--- /dev/null
+++ b/examples/threaded_crash.py
@@ -0,0 +1,93 @@
+"""A threaded-scheduler pipeline that ACTUALLY CRASHES from memory exhaustion.
+
+Unlike ``threaded_big.py`` (which hoards a bounded amount to show the staircase),
+this one grows resident memory **without bound** until the process dies — a real
+``MemoryError`` (or an OS OOM-kill). It's the "my script just died and I don't
+know why" scenario: DaskGenie's memory-over-time chart shows the climb right up
+to the ceiling, and the deep Memory tab names the line doing the allocating.
+
+This is *self-limiting*: it grows resident memory to a ceiling (default 55% of
+this machine's RAM), then attempts one allocation larger than the memory that's
+left — which raises a real ``MemoryError`` and crashes the Python process,
+WITHOUT dragging the whole machine (or WSL) down with it. That's the honest "my
+job died of memory" event, safely.
+
+Run (start the stack first with ``docker compose up -d``):
+
+ uv run --extra demo --extra deep python examples/threaded_crash.py
+
+Because the sampler flushes every second, the memory curve and the deep
+allocation line are already in the collector by the time the process dies — open
+the run and you'll see exactly how big it got and which line got it there.
+
+Tunables (env): DG_STEP_MB (per-step MiB, default 1024), DG_N/DG_CHUNK (tile),
+DG_MAX_GB (hoard ceiling; default ~55% of RAM), DG_THREADS.
+"""
+
+from __future__ import annotations
+
+import os
+
+import numpy as np
+import psutil
+
+import daskgenie as dg
+
+COLLECTOR = os.environ.get("DG_COLLECTOR", "http://localhost:8765")
+STEP_MB = int(os.environ.get("DG_STEP_MB", "1024")) # ~1 GiB added per step
+N = int(os.environ.get("DG_N", "6000"))
+CHUNK = int(os.environ.get("DG_CHUNK", "1000"))
+THREADS = int(os.environ.get("DG_THREADS", "4"))
+# Ceiling for the hoard, defaulting to a safe fraction of physical RAM so we
+# crash *this process*, not the machine. Override with DG_MAX_GB.
+_DEFAULT_MAX_GB = round(psutil.virtual_memory().total * 0.55 / 1e9, 1)
+MAX_GB = float(os.environ.get("DG_MAX_GB", str(_DEFAULT_MAX_GB)))
+
+
+def heavy_block(block: np.ndarray) -> np.ndarray:
+ # real per-tile compute so this isn't just np.ones — normalise + smooth
+ b = (block - block.mean()) / (block.std() + 1e-6)
+ return np.tanh(b).astype(np.float32)
+
+
+def main() -> None:
+ import dask.array as da
+
+ # one float64 array of this many elements ~ STEP_MB per step
+ side = int((STEP_MB * 1024 * 1024 / 8) ** 0.5)
+ leak: list[np.ndarray] = [] # never freed → the runaway growth
+
+ print(f"threaded scheduler, {THREADS} threads — growing ~{STEP_MB} MiB/step")
+ print(f"will crash (MemoryError) once it holds ~{MAX_GB} GB (DG_MAX_GB to change)\n")
+
+ with dg.LocalProfiler(
+ COLLECTOR, run_name="threaded CRASH (OOM)", deep=True, deep_epoch_seconds=1.0
+ ) as prof:
+ print(f"run_id = {prof.run_id} — open the dashboard and watch it climb to the wall\n")
+ step = 0
+ while True:
+ step += 1
+ with dg.track():
+ scene = da.random.random((N, N), chunks=(CHUNK, CHUNK)).astype(np.float32)
+ _ = (
+ scene.map_blocks(heavy_block, dtype=np.float32)
+ .sum()
+ .compute(scheduler="threads", num_workers=THREADS)
+ )
+ # THE LEAK: a big block that never leaves memory. This is the line the
+ # deep Memory tab will blame for the runaway growth.
+ leak.append(np.ones((side, side), dtype=np.float64))
+ held_gb = sum(a.nbytes for a in leak) / 1e9
+ print(f" step {step:3d} holding {held_gb:6.2f} GB resident", flush=True)
+
+ if held_gb >= MAX_GB:
+ # Tip over the edge: ask for more than the RAM that's left. numpy
+ # raises MemoryError immediately — a real crash, machine intact.
+ avail = psutil.virtual_memory().available
+ boom = int((avail * 1.5) ** 0.5) + 1
+ print(f" ceiling reached — allocating past the wall ({avail / 1e9:.1f} GB free)")
+ _ = np.ones((boom, boom), dtype=np.float64) # -> MemoryError, crashes here
+
+
+if __name__ == "__main__":
+ main()
diff --git a/examples/xarray_netcdf.py b/examples/xarray_netcdf.py
new file mode 100644
index 0000000..1bb53bf
--- /dev/null
+++ b/examples/xarray_netcdf.py
@@ -0,0 +1,63 @@
+"""Profile an **xarray-on-NetCDF** pipeline (threaded scheduler).
+
+Same shape as the Zarr example but over a NetCDF file (HDF5 via ``h5netcdf``) —
+the format most climate/ocean data actually ships in. Opens the file with Dask
+chunks, computes a rolling-mean smoothing and a reduction, and DaskGenie
+attributes the memory to your lines.
+
+Needs the ``examples`` + ``deep`` extras. Start the stack, then:
+
+ uv run --extra examples --extra deep python examples/xarray_netcdf.py
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+
+import numpy as np
+import xarray as xr
+
+import daskgenie as dg
+
+COLLECTOR = os.environ.get("DG_COLLECTOR", "http://localhost:8765")
+NT = int(os.environ.get("DG_NT", "200"))
+NY = int(os.environ.get("DG_NY", "512"))
+NX = int(os.environ.get("DG_NX", "1024"))
+
+
+def build_netcdf(path: str) -> None:
+ ds = xr.DataArray(
+ np.random.default_rng(1).random((NT, NY, NX), dtype=np.float32),
+ dims=("time", "lat", "lon"),
+ name="ssh",
+ ).to_dataset()
+ ds.to_netcdf(path, engine="h5netcdf")
+
+
+def main() -> None:
+ tmp = os.path.join(tempfile.gettempdir(), "daskgenie_field.nc")
+ print(f"writing a {NT}×{NY}×{NX} NetCDF to {tmp} ...")
+ build_netcdf(tmp)
+
+ with dg.track() as source_map:
+ ds = xr.open_dataset(tmp, engine="h5netcdf", chunks={"time": 20})
+ smooth = ds.ssh.rolling(time=5, min_periods=1).mean() # temporal smoothing
+ detrended = ds.ssh - smooth
+ result = detrended.var(dim="time").mean().data
+
+ with dg.LocalProfiler(
+ COLLECTOR,
+ run_name="xarray + NetCDF smoothing",
+ source_map=source_map,
+ collection=result,
+ deep=True,
+ deep_epoch_seconds=2.0,
+ ) as prof:
+ print(f"run {prof.run_id}: computing detrended variance...")
+ val = result.compute(scheduler="threads", num_workers=4)
+ print(f"done — result={float(val):.5f}; open the dashboard run {prof.run_id!r}")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/examples/xarray_zarr.py b/examples/xarray_zarr.py
new file mode 100644
index 0000000..2ebb57a
--- /dev/null
+++ b/examples/xarray_zarr.py
@@ -0,0 +1,66 @@
+"""Profile an **xarray-on-Zarr** pipeline (threaded scheduler).
+
+The canonical geoscience workflow: a chunked data cube (time × lat × lon) stored
+in Zarr, opened lazily, and reduced (climatology + anomaly). This is exactly the
+"read from disk, don't hoard in RAM" pattern — DaskGenie shows the memory staying
+bounded because the cube streams from Zarr, and attributes compute memory to the
+anomaly line.
+
+Needs the ``examples`` + ``deep`` extras. Start the stack, then:
+
+ uv run --extra examples --extra deep python examples/xarray_zarr.py
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+
+import numpy as np
+import xarray as xr
+
+import daskgenie as dg
+
+COLLECTOR = os.environ.get("DG_COLLECTOR", "http://localhost:8765")
+NT = int(os.environ.get("DG_NT", "365"))
+NY = int(os.environ.get("DG_NY", "720"))
+NX = int(os.environ.get("DG_NX", "1440"))
+
+
+def build_cube_zarr(path: str) -> None:
+ # a synthetic "daily temperature" cube, chunked and written to Zarr
+ data = xr.DataArray(
+ np.random.default_rng(0).random((NT, NY, NX), dtype=np.float32) * 30,
+ dims=("time", "lat", "lon"),
+ coords={"time": np.arange(NT)},
+ name="t2m",
+ ).chunk({"time": 30, "lat": NY, "lon": NX})
+ data.to_dataset().to_zarr(path, mode="w")
+
+
+def main() -> None:
+ tmp = os.path.join(tempfile.gettempdir(), "daskgenie_cube.zarr")
+ print(f"writing a {NT}×{NY}×{NX} cube to {tmp} ...")
+ build_cube_zarr(tmp)
+
+ with dg.track() as source_map:
+ ds = xr.open_zarr(tmp)
+ clim = ds.t2m.mean("time") # climatology
+ anomaly = ds.t2m - clim # broadcast anomaly (the compute memory)
+ result = (anomaly**2).mean().data # a dask array scalar
+
+ with dg.LocalProfiler(
+ COLLECTOR,
+ run_name="xarray + Zarr anomaly",
+ source_map=source_map,
+ collection=result,
+ deep=True,
+ deep_epoch_seconds=2.0,
+ ) as prof:
+ print(f"run {prof.run_id}: computing the anomaly variance...")
+ val = result.compute(scheduler="threads", num_workers=4)
+ print(f"done — result={float(val):.4f}; open the dashboard run {prof.run_id!r}")
+
+
+if __name__ == "__main__":
+ main()
From 6aad1abd31a682a3624d68699b8029686c1052ca Mon Sep 17 00:00:00 2001
From: polymood <36277904+polymood@users.noreply.github.com>
Date: Fri, 3 Jul 2026 16:58:37 +0200
Subject: [PATCH 4/4] chore: release workflow, contributing guide, changelog,
pre-commit
- .github/workflows/workflow-pypi.yml: on a v* tag, build + publish to PyPI via
Trusted Publishing (OIDC, no token), push collector + dashboard images to GHCR,
and create the GitHub release.
- CONTRIBUTING.md: two-trunk model, Conventional Commits, uv/ruff/mypy/pytest,
release process.
- CHANGELOG.md (Keep a Changelog) and daskgenie.__version__.
- .pre-commit-config.yaml running ruff + mypy via uv.
---
.github/workflows/workflow-pypi.yml | 86 ++++++++
.pre-commit-config.yaml | 22 ++
CHANGELOG.md | 33 +++
CONTRIBUTING.md | 311 ++++++++++++++++++++++++++++
src/daskgenie/__init__.py | 3 +
5 files changed, 455 insertions(+)
create mode 100644 .github/workflows/workflow-pypi.yml
create mode 100644 .pre-commit-config.yaml
create mode 100644 CHANGELOG.md
create mode 100644 CONTRIBUTING.md
diff --git a/.github/workflows/workflow-pypi.yml b/.github/workflows/workflow-pypi.yml
new file mode 100644
index 0000000..eb09870
--- /dev/null
+++ b/.github/workflows/workflow-pypi.yml
@@ -0,0 +1,86 @@
+name: Publish
+
+# Cut a release by pushing a version tag (e.g. v0.1.0). This builds the Python
+# distributions and publishes them to PyPI with a Trusted Publisher (OIDC, no
+# stored token), builds and pushes the collector + dashboard images to GHCR, and
+# creates the GitHub release.
+on:
+ push:
+ tags: ["v*"]
+
+jobs:
+ build:
+ name: Build distributions
+ runs-on: ubuntu-latest
+ steps:
+ - uses: actions/checkout@v4
+ - name: Install uv
+ uses: astral-sh/setup-uv@v5
+ - name: Build sdist + wheel
+ run: uv build
+ - uses: actions/upload-artifact@v4
+ with:
+ name: dist
+ path: dist/
+
+ pypi:
+ name: Publish to PyPI
+ needs: build
+ runs-on: ubuntu-latest
+ # Trusted Publishing requires the id-token permission; no API token is stored.
+ permissions:
+ id-token: write
+ steps:
+ - uses: actions/download-artifact@v4
+ with:
+ name: dist
+ path: dist/
+ - name: Publish
+ uses: pypa/gh-action-pypi-publish@release/v1
+
+ images:
+ name: Build & push images (GHCR)
+ runs-on: ubuntu-latest
+ permissions:
+ contents: read
+ packages: write
+ steps:
+ - uses: actions/checkout@v4
+ - uses: docker/setup-buildx-action@v3
+ - name: Log in to GHCR
+ uses: docker/login-action@v3
+ with:
+ registry: ghcr.io
+ username: ${{ github.actor }}
+ password: ${{ secrets.GITHUB_TOKEN }}
+ - name: Collector image
+ uses: docker/build-push-action@v6
+ with:
+ context: .
+ file: ./Dockerfile
+ push: true
+ tags: |
+ ghcr.io/${{ github.repository_owner }}/daskgenie-collector:${{ github.ref_name }}
+ ghcr.io/${{ github.repository_owner }}/daskgenie-collector:latest
+ - name: Dashboard image
+ uses: docker/build-push-action@v6
+ with:
+ context: ./web
+ file: ./web/Dockerfile
+ push: true
+ tags: |
+ ghcr.io/${{ github.repository_owner }}/daskgenie-web:${{ github.ref_name }}
+ ghcr.io/${{ github.repository_owner }}/daskgenie-web:latest
+
+ release:
+ name: GitHub release
+ needs: [pypi, images]
+ runs-on: ubuntu-latest
+ permissions:
+ contents: write
+ steps:
+ - uses: actions/checkout@v4
+ - name: Create release
+ uses: softprops/action-gh-release@v2
+ with:
+ generate_release_notes: true
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
new file mode 100644
index 0000000..73d835b
--- /dev/null
+++ b/.pre-commit-config.yaml
@@ -0,0 +1,22 @@
+# Run `uv run pre-commit install` once, then these run on every commit.
+# Local/system hooks use the project's own pinned tool versions via uv, so
+# there's no drift between pre-commit and CI.
+repos:
+ - repo: local
+ hooks:
+ - id: ruff-format
+ name: ruff format
+ entry: uv run ruff format
+ language: system
+ types: [python]
+ - id: ruff
+ name: ruff check
+ entry: uv run ruff check --fix
+ language: system
+ types: [python]
+ - id: mypy
+ name: mypy
+ entry: uv run mypy src/
+ language: system
+ types: [python]
+ pass_filenames: false
diff --git a/CHANGELOG.md b/CHANGELOG.md
new file mode 100644
index 0000000..9013392
--- /dev/null
+++ b/CHANGELOG.md
@@ -0,0 +1,33 @@
+# Changelog
+
+All notable changes to this project are documented here. The format is based on
+[Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project
+adheres to [Semantic Versioning](https://semver.org/).
+
+## [Unreleased]
+
+## [0.1.0]
+
+### Added
+
+- **Deep memory profiling** with memray driven as a library: per-source-line
+ high-water-mark attribution and full call stacks for a per-worker flamegraph /
+ memray-style tree. Opt-in `deep=True`; degrades to lightweight sampling where
+ memray isn't available.
+- **Real-time dashboard** (Next.js) streaming over WebSocket: live Workers table,
+ global + per-worker task stream, whole-graph canvas DAG, memory-over-time with
+ a click-to-inspect spike explorer, per-layer allocations over time, and the
+ deep flamegraph. Collapsible sidebar, in-app modals, Dask warm palette + logo.
+- **TimescaleDB** backend (hypertables) as the default collector store behind a
+ `StoreProtocol`, with SQLite kept for local use and tests. Prometheus
+ `/metrics` retained.
+- **Worker-death post-mortem** joining suspect tasks with chunk metadata and the
+ allocation lines at the high-water mark.
+- **Origin tracking** — each run records the client hostname and IP.
+- **Examples** across Dask: distributed/deep OOM demos, a minutes-long pipeline,
+ a self-limiting crash, and one per collection type (`dask.delayed`,
+ `dask.dataframe`, `dask.bag`, xarray on Zarr and NetCDF).
+- Packaging (PyPI metadata, MIT license), CI, and release workflows.
+
+[Unreleased]: https://github.com/polymood/DaskGenie/compare/v0.1.0...HEAD
+[0.1.0]: https://github.com/polymood/DaskGenie/releases/tag/v0.1.0
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
new file mode 100644
index 0000000..81088c0
--- /dev/null
+++ b/CONTRIBUTING.md
@@ -0,0 +1,311 @@
+# Contributing to DaskGenie
+
+Thank you for your interest in contributing to DaskGenie, a memory profiler and
+live dashboard for Dask. This document describes how we work together on this
+project: how to set up your environment, how to write and commit changes, how
+branches are managed, and how releases are tagged. Please read it in full before
+opening your first pull request.
+
+The goal of these guidelines is to keep the codebase consistent, reviewable, and
+reliable. Following them helps maintainers review your work quickly and helps
+everyone trust that the `main` branch is always in a releasable state.
+
+## Table of Contents
+
+- [Code of Conduct](#code-of-conduct)
+- [Getting Started](#getting-started)
+- [Development Environment](#development-environment)
+- [Code Quality and Tooling](#code-quality-and-tooling)
+- [Branching Model](#branching-model)
+- [Commit Messages](#commit-messages)
+- [Pull Requests](#pull-requests)
+- [Testing](#testing)
+- [Versioning and Releases](#versioning-and-releases)
+- [Reporting Issues](#reporting-issues)
+
+## Code of Conduct
+
+We expect all contributors to be respectful and constructive in every
+interaction. Disagreement about technical decisions is welcome and healthy, but
+personal attacks, harassment, and dismissive behavior are not acceptable.
+Maintainers reserve the right to remove comments, commits, or contributors that
+violate this principle.
+
+## Getting Started
+
+1. Fork the repository on your account, or, if you have write access, create a
+ branch directly in the main repository.
+2. Clone your fork or the repository to your local machine.
+3. Set up the development environment as described below.
+4. Create a branch for your work following the branching model.
+5. Make your changes, ensuring all code quality checks pass.
+6. Open a pull request against the `develop` branch.
+
+## Development Environment
+
+DaskGenie has two parts: a Python package (`src/daskgenie`) and a Next.js
+dashboard (`web/`). The Python side targets Python 3.11 or newer and uses
+[uv](https://docs.astral.sh/uv/) for dependency management and virtual
+environments. The dashboard uses Node.js 20+.
+
+### Prerequisites
+
+- Python 3.11 or newer.
+- `uv` installed on your system.
+- Node.js 20 or newer (only for dashboard work).
+- Git, and Docker if you want to run the full stack locally.
+
+### Setting up (Python)
+
+Install the project together with its development and optional dependencies:
+
+```bash
+uv sync --group dev --extra collector --extra deep --extra examples
+```
+
+Prefix commands with `uv run`, for example:
+
+```bash
+uv run python -c "import daskgenie; print(daskgenie.__version__)"
+```
+
+Install the pre-commit hooks so that quality checks run automatically before
+every commit:
+
+```bash
+uv run pre-commit install
+```
+
+From this point on, the configured hooks run each time you create a commit. You
+can also run them against the whole codebase at any time:
+
+```bash
+uv run pre-commit run --all-files
+```
+
+### Setting up (dashboard)
+
+```bash
+cd web && npm install
+npm run dev # http://localhost:3000, proxies /api to the collector
+```
+
+Run the collector separately (`uv run python -m daskgenie.collector --port 8765`),
+or bring up the whole stack with `docker compose up -d --build`.
+
+## Code Quality and Tooling
+
+We hold the codebase to a consistent standard and enforce it automatically. A
+pull request will not be merged unless all of the following checks pass.
+
+### Formatting and Linting
+
+We use [Ruff](https://docs.astral.sh/ruff/) as both formatter and linter. The
+configuration lives in `pyproject.toml`.
+
+```bash
+uv run ruff format . # format in place
+uv run ruff check --fix . # lint and apply safe fixes
+```
+
+For the dashboard, `npm run build` must succeed (it type-checks and lints via
+Next.js/ESLint).
+
+### Static Type Checking
+
+All Python code must be fully type annotated. We use mypy in strict mode.
+
+```bash
+uv run mypy src/
+```
+
+Please do not silence type errors with `# type: ignore` unless it is genuinely
+unavoidable. When you do, add a specific error code and a short comment
+explaining why.
+
+### Typing Conventions
+
+- Prefer precise types over `Any`; reach for `Any` only when there is no
+ reasonable alternative, and document the reason.
+- Use the modern built-in generic syntax (`list[str]`, `dict[str, int]`).
+- Use `from __future__ import annotations` where it keeps annotations readable.
+
+### Running All Checks Locally
+
+```bash
+uv run pre-commit run --all-files
+uv run mypy src/
+uv run pytest -m "not integration"
+```
+
+## Branching Model
+
+This project uses a two-trunk model with long-lived `main` and `develop`
+branches.
+
+- `main` always reflects the latest released version. Every commit on `main`
+ corresponds to a tagged release. Nothing is committed to `main` directly except
+ release merges.
+- `develop` is the integration branch where completed work accumulates between
+ releases. It must always remain in a working, testable state.
+
+All day-to-day work happens on short-lived branches created from `develop` and
+merged back into `develop` through pull requests. A release is performed by
+merging `develop` into `main` and tagging that commit.
+
+### Branch Naming
+
+Use short, descriptive branch names that begin with a category prefix and use
+hyphens to separate words. Always branch from the current `develop`:
+
+- `feat/` for new functionality, e.g. `feat/per-key-residency`.
+- `fix/` for bug fixes, e.g. `fix/timescale-decimal-serialization`.
+- `docs/` for documentation-only changes.
+- `refactor/` for internal changes that do not alter behavior.
+- `test/` for changes that only add or adjust tests.
+- `chore/` for maintenance work such as dependency updates or tooling.
+
+When a branch addresses a tracked issue, include the issue number, e.g.
+`fix/142-oom-attribution`.
+
+```bash
+git switch develop
+git pull origin develop
+git switch -c feat/per-key-residency
+# work, commit, push, open a pull request into develop
+```
+
+### Hotfixes
+
+An urgent fix to the released version is the only exception to the rule that work
+starts from `develop`. A hotfix branch is created from `main`, named with the
+`fix/` prefix, merged into `main` as a patch release, and then merged back into
+`develop` so the fix is not lost.
+
+### Keeping Branches Current
+
+Keep your branch up to date with `develop`. We prefer rebasing over merging to
+keep the history linear:
+
+```bash
+git fetch origin
+git rebase origin/develop
+```
+
+Resolve conflicts locally, run the full set of checks again, and continue. Avoid
+merge commits inside feature branches.
+
+## Commit Messages
+
+We follow the [Conventional Commits](https://www.conventionalcommits.org/)
+specification.
+
+```
+():
+
+
+
+
+```
+
+The summary line is in the imperative mood, does not end with a period, and stays
+within roughly fifty characters. The body, when present, explains what changed
+and why rather than how, wrapped at seventy-two characters.
+
+Allowed types: `feat`, `fix`, `docs`, `style`, `refactor`, `perf`, `test`,
+`build`, `ci`, `chore`. A breaking change is indicated with a `!` after the type
+(`feat!:`) or a `BREAKING CHANGE:` footer describing the impact and migration.
+
+```
+feat(deepmem): fold memray stacks to the first user source line
+
+Allocations are now attributed to the caller's line rather than into
+numpy/dask internals, so the flamegraph reads on your own code.
+```
+
+## Pull Requests
+
+- Make sure your branch is rebased on the latest `develop` and that all checks
+ pass locally.
+- Push your branch and open a pull request against `develop`. The only pull
+ requests that target `main` are release merges and hotfixes.
+- Fill in the description: explain the motivation, summarize the changes, and
+ link related issues with a closing keyword such as `Closes #142`.
+- Keep pull requests focused on a single concern.
+- Be responsive to review feedback; push follow-up commits during review, and we
+ squash them on merge so the final history stays clean.
+
+### Review and Merge
+
+- At least one maintainer approval is required.
+- Continuous integration must be green.
+- We merge with the squash strategy so each pull request becomes a single commit
+ on `develop`. The squash commit message must follow Conventional Commits.
+
+## Testing
+
+We use pytest. New features must be accompanied by tests, and bug fixes should
+include a regression test that fails before the fix and passes after it.
+
+```bash
+uv run pytest -m "not integration" # fast, deterministic suite
+uv run pytest -m integration # spins up a real LocalCluster + collector
+```
+
+Tests marked `integration` start real clusters and are excluded from the default
+run so the standard suite stays fast. Aim for meaningful coverage of the behavior
+you add, including error paths (worker death, schema mismatch, memray-unavailable
+degradation).
+
+## Versioning and Releases
+
+This project follows [Semantic Versioning](https://semver.org/). While below
+`1.0.0` the public interface should be considered unstable, and minor versions
+may include breaking changes.
+
+### Cutting a Release
+
+Releases are cut by a maintainer by promoting `develop` to `main` and tagging the
+result:
+
+1. Ensure `develop` is green and contains all changes intended for the release.
+2. On `develop`, update the version in `pyproject.toml` and in
+ `src/daskgenie/__init__.py`, and move the entries under the **Unreleased**
+ heading of `CHANGELOG.md` into a new section for the version. Commit this.
+3. Open a pull request from `develop` into `main` and merge it (with a regular
+ merge commit, not squashed) once approved and green.
+4. Check out `main`, pull the merge, and create an annotated tag prefixed with
+ `v`:
+
+ ```bash
+ git switch main
+ git pull origin main
+ git tag -a v0.2.0 -m "Release 0.2.0"
+ git push origin v0.2.0
+ ```
+
+Pushing the tag triggers the publish workflow (`.github/workflows/workflow-pypi.yml`),
+which builds the distributions, publishes them to PyPI using a Trusted Publisher
+(so no API token is stored in the repository), builds and pushes the collector and
+dashboard images to GHCR, and creates the GitHub release. No manual release
+command is needed.
+
+Tags must always point to a commit on `main` and must never be moved or deleted
+once published. After the release, `main` is merged back into `develop` if the
+release introduced commits not already present there.
+
+## Reporting Issues
+
+If you find a bug or want to request a feature, please open an issue. A good
+report includes:
+
+- A clear and descriptive title.
+- The version of DaskGenie and of Python (and OS) you are using.
+- The steps required to reproduce the problem.
+- What you expected to happen and what actually happened.
+- Any relevant logs or tracebacks.
+
+If you believe you have found a security vulnerability, report it privately to
+the maintainers rather than opening a public issue.
+
+Thank you for helping make DaskGenie better.
diff --git a/src/daskgenie/__init__.py b/src/daskgenie/__init__.py
index 07f63c9..e243e7e 100644
--- a/src/daskgenie/__init__.py
+++ b/src/daskgenie/__init__.py
@@ -9,6 +9,8 @@
from daskgenie.local_profiler import LocalProfiler
from daskgenie.report import create_run, upload_graph
+__version__ = "0.1.0"
+
__all__ = [
"SourceLocation",
"GraphInfo",
@@ -19,4 +21,5 @@
"LocalProfiler",
"upload_graph",
"create_run",
+ "__version__",
]
- setSelected(node.id)}
- >
- (n.data as NodeData).color} />
-
-
+ {useCanvas ? (
+
+ setSelected(node.id)}
+ >
+ (n.data as NodeData).color} />
+
+
+ )}
{selected && (
No task spans yet.
;
+
+ const zoomed = view.current.lo > domain.lo + 1e-6 || view.current.hi < domain.hi - 1e-6;
+
+ return (
+
+
+ );
+}
diff --git a/web/components/WorkersView.tsx b/web/components/WorkersView.tsx
new file mode 100644
index 0000000..f7f97ec
--- /dev/null
+++ b/web/components/WorkersView.tsx
@@ -0,0 +1,87 @@
+"use client";
+
+import { useMemo } from "react";
+import { useLive } from "@/lib/live";
+import { bytes } from "@/lib/format";
+import type { WorkerStatus } from "@/lib/types";
+
+// A memory-pressure colour: green under half, amber past 70%, red past 90% of
+// the worker's limit — the same read the native Dask workers page gives you.
+function pressure(frac: number): string {
+ if (frac >= 0.9) return "var(--danger)";
+ if (frac >= 0.7) return "var(--warn)";
+ return "var(--ok)";
+}
+
+function Bar({ used, limit }: { used: number; limit: number }) {
+ const frac = limit > 0 ? Math.min(1, used / limit) : 0;
+ return (
+ 0 ? `${bytes(used)} / ${bytes(limit)}` : bytes(used)}>
+
+ {limit > 0 ? `${Math.round(frac * 100)}%` : bytes(used)}
+
+ );
+}
+
+export function WorkersView() {
+ const { statuses } = useLive();
+ const rows = useMemoNo worker heartbeats yet.
;
+
+ const totalRss = rows.reduce((a, w) => a + w.rss_bytes, 0);
+ const totalExec = rows.reduce((a, w) => a + w.executing, 0);
+
+ return (
+ <>
+
+
+
+
+
+ {rows.length}
+ Workers
+
+
+ {bytes(totalRss)}
+ Total RSS
+
+
+ {totalExec}
+ Tasks executing
+ | Worker | +Memory | +Managed | +CPU | +Threads | +Executing | +Ready | +
|---|---|---|---|---|---|---|
| + 0 ? "busy" : "idle"}`} /> {w.worker} + | +
+ |
+ {w.managed_bytes ? bytes(w.managed_bytes) : "—"} | +{w.cpu.toFixed(0)}% | +{w.nthreads || "—"} | +{w.executing} | +{w.ready} | +
+ 
+