diff --git a/examples/cuda-fanout/gpu-worker.lsp b/examples/cuda-fanout/gpu-worker.lsp index 23fa1c3..3660faa 100644 --- a/examples/cuda-fanout/gpu-worker.lsp +++ b/examples/cuda-fanout/gpu-worker.lsp @@ -350,11 +350,98 @@ ((eof-object? line) 0) (else (parse-leading-number line)))))))))) +;;; -- pool / queue health ------------------------------------------- +;;; +;;; bend co-lives with an ecdsa-emit-pool on each foxhop production host. +;;; When the pool dies but bend stays up, .lsp cells pile up un-emitted +;;; and bend sits idle waiting for .ready bins that never arrive. The +;;; feeder daemon polls bend's (health) to know whether to push more +;;; work — adding pool/queue counts here means a single RPC tells the +;;; feeder everything: bend health, pool health, queue depth. +;;; +;;; BEND_QUEUE_DIR env var — set when bend is launched on a host with +;;; an associated pool. Absent on hosts that only run bend without a +;;; pool. When absent, queue counts return -1 (caller treats as +;;; "unknown / not applicable"). + +(define (run-and-count cmd-args) + ;; Spawn cmd-args, read all stdout lines, return integer line count. + ;; Returns 0 on spawn failure. Used to count ls / pgrep output. + (cond + ((not (file-exists? (car cmd-args))) 0) + (else + (let ((pair (spawn-process-stdio (car cmd-args) (cdr cmd-args)))) + (cond + ((eq? pair #f) 0) + (else + (let loop ((count 0)) + (let ((line (read-line (cdr pair)))) + (cond + ((or (eq? line #f) (eof-object? line)) + (close-port (car pair)) + (close-port (cdr pair)) + count) + (else (loop (+ count 1)))))))))))) + +(define (health-pool-procs) + ;; Count of ecdsa-emit-pool processes on this host. 0 = pool dead. + (let* ((pair (spawn-process-stdio "/usr/bin/pgrep" + '("-cf" "ecdsa-emit-pool")))) + (cond + ((eq? pair #f) 0) + (else + (let ((line (read-line (cdr pair)))) + (close-port (car pair)) + (close-port (cdr pair)) + (cond + ((or (eq? line #f) (eof-object? line)) 0) + (else (or (string->number (parse-trim line)) 0)))))))) + +(define (parse-trim s) + ;; Strip leading/trailing whitespace from a single-line string. + (let* ((n (string-length s)) + (lo (let ll ((i 0)) + (cond + ((>= i n) i) + ((char-whitespace? (string-ref s i)) (ll (+ i 1))) + (else i)))) + (hi (let lh ((i (- n 1))) + (cond + ((< i lo) lo) + ((char-whitespace? (string-ref s i)) (lh (- i 1))) + (else (+ i 1)))))) + (substring s lo hi))) + +(define (health-queue-count ext) + ;; Count of files matching $BEND_QUEUE_DIR/*.. Returns -1 when + ;; BEND_QUEUE_DIR env is unset (caller treats as unknown). + (let ((qdir (get-environment-variable "BEND_QUEUE_DIR"))) + (cond + ((or (eq? qdir #f) (string=? qdir "")) -1) + (else + (let* ((pattern (string-append qdir "/*." ext)) + (pair (spawn-process-stdio "/bin/sh" + (list "-c" + (string-append "ls " pattern " 2>/dev/null | wc -l"))))) + (cond + ((eq? pair #f) 0) + (else + (let ((line (read-line (cdr pair)))) + (close-port (car pair)) + (close-port (cdr pair)) + (cond + ((or (eq? line #f) (eof-object? line)) 0) + (else (or (string->number (parse-trim line)) 0))))))))))) + (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)))) + (list 'uptime-ms (current-time-ms)) + (list 'pool-procs (health-pool-procs)) + (list 'queue-ready (health-queue-count "ready")) + (list 'queue-emitting (health-queue-count "emitting")) + (list 'queue-done (health-queue-count "done")))) ;;; -- dispatch --------------------------------------------------