diff --git a/cmux-tui/crates/cmux-tui-core/src/mux/tests/read_paths.rs b/cmux-tui/crates/cmux-tui-core/src/mux/tests/read_paths.rs index 39f81a7454e1..c43506eb9312 100644 --- a/cmux-tui/crates/cmux-tui-core/src/mux/tests/read_paths.rs +++ b/cmux-tui/crates/cmux-tui-core/src/mux/tests/read_paths.rs @@ -176,3 +176,85 @@ fn session_ping_reads_only_the_topology_frame() { .collect::>(); assert!(extra.is_empty(), "session.ping read {extra:?}"); } + +/// Tables a pane or tab list of one pane, screen or workspace has no use +/// for: terminal hosts, projections and the journal (a tab row joins its +/// terminal's resource row for the host id). +const NOT_A_TAB_TABLE: &[&str] = &[ + "terminal_hosts", + "resource_agent_projections", + "resource_effect_receipts", + "frontend_projections", + "session_journal", +]; + +/// CLI-1 (TYPING UNDER LOAD, STATUS 6237): `pane.list` and `tab.list` built +/// the whole session snapshot under the registry and state locks, then kept +/// one pane's rows; under the L1 CLI storm that held both locks 37 % of the +/// time. A scoped list reads only its workspace's rows, the same values. +#[test] +fn scoped_pane_and_tab_lists_read_only_their_workspace() { + let mux = test_mux(); + for _ in 0..2 { + let first = mux.new_workspace(None, Some((80, 24))).unwrap(); + let pane = mux.with_state(|state| state.pane_of(first.id).unwrap()); + mux.split(pane, crate::SplitDir::Right, Some((80, 24))).unwrap(); + mux.new_tab(Some(pane), None, Some((80, 24))).unwrap(); + } + let snapshot = resource(&mux, "session.snapshot", json!({})); + let rows = |collection: &str| snapshot[collection].as_array().unwrap().clone(); + let pane = rows("panes")[0]["id"].as_str().unwrap().to_string(); + let screen = rows("panes")[0]["screen_id"].as_str().unwrap().to_string(); + let workspace = rows("screens") + .into_iter() + .find(|value| value["id"] == screen.as_str()) + .unwrap()["workspace_id"] + .as_str() + .unwrap() + .to_string(); + let screens = rows("screens") + .into_iter() + .filter(|value| value["workspace_id"] == workspace.as_str()) + .map(|value| value["id"].as_str().unwrap().to_string()) + .collect::>(); + let panes = rows("panes") + .into_iter() + .filter(|value| screens.contains(value["screen_id"].as_str().unwrap())) + .collect::>(); + let pane_ids = panes + .iter() + .map(|value| value["id"].as_str().unwrap().to_string()) + .collect::>(); + let tabs_of = |keep: &dyn Fn(&str) -> bool| { + Value::Array( + rows("tabs") + .into_iter() + .filter(|value| keep(value["pane_id"].as_str().unwrap())) + .collect(), + ) + }; + let cases = [ + ("pane.list", json!({"workspace": workspace}), Value::Array(panes.clone())), + ("tab.list", json!({"pane": pane}), tabs_of(&|id| id == pane)), + ("tab.list", json!({"workspace": workspace}), tabs_of(&|id| pane_ids.contains(id))), + ]; + assert!(pane_ids.len() >= 2, "the workspace has two panes: {snapshot}"); + for (operation, scope, expected) in cases { + assert!(expected.as_array().is_some_and(|rows| !rows.is_empty()), "{operation} {scope}"); + let reads = crate::workspace_registry::WorkspaceRegistry::full_topology_reads_for_test(); + let mut listed = Value::Null; + let touched = touched(&mux, || listed = resource(&mux, operation, scope.clone())); + assert_eq!(listed, expected, "{operation} {scope} equals the snapshot's rows"); + assert_eq!( + crate::workspace_registry::WorkspaceRegistry::full_topology_reads_for_test(), + reads, + "{operation} {scope} read the whole topology" + ); + let extra = NOT_A_TAB_TABLE + .iter() + .filter(|table| touched.read.contains(**table)) + .collect::>(); + assert!(extra.is_empty(), "{operation} {scope} read {extra:?}"); + assert!(touched.written.is_empty(), "{operation} {scope} wrote {:?}", touched.written); + } +} diff --git a/cmux-tui/crates/cmux-tui-core/src/resource_api.rs b/cmux-tui/crates/cmux-tui-core/src/resource_api.rs index 5ce4b76dee97..dc9a8e9ea187 100644 --- a/cmux-tui/crates/cmux-tui-core/src/resource_api.rs +++ b/cmux-tui/crates/cmux-tui-core/src/resource_api.rs @@ -21,7 +21,8 @@ use serde_json::{Map, Value, json}; use crate::resource::{ ContentPublicId, FrontendProjectionPublicId, MachinePublicId, PanePublicId, ResourceError, - ResourceOperation, Selector, SessionPublicId, TabPublicId, TerminalPublicId, + ResourceOperation, ScreenPublicId, Selector, SessionPublicId, TabPublicId, TerminalPublicId, + WorkspacePublicId, }; use crate::resource_screen::{public_screen_value, tabs_by_pane}; use crate::sidebar_resource::{sidebar_snapshot, sidebar_view_id}; @@ -589,7 +590,6 @@ pub(crate) fn public_session_snapshot_with_journal_head( .iter() .map(|screen| (&screen.public_id, screen)) .collect::>(); - let active_screens = topology.active_screens.iter().cloned().collect::>(); let terminals_by_id = terminal_registry .terminals .iter() @@ -610,38 +610,7 @@ pub(crate) fn public_session_snapshot_with_journal_head( .map(|screen| public_screen_value(&topology, screen, &tabs_by_pane, &panes_by_id)) .collect::>>()?; - let panes = topology - .panes - .iter() - .map(|pane| { - let screen = screens_by_id - .get(&pane.screen_id) - .ok_or_else(|| anyhow::anyhow!("pane references a missing screen"))?; - let screen_focused = topology.active_workspace.as_ref() - == Some(&screen.workspace_id) - && active_screens.get(&screen.workspace_id).and_then(Option::as_ref) - == Some(&screen.public_id); - Ok(json!({ - "id": pane.public_id, - "screen_id": pane.screen_id, - "name": pane.name, - "focused": screen_focused && screen.active_pane == pane.public_id, - "zoomed": screen.zoomed_pane.as_ref() == Some(&pane.public_id), - })) - }) - .collect::>>()?; - - let tabs = topology - .tabs - .iter() - .map(|tab| { - let pane = panes_by_id - .get(&tab.pane_id) - .ok_or_else(|| anyhow::anyhow!("tab references a missing pane"))?; - checked_index(tab.position)?; - Ok(tab.public_value(pane.active_tab.as_ref() == Some(&tab.public_id))) - }) - .collect::>>()?; + let (panes, tabs) = public_pane_and_tab_values(&topology, &screens_by_id, &panes_by_id)?; let mut terminal_order = Vec::new(); let mut seen_terminals = HashSet::new(); @@ -888,6 +857,100 @@ pub(crate) fn public_workspaces(mux: &Mux) -> Result, ResourceError> .map_err(operation_failed) } +/// The public panes and tabs of `topology`, in its order: one builder for +/// the session snapshot and the scoped lists. +fn public_pane_and_tab_values( + topology: &crate::workspace_registry::ResourceTopologySnapshot, + screens_by_id: &HashMap<&ScreenPublicId, &crate::workspace_registry::RegistryScreen>, + panes_by_id: &HashMap<&PanePublicId, &crate::workspace_registry::RegistryPane>, +) -> anyhow::Result<(Vec, Vec)> { + let active_screens = topology.active_screens.iter().cloned().collect::>(); + let panes = topology + .panes + .iter() + .map(|pane| { + let screen = screens_by_id + .get(&pane.screen_id) + .ok_or_else(|| anyhow::anyhow!("pane references a missing screen"))?; + let screen_focused = topology.active_workspace.as_ref() == Some(&screen.workspace_id) + && active_screens.get(&screen.workspace_id).and_then(Option::as_ref) + == Some(&screen.public_id); + Ok(json!({ + "id": pane.public_id, + "screen_id": pane.screen_id, + "name": pane.name, + "focused": screen_focused && screen.active_pane == pane.public_id, + "zoomed": screen.zoomed_pane.as_ref() == Some(&pane.public_id), + })) + }) + .collect::>>()?; + let tabs = topology + .tabs + .iter() + .map(|tab| { + let pane = panes_by_id + .get(&tab.pane_id) + .ok_or_else(|| anyhow::anyhow!("tab references a missing pane"))?; + checked_index(tab.position)?; + Ok(tab.public_value(pane.active_tab.as_ref() == Some(&tab.public_id))) + }) + .collect::>>()?; + Ok((panes, tabs)) +} + +/// The session snapshot's `screens` (id and workspace only), `panes` and +/// `tabs` of one workspace, for a `pane.list` or `tab.list` scoped to a +/// workspace, screen or pane (CLI-1): the same values in the same order, +/// from the workspace's stored subtree. It reads no other workspace, no +/// terminals, projections or journal, and holds only the registry lock, not +/// state, so a list costs the workspace's size, not the session's. +pub(crate) fn public_workspace_topology( + mux: &Mux, + workspace: &WorkspacePublicId, +) -> Result { + let registry = mux.workspace_registry.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + // One connection hold for the whole read, as `public_workspaces`. + let _connection = registry.connection.get(); + (|| -> anyhow::Result { + let mut topology = registry.resource_topology_scoped(std::slice::from_ref(workspace))?; + // The full load's order (topology_load.rs): panes by screen id, then + // creation; tabs by pane id, then position. The scoped load walks + // screens by position, so sort to match. + topology.panes.sort_by(|left, right| { + (&left.screen_id, left.creation_ordinal, &left.public_id).cmp(&( + &right.screen_id, + right.creation_ordinal, + &right.public_id, + )) + }); + topology.tabs.sort_by(|left, right| { + (&left.pane_id, left.position).cmp(&(&right.pane_id, right.position)) + }); + let screens_by_id = topology + .screens + .iter() + .map(|screen| (&screen.public_id, screen)) + .collect::>(); + let panes_by_id = + topology.panes.iter().map(|pane| (&pane.public_id, pane)).collect::>(); + let (panes, mut tabs) = + public_pane_and_tab_values(&topology, &screens_by_id, &panes_by_id)?; + registry.read_state(|connection| { + for tab in &mut tabs { + crate::state::values::decorate_value(connection, "tab", tab)?; + } + Ok(()) + })?; + let screens = topology + .screens + .iter() + .map(|screen| json!({"id": screen.public_id, "workspace_id": screen.workspace_id})) + .collect::>(); + Ok(json!({"screens": screens, "panes": panes, "tabs": tabs})) + })() + .map_err(operation_failed) +} + /// The session snapshot's `cursor` alone (`session.ping`): the topology /// generation and resource revision, from the topology frame. pub(crate) fn public_session_cursor(mux: &Mux) -> Result { diff --git a/cmux-tui/crates/cmux-tui-core/src/resource_router/topology.rs b/cmux-tui/crates/cmux-tui-core/src/resource_router/topology.rs index 3c47bff8ee24..253a967bdd8e 100644 --- a/cmux-tui/crates/cmux-tui-core/src/resource_router/topology.rs +++ b/cmux-tui/crates/cmux-tui-core/src/resource_router/topology.rs @@ -128,7 +128,16 @@ fn list_resources( operation: &str, ) -> Result { let path = mux.resolve_resource_path(scope, selectors)?; - let snapshot = public_session_snapshot(mux)?; + // A pane or tab list inside one workspace reads that workspace's rows + // only, under the registry lock alone (CLI-1: the whole snapshot held + // both global locks for each CLI `tab list`); every other list reads + // the session snapshot. + let snapshot = match &path.workspace { + Some(workspace) if matches!(collection, "panes" | "tabs") => { + crate::resource_api::public_workspace_topology(mux, workspace)? + } + _ => public_session_snapshot(mux)?, + }; let path_index = SnapshotPathIndex::new(&snapshot); let values = snapshot[collection] .as_array() diff --git a/cmux-tui/scripts/bench-daemon-latency.py b/cmux-tui/scripts/bench-daemon-latency.py index 25bc1dbdcd51..35822c018d97 100755 --- a/cmux-tui/scripts/bench-daemon-latency.py +++ b/cmux-tui/scripts/bench-daemon-latency.py @@ -32,6 +32,18 @@ script, which passes only --scenarios (perf-10x-daemon.sh --arg=scenarios=agents-hooks). +Scenario "cli-storm" (not in the default list) is the TYPING UNDER LOAD L1 CLI storm against a +session the size of L5 HEAVY (40 workspaces, 200 terminals, 5% printing): 8 workers run the real +`cmux` CLI (one process per call, as scripts/journeys/tui-kit/rust/src/storm.rs does) in the +storm's weighted mix (workspace, pane, tab and terminal list; tab create and close, each worker +closing only its own tabs; terminal write; screen read; notify; workspace status, progress and +log) on one shared schedule at --storm-rate calls a second. It first times each CLI kind alone, +idle, at 1 workspace and at full size (the per-call cost and how it grows with the session), and +a raw socket probe sends the same list reads during the storm (daemon time without the CLI +process). It reports per kind p50/p95/max, the achieved rate, both daemon locks by site +(`server-stats` `locks`) and the topology projections (`resource_projection`: full against +scoped, read/index/diff time, changes per commit). + Scenario "resize-mirror" (not in the default list) is a terminal resized in bursts while a byte mirror is attached, the cost cmux-ad H10 names: a terminal with 10,000 lines of history, one connection attached to it as a @@ -511,6 +523,228 @@ def print_locks(name, st): f"n={site.get('acquisitions')} hold_total_us={site.get('hold_total_us')} " f"hold_max_us={site.get('hold_max_us')}", flush=True) + # -- the "cli-storm" scenario (TYPING UNDER LOAD CLI-1) + + STORM_MIX = (("workspace.list", 3), ("pane.list", 2), ("tab.list", 3), ("terminal.list", 2), + ("tab.create", 2), ("tab.close", 2), ("terminal.write", 4), ("screen.read", 4), + ("notify", 1), ("status.set", 3), ("progress.set", 3), ("log.append", 3)) + + def storm_cli(self, args): + """One `cmux` CLI call on this daemon: (ms, ok, stdout).""" + env = {"HOME": self.env["HOME"], "PATH": "/usr/bin:/bin:/usr/sbin:/sbin", "CMUX_QUIET": "1"} + t = time.perf_counter() + try: + p = subprocess.run([self.args.bin, "--socket", self.sock_path, "--json", *args], env=env, + stdin=subprocess.DEVNULL, capture_output=True, timeout=30) + ok, out = p.returncode == 0, p.stdout.decode(errors="replace") + if not ok: + out += p.stderr.decode(errors="replace") + except subprocess.TimeoutExpired: + ok, out = False, "timeout" + return (time.perf_counter() - t) * 1000, ok, out + + def storm_target(self): + _, ok, out = self.storm_cli(["workspace", "list"]) + ws = json.loads(out) + ws = (ws.get("result") if isinstance(ws, dict) else ws) or ws + workspace = ws[0]["id"] + _, ok, out = self.storm_cli(["pane", "list", "--workspace", workspace]) + panes = json.loads(out) + panes = (panes.get("result") if isinstance(panes, dict) else panes) or panes + pane = panes[0]["id"] + _, ok, out = self.storm_cli(["terminal", "list", "--pane", pane]) + terms = json.loads(out) + terms = (terms.get("result") if isinstance(terms, dict) else terms) or terms + return workspace, pane, terms[0]["id"] + + def storm_argv(self, kind, target, worker, n, close): + ws, pane, term = target + return { + "workspace.list": ["workspace", "list"], + "pane.list": ["pane", "list", "--workspace", ws], + "tab.list": ["tab", "list", "--pane", pane], + "terminal.list": ["terminal", "list", "--pane", pane], + "tab.create": ["tab", "create", "terminal", "--workspace", ws, "--pane", pane, + "--name", f"storm-{worker}-{n}"], + "tab.close": ["tab", close, "close"] if close else None, + "terminal.write": ["terminal", term, "write", "--text", f"echo storm {worker} {n}\n"], + "screen.read": ["terminal", term, "screen", "read"], + "notify": ["notify", "--title", f"bench {n}", "--body", f"worker {worker}", "--workspace", ws, + "--desktop", "false"], + "status.set": ["workspace", ws, "status", "set", f"storm{worker}", f"call {n}"], + "progress.set": ["workspace", ws, "progress", "set", f"{(n % 100) / 100:.2f}", "--label", "storm"], + "log.append": ["workspace", ws, "log", "append", f"storm {worker} call {n}", "--level", "info", + "--source", "bench"], + }[kind] + + @staticmethod + def tab_id_of(out): + try: + value = json.loads(out) + except ValueError: + return None + stack = [value] + while stack: + v = stack.pop() + if isinstance(v, dict): + if isinstance(v.get("tab_id"), str): + return v["tab_id"] + stack.extend(v.values()) + elif isinstance(v, list): + stack.extend(v) + return None + + def storm_idle(self, target, label, reps=15): + """Each kind alone, back to back: the per-call cost with no contention.""" + out = {} + for kind, _ in self.STORM_MIX: + vals = [] + for i in range(reps): + close = None + if kind == "tab.close": + _, ok, made = self.storm_cli(self.storm_argv("tab.create", target, 99, i, None)) + close = self.tab_id_of(made) if ok else None + if not close: + continue + ms, ok, text = self.storm_cli(self.storm_argv(kind, target, 99, i, close)) + if not ok: + log(label, kind, "failed", text[-300:]) + continue + vals.append(ms) + if kind == "tab.create": + made = self.tab_id_of(text) + if made: + self.storm_cli(["tab", made, "close"]) + out[f"{label}-{kind}"] = stats(vals) + t = [] + for _ in range(reps): + t0 = time.perf_counter() + subprocess.run([self.args.bin, "--version"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + timeout=30) + t.append((time.perf_counter() - t0) * 1000) + out[f"{label}-cli-version-no-socket"] = stats(t) + return out + + def cli_storm(self, name="cli-storm"): + log("scenario", name) + res = {} + c = Conn(self.sock_path, timeout=300) + must(c.call("new-workspace", name="storm", cols=80, rows=24), "new-workspace") + c.close() + target = self.storm_target() + res.update(self.storm_idle(target, "idle-1ws")) + self.grow(40, 200, 0.05) + time.sleep(self.args.settle) + res.update(self.storm_idle(target, "idle-40ws")) + c = Conn(self.sock_path, timeout=300) + before = c.call("server-stats", include=["locks", "write_path", "resource_projection"]).get("data") or {} + rate, workers, seconds = self.args.storm_rate, 8, self.args.load_seconds + weights = [k for k, w in self.STORM_MIX for _ in range(w)] + lat = {k: [] for k, _ in self.STORM_MIX} + raw = {"raw-workspace.list": [], "raw-tab.list": []} + failed = {k: 0 for k in lat} + errors = [] + lock = threading.Lock() + counter = [0] + start = time.monotonic() + 0.5 + end = start + seconds + stop = threading.Event() + + def worker(index): + import random + rng = random.Random(7 + index) + mine = [] + while True: + with lock: + n = counter[0] + counter[0] += 1 + due = start + n / rate + if due >= end: + break + delay = due - time.monotonic() + if delay > 0: + time.sleep(delay) + kind = rng.choice(weights) + if kind == "tab.close" and not mine: + kind = "tab.create" + close = mine.pop(0) if kind == "tab.close" else None + ms, ok, text = self.storm_cli(self.storm_argv(kind, target, index, n, close)) + with lock: + lat[kind].append(ms) + if not ok: + failed[kind] += 1 + if len(errors) < 10: + errors.append(f"{kind}: {text[-200:]}") + if ok and kind == "tab.create": + made = self.tab_id_of(text) + if made: + mine.append(made) + for tab in mine: + self.storm_cli(["tab", tab, "close"]) + + def probe(): + conn = Conn(self.sock_path, timeout=300) + ws, pane, _ = target + k = 0 + while not stop.is_set(): + k += 1 + for op, params in (("workspace.list", {}), ("tab.list", {"pane": pane})): + rid = f"probe-{k}-{op}" + request = {"protocol": "cmux.protocol/2", "type": "request", "id": rid, "operation": op, + "params": {"machine": "current", "session": "current", **params}} + t = time.perf_counter() + conn.sock.sendall((json.dumps(request) + "\n").encode()) + while True: + msg = conn.read_line() + if msg.get("id") == rid and msg.get("type") == "response": + break + with lock: + raw[f"raw-{op}"].append((time.perf_counter() - t) * 1000) + stop.wait(0.1) + conn.close() + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(workers)] + prober = threading.Thread(target=probe) + prober.start() + for t in threads: + t.start() + for t in threads: + t.join() + stop.set() + prober.join() + after = c.call("server-stats", include=["locks", "write_path", "resource_projection"]).get("data") or {} + c.close() + calls = sum(len(v) for v in lat.values()) + all_ms = [v for vals in lat.values() for v in vals] + for kind, vals in lat.items(): + res[f"storm-{kind}"] = {**stats(vals), "failed": failed[kind]} + res["storm-all"] = stats(all_ms) + for k, vals in raw.items(): + res[f"storm-{k}"] = stats(vals) + res["storm_rate"] = {"asked_per_s": rate, "ok_per_s": round(calls / seconds, 2), "calls": calls, + "failed": sum(failed.values())} + res["errors"] = errors + res["server_stats_before"] = before + res["server_stats_after"] = after + self.result["scenarios"][name] = res + for metric, s in res.items(): + if isinstance(s, dict) and "n" in s: + print(f"PERF-DAEMON {name} {metric} n={s['n']} p50={s.get('p50_ms')} p95={s.get('p95_ms')} " + f"p99={s.get('p99_ms')} max={s.get('max_ms')} mean={s.get('mean_ms')}" + + (f" failed={s['failed']}" if "failed" in s else ""), flush=True) + print(f"PERF-DAEMON {name} rate {json.dumps(res['storm_rate'], sort_keys=True)}", flush=True) + if errors: + print(f"PERF-DAEMON {name} errors {errors}", flush=True) + self.print_locks(name, after) + rp = after.get("resource_projection") or {} + rp0 = before.get("resource_projection") or {} + delta = {k: (rp.get(k) or 0) - (rp0.get(k) or 0) for k in + ("projections", "full_projections", "scoped_projections", "scope_fallbacks", "commits")} + hist = {k: {q: (rp.get(k) or {}).get(q) for q in ("p50", "p99", "max")} + for k in ("read_us", "index_us", "diff_us", "projected_changes", "commit_us", "written_changes")} + print(f"PERF-DAEMON-PROJECTION {name} during_storm={json.dumps(delta, sort_keys=True)} " + f"since_start={json.dumps(hist, sort_keys=True)}", flush=True) + def resize_mirror(self, name="resize-mirror"): """Resize bursts of a terminal with deep history and one byte mirror.""" steps, bursts = 24, 5 @@ -663,6 +897,8 @@ def main(): ap.add_argument("--load-seconds", type=float, default=60.0) ap.add_argument("--agent-cli", type=int, choices=(0, 1), default=1, help="1: each fake agent also runs `cmux workspace list` once per turn; 0: hooks only") + ap.add_argument("--storm-rate", type=float, default=120.0, + help="cli-storm: calls a second asked of the 8 CLI workers in total") ap.add_argument("--resize-rounds", type=int, default=4, help="resize-mirror: how many times to run its 5 bursts per case") args = ap.parse_args() @@ -679,6 +915,8 @@ def main(): b.agents() if "resize-mirror" in wanted: b.resize_mirror() + if "cli-storm" in wanted: + b.cli_storm() if "agents-hooks" in wanted: # The hooks only: no shipped agent hook calls the CLI per turn. args.agent_cli = 0