Compare commits

...
Author SHA1 Message Date
Alex Cheema d1f4b24d5e fix(profilers): use local RDMA interfaces in probe matrix 2026-05-03 17:49:41 +01:00
Alex CheemaandClaude Opus 4.7 c468fd5c14 TopologyGraph: position edge labels on outer side, drop ↑↓ glyphs
Two changes that fix the label-pile-up at the cluster centroid and
make labels actually adjacent to their arrows:

- Push labels AWAY from the viewport centroid (outer side of each
  edge), not toward it. Previously every edge in a 4-node diamond
  ended up with its labels piled near the centroid because
  `towardCenter` pointed there. For diagonals whose midpoint *is*
  the centroid we pick a stable side based on edge direction.

- Latency also halved in the tooltip (jitter/2 alongside RTT/2),
  for consistency with the topology edge label.

End result: each perimeter edge has its A→B bandwidth, latency,
and B→A bandwidth strung along the outer side, each adjacent to
its arrow head. Crossing diagonals' labels form a "+" pattern at
the center instead of overlapping each other.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 15:21:18 +01:00
Alex CheemaandClaude Opus 4.7 d366e035ae dashboard: drop GPU rich/poor bar, split edge labels per direction, RTT/2
GpuRichBar simplified to "ClusterStats": dropped the gradient bar +
scoring and kept just the three aggregate tiles (FP16 compute,
memory bandwidth, total memory). Centered, inline.

Edge labels reworked: instead of one combined "↑X ↓Y · Z RTT" label
per pair (which collided on tight layouts and forced ↑↓ glyphs to
disambiguate direction), now we render up to three labels per edge:

- Per-direction bandwidth — placed next to its arrow head, on the
  matching side of the midpoint. The arrow direction implies which
  way the number applies, so no glyphs needed.
- Latency centered at midpoint, on the *other* side of the edge so
  the eye doesn't have to disambiguate it from the bandwidth labels.

Latency display also switched from RTT to RTT/2 (one-way
approximation) — the topology edge label and the tooltip both show
RTT/2 now, with the column header explicitly labeled "RTT/2" so the
semantic is clear.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 15:01:28 +01:00
Alex CheemaandClaude Opus 4.7 1fafdf99d0 profilers: measure link jitter alongside RTT (RFC 3550 / iperf3 style)
Added `latency_jitter_ms` to both SocketLinkProfile and
RDMALinkProfile. Defined as the mean of |Δ| between consecutive
RTT samples — same convention iperf3 reports, captures short-term
variance better than stddev.

- Socket: bumped LATENCY_SAMPLES from 5 → 10 so the mean-of-deltas
  is meaningful (4 deltas was thin).
- RDMA: 50 samples already, just compute the deltas alongside the
  median.
- State + apply: plumb `latency_jitter_ms` through. NodeSocketLinkProfile
  defaults to 0.0; NodeRdmaLinkProfile to None (matches the rest of
  its optional fields).

Dashboard: new "Jitter" column in the hover tooltip. Edge label
left alone — keeping it short.

Tooltip styling fixes pulled in along the way:
- `position: fixed` so it can escape `overflow: hidden` on the
  topology container — multi-row tooltips were getting clipped at
  the box edge.
- `white-space: nowrap` on table cells; bandwidth numbers were
  wrapping onto two lines and overlapping. Slightly wider per-cell
  padding for breathing room.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 12:25:16 +01:00
Alex CheemaandClaude Opus 4.7 e0dc64bee9 apply: dedup link profiles per (transport, identity) not just transport
Previously the apply layer's link-profile dedup keyed on transport
alone, so a peer reachable on multiple IPs (LAN + Tailscale +
link-local + ...) would have all of its socket profiles collapsed
into one slot — and the reconciler probing each IP in turn would
overwrite the previous, making the displayed bandwidth and
classification bounce ("Ethernet 1.1 Gbps" → "Unknown 400 Mbps" →
"Ethernet 1.1 Gbps" → ...).

Switch the dedup to the natural identity per transport:
- socket: (transport, sink_ip)  — one row per IP
- rdma:   (transport, source_iface, sink_iface)

Each connection now gets its own stable row. The dashboard's edge
label still shows max-up / max-down / min-RTT across all profiles,
so the summary is the best path; the hover tooltip shows the full
breakdown per connection.

Verified live on the 4-node M3 Ultra cluster: james -> s14 has 4
distinct socket rows (link-local, LAN, Tailscale, and a slow path)
plus the RDMA row, all stable across 5+ minutes of probes.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 12:06:39 +01:00
Alex CheemaandClaude Opus 4.7 517fb06ce0 connection-type: sticky cache to stop Ethernet ↔ Unknown flapping
The dashboard re-derives the connection type on every reactive
update. The source data — `nodeNetwork[peer].interfaces` — is
re-parsed on the backend every 10 s from `networksetup` and
occasionally drops an entry for one tick before the next refresh
puts it back. Without this fix the user sees the label flicker
between "Ethernet" and "Unknown" several times a minute.

Solution: cache the last *concrete* (non-"Unknown") classification
per (sinkNodeId, sinkIp). When a fresh lookup returns "Unknown"
we ignore it in favour of the cached answer. Concrete answers
always update the cache, so a real network change propagates
immediately.

Cache is bounded by O(N²) for N nodes (one entry per directed
edge × IP), so no leak.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 11:55:18 +01:00
Alex CheemaandClaude Opus 4.7 2124b31ab0 gpu_profiler: use mx.zeros for bandwidth buffer to avoid fp32 temp
`mx.random.uniform(shape=..., dtype=float16)` internally generates
fp32 then casts to fp16, which doubles the peak Metal allocation —
our 2 GiB fp16 buffer briefly needs 4 GiB of heap. CI's macOS
runner has max_buffer_size = 3.5 GiB and rejected the alloc:

    RuntimeError: [metal::malloc] Attempting to allocate 4294967296
    bytes which is greater than the maximum allowed buffer size of
    3758096384 bytes.

DRAM bandwidth is independent of the values being streamed, so we
just allocate `mx.zeros` directly. No fp32 temp, peak heap = exactly
the 2 GiB we want.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 01:30:51 +01:00
Alex CheemaandClaude Opus 4.7 e5760dd50b TopologyGraph: drop "RTT" suffix from edge label, keep it in tooltip
The edge metrics label was "↑X ↓Y · Z µs RTT" — the "RTT" suffix
was redundant once the unit was already there and made the label
crowded. Keep the "RTT" column header in the hover tooltip where
the explicit semantic still helps readers parse the table.

Also picks up minor nix-fmt comment-spacing tweaks across the
profiler modules.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 01:23:02 +01:00
Alex CheemaandClaude Opus 4.7 ef97ef7088 GpuRichBar: replace draggable-looking thumb with a tick marker
The white circle reads as a slider thumb that the user can grab and
drag, but the bar is a passive readout. Apple HIG distinguishes
controls (slider with prominent thumb) from indicators (gauge, with
a tick or filled portion). For a "where do you fall on this scale"
display, the indicator pattern is the right one.

Replace the circular thumb with:
- A thin 2px vertical tick line through the track at the value, with
  a dark halo so it stays legible across the red->green gradient.
- The gradient past the marker is dimmed so the eye lands on the
  position instead of perceiving a static spectrum with a tick on it.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 01:21:18 +01:00
Alex CheemaandClaude Opus 4.7 283d8d4573 profilers: drop the per-node single-flight RDMA probe lock
The lock was a defensive guess that simultaneous jaccl/RDMA probes
would fight for the Thunderbolt RDMA hardware. They don't —
concurrent processes get independent QPs. Verified end-to-end on a
2x M3 Ultra cluster: both nodes start probes simultaneously at
discovery and both complete cleanly with sensible numbers.

Keep the `state.runners` gate — that one is still real (don't compete
with active inference traffic).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 01:21:18 +01:00
Alex Cheema 871414eae8 chore: remove temporary PR screenshot files 2026-05-01 01:21:18 +01:00
Alex Cheema 7452a943e5 temp: add screenshots for PR description 2026-05-01 01:21:18 +01:00
Alex CheemaandClaude Opus 4.7 15f1e4a3eb feat(profilers): GPU + link profiling with directional measurements
Adds an active-probe profiler subsystem alongside the existing info
gatherer. Each node measures:

- GPU FP16 TFLOPS (square FP16 matmul) and memory bandwidth (large
  `mx.sum` streaming read). Long warm-up + best-of-N timing keeps two
  M3 Ultras agreeing to within ~2%.
- Per-edge socket bandwidth (upload + download separately, server
  times the receive on uploads) and RTT, via new `/profile/echo`,
  `/profile/upload`, and `/profile/download` endpoints.
- Per-edge RDMA bandwidth (rank0->rank1 and rank1->rank0 via
  `mx.distributed.send`/`recv`) and RTT (tiny-payload `all_sum`
  ping-pong), in a child process so jaccl init doesn't poison the
  worker process.

Scheduling is a state-driven reconciliation loop (15s tick) so the
event log can be replayed without re-firing probes. TTLs: GPU 1h,
socket 5m, RDMA 6h. GPU and RDMA probes skip while runners are active;
all RDMA work serialised through a process-global lock.

Dashboard:
- New GpuRichBar component at the top of the topology view: gradient
  thumb scored as a weighted blend of TFLOPS / bandwidth / memory
  against an 8x H100 anchor (TFLOPS heaviest, memory least), with a
  per-dimension cap so a single insanely high axis still pulls the
  thumb right.
- Per-node TFLOPS / memory bandwidth labels under each device.
- Edge label shows max upload / max download / min RTT across all
  profiles for the pair.
- Hovering an edge shows a breakdown table: direction, inferred
  connection type (Wi-Fi / Ethernet / Thunderbolt N / RDMA),
  upload / download / RTT per profile.
