diff --git a/examples/cuda-fanout/bend.lsp b/examples/cuda-fanout/bend.lsp index c027c33..ee1e846 100644 --- a/examples/cuda-fanout/bend.lsp +++ b/examples/cuda-fanout/bend.lsp @@ -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) diff --git a/examples/cuda-fanout/gpu-worker.lsp b/examples/cuda-fanout/gpu-worker.lsp index 8973d8a..3158581 100644 --- a/examples/cuda-fanout/gpu-worker.lsp +++ b/examples/cuda-fanout/gpu-worker.lsp @@ -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))))))))