engineering/lumbda
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

bend: worker health heartbeat — (health) op + cache + VRAM-ranked pick

Browse source 
Adds a (health) op handler on the worker side & a lazy-refresh
health cache + VRAM-aware selection on the bend client side.

WORKER (gpu-worker.lsp)
  (health) returns (ok (load-avg L) (vram-free-mb V) (uptime-ms U))
  - L from /proc/loadavg first field
  - V from `nvidia-smi --query-gpu=memory.free --format=csv,noheader,nounits`
    (returns 0 when nvidia-smi missing — host w/o NVIDIA GPU)
  - U from current-time-ms; client detects a worker that hung
    & restarted between probes via uptime jump
  Backward-compat: workers without (health) return
  (error (unknown-op health)); client treats that as ok+vram=0.

CLIENT (bend.lsp)
  *worker-health* alist keyed "host:port" → (last-checked-ms status vram-mb)
  Cache TTL on ok = 5 s; cooldown on down = 30 s.
  bend-pick-worker now:
    - filters out workers in down-cooldown
    - sorts healthy peers by free VRAM descending
    - falls back to round-robin if every worker is in cooldown
  bend-dispatch-to-gpu flips workers to down on tcp-connect-fail
    or empty-reply so a transient failure costs at most one call.

Two lumbda quirks caught while building:
  - (eq? 0 #f) → #t in lumbda. worker-probe-health returns 0
    (a number) for the unknown-op fallback, but if we'd checked
    (eq? vram #f) we would have mis-marked the worker down.
    Now uses (number? vram) instead.
  - tcp-connect raises a Python ConnectionRefusedError (NOT a
    LispErr) on dead-host probes. lumbda's `guard` only catches
    LispErr; only with-exception-handler catches Python
    exceptions. Probe now wraps via with-exception-handler so a
    single dead worker never aborts a fleet iteration.

Smoke on Python tier:
  mixed (127.0.0.1:1 dead + 3090-ai live) → cache shows down for
  the dead one (30s cooldown), ok for live (22777 MB free VRAM,
  measured by the worker's nvidia-smi probe).
This commit is contained in:
russell@unturf.com 2026-06-06 09:25:08 -04:00
parent 99701c863e
commit 58fd787ebf
No known key found for this signature in database
2 changed files with 226 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

173
examples/cuda-fanout/bend.lsp
View file

@ -101,16 +101,166 @@
(define (bend-set-workers! lst)
(set! *bend-workers* lst)
(set! *bend-rr-idx* 0))
(set! *bend-rr-idx* 0)
(set! *worker-health* '()))
;; Round-robin pick a worker from *bend-workers*. Returns (host . port).
;; Caller asserts *bend-workers* non-empty.
;;; -- worker-health heartbeat -----------------------------------
;;;
;;; *worker-health* is an alist keyed by "host:port" -> a record
;;; (last-checked-ms status vram-mb)
;;; status ∈ {ok, down}
;;;
;;; bend-pick-worker filters workers whose cached status=down sits
;;; within *health-down-cooldown-ms*, falls through to a fresh
;;; (health) probe whenever a cache entry ages past
;;; *health-cache-ttl-ms*, & ranks healthy peers by free VRAM
;;; descending so we route long-running jobs to whichever box
;;; carries the most spare GPU memory.
(define *worker-health* '())
(define *health-cache-ttl-ms* 5000)
(define *health-down-cooldown-ms* 30000)
(define *health-probe-timeout-ms* 2000)
(define (worker-key w)
(string-append (car w) ":" (number->string (cdr w))))
(define (worker-health-get w)
(let ((e (assoc (worker-key w) *worker-health*)))
(cond ((eq? e #f) #f) (else (cdr e)))))
(define (worker-health-set! w status vram-mb)
(let ((key (worker-key w))
(entry (list (current-time-ms) status vram-mb)))
(set! *worker-health*
(cons (cons key entry)
(let drop ((rest *worker-health*))
(cond
((null? rest) '())
((string=? (car (car rest)) key) (drop (cdr rest)))
(else (cons (car rest) (drop (cdr rest))))))))))
(define (worker-mark-down! w)
(worker-health-set! w 'down 0))
(define (worker-mark-ok! w vram-mb)
(worker-health-set! w 'ok vram-mb))
;; Extract (vram-free-mb N) from a (health) response form.
(define (health-find-vram form-list)
(let loop ((rest form-list))
(cond
((null? rest) 0)
((and (pair? (car rest)) (eq? (car (car rest)) 'vram-free-mb)
(pair? (cdr (car rest))) (number? (car (cdr (car rest)))))
(car (cdr (car rest))))
(else (loop (cdr rest))))))
;; Send (health) over a fresh tcp-connect. Returns vram-free-mb on
;; success (a number, possibly 0), or symbol 'transport-fail on
;; failure. Lumbda's (eq? 0 #f) reads #t — we deliberately do NOT
;; signal failure via #f so a healthy worker reporting 0 free VRAM
;; (no GPU on host, or VRAM fully consumed by a co-tenant) still
;; ranks as up; calling code checks with `number?`.
;;
;; Backward-compat: a worker without the health handler returns
;; (error (unknown-op health)); treat that as ok with vram=0 so
;; older builds still rank as available.
(define (worker-probe-health-body w)
(let ((sock (tcp-connect (car w) (cdr w))))
(cond
((eq? sock #f) 'transport-fail)
(else
(wire-send sock '(health))
(let ((reply (wire-recv sock)))
(tcp-close sock)
(cond
((eq? reply #f) 'transport-fail)
((not (pair? reply)) 'transport-fail)
((eq? (car reply) 'ok)
(health-find-vram (cdr reply)))
((and (eq? (car reply) 'error)
(pair? (cdr reply))
(pair? (car (cdr reply)))
(eq? (car (car (cdr reply))) 'unknown-op))
0)
(else 'transport-fail)))))))
;; tcp-connect raises a Python ConnectionRefusedError / gaierror on
;; dead-host probes. lumbda's `guard` only catches LispErr; we use
;; with-exception-handler which catches both LispErr & Python
;; Exception. Net effect: probing a single dead worker never
;; aborts iteration over our fleet.
(define (worker-probe-health w)
(with-exception-handler
(lambda (exn) 'transport-fail)
(lambda () (worker-probe-health-body w))))
(define (worker-cache-fresh? entry now)
(let ((age (- now (car entry)))
(status (car (cdr entry))))
(cond
((eq? status 'ok) (< age *health-cache-ttl-ms*))
((eq? status 'down) (< age *health-down-cooldown-ms*))
(else #f))))
;; #t if worker is reachable & not in down-cooldown. Refreshes cache
;; lazily when a probe is needed.
(define (worker-healthy? w)
(let* ((entry (worker-health-get w))
(now (current-time-ms)))
(cond
((and entry (worker-cache-fresh? entry now))
(eq? (car (cdr entry)) 'ok))
(else
(let ((vram (worker-probe-health w)))
(cond
((number? vram) (worker-mark-ok! w vram) #t)
(else (worker-mark-down! w) #f)))))))
(define (worker-cached-vram w)
(let ((entry (worker-health-get w)))
(cond ((eq? entry #f) 0) (else (car (cdr (cdr entry)))))))
;; In-place insertion sort by cached-vram descending.
(define (workers-rank-by-vram lst)
(let outer ((rest lst) (acc '()))
(cond
((null? rest) acc)
(else
(let ((w (car rest)))
(outer (cdr rest)
(let inner ((sorted acc))
(cond
((null? sorted) (list w))
((>= (worker-cached-vram w)
(worker-cached-vram (car sorted)))
(cons w sorted))
(else (cons (car sorted) (inner (cdr sorted))))))))))))
(define (workers-filter-healthy lst)
(let loop ((rest lst) (acc '()))
(cond
((null? rest) (reverse acc))
((worker-healthy? (car rest))
(loop (cdr rest) (cons (car rest) acc)))
(else (loop (cdr rest) acc)))))
;; Pick the healthiest worker (highest free VRAM). If every worker
;; is in down-cooldown, fall back to round-robin so a recovering
;; box gets a real attempt after cooldown — better than refusing
;; to dispatch at all.
(define (bend-pick-worker)
(let* ((n (length *bend-workers*))
(idx (if (= n 0) 0 (remainder *bend-rr-idx* n)))
(pick (list-ref *bend-workers* idx)))
(set! *bend-rr-idx* (+ *bend-rr-idx* 1))
pick))
(let* ((healthy (workers-filter-healthy *bend-workers*))
(n (length healthy)))
(cond
((= n 0)
(let* ((all-n (length *bend-workers*))
(idx (if (= all-n 0) 0 (remainder *bend-rr-idx* all-n))))
(set! *bend-rr-idx* (+ *bend-rr-idx* 1))
(list-ref *bend-workers* idx)))
((= n 1) (car healthy))
(else (car (workers-rank-by-vram healthy))))))
;; Parse "host:port,host:port" into a list of (host . port) pairs. Skips
;; malformed entries silently rather than raising — keeps boot path resilient.
@ -168,13 +318,18 @@
(sock (tcp-connect (car target) (cdr target))))
(cond
((eq? sock #f)
;; Worker unreachable — mark down so the next pick skips it
;; for *health-down-cooldown-ms* before re-probing.
(cond ((not (null? *bend-workers*)) (worker-mark-down! target)))
(bend-error "bend: tcp-connect failed to" target))
(else
(wire-send sock quoted-form)
(let ((reply (wire-recv sock)))
(tcp-close sock)
(cond
((eq? reply #f) (bend-error "bend: worker closed connection"))
((eq? reply #f)
(cond ((not (null? *bend-workers*)) (worker-mark-down! target)))
(bend-error "bend: worker closed connection"))
((not (pair? reply)) (bend-error "bend: malformed reply" reply))
((eq? (car reply) 'ok) (car (cdr reply)))
((eq? (car reply) 'error)

62
examples/cuda-fanout/gpu-worker.lsp
View file

@ -274,6 +274,67 @@
(display wall-ms) (newline)
(list 'error (list 'no-portal portal-path))))))))))
;;; -- op handler -- health --------------------------------------
;;;
;;; (health) returns
;;; (ok (load-avg L) (vram-free-mb V) (uptime-ms U))
;;;
;;; L = 1-minute load average parsed from /proc/loadavg
;;; V = MB of free VRAM from `nvidia-smi --query-gpu=memory.free`
;;; U = current-time-ms (millisecond wallclock, lets a client detect
;;; a worker that hung & restarted between probes)
;;;
;;; Cheap (sub-millisecond load-avg + a single nvidia-smi exec for
;;; VRAM). Cache TTL on a client makes per-dispatch polling free.
(define (parse-leading-number s)
;; Longest digit-or-decimal-point run from index 0; 0 on failure.
(let* ((n (string-length s))
(end (let scan ((i 0))
(cond
((>= i n) i)
((let ((c (string-ref s i)))
(or (char-numeric? c) (char=? c #\.)))
(scan (+ i 1)))
(else i)))))
(cond
((= end 0) 0)
(else (or (string->number (substring s 0 end)) 0)))))
(define (health-load-avg)
(let ((s (file->string "/proc/loadavg")))
(cond
((eq? s #f) 0)
(else (parse-leading-number s)))))
(define (health-vram-free-mb)
;; nvidia-smi --query-gpu=memory.free --format=csv,noheader,nounits
;; one number per GPU, one per line; first line suffices.
;; Returns 0 when nvidia-smi missing (host without an NVIDIA GPU)
;; or when the spawn output cannot be parsed — never raises.
(cond
((not (file-exists? "/usr/bin/nvidia-smi")) 0)
(else
(let ((pair (spawn-process-stdio "/usr/bin/nvidia-smi"
'("--query-gpu=memory.free"
"--format=csv,noheader,nounits"))))
(cond
((eq? pair #f) 0)
(else
(let ((line (read-line (cdr pair))))
(close-port (car pair))
(close-port (cdr pair))
(cond
((eq? line #f) 0)
((eof-object? line) 0)
(else (parse-leading-number line))))))))))
(define (handle-health)
(list 'ok
(list 'load-avg (health-load-avg))
(list 'vram-free-mb (health-vram-free-mb))
(list 'uptime-ms (current-time-ms))))
;;; -- dispatch --------------------------------------------------
(define (handle-request sexp)
@ -284,6 +345,7 @@
(cond
((eq? op 'cuda-shake-fanout) (handle-cuda-shake-fanout args))
((eq? op 'cuda-sim-ops-bin) (handle-cuda-sim-ops-bin args))
((eq? op 'health) (handle-health))
((eq? op 'ping) (list 'ok 'pong))
(else (list 'error (list 'unknown-op op))))))))