- Connection-type inference reads `nodeNetwork.interfaceType` for
  socket edges and parses `nodeThunderbolt.linkSpeed` (e.g. "Up to
  80 Gb/s") to distinguish TB4 from TB5 on RDMA edges.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 01:21:18 +01:00
ciaranbor 8dae3ecb9a A few targeted tweaks to address HF rate limits (#2009)
## Motivation

- exo bursts ~200 HF Hub-API requests on every cold start, blowing past
the anonymous 500-req/5-min budget.
- The existing retry loop catches 429 generically and gives up in ~3s —
well before HF's reset window.
- `file_meta` and `_download_file` had no 429 handling at all (became
`AssertionError`).
- Disk file-list cache was bypassed on every process restart.

## Changes

All in `src/exo/download/download_utils.py` + tests.

- Parse `t=` from HF's `RateLimit` header on 429; sleep `min(t, 300s) +
jitter`.
- Handle 429 at all three call sites (`_fetch_file_list`, `file_meta`,
`_download_file`).
- `n_attempts`: 3 → 5.
- Disk cache now primary across restarts (24h mtime TTL).
- `?recursive=true` instead of N+1 subdir walks.

## Why It Works

`t=<seconds>` is HF's "wait this long and you'll be unblocked" —
sleeping that long lets the window reset. Disk-cache-as-primary plus
recursive listing cuts cold-start Hub-API traffic by ~10×.

## Test Plan

### Manual Testing

MacBook Pro M1 Max. Tripped the real HF 429. Pre-fix: failed in 3.4s.
Post-fix: slept (HF returned `t=158`) and recovered.

### Automated Testing

- New `test_rate_limit_handling.py` (19 tests) — header parsing,
retry-loop behaviour, plus HTTP-level coverage that mocks aiohttp to
return a 429 and asserts each call site raises
`HuggingFaceRateLimitError(retry_after=52.0)`.
- New `TestFileListCacheTTL` in `test_offline_mode.py` — fresh cache
hits, stale cache refetches.
- 421 tests pass; basedpyright / ruff / nix fmt clean.
2026-04-30 18:06:15 +00:00
Alex CheemaandClaude Opus 4.7 fb12b403ea fix(app): tighten Share Bug Report prompt layout (#2008)
## Summary

Follow-ups to #2003 based on feedback that the Share Bug Report window
felt visually weighty: too much padding above and below, and a
description editor that invited an essay rather than a one-liner.

## Changes (one file)

`app/EXO/EXO/Views/BugReportWindowController.swift`:

- **Auto-size the window to its content.** Switched from `NSHostingView`
+ fixed `contentRect: 480x380` + SwiftUI `frame(minHeight: 320)` to
`NSHostingController` with `sizingOptions = [.preferredContentSize,
.minSize]`. The fixed-min combo was centering the form in dead vertical
space.
- **Smaller, lower-pressure editor.** Field is now labeled `Description
(optional)` with a placeholder hint (`What were you doing when it
broke?`) inside the editor. Editor height fixed at 72pt (was 120pt min).
Replaced the long lead-in paragraph and headline with a single one-line
caption between field and buttons: `Diagnostic logs will be uploaded
with your report.`
- **Tighter spacing.** Outer padding 20 -> 16, root spacing 16 -> 12,
prompting-section spacing 12 -> 8.
- **Remove em dash from copy.**

`BugReportService` and the menu wiring are unchanged.

## Test plan

- [ ] Click `Share Bug Report...` from the menu bar.
- [ ] The window opens centered and sized to its content (no big empty
bands top/bottom).
- [ ] Description editor is visibly compact, with the placeholder hint
showing when empty.
- [ ] The optional-ness is conveyed by the field label (no separate help
paragraph).
- [ ] Caption `Diagnostic logs will be uploaded with your report.`
appears in `.caption` style under the editor, above the buttons.
- [ ] Resize the window: persists across re-opens (frame autosave still
works).
- [ ] Send/Cancel/Try Again/Done flows behave the same as before.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-30 15:10:26 +01:00
Alex CheemaandClaude Opus 4.7 1606e63816 feat(app): open Share Bug Report in a dedicated window (#2003)
## Summary

- Adds a top-level **Share Bug Report…** menu item to the macOS popover
(between *Check for Updates* and *Quit*) with SF Symbol `ladybug`.
- Clicking it opens a dedicated resizable `NSWindow` ("Send a Bug
Report") that hosts the prompting / sending / success / failure flow.
- Removes the description-less duplicate from Settings → Debug Info, and
the dead `debugSection` it nominally lived behind.

## Why

PR #1959 added a user-description prompt to the bug-report flow, but its
trigger lived inside `ContentView.debugSection` — a view that's defined
but never rendered in the body. The path users actually hit was
`SettingsView.sendBugReportButton`, which called
`BugReportService.sendReport(isManual: true)` without ever passing
`userDescription`. So the description prompt was unreachable in the
built app.

## Approach

Per Apple HIG, an action that requires further input before completing
should open a dialog, not transform the menu inline. So:

- Add a top-level menu entry that ends in `…` (HIG: ellipsis indicates
"further input required").
- Move the prompting/sending/success/failure state machine into a
standalone `BugReportWindowController` modeled after the existing
`SettingsWindowController`.
- Single-instance window with frame-autosave name, sensible
`contentMinSize`, resizable, native button layout (`.cancelAction` /
`.defaultAction` keyboard shortcuts), light/dark-mode-correct
`.textBackgroundColor` and `.separatorColor`.
- Auto-focus the description field on open. `Try Again` from failure,
`Open GitHub Issue` + `Done` from success.

## Files

- `app/EXO/EXO/Views/BugReportWindowController.swift` (new) — controller
+ view.
- `app/EXO/EXO/EXOApp.swift` — wire `BugReportWindowController` as a
`@StateObject` and inject as environment object.
- `app/EXO/EXO/ContentView.swift` — replace inline state machine with
menu item that calls `bugReportWindowController.open()`. Remove
now-unused state, helpers, and dead `debugSection`.
- `app/EXO/EXO/Views/SettingsView.swift` — remove duplicate
`sendBugReportButton`, `sendBugReport()`, and related `@State`. Section
"Debug Info" keeps Thunderbolt / interface / RDMA info.

`BugReportService` is unchanged.

## Test plan

- [ ] Open the menu-bar popover → confirm **Share Bug Report…** appears
between *Check for Updates* and *Quit*, with a ladybug icon.
- [ ] Click it → a window titled "Send a Bug Report" appears, centered,
with the description editor focused.
- [ ] Resize the window → size persists across re-opens (frame
autosave).
- [ ] Type a description, press Return → upload succeeds, success card
with **Open GitHub Issue** + **Done** appears.
- [ ] Click **Open GitHub Issue** → browser opens with the description
pre-filled into the issue template.
- [ ] Send with empty description → upload still succeeds.
- [ ] Press Esc from the prompting state → window closes.
- [ ] On failure (e.g., offline) → error card with **Try Again** +
**Close** appears; Try Again returns to the editor with the description
preserved.
- [ ] Open the Settings window → Debug Info section is unchanged except
the Send Bug Report button is gone.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-30 13:05:07 +01:00
30 changed files with 3454 additions and 303 deletions

No files matched your search

+8 -227
View File
@@ -16,22 +16,13 @@ struct ContentView: View {
@EnvironmentObject private var updater: SparkleUpdater
@EnvironmentObject private var thunderboltBridgeService: ThunderboltBridgeService
@EnvironmentObject private var settingsWindowController: SettingsWindowController
@EnvironmentObject private var bugReportWindowController: BugReportWindowController
@State private var focusedNode: NodeViewModel?
@State private var deletingInstanceIDs: Set<String> = []
@State private var showAllNodes = false
@State private var showAllInstances = false
@State private var baseURLCopied = false
@State private var showAdvanced = false
@State private var showDebugInfo = false
private enum BugReportPhase: Equatable {
case idle
case prompting
case sending(String)
case success(String)
case failure(String)
}
@State private var bugReportPhase: BugReportPhase = .idle
@State private var bugReportUserDescription: String = ""
@State private var uninstallInProgress = false
@State private var pendingNamespace: String = ""
@State private var pendingHFToken: String = ""
@@ -294,6 +285,13 @@ struct ContentView: View {
) {
updater.checkForUpdates()
}
HoverButton(
title: "Share Bug Report…",
tint: .primary,
trailingSystemImage: "ladybug"
) {
bugReportWindowController.open()
}
.padding(.bottom, 8)
HoverButton(title: "Quit", tint: .secondary) {
controller.stop()
@@ -477,40 +475,6 @@ struct ContentView: View {
}
}
private var debugSection: some View {
VStack(alignment: .leading, spacing: 4) {
HoverButton(
title: "Debug Info",
tint: .primary,
trailingSystemImage: showDebugInfo ? "chevron.up" : "chevron.down",
small: true
) {
showDebugInfo.toggle()
}
if showDebugInfo {
VStack(alignment: .leading, spacing: 4) {
Text("Version: \(buildTag)")
.font(.caption2)
.foregroundColor(.secondary)
Text("Commit: \(buildCommit)")
.font(.caption2)
.foregroundColor(.secondary)
Text(thunderboltStatusText)
.font(.caption2)
.foregroundColor(thunderboltStatusColor)
clusterThunderboltBridgeView
interfaceIpList
rdmaStatusView
sendBugReportButton
.padding(.top, 6)
}
.padding(.leading, 8)
.transition(.opacity)
}
}
.animation(.easeInOut(duration: 0.25), value: showDebugInfo)
}
private var rdmaStatusView: some View {
let rdmaStatuses = stateService.latestSnapshot?.nodeRdmaCtl ?? [:]
let localNodeId = stateService.localNodeId
@@ -559,127 +523,6 @@ struct ContentView: View {
}
}
private var sendBugReportButton: some View {
VStack(alignment: .leading, spacing: 6) {
switch bugReportPhase {
case .idle:
Button {
bugReportPhase = .prompting
bugReportUserDescription = ""
} label: {
HStack {
Text("Send Bug Report")
.font(.caption)
.fontWeight(.semibold)
Spacer()
}
.padding(.vertical, 6)
.padding(.horizontal, 8)
.background(
RoundedRectangle(cornerRadius: 6)
.fill(Color.accentColor.opacity(0.12))
)
}
.buttonStyle(.plain)
case .prompting:
VStack(alignment: .leading, spacing: 6) {
VStack(alignment: .leading, spacing: 2) {
Text("Tell us what went wrong (optional)")
.font(.caption2)
.foregroundColor(.secondary)
Text(
"A quick description of what you were doing and what happened helps us track down the bug for you."
)
.font(.caption2)
.foregroundColor(.secondary)
.opacity(0.8)
.fixedSize(horizontal: false, vertical: true)
}
TextEditor(text: $bugReportUserDescription)
.font(.caption2)
.frame(height: 60)
.overlay(
RoundedRectangle(cornerRadius: 4)
.stroke(Color.secondary.opacity(0.3), lineWidth: 1)
)
HStack(spacing: 8) {
Button("Send") {
Task {
await sendBugReport()
}
}
.font(.caption2)
.buttonStyle(.borderedProminent)
.controlSize(.small)
Button("Cancel") {
bugReportPhase = .idle
}
.font(.caption2)
.buttonStyle(.bordered)
.controlSize(.small)
}
}
.padding(8)
.background(
RoundedRectangle(cornerRadius: 6)
.fill(Color.accentColor.opacity(0.06))
)
case .sending(let message):
HStack(spacing: 6) {
ProgressView()
.scaleEffect(0.6)
Text(message)
.font(.caption2)
.foregroundColor(.secondary)
}
case .success(let message):
VStack(alignment: .leading, spacing: 6) {
Text(message)
.font(.caption2)
.foregroundColor(.secondary)
.fixedSize(horizontal: false, vertical: true)
Button {
openGitHubIssue()
} label: {
HStack(spacing: 4) {
Image(systemName: "arrow.up.right.square")
.imageScale(.small)
Text("Create GitHub Issue")
.font(.caption2)
}
}
.buttonStyle(.bordered)
.controlSize(.small)
Button("Done") {
bugReportPhase = .idle
bugReportUserDescription = ""
}
.font(.caption2)
.buttonStyle(.plain)
.foregroundColor(.secondary)
}
case .failure(let message):
VStack(alignment: .leading, spacing: 4) {
Text(message)
.font(.caption2)
.foregroundColor(.red)
.fixedSize(horizontal: false, vertical: true)
Button("Dismiss") {
bugReportPhase = .idle
}
.font(.caption2)
.buttonStyle(.plain)
.foregroundColor(.secondary)
}
}
}
.animation(.easeInOut(duration: 0.2), value: bugReportPhase)
}
private var processToggleBinding: Binding<Bool> {
Binding(
get: {
@@ -720,61 +563,6 @@ struct ContentView: View {
)
}
private func sendBugReport() async {
bugReportPhase = .sending("Collecting logs...")
let service = BugReportService()
let description = bugReportUserDescription.trimmingCharacters(in: .whitespacesAndNewlines)
do {
let outcome = try await service.sendReport(
isManual: true,
userDescription: description.isEmpty ? nil : description
)
if outcome.success {
bugReportPhase = .success(outcome.message)
} else {
bugReportPhase = .failure(outcome.message)
}
} catch {
bugReportPhase = .failure(error.localizedDescription)
}
}
private func openGitHubIssue() {
let description = bugReportUserDescription.trimmingCharacters(in: .whitespacesAndNewlines)
var bodyParts: [String] = []
bodyParts.append("## Describe the bug")
bodyParts.append("")
if !description.isEmpty {
bodyParts.append(description)
} else {
bodyParts.append("A clear and concise description of what the bug is.")
}
bodyParts.append("")
bodyParts.append("## Environment")
bodyParts.append("")
bodyParts.append("- macOS Version: \(ProcessInfo.processInfo.operatingSystemVersionString)")
bodyParts.append("- EXO Version: \(buildTag) (\(buildCommit))")
bodyParts.append("")
bodyParts.append("## Additional context")
bodyParts.append("")
bodyParts.append("A bug report with diagnostic logs was submitted via the app.")
let body = bodyParts.joined(separator: "\n")
var components = URLComponents(string: "https://github.com/exo-explore/exo/issues/new")!
components.queryItems = [
URLQueryItem(name: "template", value: "bug_report.md"),
URLQueryItem(name: "title", value: "[BUG] "),
URLQueryItem(name: "body", value: body),
URLQueryItem(name: "labels", value: "bug"),
]
if let url = components.url {
NSWorkspace.shared.open(url)
}
}
private func showUninstallConfirmationAlert() {
let alert = NSAlert()
alert.messageText = "Uninstall EXO"
@@ -857,13 +645,6 @@ struct ContentView: View {
}
}
private var buildTag: String {
Bundle.main.infoDictionary?["EXOBuildTag"] as? String ?? "unknown"
}
private var buildCommit: String {
Bundle.main.infoDictionary?["EXOBuildCommit"] as? String ?? "unknown"
}
}
private struct HoverButton: View {
+3
View File
@@ -22,6 +22,7 @@ struct EXOApp: App {
@StateObject private var updater: SparkleUpdater
@StateObject private var thunderboltBridgeService: ThunderboltBridgeService
@StateObject private var settingsWindowController: SettingsWindowController
@StateObject private var bugReportWindowController: BugReportWindowController
private let terminationObserver: TerminationObserver
private let firstLaunchPopout = FirstLaunchPopout()
private let ciContext = CIContext(options: nil)
@@ -46,6 +47,7 @@ struct EXOApp: App {
let thunderboltBridge = ThunderboltBridgeService(clusterStateService: service)
_thunderboltBridgeService = StateObject(wrappedValue: thunderboltBridge)
_settingsWindowController = StateObject(wrappedValue: SettingsWindowController())
_bugReportWindowController = StateObject(wrappedValue: BugReportWindowController())
enableLaunchAtLoginIfNeeded()
// Install LaunchDaemon to disable Thunderbolt Bridge on startup (prevents network loops)
NetworkSetupHelper.promptAndInstallIfNeeded()
@@ -66,6 +68,7 @@ struct EXOApp: App {
.environmentObject(updater)
.environmentObject(thunderboltBridgeService)
.environmentObject(settingsWindowController)
.environmentObject(bugReportWindowController)
} label: {
menuBarIcon
.onReceive(controller.$isFirstLaunchReady) { ready in
@@ -0,0 +1,242 @@
import AppKit
import SwiftUI
/// Manages a standalone window for the bug-report flow.
/// Ensures only one instance exists and brings it to front on repeated opens.
@MainActor
final class BugReportWindowController: ObservableObject {
private var window: NSWindow?
func open() {
if let existing = window, existing.isVisible {
existing.makeKeyAndOrderFront(nil)
NSApp.activate()
return
}
let view = BugReportView(onDismiss: { [weak self] in
self?.window?.close()
})
let hostingController = NSHostingController(rootView: view)
hostingController.sizingOptions = [.preferredContentSize, .minSize]
let newWindow = NSWindow(contentViewController: hostingController)
newWindow.styleMask = [.titled, .closable, .resizable]
newWindow.title = "Send a Bug Report"
newWindow.center()
newWindow.setFrameAutosaveName("ExoBugReportWindow")
newWindow.isReleasedWhenClosed = false
newWindow.makeKeyAndOrderFront(nil)
NSApp.activate()
window = newWindow
}
}
private struct BugReportView: View {
fileprivate enum Phase: Equatable {
case prompting
case sending(String)
case success(String)
case failure(String)
}
let onDismiss: () -> Void
@State private var phase: Phase = .prompting
@State private var userDescription: String = ""
@FocusState private var descriptionFocused: Bool
var body: some View {
VStack(alignment: .leading, spacing: 12) {
switch phase {
case .prompting:
promptingView
case .sending(let message):
sendingView(message: message)
case .success(let message):
successView(message: message)
case .failure(let message):
failureView(message: message)
}
}
.padding(16)
.frame(minWidth: 380)
.animation(.easeInOut(duration: 0.2), value: phase)
.onAppear { descriptionFocused = true }
}
private var promptingView: some View {
VStack(alignment: .leading, spacing: 8) {
Text("Description (optional)")
.font(.subheadline)
.foregroundColor(.secondary)
ZStack(alignment: .topLeading) {
if userDescription.isEmpty {
Text("What were you doing when it broke?")
.font(.body)
.foregroundColor(Color(nsColor: .placeholderTextColor))
.padding(.horizontal, 10)
.padding(.vertical, 8)
.allowsHitTesting(false)
}
TextEditor(text: $userDescription)
.font(.body)
.scrollContentBackground(.hidden)
.padding(4)
.frame(height: 72)
.focused($descriptionFocused)
}
.background(
RoundedRectangle(cornerRadius: 6)
.fill(Color(nsColor: .textBackgroundColor))
)
.overlay(
RoundedRectangle(cornerRadius: 6)
.strokeBorder(Color(nsColor: .separatorColor), lineWidth: 1)
)
Text("Diagnostic logs will be uploaded with your report.")
.font(.caption)
.foregroundColor(.secondary)
HStack {
Spacer()
Button("Cancel") { onDismiss() }
.keyboardShortcut(.cancelAction)
Button("Send") {
Task { await send() }
}
.keyboardShortcut(.defaultAction)
}
.padding(.top, 4)
}
}
private func sendingView(message: String) -> some View {
VStack(alignment: .leading, spacing: 12) {
HStack(spacing: 10) {
ProgressView().controlSize(.small)
Text(message)
.foregroundColor(.secondary)
}
HStack {
Spacer()
Button("Cancel") { onDismiss() }
.keyboardShortcut(.cancelAction)
.disabled(true)
Button("Send") {}
.disabled(true)
}
}
}
private func successView(message: String) -> some View {
VStack(alignment: .leading, spacing: 12) {
HStack(alignment: .top, spacing: 10) {
Image(systemName: "checkmark.circle.fill")
.foregroundColor(.green)
.font(.title2)
Text(message)
.fixedSize(horizontal: false, vertical: true)
}
HStack {
Button {
openGitHubIssue()
} label: {
HStack(spacing: 4) {
Image(systemName: "arrow.up.right.square")
Text("Open GitHub Issue")
}
}
Spacer()
Button("Done") { onDismiss() }
.keyboardShortcut(.defaultAction)
}
}
}
private func failureView(message: String) -> some View {
VStack(alignment: .leading, spacing: 12) {
HStack(alignment: .top, spacing: 10) {
Image(systemName: "exclamationmark.triangle.fill")
.foregroundColor(.orange)
.font(.title2)
Text(message)
.fixedSize(horizontal: false, vertical: true)
}
HStack {
Spacer()
Button("Try Again") {
phase = .prompting
}
Button("Close") { onDismiss() }
.keyboardShortcut(.defaultAction)
}
}
}
private func send() async {
phase = .sending("Collecting logs and uploading…")
let service = BugReportService()
let description = userDescription.trimmingCharacters(in: .whitespacesAndNewlines)
do {
let outcome = try await service.sendReport(
isManual: true,
userDescription: description.isEmpty ? nil : description
)
if outcome.success {
phase = .success(outcome.message)
} else {
phase = .failure(outcome.message)
}
} catch {
phase = .failure(error.localizedDescription)
}
}
private func openGitHubIssue() {
let description = userDescription.trimmingCharacters(in: .whitespacesAndNewlines)
var bodyParts: [String] = []
bodyParts.append("## Describe the bug")
bodyParts.append("")
if !description.isEmpty {
bodyParts.append(description)
} else {
bodyParts.append("A clear and concise description of what the bug is.")
}
bodyParts.append("")
bodyParts.append("## Environment")
bodyParts.append("")
bodyParts.append("- macOS Version: \(ProcessInfo.processInfo.operatingSystemVersionString)")
bodyParts.append("- EXO Version: \(buildTag) (\(buildCommit))")
bodyParts.append("")
bodyParts.append("## Additional context")
bodyParts.append("")
bodyParts.append("A bug report with diagnostic logs was submitted via the app.")
let body = bodyParts.joined(separator: "\n")
var components = URLComponents(string: "https://github.com/exo-explore/exo/issues/new")!
components.queryItems = [
URLQueryItem(name: "template", value: "bug_report.md"),
URLQueryItem(name: "title", value: "[BUG] "),
URLQueryItem(name: "body", value: body),
URLQueryItem(name: "labels", value: "bug"),
]
if let url = components.url {
NSWorkspace.shared.open(url)
}
}
private var buildTag: String {
Bundle.main.infoDictionary?["EXOBuildTag"] as? String ?? "unknown"
}
private var buildCommit: String {
Bundle.main.infoDictionary?["EXOBuildCommit"] as? String ?? "unknown"
}
}
-46
View File
@@ -21,8 +21,6 @@ struct SettingsView: View {
@State private var pendingReadOnlyModelsDirs: String = ""
@State private var pendingCustomEnvironmentVariables: [CustomEnvironmentVariable] = []
@State private var needsRestart = false
@State private var bugReportInFlight = false
@State private var bugReportMessage: String?
@State private var uninstallInProgress = false
var body: some View {
@@ -202,8 +200,6 @@ struct SettingsView: View {
VStack(alignment: .leading, spacing: 2) {
rdmaStatusView
}
sendBugReportButton
}
Section("Danger Zone") {
@@ -504,50 +500,8 @@ struct SettingsView: View {
}
}
private var sendBugReportButton: some View {
VStack(alignment: .leading, spacing: 4) {
Button {
Task {
await sendBugReport()
}
} label: {
HStack {
if bugReportInFlight {
ProgressView()
.scaleEffect(0.6)
}
Text("Send Bug Report")
.font(.caption)
.fontWeight(.semibold)
Spacer()
}
}
.disabled(bugReportInFlight)
if let message = bugReportMessage {
Text(message)
.font(.caption2)
.foregroundColor(.secondary)
.fixedSize(horizontal: false, vertical: true)
}
}
}
// MARK: - Actions
private func sendBugReport() async {
bugReportInFlight = true
bugReportMessage = "Collecting logs..."
let service = BugReportService()
do {
let outcome = try await service.sendReport(isManual: true)
bugReportMessage = outcome.message
} catch {
bugReportMessage = error.localizedDescription
}
bugReportInFlight = false
}
private func showUninstallConfirmationAlert() {
let alert = NSAlert()
alert.messageText = "Uninstall EXO"
@@ -0,0 +1,96 @@
<script lang="ts">
import { nodeGpuProfile, topologyData } from "$lib/stores/app.svelte";
interface Props {
class?: string;
}
let { class: className = "" }: Props = $props();
const profiles = $derived(nodeGpuProfile());
const topology = $derived(topologyData());
const totalTflops = $derived(
Object.values(profiles).reduce((sum, p) => sum + (p?.tflopsFp16 ?? 0), 0),
);
const totalBandwidthGbps = $derived(
Object.values(profiles).reduce(
(sum, p) => sum + (p?.memoryBandwidthGbps ?? 0),
0,
),
);
const totalMemoryBytes = $derived(
Object.values(topology?.nodes ?? {}).reduce(
(sum, n) => sum + (n.system_info?.memory ?? 0),
0,
),
);
function formatTflops(value: number): string {
if (value >= 1000) return `${(value / 1000).toFixed(2)} PFLOPS`;
return `${value.toFixed(1)} TFLOPS`;
}
function formatBandwidth(value: number): string {
if (value >= 1000) return `${(value / 1000).toFixed(2)} TB/s`;
return `${value.toFixed(0)} GB/s`;
}
function formatMemory(bytes: number): string {
if (bytes <= 0) return "—";
if (bytes >= 1024 ** 4) return `${(bytes / 1024 ** 4).toFixed(2)} TB`;
if (bytes >= 1024 ** 3) return `${(bytes / 1024 ** 3).toFixed(0)} GB`;
return `${(bytes / 1024 ** 2).toFixed(0)} MB`;
}
</script>
<div class={`cluster-stats ${className}`}>
<div class="stat-block">
<span class="stat-value">{formatTflops(totalTflops)}</span>
<span class="stat-label">FP16 compute</span>
</div>
<div class="stat-block">
<span class="stat-value">{formatBandwidth(totalBandwidthGbps)}</span>
<span class="stat-label">Memory bandwidth</span>
</div>
<div class="stat-block">
<span class="stat-value">{formatMemory(totalMemoryBytes)}</span>
<span class="stat-label">Memory</span>
</div>
</div>
<style>
.cluster-stats {
width: 100%;
max-width: 560px;
margin: 0 auto;
padding: 4px 4px 2px;
background: transparent;
font-family: "SF Mono", Monaco, monospace;
display: flex;
justify-content: center;
gap: 24px;
flex-wrap: wrap;
}
.stat-block {
display: flex;
flex-direction: row;
align-items: baseline;
gap: 6px;
}
.stat-value {
color: rgba(255, 215, 0, 0.95);
font-size: 13px;
font-weight: 600;
font-variant-numeric: tabular-nums;
}
.stat-label {
color: rgba(179, 179, 179, 0.55);
font-size: 9px;
letter-spacing: 0.08em;
text-transform: uppercase;
}
</style>
@@ -8,8 +8,17 @@
nodeThunderboltBridge,
nodeRdmaCtl,
nodeIdentities,
nodeGpuProfile,
nodeLinkProfiles,
nodeNetworkRaw,
nodeThunderbolt,
type NodeInfo,
} from "$lib/stores/app.svelte";
import {
inferRdmaConnectionType,
inferSocketConnectionType,
type ConnectionType,
} from "$lib/utils/connection-type";
interface Props {
class?: string;
@@ -35,6 +44,105 @@
const tbBridgeData = $derived(nodeThunderboltBridge());
const rdmaCtlData = $derived(nodeRdmaCtl());
const identitiesData = $derived(nodeIdentities());
const gpuProfileData = $derived(nodeGpuProfile());
const linkProfilesData = $derived(nodeLinkProfiles());
const nodeNetworkData = $derived(nodeNetworkRaw());
const nodeThunderboltData = $derived(nodeThunderbolt());
// Hover state for edge tooltips. We render the tooltip as plain HTML
// overlaid on the SVG so it picks up the existing dashboard typography.
let hoveredPair = $state<{
a: string;
b: string;
x: number;
y: number;
} | null>(null);
function formatBandwidthMbps(value: number | null | undefined): string {
if (value == null || !isFinite(value)) return "—";
if (value >= 1000) return `${(value / 1000).toFixed(2)} Gbps`;
return `${value.toFixed(0)} Mbps`;
}
function formatLatencyMs(value: number | null | undefined): string {
if (value == null || !isFinite(value)) return "—";
if (value < 1) return `${(value * 1000).toFixed(0)} µs`;
return `${value.toFixed(2)} ms`;
}
interface PairProfileEntry {
fromId: string;
toId: string;
transport: "socket" | "rdma";
uploadMbps: number | null; // fromId -> toId (this row's source uploads)
downloadMbps: number | null; // toId -> fromId (this row's source downloads)
latencyMs: number | null; // round-trip; one-way ≈ RTT/2 (no clock sync)
latencyJitterMs: number | null; // mean |Δ| between adjacent RTTs
type: ConnectionType;
detail: string;
}
/** Collect every measured profile in either direction between two nodes. */
function collectPairProfiles(a: string, b: string): PairProfileEntry[] {
const result: PairProfileEntry[] = [];
const ctx = {
nodeNetwork: nodeNetworkData,
nodeThunderbolt: nodeThunderboltData,
};
for (const [fromId, toId] of [
[a, b],
[b, a],
] as const) {
const profiles = linkProfilesData[fromId]?.[toId];
if (!profiles) continue;
for (const profile of profiles) {
if (profile.transport === "socket") {
const type = inferSocketConnectionType(toId, profile.sinkIp, ctx);
result.push({
fromId,
toId,
transport: "socket",
uploadMbps: profile.uploadMbps,
downloadMbps: profile.downloadMbps,
latencyMs: profile.latencyMs,
latencyJitterMs: profile.latencyJitterMs ?? null,
type,
detail: profile.sinkIp,
});
} else {
const type = inferRdmaConnectionType(
fromId,
{
sourceRdmaIface: profile.sourceRdmaIface,
sinkRdmaIface: profile.sinkRdmaIface,
},
ctx,
);
result.push({
fromId,
toId,
transport: "rdma",
uploadMbps: profile.uploadMbps,
downloadMbps: profile.downloadMbps,
latencyMs: profile.latencyMs,
latencyJitterMs: profile.latencyJitterMs ?? null,
type,
detail: `${profile.sourceRdmaIface}${profile.sinkRdmaIface}`,
});
}
}
}
return result;
}
const hoveredPairProfiles = $derived.by(() => {
if (!hoveredPair) return [] as PairProfileEntry[];
return collectPairProfiles(hoveredPair.a, hoveredPair.b);
});
function nodeLabel(nodeId: string): string {
return data?.nodes?.[nodeId]?.friendly_name || nodeId.slice(0, 8);
}
function getNodeLabel(nodeId: string): string {
const node = data?.nodes?.[nodeId];
@@ -417,6 +525,128 @@
.attr("marker-end", "url(#arrowhead)");
}
// Per-direction labels, each placed next to the arrow that represents
// that direction. The arrow encodes "which way", so the label doesn't
// need ↑↓ glyphs. A profile from A's perspective contributes its
// uploadMbps to the A→B direction and its downloadMbps to the B→A
// direction; from B's perspective it's the other way around.
const pairProfiles = collectPairProfiles(entry.a, entry.b);
let maxAToB: number | null = null;
let maxBToA: number | null = null;
let minLat: number | null = null;
for (const p of pairProfiles) {
const aToBValue = p.fromId === entry.a ? p.uploadMbps : p.downloadMbps;
const bToAValue = p.fromId === entry.a ? p.downloadMbps : p.uploadMbps;
if (aToBValue != null && (maxAToB == null || aToBValue > maxAToB)) {
maxAToB = aToBValue;
}
if (bToAValue != null && (maxBToA == null || bToAValue > maxBToA)) {
maxBToA = bToAValue;
}
if (p.latencyMs != null && (minLat == null || p.latencyMs < minLat)) {
minLat = p.latencyMs;
}
}
// All three labels for an edge sit on its OUTER side (away from the
// viewport centroid). Pushing toward the centroid would pile up
// every edge's labels on top of each other in the middle.
const px = -uy;
const py = ux;
const dotToCenter = (centerX - mx) * px + (centerY - my) * py;
// sign = away from center. For diagonals whose midpoint *is* the
// centroid (dot==0), pick a deterministic side based on the edge
// direction so the choice is stable across renders.
const awayFromCenter =
dotToCenter === 0 ? (ux + uy >= 0 ? 1 : -1) : dotToCenter < 0 ? 1 : -1;
const perpOffset = 13;
const labelFontSize = isMinimized ? 9 : 11;
const bwColor = "rgba(255,215,0,0.95)";
const latColor = "rgba(74,222,128,0.95)";
// Place labels along a strip parallel to the edge, on its outer side.
// [A→B] [latency] [B→A]
// Distances are tuned so each bandwidth label sits next to its
// arrow head (arrows are at ±tipOffset=16 along the edge).
const bwAlong = 56;
function placeLabel(x: number, y: number, text: string, fill: string) {
linksGroup
.append("text")
.attr("x", x)
.attr("y", y)
.attr("text-anchor", "middle")
.attr("dominant-baseline", "middle")
.attr("font-size", labelFontSize)
.attr("font-family", "SF Mono, Monaco, monospace")
.attr("pointer-events", "none")
.attr("fill", fill)
.text(text);
}
const perpX = px * perpOffset * awayFromCenter;
const perpY = py * perpOffset * awayFromCenter;
// A→B bandwidth: next to the A→B arrow head (on the A side).
if (entry.aToB && maxAToB != null) {
placeLabel(
mx - ux * bwAlong + perpX,
my - uy * bwAlong + perpY,
formatBandwidthMbps(maxAToB),
bwColor,
);
}
// B→A bandwidth: next to the B→A arrow head (on the B side).
if (entry.bToA && maxBToA != null) {
placeLabel(
mx + ux * bwAlong + perpX,
my + uy * bwAlong + perpY,
formatBandwidthMbps(maxBToA),
bwColor,
);
}
// Latency centered between the two bandwidth labels, same outer side.
if (minLat != null) {
placeLabel(
mx + perpX,
my + perpY,
formatLatencyMs(minLat / 2),
latColor,
);
}
// Wide invisible hit target for hover, even when no profiles exist —
// makes "no measurements yet" debuggable from the UI.
const hitTarget = linksGroup
.append("line")
.attr("x1", posA.x)
.attr("y1", posA.y)
.attr("x2", posB.x)
.attr("y2", posB.y)
.attr("stroke", "transparent")
.attr("stroke-width", 18)
.attr("pointer-events", "stroke")
.style("cursor", "help");
hitTarget.on("mousemove", (event: MouseEvent) => {
// Viewport coords — the tooltip is `position: fixed` so it can
// escape the topology container's overflow:hidden clipping.
hoveredPair = {
a: entry.a,
b: entry.b,
x: event.clientX,
y: event.clientY,
};
});
hitTarget.on("mouseleave", () => {
if (
hoveredPair &&
hoveredPair.a === entry.a &&
hoveredPair.b === entry.b
) {
hoveredPair = null;
}
});
// Collect debug labels for later positioning at edges
if (debugEnabled && entry.connections.length > 0) {
// Determine which side of viewport based on edge midpoint
@@ -1012,6 +1242,10 @@
.text(powerText);
}
// GPU profile (TFLOPS + memory bandwidth) — only shown when we have a
// measurement; otherwise the slot collapses.
const gpuProfile = gpuProfileData[nodeInfo.id];
// Labels - adapt based on mode
if (showFullLabels) {
// FULL MODE: Name above, memory info below (1-4 nodes)
@@ -1060,6 +1294,29 @@
.append("tspan")
.attr("fill", "rgba(179,179,179,0.7)")
.text(` (${ramUsagePercent.toFixed(0)}%)`);
if (gpuProfile) {
const profileY = infoY + fontSize * 1.05;
const profileText = nodeG
.append("text")
.attr("x", nodeInfo.x)
.attr("y", profileY)
.attr("text-anchor", "middle")
.attr("font-size", fontSize * 0.8)
.attr("font-family", "SF Mono, Monaco, monospace");
profileText
.append("tspan")
.attr("fill", "rgba(74,222,128,0.95)")
.text(`${gpuProfile.tflopsFp16.toFixed(1)} TFLOPS`);
profileText
.append("tspan")
.attr("fill", "rgba(179,179,179,0.6)")
.text(" · ");
profileText
.append("tspan")
.attr("fill", "rgba(255,215,0,0.8)")
.text(`${gpuProfile.memoryBandwidthGbps.toFixed(0)} GB/s`);
}
} else if (showCompactLabels) {
// COMPACT MODE: Just name and basic info (4+ nodes)
const fontSize = Math.max(7, nodeRadius * 0.11);
@@ -1093,6 +1350,21 @@
.text(
`${ramUsagePercent.toFixed(0)}%${!isNaN(gpuTemp) ? " " + gpuTemp.toFixed(0) + "°C" : ""}`,
);
if (gpuProfile) {
const profileY = statsY + 9;
nodeG
.append("text")
.attr("x", nodeInfo.x)
.attr("y", profileY)
.attr("text-anchor", "middle")
.attr("fill", "rgba(74,222,128,0.95)")
.attr("font-size", fontSize * 0.85)
.attr("font-family", "SF Mono, Monaco, monospace")
.text(
`${gpuProfile.tflopsFp16.toFixed(0)} TFLOPS · ${gpuProfile.memoryBandwidthGbps.toFixed(0)} GB/s`,
);
}
} else {
// MINIMIZED MODE: Show name above and memory info below (like main topology)
const fontSize = 8;
@@ -1135,6 +1407,21 @@
.append("tspan")
.attr("fill", "rgba(179,179,179,0.7)")
.text(` (${ramUsagePercent.toFixed(0)}%)`);
if (gpuProfile) {
const profileY = infoY + 8;
nodeG
.append("text")
.attr("x", nodeInfo.x)
.attr("y", profileY)
.attr("text-anchor", "middle")
.attr("fill", "rgba(74,222,128,0.95)")
.attr("font-size", fontSize * 0.85)
.attr("font-family", "SF Mono, Monaco, monospace")
.text(
`${gpuProfile.tflopsFp16.toFixed(0)}T · ${gpuProfile.memoryBandwidthGbps.toFixed(0)}GB/s`,
);
}
}
// Debug mode: Show TB bridge and RDMA status
@@ -1206,6 +1493,10 @@
const _hoveredNodeId = hoveredNodeId;
const _filteredNodes = filteredNodes;
const _highlightedNodes = highlightedNodes;
const _gpu = gpuProfileData;
const _links = linkProfilesData;
const _network = nodeNetworkData;
const _tb = nodeThunderboltData;
if (_data) {
renderGraph();
}
@@ -1225,7 +1516,71 @@
});
</script>
<svg bind:this={svgContainer} class="w-full h-full {className}"></svg>
<div class="topology-root {className}">
<svg bind:this={svgContainer} class="w-full h-full"></svg>
{#if hoveredPair && hoveredPairProfiles.length > 0}
<div
class="link-tooltip"
style="left: {hoveredPair.x + 14}px; top: {hoveredPair.y + 14}px;"
>
<div class="tooltip-header">
{nodeLabel(hoveredPair.a)}{nodeLabel(hoveredPair.b)}
</div>
<table>
<thead>
<tr>
<th>Direction</th>
<th>Type</th>
<th>↑ Upload</th>
<th>↓ Download</th>
<th>RTT/2</th>
<th>Jitter</th>
</tr>
</thead>
<tbody>
{#each hoveredPairProfiles as profile (profile.fromId + profile.toId + profile.transport + profile.detail)}
<tr>
<td class="direction">
{nodeLabel(profile.fromId)}{nodeLabel(profile.toId)}
<span class="detail">{profile.detail}</span>
</td>
<td class="type">{profile.type.label}</td>
<td class="bandwidth"
>{formatBandwidthMbps(profile.uploadMbps)}</td
>
<td class="bandwidth"
>{formatBandwidthMbps(profile.downloadMbps)}</td
>
<td class="latency"
>{formatLatencyMs(
profile.latencyMs != null ? profile.latencyMs / 2 : null,
)}</td
>
<td class="latency"
>{formatLatencyMs(
profile.latencyJitterMs != null
? profile.latencyJitterMs / 2
: null,
)}</td
>
</tr>
{/each}
</tbody>
</table>
</div>
{:else if hoveredPair}
<div
class="link-tooltip empty"
style="left: {hoveredPair.x + 14}px; top: {hoveredPair.y + 14}px;"
>
<div class="tooltip-header">
{nodeLabel(hoveredPair.a)}{nodeLabel(hoveredPair.b)}
</div>
<div class="empty-message">No link measurements yet</div>
</div>
{/if}
</div>
<style>
:global(.graph-node) {
@@ -1247,4 +1602,93 @@
stroke-dashoffset: -10;
}
}
.topology-root {
position: relative;
width: 100%;
height: 100%;
}
.link-tooltip {
/* position: fixed so the tooltip can escape `overflow: hidden` on the
topology container — multi-row tooltips were being clipped. */
position: fixed;
z-index: 1000;
pointer-events: none;
background: rgba(15, 15, 15, 0.95);
border: 1px solid rgba(255, 215, 0, 0.3);
border-radius: 6px;
padding: 10px 12px;
font-family: "SF Mono", Monaco, monospace;
font-size: 11px;
color: rgba(230, 230, 230, 0.95);
box-shadow: 0 6px 18px rgba(0, 0, 0, 0.55);
backdrop-filter: blur(6px);
max-width: 720px;
}
.tooltip-header {
color: rgba(255, 215, 0, 0.95);
font-size: 12px;
margin-bottom: 6px;
letter-spacing: 0.04em;
}
.link-tooltip table {
border-collapse: collapse;
width: 100%;
}
.link-tooltip th,
.link-tooltip td {
text-align: left;
padding: 4px 14px 4px 0;
vertical-align: top;
white-space: nowrap;
}
.link-tooltip th:last-child,
.link-tooltip td:last-child {
padding-right: 0;
}
.link-tooltip th {
color: rgba(179, 179, 179, 0.65);
font-weight: 500;
text-transform: uppercase;
font-size: 9px;
letter-spacing: 0.08em;
border-bottom: 1px solid rgba(255, 215, 0, 0.18);
}
.link-tooltip .direction {
color: rgba(255, 255, 255, 0.85);
}
.link-tooltip .detail {
display: block;
color: rgba(179, 179, 179, 0.65);
font-size: 10px;
}
.link-tooltip .type {
color: rgba(74, 222, 128, 0.95);
}
.link-tooltip .bandwidth {
color: rgba(255, 215, 0, 0.95);
font-variant-numeric: tabular-nums;
}
.link-tooltip .latency {
color: rgba(74, 222, 128, 0.95);
font-variant-numeric: tabular-nums;
}
.link-tooltip.empty {
color: rgba(179, 179, 179, 0.7);
}
.empty-message {
font-size: 11px;
}
</style>
+1
View File
@@ -1,4 +1,5 @@
export { default as TopologyGraph } from "./TopologyGraph.svelte";
export { default as GpuRichBar } from "./GpuRichBar.svelte";
export { default as ChatForm } from "./ChatForm.svelte";
export { default as ChatMessages } from "./ChatMessages.svelte";
export { default as ChatAttachments } from "./ChatAttachments.svelte";
+65
View File
@@ -104,6 +104,13 @@ interface RawSystemPerformanceProfile {
ecpuUsage?: number;
}
type RawInterfaceType =
| "wifi"
| "ethernet"
| "maybe_ethernet"
| "thunderbolt"
| "unknown";
interface RawNetworkInterfaceInfo {
name?: string;
ipAddress?: string;
@@ -112,12 +119,51 @@ interface RawNetworkInterfaceInfo {
ipv6?: string;
ipAddresses?: string[];
ips?: string[];
interfaceType?: RawInterfaceType;
}
interface RawNodeNetworkInfo {
interfaces?: RawNetworkInterfaceInfo[];
}
export interface RawNodeGpuProfile {
engine: "mlx";
tflopsFp16: number;
memoryBandwidthGbps: number;
measuredAt: string;
}
export interface RawNodeSocketLinkProfile {
transport: "socket";
sinkIp: string;
latencyMs: number;
latencyJitterMs?: number;
uploadMbps: number;
downloadMbps: number;
measuredAt: string;
}
export interface RawNodeRdmaLinkProfile {
transport: "rdma";
sourceRdmaIface: string;
sinkRdmaIface: string;
uploadMbps: number | null;
downloadMbps: number | null;
payloadBytes: number | null;
latencyMs: number | null;
latencyJitterMs?: number | null;
measuredAt: string;
}
export type RawNodeLinkProfile =
| RawNodeSocketLinkProfile
| RawNodeRdmaLinkProfile;
export type RawNodeLinkProfiles = Record<
string,
Record<string, RawNodeLinkProfile[]>
>;
interface RawSocketConnection {
sinkMultiaddr?: {
address?: string;
@@ -261,6 +307,10 @@ interface RawStateResponse {
string,
{ total: { inBytes: number }; available: { inBytes: number } }
>;
// Per-node GPU compute + memory bandwidth profile.
nodeGpuProfile?: Record<string, RawNodeGpuProfile>;
// Per-edge link probe results, keyed source -> sink -> [profiles].
nodeLinkProfiles?: RawNodeLinkProfiles;
}
export interface MessageAttachment {
@@ -583,6 +633,9 @@ class AppStore {
{ enabled: boolean; exists: boolean; serviceName?: string | null }
>
>({});
nodeGpuProfile = $state<Record<string, RawNodeGpuProfile>>({});
nodeLinkProfiles = $state<RawNodeLinkProfiles>({});
nodeNetworkRaw = $state<Record<string, RawNodeNetworkInfo>>({});
// UI state
isTopologyMinimized = $state(false);
@@ -1351,6 +1404,13 @@ class AppStore {
this.thunderboltBridgeCycles = data.thunderboltBridgeCycles ?? [];
// Thunderbolt bridge status per node
this.nodeThunderboltBridge = data.nodeThunderboltBridge ?? {};
// Profiler outputs
this.nodeGpuProfile = data.nodeGpuProfile ?? {};
this.nodeLinkProfiles = data.nodeLinkProfiles ?? {};
// Raw network info — kept so the connection-type inference can use
// interfaceType, which the topology-shaped `NodeInfo.network_interfaces`
// drops in its flattening step.
this.nodeNetworkRaw = data.nodeNetwork ?? {};
this.lastUpdate = Date.now();
// Connection recovered
if (!this.isConnected) {
@@ -3609,6 +3669,11 @@ export const nodeRdmaCtl = () => appStore.nodeRdmaCtl;
export const thunderboltBridgeCycles = () => appStore.thunderboltBridgeCycles;
export const nodeThunderboltBridge = () => appStore.nodeThunderboltBridge;
// Profiler outputs
export const nodeGpuProfile = () => appStore.nodeGpuProfile;
export const nodeLinkProfiles = () => appStore.nodeLinkProfiles;
export const nodeNetworkRaw = () => appStore.nodeNetworkRaw;
// Image generation params
export const imageGenerationParams = () => appStore.getImageGenerationParams();
export const setImageGenerationParams = (
+216
View File
@@ -0,0 +1,216 @@
/**
* Infer the physical transport behind a topology edge so the dashboard can
* label it accurately ("Thunderbolt 5", "Wi-Fi", etc.).
*
* Inputs we already have in state:
* - SocketConnection edges carry an IP, which appears in the sink node's
* `nodeNetwork.interfaces[].ipAddress` together with an `interfaceType`
* (wifi / ethernet / thunderbolt / unknown).
* - RDMAConnection edges carry interface names like "rdma_en3", which appear
* in `nodeThunderbolt[node].interfaces[].rdmaInterface` alongside a
* `linkSpeed` string from `system_profiler SPThunderboltDataType`
* (e.g. "Up to 40 Gb/s x1" → TB4, "Up to 80 Gb/s x1" → TB5).
*
* The output is a string label intended for direct display, plus the
* structured pieces (kind + generation) so callers can do further styling.
*/
export type ConnectionKind =
| "thunderbolt"
| "ethernet"
| "wifi"
| "loopback"
| "unknown";
export interface ConnectionType {
/** Display label, e.g. "Thunderbolt 5 (RDMA)" or "WiFi". */
label: string;
kind: ConnectionKind;
/** "4" / "5" / undefined; only set for Thunderbolt. */
thunderboltGeneration?: "3" | "4" | "5";
/** True when the edge transports RDMA, false for plain TCP/IP. */
isRdma: boolean;
/** The matched linkSpeed string (e.g. "Up to 40 Gb/s x1"), if any. */
linkSpeedHint?: string;
}
interface RawNetworkInterface {
name?: string;
ipAddress?: string;
addresses?: Array<{ address?: string } | string>;
ipAddresses?: string[];
ips?: string[];
ipv4?: string;
ipv6?: string;
interfaceType?:
| "wifi"
| "ethernet"
| "maybe_ethernet"
| "thunderbolt"
| "unknown";
}
interface RawNodeNetwork {
interfaces?: RawNetworkInterface[];
}
interface RawThunderboltIdent {
rdmaInterface: string;
domainUuid: string;
linkSpeed: string;
}
interface RawNodeThunderbolt {
interfaces: RawThunderboltIdent[];
}
export interface ConnectionTypeContext {
nodeNetwork: Record<string, RawNodeNetwork>;
nodeThunderbolt: Record<string, RawNodeThunderbolt>;
}
/** Parse a system_profiler "Up to N Gb/s xK" string into a TB generation. */
export function thunderboltGenerationFromLinkSpeed(
linkSpeed: string | undefined,
): ConnectionType["thunderboltGeneration"] | undefined {
if (!linkSpeed) return undefined;
// Match the integer right before "Gb/s". Handles "Up to 40 Gb/s x1",
// "40 Gb/s", "80 Gb/s", etc.
const match = linkSpeed.match(/(\d+)\s*Gb\/s/i);
if (!match) return undefined;
const gbps = Number(match[1]);
if (gbps >= 80) return "5";
if (gbps >= 40) return "4";
if (gbps >= 20) return "3";
return undefined;
}
function findNetworkInterface(
network: RawNodeNetwork | undefined,
ip: string,
): RawNetworkInterface | undefined {
if (!network?.interfaces) return undefined;
for (const iface of network.interfaces) {
if (iface.ipAddress === ip) return iface;
if (iface.ipv4 === ip || iface.ipv6 === ip) return iface;
if (iface.ipAddresses?.includes(ip)) return iface;
if (iface.ips?.includes(ip)) return iface;
if (
iface.addresses?.some(
(a) => (typeof a === "string" ? a : a?.address) === ip,
)
) {
return iface;
}
}
return undefined;
}
function thunderboltIdentForIface(
thunderbolt: RawNodeThunderbolt | undefined,
rdmaInterface: string,
): RawThunderboltIdent | undefined {
if (!thunderbolt?.interfaces) return undefined;
return thunderbolt.interfaces.find((i) => i.rdmaInterface === rdmaInterface);
}
/** Connection type for an RDMA edge. */
export function inferRdmaConnectionType(
sourceNodeId: string,
edge: { sourceRdmaIface: string; sinkRdmaIface: string },
context: ConnectionTypeContext,
): ConnectionType {
// Both ends should report the same Thunderbolt generation; we read the
// source side because it's our own node and most likely to be present.
const ident = thunderboltIdentForIface(
context.nodeThunderbolt[sourceNodeId],
edge.sourceRdmaIface,
);
const generation = thunderboltGenerationFromLinkSpeed(ident?.linkSpeed);
const label = generation
? `Thunderbolt ${generation} (RDMA)`
: "Thunderbolt (RDMA)";
return {
label,
kind: "thunderbolt",
thunderboltGeneration: generation,
isRdma: true,
linkSpeedHint: ident?.linkSpeed,
};
}
// Sticky-cache keyed by (sinkNodeId, sinkIp). The backend re-derives
// nodeNetwork every 10 s from `networksetup`, and individual entries
// occasionally flicker (the IP is missing for one tick, then back). That
// would visibly bounce the label between "Ethernet" and "Unknown" between
// re-renders. Once we've classified a peer's IP as something concrete, we
// stick with that until we get a *different* concrete answer — transient
// "Unknown" readings are ignored.
const _socketTypeCache: Map<string, ConnectionType> = new Map();
/** Connection type for a SocketConnection edge identified by sink IP. */
export function inferSocketConnectionType(
sinkNodeId: string,
sinkIp: string,
context: ConnectionTypeContext,
): ConnectionType {
const fresh = _inferSocketConnectionTypeFresh(sinkNodeId, sinkIp, context);
const cacheKey = `${sinkNodeId}|${sinkIp}`;
if (fresh.kind !== "unknown") {
_socketTypeCache.set(cacheKey, fresh);
return fresh;
}
// Fresh classification couldn't determine a kind. Prefer the last good
// answer over flickering to "Unknown".
const cached = _socketTypeCache.get(cacheKey);
if (cached && cached.kind !== "unknown") {
return cached;
}
return fresh;
}
function _inferSocketConnectionTypeFresh(
sinkNodeId: string,
sinkIp: string,
context: ConnectionTypeContext,
): ConnectionType {
const iface = findNetworkInterface(context.nodeNetwork[sinkNodeId], sinkIp);
const ifType = iface?.interfaceType;
if (ifType === "thunderbolt") {
// For TCP-over-Thunderbolt we don't have a direct linkSpeed mapping —
// the rdmaInterface is named e.g. "rdma_en3" while the IP iface is "en3"
// or a bridge. Best effort: scan the node's TB identifiers and pick the
// fastest one; that's almost always the one carrying the connection.
const tb = context.nodeThunderbolt[sinkNodeId];
let bestGen: ConnectionType["thunderboltGeneration"] | undefined;
let bestSpeed: string | undefined;
for (const ident of tb?.interfaces ?? []) {
const g = thunderboltGenerationFromLinkSpeed(ident.linkSpeed);
if (g && (!bestGen || g > bestGen)) {
bestGen = g;
bestSpeed = ident.linkSpeed;
}
}
return {
label: bestGen ? `Thunderbolt ${bestGen} (TCP)` : "Thunderbolt (TCP)",
kind: "thunderbolt",
thunderboltGeneration: bestGen,
isRdma: false,
linkSpeedHint: bestSpeed,
};
}
if (ifType === "wifi") {
return { label: "WiFi", kind: "wifi", isRdma: false };
}
if (ifType === "ethernet") {
return { label: "Ethernet", kind: "ethernet", isRdma: false };
}
if (ifType === "maybe_ethernet") {
return { label: "Ethernet", kind: "ethernet", isRdma: false };
}
// Loopback IPs (Tailscale CGNAT 100.64.0.0/10, link-local, etc.) — surface
// generically rather than guessing.
return { label: "Unknown", kind: "unknown", isRdma: false };
}
+7 -2
View File
@@ -1,6 +1,7 @@
<script lang="ts">
import {
TopologyGraph,
GpuRichBar,
ChatForm,
ChatMessages,
ChatSidebar,
@@ -4743,9 +4744,10 @@
{#if topologyOnlyEnabled}
<!-- TOPOLOGY ONLY MODE: Full-screen topology -->
<div
class="flex-1 flex flex-col min-h-0 min-w-0 p-4"
class="flex-1 flex flex-col min-h-0 min-w-0 p-4 gap-3"
in:fade={{ duration: 300 }}
>
<GpuRichBar />
<div
class="flex-1 relative bg-exo-dark-gray/40 rounded-lg overflow-hidden"
>
@@ -4868,7 +4870,10 @@
out:fade={{ duration: 200 }}
>
<!-- Center: MAIN TOPOLOGY DISPLAY -->
<div class="flex-1 flex flex-col min-h-0 min-w-0 py-4">
<div class="flex-1 flex flex-col min-h-0 min-w-0 py-4 gap-3">
<div class="mx-4">
<GpuRichBar />
</div>
<!-- Topology Container - Takes most of the space -->
<div
class="flex-1 relative bg-exo-dark-gray/40 mx-4 mb-4 rounded-lg overflow-hidden"
+88 -1
View File
@@ -15,7 +15,7 @@ import anyio
from anyio import BrokenResourceError, ClosedResourceError
from fastapi import FastAPI, File, Form, HTTPException, Query, Request, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
from fastapi.responses import FileResponse, JSONResponse, Response, StreamingResponse
from fastapi.staticfiles import StaticFiles
from hypercorn.asyncio import serve # pyright: ignore[reportUnknownVariableType]
from hypercorn.config import Config
@@ -202,6 +202,12 @@ from exo.utils.banner import print_startup_banner
from exo.utils.channels import Receiver, Sender, channel
from exo.utils.disk_event_log import DiskEventLog
from exo.utils.power_sampler import PowerSampler
from exo.utils.profilers.rdma_probe import (
RdmaProbeBusyError,
RdmaProbeError,
RdmaProbeParams,
handle_rdma_probe_sink_request,
)
from exo.utils.task_group import TaskGroup
_API_EVENT_LOG_DIR = EXO_EVENT_LOG_DIR / "api"
@@ -400,6 +406,10 @@ class API:
self.app.get("/v1/traces/{task_id}/raw")(self.get_trace_raw)
self.app.get("/onboarding")(self.get_onboarding)
self.app.post("/onboarding")(self.complete_onboarding)
self.app.post("/profile/echo")(self.profile_echo)
self.app.post("/profile/upload")(self.profile_upload)
self.app.get("/profile/download")(self.profile_download)
self.app.post("/profile/rdma_probe")(self.profile_rdma_probe)
def get_state(self, path: str = ""):
if path == "":
@@ -2116,3 +2126,80 @@ class API:
ONBOARDING_COMPLETE_FILE.parent.mkdir(parents=True, exist_ok=True)
ONBOARDING_COMPLETE_FILE.write_text("true")
return JSONResponse({"completed": True})
async def profile_echo(self, request: Request) -> Response:
"""Echo the request body back unchanged.
Used by the link profiler for ping probes (RTT). Capped to keep an
attacker (or buggy peer) from exhausting memory.
"""
body = await request.body()
max_bytes = 32 * 1024 * 1024
if len(body) > max_bytes:
raise HTTPException(
status_code=413,
detail=f"echo payload exceeds {max_bytes} byte cap",
)
return Response(content=body, media_type="application/octet-stream")
async def profile_upload(self, request: Request) -> JSONResponse:
"""Discard the request body, time how long the receive took, and
return that duration. The response is intentionally tiny so that the
client's wall-clock measurement, after subtracting the round-trip,
cleanly maps to one-way upload bandwidth — but better still, we just
return our server-side duration so the client doesn't have to
subtract anything.
"""
max_bytes = 32 * 1024 * 1024
start = time.perf_counter()
bytes_received = 0
async for chunk in request.stream():
bytes_received += len(chunk)
if bytes_received > max_bytes:
raise HTTPException(
status_code=413,
detail=f"upload payload exceeds {max_bytes} byte cap",
)
recv_duration_ms = (time.perf_counter() - start) * 1000.0
return JSONResponse(
{
"bytes_received": bytes_received,
"recv_duration_ms": recv_duration_ms,
}
)
async def profile_download(
self, size: int = Query(0, alias="bytes", ge=0)
) -> Response:
"""Return `size` zeroed bytes. The client times its receive to get
one-way download bandwidth (the request itself is negligibly small).
Query param is named `bytes` for HTTP convention.
"""
max_bytes = 32 * 1024 * 1024
if size <= 0:
raise HTTPException(status_code=400, detail="bytes query param must be > 0")
if size > max_bytes:
raise HTTPException(
status_code=413,
detail=f"download size exceeds {max_bytes} byte cap",
)
return Response(
content=b"\x00" * size,
media_type="application/octet-stream",
)
async def profile_rdma_probe(self, params: RdmaProbeParams) -> Response:
"""Run the sink side of an RDMA bandwidth probe in a child process.
Returns 200 on a clean probe run, 409 if the node is busy with active
runners or another probe, 500 on any unexpected subprocess failure.
"""
try:
await handle_rdma_probe_sink_request(
params=params, runners=self.state.runners
)
except RdmaProbeBusyError as e:
raise HTTPException(status_code=409, detail=str(e)) from e
except RdmaProbeError as e:
raise HTTPException(status_code=500, detail=str(e)) from e
return Response(status_code=200)
+89 -25
View File
@@ -1,11 +1,12 @@
import asyncio
import hashlib
import os
import random
import shutil
import ssl
import time
import traceback
from collections.abc import Awaitable
from collections.abc import Awaitable, Mapping
from datetime import timedelta
from pathlib import Path
from typing import Callable, Literal
@@ -55,6 +56,36 @@ class HuggingFaceAuthenticationError(Exception):
class HuggingFaceRateLimitError(Exception):
"""429 Huggingface code"""
def __init__(self, msg: str, retry_after: float | None = None) -> None:
super().__init__(msg)
self.retry_after = retry_after
def _parse_retry_after(headers: Mapping[str, str]) -> float | None:
"""Parse seconds-to-reset from HF's RateLimit header.
HF sends e.g. ``ratelimit: "api";r=0;t=52`` on 429s; ``t`` is the wait.
Returns ``None`` if the header is missing or has no ``t`` field.
"""
raw = headers.get("RateLimit") or headers.get("ratelimit")
if raw is None:
return None
for part in raw.split(";"):
key, _, val = part.strip().partition("=")
if key == "t":
try:
return float(val)
except ValueError:
return None
return None
# reset window is 5 min
_RATE_LIMIT_MAX_SLEEP_SECS = 300.0
# 24h. Manually clear the cache (or `delete_model`) to force a refresh.
_FILE_LIST_CACHE_TTL_SECS = 24 * 60 * 60
async def _build_auth_error_message(status_code: int, model_id: ModelId) -> str:
token = await get_hf_token()
@@ -348,9 +379,6 @@ async def _build_file_list_from_local_directory(
return None
_fetched_file_lists_this_session: set[str] = set()
async def fetch_file_list_with_cache(
model_id: ModelId,
revision: str = "main",
@@ -360,13 +388,16 @@ async def fetch_file_list_with_cache(
) -> list[FileListEntry]:
target_dir = await ensure_cache_dir(model_id)
cache_file = target_dir / f"{model_id.normalize()}--{revision}--file_list.json"
cache_key = f"{model_id.normalize()}--{revision}"
if cache_key in _fetched_file_lists_this_session and await aios.path.exists(
cache_file
):
async with aiofiles.open(cache_file, "r") as f:
return TypeAdapter(list[FileListEntry]).validate_json(await f.read())
# cache survives process restarts so cold starts don't re-burst HF
if await aios.path.exists(cache_file):
try:
cache_age = time.time() - (await aios.stat(cache_file)).st_mtime
except OSError:
cache_age = float("inf")
if cache_age < _FILE_LIST_CACHE_TTL_SECS:
async with aiofiles.open(cache_file, "r") as f:
return TypeAdapter(list[FileListEntry]).validate_json(await f.read())
if skip_internet:
if await aios.path.exists(cache_file):
@@ -395,7 +426,6 @@ async def fetch_file_list_with_cache(
await f.write(
TypeAdapter(list[FileListEntry]).dump_json(file_list).decode()
)
_fetched_file_lists_this_session.add(cache_key)
return file_list
except Exception as e:
logger.opt(exception=e).warning(
@@ -426,17 +456,29 @@ async def fetch_file_list_with_retry(
recursive: bool = False,
on_connection_lost: Callable[[], None] = lambda: None,
) -> list[FileListEntry]:
n_attempts = 3
n_attempts = 5
for attempt in range(n_attempts):
try:
return await _fetch_file_list(model_id, revision, path, recursive)
except HuggingFaceAuthenticationError:
raise
except HuggingFaceRateLimitError as e:
if attempt == n_attempts - 1:
raise
sleep_for = e.retry_after if e.retry_after is not None else 2.0**attempt
sleep_for = min(sleep_for, _RATE_LIMIT_MAX_SLEEP_SECS) + random.uniform(
0, 1
)
logger.warning(
f"Rate limited by HuggingFace fetching file list for {model_id}; "
f"sleeping {sleep_for:.1f}s before retry {attempt + 2}/{n_attempts}"
)
await asyncio.sleep(sleep_for)
except Exception as e:
on_connection_lost()
if attempt == n_attempts - 1:
raise e
await asyncio.sleep(2.0**attempt)
await asyncio.sleep(2.0**attempt + random.uniform(0, 1))
raise Exception(
f"Failed to fetch file list for {model_id=} {revision=} {path=} {recursive=}"
)
@@ -447,6 +489,9 @@ async def _fetch_file_list(
) -> list[FileListEntry]:
api_url = f"{get_hf_endpoint()}/api/models/{model_id}/tree/{revision}"
url = f"{api_url}/{path}" if path else api_url
# ?recursive=true returns the whole subtree in one request
if recursive:
url = f"{url}?recursive=true"
headers = await get_download_headers()
async with (
@@ -458,7 +503,8 @@ async def _fetch_file_list(
raise HuggingFaceAuthenticationError(msg)
elif response.status == 429:
raise HuggingFaceRateLimitError(
f"Couldn't download {model_id} because of HuggingFace rate limit."
f"HuggingFace rate limit hit fetching file list for {model_id}",
retry_after=_parse_retry_after(response.headers),
)
elif response.status == 200:
data_json = await response.text()
@@ -468,10 +514,14 @@ async def _fetch_file_list(
if item.type == "file":
files.append(FileListEntry.model_validate(item))
elif item.type == "directory" and recursive:
subfiles = await _fetch_file_list(
model_id, revision, item.path, recursive
)
files.extend(subfiles)
# already inlined by ?recursive=true
continue
if recursive and len(data) >= 1000:
# HF tree endpoint paginates at 1000; we don't follow cursors
logger.warning(
f"File list for {model_id} hit the 1000-entry page cap "
"and may be truncated; cursor pagination is not implemented"
)
return files
else:
raise Exception(f"Failed to fetch file list: {response.status}")
@@ -552,6 +602,11 @@ async def file_meta(
if r.status in [401, 403]:
msg = await _build_auth_error_message(r.status, model_id)
raise HuggingFaceAuthenticationError(msg)
if r.status == 429:
raise HuggingFaceRateLimitError(
f"HuggingFace rate limit hit fetching metadata for {model_id}/{path}",
retry_after=_parse_retry_after(r.headers),
)
content_length = int(
r.headers.get("x-linked-size") or r.headers.get("content-length") or 0
)
@@ -571,7 +626,7 @@ async def download_file_with_retry(
on_connection_lost: Callable[[], None] = lambda: None,
skip_internet: bool = False,
) -> Path:
n_attempts = 3
n_attempts = 5
for attempt in range(n_attempts):
try:
return await _download_file(
@@ -583,12 +638,16 @@ async def download_file_with_retry(
raise
except HuggingFaceRateLimitError as e:
if attempt == n_attempts - 1:
raise e
logger.error(
f"Download error on attempt {attempt}/{n_attempts} for {model_id=} {revision=} {path=} {target_dir=}"
raise
sleep_for = e.retry_after if e.retry_after is not None else 2.0**attempt
sleep_for = min(sleep_for, _RATE_LIMIT_MAX_SLEEP_SECS) + random.uniform(
0, 1
)
logger.error(traceback.format_exc())
await asyncio.sleep(2.0**attempt)
logger.warning(
f"Rate limited by HuggingFace downloading {model_id}/{path}; "
f"sleeping {sleep_for:.1f}s before retry {attempt + 2}/{n_attempts}"
)
await asyncio.sleep(sleep_for)
except Exception as e:
if attempt == n_attempts - 1:
on_connection_lost()
@@ -597,7 +656,7 @@ async def download_file_with_retry(
f"Download error on attempt {attempt + 1}/{n_attempts} for {model_id=} {revision=} {path=} {target_dir=}"
)
logger.error(traceback.format_exc())
await asyncio.sleep(2.0**attempt)
await asyncio.sleep(2.0**attempt + random.uniform(0, 1))
raise Exception(
f"Failed to download file {model_id=} {revision=} {path=} {target_dir=}"
)
@@ -665,6 +724,11 @@ async def _download_file(
if r.status in [401, 403]:
msg = await _build_auth_error_message(r.status, model_id)
raise HuggingFaceAuthenticationError(msg)
if r.status == 429:
raise HuggingFaceRateLimitError(
f"HuggingFace rate limit hit downloading {model_id}/{path}",
retry_after=_parse_retry_after(r.headers),
)
assert r.status in [200, 206], (
f"Failed to download {path} from {url}: {r.status}"
)
@@ -1,5 +1,7 @@
"""Tests for offline/air-gapped mode."""
import os
import time
from collections.abc import AsyncIterator
from pathlib import Path
from unittest.mock import AsyncMock, patch
@@ -231,3 +233,64 @@ class TestFetchFileListOffline:
raise FileNotFoundError."""
with pytest.raises(FileNotFoundError, match="No internet"):
await fetch_file_list_with_cache(model_id, "main", skip_internet=True)
class TestFileListCacheTTL:
async def test_uses_fresh_cache_without_fetching(
self, model_id: ModelId, temp_models_dir: Path
) -> None:
from pydantic import TypeAdapter
cache_dir = temp_models_dir / "caches" / model_id.normalize()
await aios.makedirs(cache_dir, exist_ok=True)
cached_list = [
FileListEntry(type="file", path="model.safetensors", size=1000),
]
cache_file = cache_dir / f"{model_id.normalize()}--main--file_list.json"
async with aiofiles.open(cache_file, "w") as f:
await f.write(
TypeAdapter(list[FileListEntry]).dump_json(cached_list).decode()
)
with patch(
"exo.download.download_utils.fetch_file_list_with_retry",
new_callable=AsyncMock,
) as mock_fetch:
result = await fetch_file_list_with_cache(model_id, "main")
assert result == cached_list
mock_fetch.assert_not_called()
async def test_refetches_when_cache_older_than_ttl(
self, model_id: ModelId, temp_models_dir: Path
) -> None:
from pydantic import TypeAdapter
from exo.download.download_utils import (
_FILE_LIST_CACHE_TTL_SECS, # pyright: ignore[reportPrivateUsage]
)
cache_dir = temp_models_dir / "caches" / model_id.normalize()
await aios.makedirs(cache_dir, exist_ok=True)
stale_list = [FileListEntry(type="file", path="stale.bin", size=1)]
cache_file = cache_dir / f"{model_id.normalize()}--main--file_list.json"
async with aiofiles.open(cache_file, "w") as f:
await f.write(
TypeAdapter(list[FileListEntry]).dump_json(stale_list).decode()
)
old_mtime = time.time() - _FILE_LIST_CACHE_TTL_SECS - 60
os.utime(cache_file, (old_mtime, old_mtime))
fresh_list = [FileListEntry(type="file", path="fresh.bin", size=2)]
with patch(
"exo.download.download_utils.fetch_file_list_with_retry",
new_callable=AsyncMock,
return_value=fresh_list,
) as mock_fetch:
result = await fetch_file_list_with_cache(model_id, "main")
assert result == fresh_list
mock_fetch.assert_called_once()
@@ -0,0 +1,355 @@
"""Tests for HuggingFace 429 rate-limit handling in download_utils."""
from collections.abc import AsyncIterator
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
import aiofiles.os as aios
import pytest
from exo.download.download_utils import (
HuggingFaceRateLimitError,
_download_file, # pyright: ignore[reportPrivateUsage]
_fetch_file_list, # pyright: ignore[reportPrivateUsage]
_parse_retry_after, # pyright: ignore[reportPrivateUsage]
download_file_with_retry,
fetch_file_list_with_retry,
file_meta,
)
from exo.shared.types.common import ModelId
# captured from a real HF 429 on 2026-04-30 (header is lowercased by Cloudfront)
REAL_HF_429_HEADERS_2026_04_30 = {
"ratelimit": '"api";r=0;t=52',
"ratelimit-policy": '"fixed window";"api";q=500;w=300',
}
class TestParseRetryAfter:
def test_parses_documented_format(self) -> None:
assert _parse_retry_after({"RateLimit": '"api";r=0;t=243'}) == 243.0
def test_parses_real_hf_response(self) -> None:
assert _parse_retry_after(REAL_HF_429_HEADERS_2026_04_30) == 52.0
def test_parses_resolvers_bucket(self) -> None:
assert _parse_retry_after({"ratelimit": '"resolvers";r=0;t=120'}) == 120.0
def test_parses_pages_bucket(self) -> None:
assert _parse_retry_after({"ratelimit": '"pages";r=0;t=10'}) == 10.0
def test_returns_none_when_header_missing(self) -> None:
assert _parse_retry_after({}) is None
def test_returns_none_when_only_retry_after_present(self) -> None:
assert _parse_retry_after({"Retry-After": "60"}) is None
def test_returns_none_when_format_unrecognised(self) -> None:
assert _parse_retry_after({"ratelimit": "garbage"}) is None
def test_handles_extra_whitespace(self) -> None:
assert _parse_retry_after({"ratelimit": '"api"; r=0; t=42'}) == 42.0
class TestFetchFileListRetry:
async def test_uses_retry_after_from_error(self) -> None:
sleeps: list[float] = []
async def fake_sleep(seconds: float) -> None:
sleeps.append(seconds)
async def fake_fetch(*args: object, **kwargs: object) -> list[object]:
if not sleeps:
raise HuggingFaceRateLimitError("rate limited", retry_after=2.0)
return []
with (
patch(
"exo.download.download_utils._fetch_file_list", side_effect=fake_fetch
),
patch("exo.download.download_utils.asyncio.sleep", side_effect=fake_sleep),
):
result = await fetch_file_list_with_retry(ModelId("test/model"))
assert result == []
assert len(sleeps) == 1
assert 2.0 <= sleeps[0] < 3.0 # retry_after + jitter[0,1)
async def test_falls_back_to_exp_backoff_when_no_retry_after(self) -> None:
sleeps: list[float] = []
async def fake_sleep(seconds: float) -> None:
sleeps.append(seconds)
async def fake_fetch(*args: object, **kwargs: object) -> list[object]:
if not sleeps:
raise HuggingFaceRateLimitError("rate limited", retry_after=None)
return []
with (
patch(
"exo.download.download_utils._fetch_file_list", side_effect=fake_fetch
),
patch("exo.download.download_utils.asyncio.sleep", side_effect=fake_sleep),
):
await fetch_file_list_with_retry(ModelId("test/model"))
assert len(sleeps) == 1
assert 1.0 <= sleeps[0] < 2.0 # 2**0 + jitter[0,1)
async def test_caps_sleep_at_max_window(self) -> None:
sleeps: list[float] = []
async def fake_sleep(seconds: float) -> None:
sleeps.append(seconds)
async def fake_fetch(*args: object, **kwargs: object) -> list[object]:
if not sleeps:
raise HuggingFaceRateLimitError("rate limited", retry_after=10_000.0)
return []
with (
patch(
"exo.download.download_utils._fetch_file_list", side_effect=fake_fetch
),
patch("exo.download.download_utils.asyncio.sleep", side_effect=fake_sleep),
):
await fetch_file_list_with_retry(ModelId("test/model"))
assert len(sleeps) == 1
assert 300.0 <= sleeps[0] < 301.0 # cap + jitter[0,1)
async def test_retries_up_to_five_times(self) -> None:
sleeps: list[float] = []
async def fake_sleep(seconds: float) -> None:
sleeps.append(seconds)
async def fake_fetch(*args: object, **kwargs: object) -> list[object]:
raise HuggingFaceRateLimitError("rate limited", retry_after=1.0)
with (
patch(
"exo.download.download_utils._fetch_file_list", side_effect=fake_fetch
),
patch("exo.download.download_utils.asyncio.sleep", side_effect=fake_sleep),
pytest.raises(HuggingFaceRateLimitError),
):
await fetch_file_list_with_retry(ModelId("test/model"))
assert len(sleeps) == 4 # 5 attempts -> 4 sleeps before giving up
class TestDownloadFileRetry:
@pytest.fixture
async def target_dir(self, tmp_path: Path) -> AsyncIterator[Path]:
target = tmp_path / "downloads"
await aios.makedirs(target, exist_ok=True)
yield target
async def test_uses_retry_after_from_error(self, target_dir: Path) -> None:
sleeps: list[float] = []
results: list[Path] = [target_dir / "file.bin"]
async def fake_sleep(seconds: float) -> None:
sleeps.append(seconds)
async def fake_download(*args: object, **kwargs: object) -> Path:
if not sleeps:
raise HuggingFaceRateLimitError("rate limited", retry_after=5.0)
return results[0]
with (
patch(
"exo.download.download_utils._download_file",
side_effect=fake_download,
),
patch("exo.download.download_utils.asyncio.sleep", side_effect=fake_sleep),
):
result = await download_file_with_retry(
ModelId("test/model"), "main", "file.bin", target_dir
)
assert result == results[0]
assert len(sleeps) == 1
assert 5.0 <= sleeps[0] < 6.0
async def test_caps_sleep_at_max_window(self, target_dir: Path) -> None:
sleeps: list[float] = []
results: list[Path] = [target_dir / "file.bin"]
async def fake_sleep(seconds: float) -> None:
sleeps.append(seconds)
async def fake_download(*args: object, **kwargs: object) -> Path:
if not sleeps:
raise HuggingFaceRateLimitError("rate limited", retry_after=99_999.0)
return results[0]
with (
patch(
"exo.download.download_utils._download_file",
side_effect=fake_download,
),
patch("exo.download.download_utils.asyncio.sleep", side_effect=fake_sleep),
):
await download_file_with_retry(
ModelId("test/model"), "main", "file.bin", target_dir
)
assert len(sleeps) == 1
assert 300.0 <= sleeps[0] < 301.0
async def test_retries_up_to_five_times(self, target_dir: Path) -> None:
sleeps: list[float] = []
async def fake_sleep(seconds: float) -> None:
sleeps.append(seconds)
with (
patch(
"exo.download.download_utils._download_file",
new_callable=AsyncMock,
side_effect=HuggingFaceRateLimitError("rate limited", retry_after=1.0),
),
patch("exo.download.download_utils.asyncio.sleep", side_effect=fake_sleep),
pytest.raises(HuggingFaceRateLimitError),
):
await download_file_with_retry(
ModelId("test/model"), "main", "file.bin", target_dir
)
assert len(sleeps) == 4
def _make_mock_session_returning(
response_attrs: dict[str, object], method: str = "get"
) -> MagicMock:
"""Build a MagicMock that mimics ``create_http_session`` returning a
response whose ``status`` / ``headers`` are set from ``response_attrs``.
Mocks the chain ``create_http_session().__aenter__() -> session``, and
``session.<method>().__aenter__() -> response``.
"""
mock_response = MagicMock()
for k, v in response_attrs.items():
setattr(mock_response, k, v)
mock_session = MagicMock()
method_mock = getattr(mock_session, method) # pyright: ignore[reportAny]
method_mock.return_value.__aenter__ = AsyncMock( # pyright: ignore[reportAny]
return_value=mock_response
)
method_mock.return_value.__aexit__ = AsyncMock( # pyright: ignore[reportAny]
return_value=None
)
mock_factory = MagicMock()
mock_factory.return_value.__aenter__ = AsyncMock( # pyright: ignore[reportAny]
return_value=mock_session
)
mock_factory.return_value.__aexit__ = AsyncMock( # pyright: ignore[reportAny]
return_value=None
)
return mock_factory
REAL_HF_429_HEADER_DICT = {"ratelimit": '"api";r=0;t=52'}
class TestRateLimitAtHttpCallSites:
"""Verify each HF call site translates an HTTP 429 into a
``HuggingFaceRateLimitError`` carrying the parsed ``retry_after``.
These tests would catch regressions where (a) the 429 branch is
deleted, (b) ``_parse_retry_after`` stops being called, or
(c) the wrong header object is passed to it.
"""
async def test_fetch_file_list_maps_429_to_rate_limit_error(self) -> None:
mock_factory = _make_mock_session_returning(
{"status": 429, "headers": REAL_HF_429_HEADER_DICT}
)
with (
patch("exo.download.download_utils.create_http_session", mock_factory),
pytest.raises(HuggingFaceRateLimitError) as exc_info,
):
await _fetch_file_list(ModelId("test/model"), "main")
assert exc_info.value.retry_after == 52.0
async def test_file_meta_maps_429_to_rate_limit_error(self) -> None:
mock_factory = _make_mock_session_returning(
{"status": 429, "headers": REAL_HF_429_HEADER_DICT}, method="head"
)
with (
patch("exo.download.download_utils.create_http_session", mock_factory),
pytest.raises(HuggingFaceRateLimitError) as exc_info,
):
await file_meta(ModelId("test/model"), "main", "weights.safetensors")
assert exc_info.value.retry_after == 52.0
async def test_file_meta_maps_429_after_307_redirect(self) -> None:
"""When the initial HEAD 307s and the redirected HEAD then 429s,
the 429 must still surface as ``HuggingFaceRateLimitError``."""
# First HEAD -> 307 with a Location header pointing somewhere new.
first_response = MagicMock()
first_response.status = 307
first_response.headers = {"location": "/redirected/url"}
# Second HEAD (the recursive call) -> 429 with the real-HF header.
second_response = MagicMock()
second_response.status = 429
second_response.headers = REAL_HF_429_HEADER_DICT
responses = iter([first_response, second_response])
mock_session = MagicMock()
mock_session.head.return_value.__aenter__ = AsyncMock( # pyright: ignore[reportAny]
side_effect=lambda: next(responses)
)
mock_session.head.return_value.__aexit__ = AsyncMock( # pyright: ignore[reportAny]
return_value=None
)
mock_factory = MagicMock()
mock_factory.return_value.__aenter__ = AsyncMock( # pyright: ignore[reportAny]
return_value=mock_session
)
mock_factory.return_value.__aexit__ = AsyncMock( # pyright: ignore[reportAny]
return_value=None
)
with (
patch("exo.download.download_utils.create_http_session", mock_factory),
pytest.raises(HuggingFaceRateLimitError) as exc_info,
):
await file_meta(ModelId("test/model"), "main", "weights.safetensors")
assert exc_info.value.retry_after == 52.0
async def test_download_file_maps_429_to_rate_limit_error(
self, tmp_path: Path
) -> None:
target_dir = tmp_path / "downloads"
await aios.makedirs(target_dir, exist_ok=True)
# No local file -> _download_file goes straight to file_meta then GET.
# We need both calls to succeed enough to reach the GET branch:
# - file_meta returns a non-429 (size, etag) so we proceed.
# - the GET then 429s.
with (
patch(
"exo.download.download_utils.file_meta",
new_callable=AsyncMock,
return_value=(100, "abc123"),
),
patch(
"exo.download.download_utils.create_http_session",
_make_mock_session_returning(
{"status": 429, "headers": REAL_HF_429_HEADER_DICT}
),
),
pytest.raises(HuggingFaceRateLimitError) as exc_info,
):
await _download_file(
ModelId("test/model"), "main", "weights.safetensors", target_dir
)
assert exc_info.value.retry_after == 52.0
+108
View File
@@ -33,9 +33,13 @@ from exo.shared.types.events import (
)
from exo.shared.types.instance_link import InstanceLink, InstanceLinkId
from exo.shared.types.profiling import (
NodeGpuProfile,
NodeIdentity,
NodeLinkProfile,
NodeNetworkInfo,
NodeRdmaCtlStatus,
NodeRdmaLinkProfile,
NodeSocketLinkProfile,
NodeThunderboltInfo,
ThunderboltBridgeStatus,
)
@@ -63,6 +67,8 @@ from exo.utils.info_gatherer.info_gatherer import (
StaticNodeInformation,
ThunderboltBridgeInfo,
)
from exo.utils.profilers.gpu_profiler import GpuProfile
from exo.utils.profilers.link_profiler import RDMALinkProfile, SocketLinkProfile
def event_apply(event: Event, state: State) -> State:
@@ -304,6 +310,22 @@ def apply_node_timed_out(event: NodeTimedOut, state: State) -> State:
node_rdma_ctl = {
key: value for key, value in state.node_rdma_ctl.items() if key != event.node_id
}
node_gpu_profile = {
key: value
for key, value in state.node_gpu_profile.items()
if key != event.node_id
}
# Drop the leaving node both as source (outer key) and as sink (inner key).
node_link_profiles: dict[NodeId, Mapping[NodeId, Sequence[NodeLinkProfile]]] = {}
for source_id, sinks in state.node_link_profiles.items():
if source_id == event.node_id:
continue
filtered = {
sink_id: profiles
for sink_id, profiles in sinks.items()
if sink_id != event.node_id
}
node_link_profiles[source_id] = filtered
# Only recompute cycles if the leaving node had TB bridge enabled
leaving_node_status = state.node_thunderbolt_bridge.get(event.node_id)
leaving_node_had_tb_enabled = (
@@ -326,6 +348,8 @@ def apply_node_timed_out(event: NodeTimedOut, state: State) -> State:
"node_thunderbolt": node_thunderbolt,
"node_thunderbolt_bridge": node_thunderbolt_bridge,
"node_rdma_ctl": node_rdma_ctl,
"node_gpu_profile": node_gpu_profile,
"node_link_profiles": node_link_profiles,
"thunderbolt_bridge_cycles": thunderbolt_bridge_cycles,
}
)
@@ -432,10 +456,94 @@ def apply_node_gathered_info(event: NodeGatheredInfo, state: State) -> State:
**state.node_rdma_ctl,
event.node_id: NodeRdmaCtlStatus(enabled=info.enabled),
}
case GpuProfile():
measured_at = datetime.fromisoformat(event.when)
update["node_gpu_profile"] = {
**state.node_gpu_profile,
event.node_id: NodeGpuProfile(
engine=info.engine,
tflops_fp16=info.tflops_fp16,
memory_bandwidth_gbps=info.memory_bandwidth_gbps,
measured_at=measured_at,
),
}
case SocketLinkProfile():
measured_at = datetime.fromisoformat(event.when)
new_entry = NodeSocketLinkProfile(
sink_ip=info.sink_ip,
latency_ms=info.latency_ms,
latency_jitter_ms=info.latency_jitter_ms,
upload_mbps=info.upload_mbps,
download_mbps=info.download_mbps,
measured_at=measured_at,
)
update["node_link_profiles"] = _merge_link_profile(
state.node_link_profiles, event.node_id, info.sink_node_id, new_entry
)
case RDMALinkProfile():
measured_at = datetime.fromisoformat(event.when)
new_entry = NodeRdmaLinkProfile(
source_rdma_iface=info.source_rdma_iface,
sink_rdma_iface=info.sink_rdma_iface,
upload_mbps=info.upload_mbps,
download_mbps=info.download_mbps,
payload_bytes=info.payload_bytes,
latency_ms=info.latency_ms,
latency_jitter_ms=info.latency_jitter_ms,
measured_at=measured_at,
)
update["node_link_profiles"] = _merge_link_profile(
state.node_link_profiles, event.node_id, info.sink_node_id, new_entry
)
return state.model_copy(update=update)
def _link_profile_dedup_key(
profile: NodeLinkProfile,
) -> tuple[str, ...]:
"""Identity for in-place replacement.
Sockets are deduped on the *IP* (one entry per (transport, sink_ip)) — a
peer with both a LAN IP and a Tailscale IP gets two rows, not one that
bounces between them as the reconciler probes each IP in turn.
RDMA edges are deduped on the rdma interface pair, since a node can have
multiple Thunderbolt cables to the same peer.
"""
if isinstance(profile, NodeSocketLinkProfile):
return ("socket", profile.sink_ip)
return ("rdma", profile.source_rdma_iface, profile.sink_rdma_iface)
def _merge_link_profile(
existing: Mapping[NodeId, Mapping[NodeId, Sequence[NodeLinkProfile]]],
source_node_id: NodeId,
sink_node_id: NodeId,
new_entry: NodeLinkProfile,
) -> Mapping[NodeId, Mapping[NodeId, Sequence[NodeLinkProfile]]]:
"""Insert/replace a per-edge link profile, keyed by transport+identity.
A node may have both a socket profile and an RDMA profile to the same peer
(e.g. Wi-Fi + Thunderbolt), and may have multiple sockets (LAN IP +
Tailscale + link-local + ...). We replace any existing entry that shares the
same dedup key (see `_link_profile_dedup_key`), and append otherwise.
"""
source_map = dict(existing.get(source_node_id, {}))
current = list(source_map.get(sink_node_id, ()))
new_key = _link_profile_dedup_key(new_entry)
replaced = False
for i, profile in enumerate(current):
if _link_profile_dedup_key(profile) == new_key:
current[i] = new_entry
replaced = True
break
if not replaced:
current.append(new_entry)
source_map[sink_node_id] = current
return {**existing, source_node_id: source_map}
def apply_topology_edge_created(event: TopologyEdgeCreated, state: State) -> State:
topology = copy.deepcopy(state.topology)
topology.add_connection(event.conn)
+48
View File
@@ -1,5 +1,6 @@
import shutil
from collections.abc import Sequence
from datetime import datetime
from pathlib import Path
from typing import Literal, Self
@@ -109,3 +110,50 @@ class ThunderboltBridgeStatus(FrozenModel):
enabled: bool
exists: bool
service_name: str | None = None
class NodeGpuProfile(FrozenModel):
"""Measured GPU compute throughput and memory bandwidth for a node."""
engine: Literal["mlx"]
tflops_fp16: float
memory_bandwidth_gbps: float
measured_at: datetime
class NodeSocketLinkProfile(FrozenModel):
"""Per-direction TCP/IP bandwidth + round-trip latency (with jitter).
`upload_mbps` is source -> sink (this node sending), `download_mbps` is
sink -> source. `latency_jitter_ms` is the mean of |Δ| between
consecutive RTT samples (RFC 3550 / iperf3 jitter convention).
"""
transport: Literal["socket"] = "socket"
sink_ip: str
latency_ms: float
latency_jitter_ms: float = 0.0
upload_mbps: float
download_mbps: float
measured_at: datetime
class NodeRdmaLinkProfile(FrozenModel):
"""Per-direction RDMA bandwidth + round-trip latency (with jitter) over a TB edge.
All numeric fields are None when the most recent probe was skipped (peer
busy) or failed. We never substitute synthetic numbers.
"""
transport: Literal["rdma"] = "rdma"
source_rdma_iface: str
sink_rdma_iface: str
upload_mbps: float | None
download_mbps: float | None
payload_bytes: int | None
latency_ms: float | None = None
latency_jitter_ms: float | None = None
measured_at: datetime
NodeLinkProfile = NodeSocketLinkProfile | NodeRdmaLinkProfile
+10
View File
@@ -11,7 +11,9 @@ from exo.shared.types.instance_link import InstanceLink, InstanceLinkId
from exo.shared.types.profiling import (
DiskUsage,
MemoryUsage,
NodeGpuProfile,
NodeIdentity,
NodeLinkProfile,
NodeNetworkInfo,
NodeRdmaCtlStatus,
NodeThunderboltInfo,
@@ -59,6 +61,14 @@ class State(FrozenModel):
node_thunderbolt_bridge: Mapping[NodeId, ThunderboltBridgeStatus] = {}
node_rdma_ctl: Mapping[NodeId, NodeRdmaCtlStatus] = {}
# Profiler results.
# node_gpu_profile: per-node GPU TFLOPS / memory bandwidth measurement.
# node_link_profiles: per-edge link probe; outer key = source node, inner
# key = sink node. May contain both a socket and an RDMA profile to the
# same peer, so the inner value is a sequence.
node_gpu_profile: Mapping[NodeId, NodeGpuProfile] = {}
node_link_profiles: Mapping[NodeId, Mapping[NodeId, Sequence[NodeLinkProfile]]] = {}
# Detected cycles where all nodes have Thunderbolt bridge enabled (>2 nodes)
thunderbolt_bridge_cycles: Sequence[Sequence[NodeId]] = []
@@ -27,6 +27,8 @@ from exo.shared.types.thunderbolt import (
ThunderboltIdentifier,
)
from exo.utils.channels import Sender
from exo.utils.profilers.gpu_profiler import GpuProfile
from exo.utils.profilers.link_profiler import RDMALinkProfile, SocketLinkProfile
from exo.utils.pydantic_ext import TaggedModel
from exo.utils.task_group import TaskGroup
@@ -365,6 +367,9 @@ GatheredInfo = (
| MiscData
| StaticNodeInformation
| NodeDiskUsage
| GpuProfile
| SocketLinkProfile
| RDMALinkProfile
)
View File
Whitespace-only changes.
+158
View File
@@ -0,0 +1,158 @@
"""GPU compute and memory bandwidth probe.
Runs a small dense FP16 matmul to estimate TFLOPS and a large `mx.sum` to
estimate memory bandwidth. Both are measured in a worker thread (MLX `eval`
is blocking) so the event loop is unaffected.
The probe is "active" — it loads the GPU. The caller (ProfilerManager) is
responsible for not running it while inference is active.
## Design
- **Why `mx.sum` instead of GEMV for bandwidth.** GEMV pays for an inner
reduction plus a vector broadcast and caps at ~58% of peak DRAM bandwidth
on Apple Silicon. A pure streaming reduction (`mx.sum`) over a multi-GB
buffer hits ~80% of published peak — closer to the ceiling that's useful
for placement decisions. M3 Ultra publishes 819 GB/s; a 2 GB `mx.sum`
reports ~650 GB/s when the GPU is in its peak performance state.
- **The buffer must defeat the SLC.** M3 Ultra has ~96 MB system-level
cache. 2 GB sits well past the cache plateau.
- **Variance control.** Apple Silicon GPUs aggressively scale clocks based
on demand. A short cold benchmark catches the GPU mid-ramp-up and
reports a number 15-20% lower than the hardware can do. Two M3 Ultras
measured cold can differ by 100 GB/s on the *same* chip-and-board pair
even though peak silicon performance is identical.
The fix is what every hardware reviewer does:
1. **Long warm-up** — drive the workload for ~1 second to let macOS
lock the GPU into its peak performance state.
2. **Best-of-N** — time several short measurement passes and report
the *fastest*. The best run represents what the silicon can do
when it isn't being interrupted by Spotlight, the WindowServer,
or thermal pressure. Slower runs are noise from the rest of the
system, not a property of the GPU.
This makes two healthy M3 Ultras agree to within a few percent.
"""
import time
from collections.abc import Callable
from typing import Literal, Self, final
import mlx.core as mx
from anyio import to_thread
from exo.utils.pydantic_ext import TaggedModel
_MATMUL_DIM = 4096
_MATMUL_ITERATIONS_PER_PASS = 8
_BANDWIDTH_BYTES = 2 * 1024 * 1024 * 1024 # 2 GiB streaming buffer
_BANDWIDTH_ITERATIONS_PER_PASS = 16
_DTYPE = mx.float16
_BYTES_PER_ELEMENT = 2
# Drive the workload until the GPU is in its peak performance state. macOS
# typically clocks up within a few hundred ms of sustained load; 1.0 s is
# generous.
_WARMUP_SECONDS = 1.0
# Number of timed passes; we report the best one.
_MEASUREMENT_PASSES = 5
@final
class GpuProfile(TaggedModel):
"""Wire format for a measured GPU profile, gathered locally on a node."""
engine: Literal["mlx"]
tflops_fp16: float
memory_bandwidth_gbps: float
@classmethod
async def measure(cls) -> Self | None:
if not mx.metal.is_available():
return None
return await to_thread.run_sync(cls._measure_blocking)
@classmethod
def _measure_blocking(cls) -> Self:
return cls(
engine="mlx",
tflops_fp16=_measure_matmul_tflops(),
memory_bandwidth_gbps=_measure_streaming_bandwidth_gbps(),
)
def _warm_up_with(do_op: Callable[[], mx.array], deadline: float) -> None:
"""Repeatedly run `do_op` and `mx.eval` it until `deadline` (perf_counter
seconds), so that the GPU reaches its peak performance state before any
timed iteration begins. We discard the return value.
"""
while time.perf_counter() < deadline:
mx.eval(do_op())
def _measure_matmul_tflops() -> float:
"""Time a square FP16 matmul and return effective TFLOPS.
`mx.eval` is called inside the loop because MLX's lazy graph would
otherwise fuse all iterations into a single matmul on the trailing eval —
we'd then divide one matmul's runtime by N and overestimate TFLOPS.
"""
a = mx.random.uniform(shape=(_MATMUL_DIM, _MATMUL_DIM), dtype=_DTYPE)
b = mx.random.uniform(shape=(_MATMUL_DIM, _MATMUL_DIM), dtype=_DTYPE)
mx.eval(a, b)
_warm_up_with(lambda: mx.matmul(a, b), time.perf_counter() + _WARMUP_SECONDS)
flops_per_iteration = 2 * _MATMUL_DIM * _MATMUL_DIM * _MATMUL_DIM
best_tflops = 0.0
for _ in range(_MEASUREMENT_PASSES):
start = time.perf_counter()
for _ in range(_MATMUL_ITERATIONS_PER_PASS):
mx.eval(mx.matmul(a, b))
elapsed = time.perf_counter() - start
if elapsed <= 0:
continue
total_flops = flops_per_iteration * _MATMUL_ITERATIONS_PER_PASS
tflops = total_flops / elapsed / 1e12
if tflops > best_tflops:
best_tflops = tflops
return best_tflops
def _measure_streaming_bandwidth_gbps() -> float:
"""Time a pure-read reduction and infer memory bandwidth from bytes streamed.
`mx.sum` over a multi-GB buffer is a near-pure memory read with a tiny
add per element — the bandwidth-bound code path that comes closest to
the chip's DRAM ceiling on Apple Silicon. Same eval-per-iteration
discipline as the matmul: without it MLX fuses the loop into one
reduction and we'd report a wildly optimistic number.
Uses `mx.zeros` rather than `mx.random.uniform` for the buffer because
DRAM bandwidth is independent of the values being streamed, and `uniform`
allocates an FP32 temporary which doubles the peak allocation — a 2 GiB
FP16 buffer would briefly need 4 GiB of Metal heap, exceeding the
`max_buffer_size` on smaller Apple devices (3.5 GiB on the CI runner).
"""
n_elements = _BANDWIDTH_BYTES // _BYTES_PER_ELEMENT
buffer = mx.zeros(shape=(n_elements,), dtype=_DTYPE)
mx.eval(buffer)
_warm_up_with(lambda: mx.sum(buffer), time.perf_counter() + _WARMUP_SECONDS)
best_gbps = 0.0
for _ in range(_MEASUREMENT_PASSES):
start = time.perf_counter()
for _ in range(_BANDWIDTH_ITERATIONS_PER_PASS):
mx.eval(mx.sum(buffer))
elapsed = time.perf_counter() - start
if elapsed <= 0:
continue
total_bytes = _BANDWIDTH_BYTES * _BANDWIDTH_ITERATIONS_PER_PASS
gbps = total_bytes / elapsed / 1e9
if gbps > best_gbps:
best_gbps = gbps
return best_gbps
+296
View File
@@ -0,0 +1,296 @@
"""Link profilers — measure latency and bandwidth between this node and a peer.
Two transports, dispatched per topology edge:
- `SocketLinkProfile` — HTTP over the API port; measures the kernel TCP path.
- `RDMALinkProfile` — subprocessisolated MLX `jaccl`; measures the RDMA
path actually used for inference traffic.
Both profiles carry directional bandwidth (`upload_mbps` = source→sink,
`download_mbps` = sink→source) so the dashboard can surface real-world
asymmetries (Wi-Fi up vs down, NIC tx/rx, RDMA controller path skew).
Latency is reported as round-trip (`latency_ms`) — true one-way latency
needs sub-microsecond clock sync that we don't have, so we don't fake it.
Both return `None` on any failure (timeout, peer mismatch, subprocess
crash). Missing values are a legitimate "next tick will retry" state.
"""
import statistics
import time
from typing import Self, final
import httpx
from pydantic import BaseModel, ConfigDict, ValidationError
from exo.shared.types.common import NodeId
from exo.utils.profilers.rdma_probe import (
RdmaProbeError,
RdmaProbeParams,
run_rdma_probe_source_side,
)
from exo.utils.pydantic_ext import TaggedModel
class _UploadResponse(BaseModel):
"""Parsed body of a POST /profile/upload response."""
model_config = ConfigDict(extra="ignore", strict=False)
bytes_received: int
recv_duration_ms: float
LATENCY_PAYLOAD_BYTES = 64
LATENCY_SAMPLES = 10
BANDWIDTH_PAYLOAD_BYTES = 8 * 1024 * 1024
PROBE_TIMEOUT_SECONDS = 30.0
@final
class SocketLinkProfile(TaggedModel):
"""Measured TCP/IP RTT (median + jitter) and per-direction bandwidth."""
sink_node_id: NodeId
sink_ip: str
latency_ms: float
# Mean of |Δ| between consecutive RTT samples (RFC 3550 / iperf3 jitter
# convention). 0 for a perfectly stable link, sub-ms on a quiet LAN,
# tens of ms over Wi-Fi or NAT-relayed paths.
latency_jitter_ms: float
upload_mbps: float
download_mbps: float
@classmethod
async def measure(
cls,
*,
client: httpx.AsyncClient,
sink_ip: str,
expected_sink_node_id: NodeId,
api_port: int,
) -> Self | None:
if not await _peer_node_id_matches(
client, sink_ip, api_port, expected_sink_node_id
):
return None
latency = await _measure_latency_ms(client, sink_ip, api_port)
if latency is None:
return None
latency_ms, latency_jitter_ms = latency
upload_mbps = await _measure_upload_mbps(client, sink_ip, api_port)
if upload_mbps is None:
return None
download_mbps = await _measure_download_mbps(client, sink_ip, api_port)
if download_mbps is None:
return None
return cls(
sink_node_id=expected_sink_node_id,
sink_ip=sink_ip,
latency_ms=latency_ms,
latency_jitter_ms=latency_jitter_ms,
upload_mbps=upload_mbps,
download_mbps=download_mbps,
)
@final
class RDMALinkProfile(TaggedModel):
"""Measured RDMA bandwidth (per direction) + RTT (with jitter) over a TB edge.
All numeric fields are None when the most recent probe was skipped
(the local node, the peer, or both had active runners) or failed.
"""
sink_node_id: NodeId
source_rdma_iface: str
sink_rdma_iface: str
upload_mbps: float | None
download_mbps: float | None
payload_bytes: int | None
latency_ms: float | None
latency_jitter_ms: float | None
@classmethod
async def measure(
cls,
*,
client: httpx.AsyncClient,
sink_ip: str,
sink_node_id: NodeId,
api_port: int,
source_rdma_iface: str,
sink_rdma_iface: str,
coordinator_ip: str,
) -> Self | None:
params = RdmaProbeParams(
source_rdma_iface=source_rdma_iface,
sink_rdma_iface=sink_rdma_iface,
coordinator_ip=coordinator_ip,
)
try:
result = await run_rdma_probe_source_side(
client=client,
params=params,
sink_ip=sink_ip,
api_port=api_port,
)
except RdmaProbeError:
return None
if result is None:
return cls(
sink_node_id=sink_node_id,
source_rdma_iface=source_rdma_iface,
sink_rdma_iface=sink_rdma_iface,
upload_mbps=None,
download_mbps=None,
payload_bytes=None,
latency_ms=None,
latency_jitter_ms=None,
)
return cls(
sink_node_id=sink_node_id,
source_rdma_iface=source_rdma_iface,
sink_rdma_iface=sink_rdma_iface,
upload_mbps=result.upload_mbps,
download_mbps=result.download_mbps,
payload_bytes=result.payload_bytes,
latency_ms=result.latency_ms,
latency_jitter_ms=result.latency_jitter_ms,
)
LinkProfile = SocketLinkProfile | RDMALinkProfile
def _bracketed(ip: str) -> str:
return f"[{ip}]" if ":" in ip else ip
def _echo_url(sink_ip: str, api_port: int) -> str:
return f"http://{_bracketed(sink_ip)}:{api_port}/profile/echo"
def _upload_url(sink_ip: str, api_port: int) -> str:
return f"http://{_bracketed(sink_ip)}:{api_port}/profile/upload"
def _download_url(sink_ip: str, api_port: int, n_bytes: int) -> str:
return f"http://{_bracketed(sink_ip)}:{api_port}/profile/download?bytes={n_bytes}"
def _node_id_url(sink_ip: str, api_port: int) -> str:
return f"http://{_bracketed(sink_ip)}:{api_port}/node_id"
async def _peer_node_id_matches(
client: httpx.AsyncClient,
sink_ip: str,
api_port: int,
expected_sink_node_id: NodeId,
) -> bool:
"""Confirm the peer reachable at `sink_ip` is the one we expect.
IP addresses outlive node memberships (e.g. DHCP rebind, node restart with
a fresh node_id), so we reverify before attributing bandwidth to a peer.
"""
try:
response = await client.get(
_node_id_url(sink_ip, api_port), timeout=PROBE_TIMEOUT_SECONDS
)
except httpx.HTTPError:
return False
if response.status_code != 200:
return False
return response.text.strip().strip('"') == expected_sink_node_id
async def _measure_latency_ms(
client: httpx.AsyncClient, sink_ip: str, api_port: int
) -> tuple[float, float] | None:
"""Round-trip latency (median ms) and jitter (mean |Δ| between adjacent
samples, ms — RFC 3550 / iperf3 convention) over K small-payload echoes.
"""
samples_ms: list[float] = []
payload = b"\x00" * LATENCY_PAYLOAD_BYTES
url = _echo_url(sink_ip, api_port)
for _ in range(LATENCY_SAMPLES):
start = time.perf_counter()
try:
response = await client.post(
url, content=payload, timeout=PROBE_TIMEOUT_SECONDS
)
except httpx.HTTPError:
return None
elapsed_ms = (time.perf_counter() - start) * 1000
if (
response.status_code != 200
or len(response.content) != LATENCY_PAYLOAD_BYTES
):
return None
samples_ms.append(elapsed_ms)
median = statistics.median(samples_ms)
deltas = [abs(samples_ms[i] - samples_ms[i - 1]) for i in range(1, len(samples_ms))]
jitter = statistics.fmean(deltas) if deltas else 0.0
return (median, jitter)
async def _measure_upload_mbps(
client: httpx.AsyncClient, sink_ip: str, api_port: int
) -> float | None:
"""One-way upload bandwidth (source -> sink). Server times its own
receive duration so the small response's RTT doesn't pollute the result.
"""
payload = b"\x00" * BANDWIDTH_PAYLOAD_BYTES
try:
response = await client.post(
_upload_url(sink_ip, api_port),
content=payload,
timeout=PROBE_TIMEOUT_SECONDS,
)
except httpx.HTTPError:
return None
if response.status_code != 200:
return None
try:
body = _UploadResponse.model_validate_json(response.content)
except ValidationError:
return None
if body.bytes_received != BANDWIDTH_PAYLOAD_BYTES or body.recv_duration_ms <= 0:
return None
return BANDWIDTH_PAYLOAD_BYTES * 8 / (body.recv_duration_ms / 1000.0) / 1e6
async def _measure_download_mbps(
client: httpx.AsyncClient, sink_ip: str, api_port: int
) -> float | None:
"""One-way download bandwidth (sink -> source). The request is tiny so
the client's wall-clock receive time is dominated by the response transit.
"""
url = _download_url(sink_ip, api_port, BANDWIDTH_PAYLOAD_BYTES)
start = time.perf_counter()
try:
response = await client.get(url, timeout=PROBE_TIMEOUT_SECONDS)
except httpx.HTTPError:
return None
elapsed = time.perf_counter() - start
if (
response.status_code != 200
or len(response.content) != BANDWIDTH_PAYLOAD_BYTES
or elapsed <= 0
):
return None
return BANDWIDTH_PAYLOAD_BYTES * 8 / elapsed / 1e6
__all__ = [
"BANDWIDTH_PAYLOAD_BYTES",
"LATENCY_PAYLOAD_BYTES",
"LATENCY_SAMPLES",
"LinkProfile",
"RDMALinkProfile",
"SocketLinkProfile",
]
+295
View File
@@ -0,0 +1,295 @@
"""State-driven reconciler for hardware and link profiles.
Every tick the manager reads the current State and decides what's missing or
stale, then probes it. This is the same controller pattern used by
Kubernetes: the desired state is "every node has a fresh GPU profile and
every outgoing topology edge has a fresh link profile". Reactive code that
fired off probes in response to specific events would be wrong here because
EXO replays the event log on master changes — replays must not have side
effects beyond updating state.
Cancellation order: `shutdown()` cancels the manager's task group, which
in turn cancels in-flight probe tasks. Since the worker shares its task
group with this manager (see `Worker.run`), shutting down the worker shuts
down the manager.
"""
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
import anyio
import httpx
from anyio import fail_after
from loguru import logger
from exo.shared.types.common import NodeId
from exo.shared.types.profiling import (
NodeRdmaLinkProfile,
NodeSocketLinkProfile,
)
from exo.shared.types.state import State
from exo.shared.types.topology import (
Connection,
RDMAConnection,
SocketConnection,
)
from exo.utils.channels import Sender
from exo.utils.info_gatherer.info_gatherer import GatheredInfo
from exo.utils.profilers.gpu_profiler import GpuProfile
from exo.utils.profilers.link_profiler import (
PROBE_TIMEOUT_SECONDS,
RDMALinkProfile,
SocketLinkProfile,
)
from exo.utils.profilers.rdma_probe import RdmaProbeBusyError
from exo.utils.task_group import TaskGroup
GPU_TTL = timedelta(hours=1)
SOCKET_LINK_TTL = timedelta(minutes=5)
RDMA_LINK_TTL = timedelta(hours=6)
RECONCILE_TICK_SECONDS = 15.0
GPU_PROBE_HARD_TIMEOUT_SECONDS = 60.0
SOCKET_PROBE_HARD_TIMEOUT_SECONDS = 30.0
RDMA_PROBE_HARD_TIMEOUT_SECONDS = 90.0
# (source, sink, transport, edge_discriminator) — uniquely identifies one edge
# we are currently probing. transport is "socket" | "rdma"; the discriminator
# is the sink IP for socket edges, or the (source_iface, sink_iface) tuple
# for RDMA edges.
LinkKey = tuple[NodeId, NodeId, str, str]
@dataclass
class ProfilerManager:
info_sender: Sender[GatheredInfo]
node_id: NodeId
api_port: int
state_view: Callable[[], State]
_tg: TaskGroup = field(init=False, default_factory=TaskGroup)
_gpu_in_flight: bool = field(init=False, default=False)
_link_in_flight: set[LinkKey] = field(init=False, default_factory=set)
async def run(self) -> None:
async with self._tg as tg:
tg.start_soon(self._reconcile_gpu, RECONCILE_TICK_SECONDS)
tg.start_soon(self._reconcile_links, RECONCILE_TICK_SECONDS)
def shutdown(self) -> None:
self._tg.cancel_tasks()
# ----- GPU --------------------------------------------------------------
async def _reconcile_gpu(self, tick_seconds: float) -> None:
while True:
try:
self._maybe_start_gpu_probe()
except Exception as e:
logger.opt(exception=e).warning("GPU reconcile error")
await anyio.sleep(tick_seconds)
def _maybe_start_gpu_probe(self) -> None:
state = self.state_view()
if self._gpu_in_flight:
return
if state.runners:
return
existing = state.node_gpu_profile.get(self.node_id)
if existing is not None and not _is_stale(existing.measured_at, GPU_TTL):
return
self._gpu_in_flight = True
self._tg.start_soon(self._do_gpu_probe)
async def _do_gpu_probe(self) -> None:
try:
with fail_after(GPU_PROBE_HARD_TIMEOUT_SECONDS):
profile = await GpuProfile.measure()
if profile is not None:
await self.info_sender.send(profile)
except Exception as e:
logger.opt(exception=e).warning("GPU probe failed")
finally:
self._gpu_in_flight = False
# ----- Links ------------------------------------------------------------
async def _reconcile_links(self, tick_seconds: float) -> None:
timeout = httpx.Timeout(timeout=PROBE_TIMEOUT_SECONDS)
async with httpx.AsyncClient(timeout=timeout, verify=False) as client:
while True:
try:
self._maybe_start_link_probes(client)
except Exception as e:
logger.opt(exception=e).warning("Link reconcile error")
await anyio.sleep(tick_seconds)
def _maybe_start_link_probes(self, client: httpx.AsyncClient) -> None:
state = self.state_view()
for connection in state.topology.out_edges(self.node_id):
self._maybe_start_one_link_probe(client, state, connection)
def _maybe_start_one_link_probe(
self, client: httpx.AsyncClient, state: State, connection: Connection
) -> None:
edge = connection.edge
sink = connection.sink
existing_profiles = list(
state.node_link_profiles.get(self.node_id, {}).get(sink, ())
)
match edge:
case SocketConnection():
key: LinkKey = (
self.node_id,
sink,
"socket",
edge.sink_multiaddr.ip_address,
)
if key in self._link_in_flight:
return
fresh = any(
isinstance(p, NodeSocketLinkProfile)
and p.sink_ip == edge.sink_multiaddr.ip_address
and not _is_stale(p.measured_at, SOCKET_LINK_TTL)
for p in existing_profiles
)
if fresh:
return
self._link_in_flight.add(key)
self._tg.start_soon(
self._do_socket_probe,
client,
key,
sink,
edge.sink_multiaddr.ip_address,
)
case RDMAConnection():
if state.runners:
return
key = (
self.node_id,
sink,
"rdma",
f"{edge.source_rdma_iface}/{edge.sink_rdma_iface}",
)
if key in self._link_in_flight:
return
fresh = any(
isinstance(p, NodeRdmaLinkProfile)
and p.source_rdma_iface == edge.source_rdma_iface
and p.sink_rdma_iface == edge.sink_rdma_iface
and p.upload_mbps is not None
and p.download_mbps is not None
and not _is_stale(p.measured_at, RDMA_LINK_TTL)
for p in existing_profiles
)
if fresh:
return
coordinator_ip = _resolve_coordinator_ip(state, self.node_id, sink)
sink_ip = _resolve_socket_sink_ip(state, self.node_id, sink)
if coordinator_ip is None or sink_ip is None:
# No reachable socket path on which to coordinate jaccl;
# skip until reachability info catches up.
return
self._link_in_flight.add(key)
self._tg.start_soon(
self._do_rdma_probe,
client,
key,
sink,
sink_ip,
coordinator_ip,
edge,
)
async def _do_socket_probe(
self,
client: httpx.AsyncClient,
key: LinkKey,
sink_node_id: NodeId,
sink_ip: str,
) -> None:
try:
with fail_after(SOCKET_PROBE_HARD_TIMEOUT_SECONDS):
profile = await SocketLinkProfile.measure(
client=client,
sink_ip=sink_ip,
expected_sink_node_id=sink_node_id,
api_port=self.api_port,
)
if profile is not None:
await self.info_sender.send(profile)
except Exception as e:
logger.opt(exception=e).warning(
f"Socket link probe to {sink_node_id} via {sink_ip} failed"
)
finally:
self._link_in_flight.discard(key)
async def _do_rdma_probe(
self,
client: httpx.AsyncClient,
key: LinkKey,
sink_node_id: NodeId,
sink_ip: str,
coordinator_ip: str,
edge: RDMAConnection,
) -> None:
try:
with fail_after(RDMA_PROBE_HARD_TIMEOUT_SECONDS):
profile = await RDMALinkProfile.measure(
client=client,
sink_ip=sink_ip,
sink_node_id=sink_node_id,
api_port=self.api_port,
source_rdma_iface=edge.source_rdma_iface,
sink_rdma_iface=edge.sink_rdma_iface,
coordinator_ip=coordinator_ip,
)
if profile is not None:
await self.info_sender.send(profile)
except RdmaProbeBusyError:
# Local lock held — try again next tick.
pass
except Exception as e:
logger.opt(exception=e).warning(
f"RDMA link probe to {sink_node_id} via {coordinator_ip} failed"
)
finally:
self._link_in_flight.discard(key)
def _is_stale(measured_at: datetime, ttl: timedelta) -> bool:
if measured_at.tzinfo is None:
measured_at = measured_at.replace(tzinfo=timezone.utc)
return datetime.now(tz=timezone.utc) - measured_at > ttl
def _resolve_socket_sink_ip(state: State, source: NodeId, sink: NodeId) -> str | None:
"""Pick an IP from the topology to reach `sink` from `source`.
Used as the address for the HTTP rendezvous request when initiating an
RDMA probe. Any reachable socket edge will do — the OS picks the route.
"""
for connection in state.topology.out_edges(source):
if connection.sink != sink:
continue
if isinstance(connection.edge, SocketConnection):
return connection.edge.sink_multiaddr.ip_address
return None
def _resolve_coordinator_ip(state: State, source: NodeId, sink: NodeId) -> str | None:
"""Find an IP of `source` that `sink` can reach, for the jaccl coordinator.
We invert the direction: the topology stores `(sink → source)` edges from
sink's perspective, so the sink_multiaddr on those edges is *source's* IP
as the sink sees it.
"""
for connection in state.topology.out_edges(sink):
if connection.sink != source:
continue
if isinstance(connection.edge, SocketConnection):
return connection.edge.sink_multiaddr.ip_address
return None
+222
View File
@@ -0,0 +1,222 @@
"""RDMA probe orchestration shared between source and sink.
The actual `mx.distributed.init(backend="jaccl")` runs in a child process
(`rdma_probe_main`) so it does not contaminate the worker process — `init`
is a oneshot global that conflicts with active inference groups.
This module owns:
- `RdmaProbeParams` — request body for both ends
- `RdmaProbeResult` — what the source side parses out of the subprocess stdout
- (Note: there is no per-node single-flight lock; jaccl + Apple's
Thunderbolt RDMA handle concurrent QPs fine, so the
reconciler tick and peer-initiated probes can run
simultaneously.)
- `run_rdma_probe_source_side` — issued by `RDMALinkProfile.measure()`
- `handle_rdma_probe_sink_request` — invoked by the `/profile/rdma_probe` API
"""
import sys
from typing import final
import anyio
import httpx
from anyio import fail_after
from loguru import logger
from exo.utils.ports import random_ephemeral_port
from exo.utils.pydantic_ext import FrozenModel
PAYLOAD_BYTES = 64 * 1024 * 1024
ITERATIONS = 4
SUBPROCESS_TIMEOUT_SECONDS = 60.0
SINK_HTTP_CONNECT_TIMEOUT_SECONDS = 10.0
# Note: there used to be a process-global single-flight lock here, on the
# theory that two simultaneous jaccl probes on the same machine would step
# on each other at the RDMA hardware level. Empirically jaccl + Apple's
# Thunderbolt RDMA implementation handle multiple concurrent QPs fine, so
# we let probes run in parallel — no lock. The reconciler still gates RDMA
# probes on `state.runners` being empty so we don't compete with active
# inference traffic.
@final
class RdmaProbeParams(FrozenModel):
"""Body sent over the wire from source to sink before the probe runs."""
source_rdma_iface: str
sink_rdma_iface: str
coordinator_ip: str
coordinator_port: int = 0 # 0 = source picks an ephemeral port
payload_bytes: int = PAYLOAD_BYTES
iterations: int = ITERATIONS
@final
class RdmaProbeResult(FrozenModel):
# Per-direction bandwidth from rank 0's (source) perspective. Upload =
# source -> sink, download = sink -> source. Apple Silicon TB5 is
# symmetric in spec (~80 Gb/s each way) but the controller's tx/rx
# pipelines can drift, so we measure each independently with send/recv.
upload_mbps: float
download_mbps: float
payload_bytes: int
iterations: int
# Round-trip latency over the same edge, measured with a tiny-payload
# all_sum ping-pong. None when the latency loop was skipped or failed.
latency_ms: float | None = None
# Jitter = mean of |Δ| between consecutive RTT samples (RFC 3550).
latency_jitter_ms: float | None = None
class RdmaProbeError(Exception):
"""Raised when an RDMA probe could not be initiated cleanly.
This is distinct from "the probe ran and failed" — for that we just return
None. RdmaProbeError signals a precondition violation (busy, lock held,
invalid params) that the caller surfaces as 409.
"""
class RdmaProbeBusyError(RdmaProbeError):
pass
def is_node_busy(state_runners: object) -> bool:
"""True if there are active runners on this node.
Accepts the runners mapping by abstract type to avoid an import cycle with
`exo.shared.types.state`.
"""
try:
return bool(len(state_runners)) # pyright: ignore[reportArgumentType]
except TypeError:
return False
async def run_rdma_probe_source_side(
*,
client: httpx.AsyncClient,
params: RdmaProbeParams,
sink_ip: str,
api_port: int,
) -> RdmaProbeResult | None:
"""Coordinate an RDMA probe with the peer at `sink_ip` and return the result.
Both ranks must run *concurrently* — jaccl init blocks each side until the
other rendezvous over the coordinator socket. So we kick off our own
rank-0 subprocess and the peer's rank-1 subprocess in parallel, with the
coordinator bound on the source. If the peer refuses (busy / error), we
cancel the local subprocess to avoid waiting on a rendezvous that will
never happen.
Returns None when the probe was skipped (peer busy) or failed (timeout,
subprocess crash, parse error).
"""
coordinator_port = params.coordinator_port or random_ephemeral_port()
sink_params = params.model_copy(update={"coordinator_port": coordinator_port})
result_holder: list[RdmaProbeResult | None] = [None]
async def _run_source() -> None:
result_holder[0] = await _spawn_probe_subprocess(rank=0, params=sink_params)
async def _ask_sink(cancel_scope: anyio.CancelScope) -> None:
try:
response = await client.post(
_rdma_probe_url(sink_ip, api_port),
content=sink_params.model_dump_json(),
headers={"Content-Type": "application/json"},
timeout=SUBPROCESS_TIMEOUT_SECONDS,
)
except httpx.HTTPError as e:
logger.debug(f"RDMA probe sink request failed: {e}")
cancel_scope.cancel()
return
if response.status_code != 200:
logger.debug(
f"RDMA probe sink returned {response.status_code}: "
f"{response.text[:200]}"
)
cancel_scope.cancel()
async with anyio.create_task_group() as tg:
tg.start_soon(_run_source)
tg.start_soon(_ask_sink, tg.cancel_scope)
return result_holder[0]
async def handle_rdma_probe_sink_request(
*, params: RdmaProbeParams, runners: object
) -> RdmaProbeResult:
"""Run the sink side of an RDMA probe in response to an HTTP request.
Raises RdmaProbeBusyError when runners are active (would compete with
inference for the same RDMA NIC). The caller translates that to 409.
"""
if is_node_busy(runners):
raise RdmaProbeBusyError("node has active runners")
result = await _spawn_probe_subprocess(rank=1, params=params)
if result is None:
raise RdmaProbeError("rdma probe subprocess produced no result")
return result
async def _spawn_probe_subprocess(
*, rank: int, params: RdmaProbeParams
) -> RdmaProbeResult | None:
"""Run rdma_probe_main with the given rank and parameters.
Returns the parsed result on a clean exit. Returns None on any subprocess
failure (timeout, non-zero exit, malformed output) — callers treat None
as "try again on the next reconciler tick".
"""
cmd = [
sys.executable,
"-m",
"exo.utils.profilers.rdma_probe_main",
str(rank),
params.model_dump_json(),
]
try:
with fail_after(SUBPROCESS_TIMEOUT_SECONDS):
completed = await anyio.run_process(cmd, check=False)
except TimeoutError:
logger.warning(f"RDMA probe subprocess (rank={rank}) timed out")
return None
except Exception as e:
logger.opt(exception=e).warning(
f"RDMA probe subprocess (rank={rank}) failed to launch"
)
return None
if completed.returncode != 0:
stderr_text = completed.stderr.decode("utf-8", errors="replace")[:512]
logger.warning(
f"RDMA probe subprocess (rank={rank}) exited "
f"{completed.returncode}: {stderr_text}"
)
return None
stdout_text = completed.stdout.decode("utf-8", errors="replace").strip()
last_line = stdout_text.splitlines()[-1] if stdout_text else ""
try:
return RdmaProbeResult.model_validate_json(last_line)
except ValueError as e:
# If stdout was empty there's almost always something illuminating in
# stderr (jaccl init failure, bad iface name, etc.). Log it so the
# operator can debug without re-running the subprocess by hand.
stderr_text = completed.stderr.decode("utf-8", errors="replace")[:1024]
logger.warning(
f"RDMA probe (rank={rank}) stdout unparseable: {e}; "
f"stdout={last_line!r}; stderr={stderr_text!r}"
)
return None
def _rdma_probe_url(sink_ip: str, api_port: int) -> str:
bracketed = f"[{sink_ip}]" if ":" in sink_ip else sink_ip
return f"http://{bracketed}:{api_port}/profile/rdma_probe"
+222
View File
@@ -0,0 +1,222 @@
"""RDMA probe subprocess entry point.
Invoked as `python -m exo.utils.profilers.rdma_probe_main <rank> <params_json>`.
Initialises an MLX `jaccl` distributed group with two ranks (rank 0 is the
node that initiated the probe and the jaccl coordinator). Runs three
back-to-back micro-benchmarks and prints the result on rank 0:
1. **Upload bandwidth (rank 0 -> rank 1)**: `mx.distributed.send` on
rank 0, `recv` on rank 1, large payload. Time on rank 0 → bytes/sec.
2. **Download bandwidth (rank 1 -> rank 0)**: same pattern reversed.
3. **Latency**: tiny-payload `all_sum` ping-pong (one round-trip per
iteration on a 2-rank group). Time/iter = RTT.
Runs in a *child process* because `mx.distributed.init` is process-global —
calling it inside the worker would conflict with active inference groups.
"""
import json
import os
import statistics
import sys
import tempfile
import time
from pathlib import Path
from pydantic import ValidationError
from exo.utils.profilers.rdma_probe import RdmaProbeParams
WARMUP_ITERATIONS = 1
LATENCY_PAYLOAD_BYTES = 64
LATENCY_ITERATIONS = 50
def build_two_rank_ibv_devs(
*, source_iface: str, sink_iface: str
) -> list[list[str | None]]:
"""Build the MLX_IBV_DEVICES matrix for a two-rank source/sink probe.
`ibv_devs[i][j]` is the RDMA interface on rank i used to reach rank j.
Rank 0 is the source and rank 1 is the sink, so row 0 must contain the
source's local interface and row 1 must contain the sink's local interface.
"""
return [[None, source_iface], [sink_iface, None]]
def main() -> int:
if len(sys.argv) != 3:
print(
f"usage: {sys.argv[0]} <rank> <params_json>",
file=sys.stderr,
)
return 2
rank_arg, params_json = sys.argv[1], sys.argv[2]
try:
rank = int(rank_arg)
params = RdmaProbeParams.model_validate_json(params_json)
except (ValueError, ValidationError) as e:
print(f"invalid arguments: {e}", file=sys.stderr)
return 2
if rank not in (0, 1):
print(f"rank must be 0 or 1, got {rank}", file=sys.stderr)
return 2
source_iface = params.source_rdma_iface
sink_iface = params.sink_rdma_iface
coordinator_ip = params.coordinator_ip
coordinator_port = params.coordinator_port
payload_bytes = params.payload_bytes
iterations = params.iterations
ibv_devs = build_two_rank_ibv_devs(source_iface=source_iface, sink_iface=sink_iface)
with tempfile.NamedTemporaryFile(
prefix="exo_rdma_probe_ibv_devs_", suffix=".json", mode="w", delete=False
) as f:
json.dump(ibv_devs, f)
ibv_devs_path = f.name
try:
os.environ["MLX_IBV_DEVICES"] = ibv_devs_path
os.environ["MLX_RANK"] = str(rank)
os.environ["MLX_JACCL_COORDINATOR"] = f"{coordinator_ip}:{coordinator_port}"
import mlx.core as mx # imported here so failures surface as a clean exit
try:
group = mx.distributed.init(backend="jaccl", strict=True)
except Exception as e: # noqa: BLE001
print(f"jaccl init failed: {e!r}", file=sys.stderr)
return 1
if group.size() != 2:
print(
f"expected jaccl group size 2, got {group.size()}",
file=sys.stderr,
)
return 1
try:
bytes_per_element = 2 # float16
n_elements = max(1, payload_bytes // bytes_per_element)
shape = (n_elements,)
dtype = mx.float16
tensor = mx.zeros(shape=shape, dtype=dtype)
mx.eval(tensor)
# Warm-up the distributed transport with one round-trip in each
# direction so kernel-launch / connection-setup overhead doesn't
# contaminate the first timed iteration.
if rank == 0:
mx.eval(mx.distributed.send(tensor, dst=1, group=group))
mx.eval(
mx.distributed.recv(shape=shape, dtype=dtype, src=1, group=group)
)
else:
mx.eval(
mx.distributed.recv(shape=shape, dtype=dtype, src=0, group=group)
)
mx.eval(mx.distributed.send(tensor, dst=0, group=group))
# ---- Upload bandwidth: rank 0 -> rank 1 ----
# `mx.eval` inside the loop because MLX is lazy: without it only
# the final op actually executes, overestimating throughput by Nx.
start = time.perf_counter()
if rank == 0:
for _ in range(iterations):
mx.eval(mx.distributed.send(tensor, dst=1, group=group))
else:
for _ in range(iterations):
mx.eval(
mx.distributed.recv(
shape=shape, dtype=dtype, src=0, group=group
)
)
upload_elapsed = time.perf_counter() - start
# ---- Download bandwidth: rank 1 -> rank 0 ----
start = time.perf_counter()
if rank == 0:
for _ in range(iterations):
mx.eval(
mx.distributed.recv(
shape=shape, dtype=dtype, src=1, group=group
)
)
else:
for _ in range(iterations):
mx.eval(mx.distributed.send(tensor, dst=0, group=group))
download_elapsed = time.perf_counter() - start
total_bits = payload_bytes * 8 * iterations
upload_mbps = (
total_bits / upload_elapsed / 1e6 if upload_elapsed > 0 else 0.0
)
download_mbps = (
total_bits / download_elapsed / 1e6 if download_elapsed > 0 else 0.0
)
# ---- Latency: tiny-payload all_sum is a ping-pong on a 2-rank
# group. Time each iter individually so we can report a median
# (resistant to a stray scheduler hiccup at loop start).
n_lat_elements = max(1, LATENCY_PAYLOAD_BYTES // bytes_per_element)
lat_tensor = mx.zeros(shape=(n_lat_elements,), dtype=dtype)
mx.eval(lat_tensor)
for _ in range(WARMUP_ITERATIONS):
mx.eval(mx.distributed.all_sum(lat_tensor, group=group))
per_iter_ms: list[float] = []
for _ in range(LATENCY_ITERATIONS):
t0 = time.perf_counter()
mx.eval(mx.distributed.all_sum(lat_tensor, group=group))
per_iter_ms.append((time.perf_counter() - t0) * 1000.0)
if per_iter_ms:
latency_ms = statistics.median(per_iter_ms)
_deltas = [
abs(per_iter_ms[i] - per_iter_ms[i - 1])
for i in range(1, len(per_iter_ms))
]
latency_jitter_ms = statistics.fmean(_deltas) if _deltas else 0.0
else:
latency_ms = None
latency_jitter_ms = None
except Exception as e: # noqa: BLE001
# Without this catch the rank-0 process can exit with returncode 0
# (because the parent `try/finally` only protects file cleanup) but
# an empty stdout — the caller would see "stdout unparseable" with
# no clue what actually went wrong. Surface the exception on stderr.
print(
f"rdma probe op failed (rank={rank}): {type(e).__name__}: {e}",
file=sys.stderr,
)
return 1
# Both ranks print the result. The sink-side HTTP handler also calls
# `_spawn_probe_subprocess` and parses stdout to confirm the probe
# completed cleanly — without rank-1 emitting JSON the handler 500s
# back to the source even on success. Both ranks measured the same
# loops, so the numbers should be near-identical.
print(
json.dumps(
{
"upload_mbps": upload_mbps,
"download_mbps": download_mbps,
"payload_bytes": payload_bytes,
"iterations": iterations,
"latency_ms": latency_ms,
"latency_jitter_ms": latency_jitter_ms,
}
),
flush=True,
)
return 0
finally:
Path(ibv_devs_path).unlink(missing_ok=True)
if __name__ == "__main__":
sys.exit(main())
Whitespace-only changes.
@@ -0,0 +1,217 @@
"""Verify the apply layer wires GpuProfile / SocketLinkProfile / RDMALinkProfile
through to the granular state mappings."""
from datetime import datetime, timezone
from exo.shared.apply import apply
from exo.shared.types.common import NodeId
from exo.shared.types.events import (
EventId,
IndexedEvent,
NodeGatheredInfo,
NodeTimedOut,
)
from exo.shared.types.profiling import NodeSocketLinkProfile
from exo.shared.types.state import State
from exo.utils.profilers.gpu_profiler import GpuProfile
from exo.utils.profilers.link_profiler import RDMALinkProfile, SocketLinkProfile
NODE_A = NodeId("a")
NODE_B = NodeId("b")
WHEN = str(datetime(2026, 1, 1, tzinfo=timezone.utc))
def _wrap(idx: int, info: object, when: str = WHEN) -> IndexedEvent:
return IndexedEvent(
idx=idx,
event=NodeGatheredInfo(
event_id=EventId(),
node_id=NODE_A,
when=when,
info=info, # pyright: ignore[reportArgumentType]
),
)
def test_apply_gpu_profile_writes_node_gpu_profile():
state = State()
profile = GpuProfile(engine="mlx", tflops_fp16=42.0, memory_bandwidth_gbps=400.0)
new_state = apply(state, _wrap(0, profile))
entry = new_state.node_gpu_profile[NODE_A]
assert entry.tflops_fp16 == 42.0
assert entry.memory_bandwidth_gbps == 400.0
assert entry.engine == "mlx"
def test_apply_socket_link_profile_keys_by_source_and_sink():
state = State()
profile = SocketLinkProfile(
sink_node_id=NODE_B,
sink_ip="10.0.0.5",
latency_ms=1.2,
latency_jitter_ms=0.1,
upload_mbps=420.0,
download_mbps=900.0,
)
new_state = apply(state, _wrap(0, profile))
profiles_to_b = new_state.node_link_profiles[NODE_A][NODE_B]
socket_profiles = [p for p in profiles_to_b if isinstance(p, NodeSocketLinkProfile)]
assert len(socket_profiles) == 1
assert socket_profiles[0].latency_ms == 1.2
assert socket_profiles[0].upload_mbps == 420.0
assert socket_profiles[0].download_mbps == 900.0
def test_apply_replaces_socket_profile_for_same_transport():
"""A second socket measurement to the same peer should overwrite, not duplicate."""
state = State()
p1 = SocketLinkProfile(
sink_node_id=NODE_B,
sink_ip="10.0.0.5",
latency_ms=1.0,
latency_jitter_ms=0.1,
upload_mbps=50.0,
download_mbps=100.0,
)
p2 = SocketLinkProfile(
sink_node_id=NODE_B,
sink_ip="10.0.0.5",
latency_ms=2.0,
latency_jitter_ms=0.1,
upload_mbps=120.0,
download_mbps=200.0,
)
state = apply(state, _wrap(0, p1))
state = apply(state, _wrap(1, p2))
profiles = state.node_link_profiles[NODE_A][NODE_B]
socket_profiles = [p for p in profiles if isinstance(p, NodeSocketLinkProfile)]
assert len(socket_profiles) == 1
assert socket_profiles[0].upload_mbps == 120.0
assert socket_profiles[0].download_mbps == 200.0
def test_apply_keeps_separate_socket_profiles_per_ip():
"""A node reachable on multiple IPs (LAN + Tailscale + link-local) gets one
row per IP — they are NOT deduped down to a single socket profile that
bounces between paths as the reconciler probes each IP in turn.
"""
state = State()
lan = SocketLinkProfile(
sink_node_id=NODE_B,
sink_ip="10.0.0.5",
latency_ms=1.0,
latency_jitter_ms=0.1,
upload_mbps=1000.0,
download_mbps=1000.0,
)
tailscale = SocketLinkProfile(
sink_node_id=NODE_B,
sink_ip="100.88.70.34",
latency_ms=5.0,
latency_jitter_ms=0.1,
upload_mbps=400.0,
download_mbps=400.0,
)
state = apply(state, _wrap(0, lan))
state = apply(state, _wrap(1, tailscale))
profiles = state.node_link_profiles[NODE_A][NODE_B]
socket_profiles = [p for p in profiles if isinstance(p, NodeSocketLinkProfile)]
assert len(socket_profiles) == 2
by_ip = {p.sink_ip: p for p in socket_profiles}
assert by_ip["10.0.0.5"].upload_mbps == 1000.0
assert by_ip["100.88.70.34"].upload_mbps == 400.0
def test_apply_keeps_socket_and_rdma_profiles_to_same_peer():
"""A node may have two transports to the same peer (Wi-Fi + Thunderbolt)."""
state = State()
socket_p = SocketLinkProfile(
sink_node_id=NODE_B,
sink_ip="10.0.0.5",
latency_ms=2.0,
latency_jitter_ms=0.1,
upload_mbps=400.0,
download_mbps=900.0,
)
rdma_p = RDMALinkProfile(
sink_node_id=NODE_B,
source_rdma_iface="rdma_en2",
sink_rdma_iface="rdma_en3",
latency_ms=0.05,
latency_jitter_ms=0.1,
upload_mbps=20_000.0,
download_mbps=18_000.0,
payload_bytes=64 * 1024 * 1024,
)
state = apply(state, _wrap(0, socket_p))
state = apply(state, _wrap(1, rdma_p))
profiles = state.node_link_profiles[NODE_A][NODE_B]
transports = sorted(p.transport for p in profiles)
assert transports == ["rdma", "socket"]
def test_apply_node_timed_out_drops_profiles():
state = State()
state = apply(
state,
_wrap(0, GpuProfile(engine="mlx", tflops_fp16=1, memory_bandwidth_gbps=1)),
)
state = apply(
state,
_wrap(
1,
SocketLinkProfile(
sink_node_id=NODE_B,
sink_ip="10.0.0.5",
latency_ms=1,
latency_jitter_ms=0.1,
upload_mbps=1,
download_mbps=1,
),
),
)
timed_out = IndexedEvent(
idx=2,
event=NodeTimedOut(event_id=EventId(), node_id=NODE_A),
)
state = apply(state, timed_out)
assert NODE_A not in state.node_gpu_profile
assert NODE_A not in state.node_link_profiles
def test_apply_node_timed_out_drops_inverse_link_profiles():
"""Removing node B should also remove any profiles where B is the sink."""
state = State()
state = apply(
state,
_wrap(
0,
SocketLinkProfile(
sink_node_id=NODE_B,
sink_ip="10.0.0.5",
latency_ms=1,
latency_jitter_ms=0.1,
upload_mbps=1,
download_mbps=1,
),
),
)
timed_out = IndexedEvent(
idx=1,
event=NodeTimedOut(event_id=EventId(), node_id=NODE_B),
)
state = apply(state, timed_out)
profiles_from_a = state.node_link_profiles.get(NODE_A, {})
assert NODE_B not in profiles_from_a
def test_apply_uses_event_when_for_measured_at():
state = State()
when = "2026-04-30T12:00:00+00:00"
profile = GpuProfile(engine="mlx", tflops_fp16=1.0, memory_bandwidth_gbps=1.0)
state = apply(state, _wrap(0, profile, when=when))
assert state.node_gpu_profile[NODE_A].measured_at == datetime.fromisoformat(when)
@@ -0,0 +1,36 @@
import mlx.core as mx
import pytest
from exo.utils.profilers.gpu_profiler import GpuProfile
@pytest.mark.skipif(
not mx.metal.is_available(),
reason="GPU profile requires Metal — skip on Linux/CPU",
)
async def test_gpu_profile_returns_plausible_numbers():
"""End-to-end: actually run the matmul + GEMV on the local GPU.
We only assert that the numbers are positive and within a sane range.
Hard-coding "expect X TFLOPS on chip Y" would just create flaky tests.
"""
profile = await GpuProfile.measure()
assert profile is not None
assert profile.engine == "mlx"
assert profile.tflops_fp16 > 0
# 1000 TFLOPS would mean we're miscomputing; nothing on the market touches it.
assert profile.tflops_fp16 < 1000
assert profile.memory_bandwidth_gbps > 0
assert profile.memory_bandwidth_gbps < 100_000
def test_gpu_profile_serializes_with_class_tag():
"""Wire format includes the class name as the discriminator (TaggedModel)."""
profile = GpuProfile(
engine="mlx",
tflops_fp16=12.3,
memory_bandwidth_gbps=400.0,
)
dumped = profile.model_dump()
assert "GpuProfile" in dumped
assert dumped["GpuProfile"]["tflops_fp16"] == 12.3
@@ -0,0 +1,140 @@
"""Tests for the socket link profiler.
We use httpx.MockTransport to simulate the peer's API. Two probes happen
inside `measure()` — a `/node_id` GET and a `/profile/echo` POST, the
latter once for latency (small payload) and once for bandwidth.
"""
import httpx
import pytest
from exo.shared.types.common import NodeId
from exo.utils.profilers.link_profiler import (
BANDWIDTH_PAYLOAD_BYTES,
LATENCY_PAYLOAD_BYTES,
SocketLinkProfile,
)
EXPECTED_NODE_ID = NodeId("alice")
WRONG_NODE_ID = NodeId("mallory")
SINK_IP = "10.0.0.5"
API_PORT = 52415
def _make_mock_transport(*, node_id: str) -> httpx.MockTransport:
def handler(request: httpx.Request) -> httpx.Response:
if request.url.path == "/node_id":
return httpx.Response(200, text=node_id)
if request.url.path == "/profile/echo":
body = request.read()
return httpx.Response(
200,
content=body,
headers={"Content-Type": "application/octet-stream"},
)
if request.url.path == "/profile/upload":
body = request.read()
return httpx.Response(
200,
json={
"bytes_received": len(body),
"recv_duration_ms": 1.0,
},
)
if request.url.path == "/profile/download":
from typing import cast as _cast
n_bytes_str = _cast(str, request.url.params.get("bytes", "0"))
n_bytes = int(n_bytes_str) if n_bytes_str.isdigit() else 0
return httpx.Response(
200,
content=b"\x00" * n_bytes,
headers={"Content-Type": "application/octet-stream"},
)
return httpx.Response(404)
return httpx.MockTransport(handler)
async def test_socket_profile_returns_finite_measurements():
transport = _make_mock_transport(node_id=str(EXPECTED_NODE_ID))
async with httpx.AsyncClient(transport=transport) as client:
profile = await SocketLinkProfile.measure(
client=client,
sink_ip=SINK_IP,
expected_sink_node_id=EXPECTED_NODE_ID,
api_port=API_PORT,
)
assert profile is not None
assert profile.sink_node_id == EXPECTED_NODE_ID
assert profile.sink_ip == SINK_IP
assert profile.latency_ms > 0
assert profile.latency_jitter_ms >= 0
assert profile.upload_mbps > 0
assert profile.download_mbps > 0
async def test_socket_profile_rejects_node_id_mismatch():
"""If the IP is reused by a different node, we must not attribute the bandwidth."""
transport = _make_mock_transport(node_id=str(WRONG_NODE_ID))
async with httpx.AsyncClient(transport=transport) as client:
profile = await SocketLinkProfile.measure(
client=client,
sink_ip=SINK_IP,
expected_sink_node_id=EXPECTED_NODE_ID,
api_port=API_PORT,
)
assert profile is None
async def test_socket_profile_returns_none_on_short_echo():
"""Echo endpoint must round-trip the exact payload — anything else is a bug."""
def handler(request: httpx.Request) -> httpx.Response:
if request.url.path == "/node_id":
return httpx.Response(200, text=str(EXPECTED_NODE_ID))
# Truncated echo — peer is misbehaving.
return httpx.Response(200, content=b"")
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
profile = await SocketLinkProfile.measure(
client=client,
sink_ip=SINK_IP,
expected_sink_node_id=EXPECTED_NODE_ID,
api_port=API_PORT,
)
assert profile is None
async def test_socket_profile_returns_none_on_http_error():
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(503)
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
profile = await SocketLinkProfile.measure(
client=client,
sink_ip=SINK_IP,
expected_sink_node_id=EXPECTED_NODE_ID,
api_port=API_PORT,
)
assert profile is None
@pytest.mark.parametrize("ip", ["10.0.0.5", "fe80::1"])
async def test_socket_profile_supports_v4_and_v6(ip: str):
transport = _make_mock_transport(node_id=str(EXPECTED_NODE_ID))
async with httpx.AsyncClient(transport=transport) as client:
profile = await SocketLinkProfile.measure(
client=client,
sink_ip=ip,
expected_sink_node_id=EXPECTED_NODE_ID,
api_port=API_PORT,
)
assert profile is not None
assert profile.sink_ip == ip
def test_payload_constants_are_consistent():
# Latency must be tiny (well under MTU); bandwidth must be much larger.
assert LATENCY_PAYLOAD_BYTES < 1024
assert BANDWIDTH_PAYLOAD_BYTES > 1024 * 1024
@@ -0,0 +1,10 @@
from exo.utils.profilers.rdma_probe_main import build_two_rank_ibv_devs
def test_build_two_rank_ibv_devs_uses_local_iface_per_rank():
assert build_two_rank_ibv_devs(
source_iface="rdma_source", sink_iface="rdma_sink"
) == [
[None, "rdma_source"],
["rdma_sink", None],
]
+9 -1
View File
@@ -55,6 +55,7 @@ from exo.utils.channels import Receiver, Sender, channel
from exo.utils.info_gatherer.info_gatherer import GatheredInfo, InfoGatherer
from exo.utils.info_gatherer.net_profile import check_reachable
from exo.utils.keyed_backoff import KeyedBackoff
from exo.utils.profilers.profiler_manager import ProfilerManager
from exo.utils.task_group import TaskGroup
from exo.worker.plan import plan
from exo.worker.runner.supervisor import RunnerSupervisor
@@ -101,11 +102,18 @@ class Worker:
logger.info("Starting Worker")
info_send, info_recv = channel[GatheredInfo]()
info_gatherer: InfoGatherer = InfoGatherer(info_send)
info_gatherer: InfoGatherer = InfoGatherer(info_send.clone())
profiler_manager: ProfilerManager = ProfilerManager(
info_sender=info_send,
node_id=self.node_id,
api_port=self.api_port,
state_view=lambda: self.state,
)
try:
async with self._tg as tg:
tg.start_soon(info_gatherer.run)
tg.start_soon(profiler_manager.run)
tg.start_soon(self._forward_info, info_recv)
tg.start_soon(self.plan_step)
tg.start_soon(self._event_applier)