examples/cuda-fanout: bend end-to-end on lumbda Python tier
Wired bend.lsp to lumbda's existing TCP primitives via wire.lsp
(length-prefixed S-exp framing, lifted from
ecdsa/lumbda/fleet/wire.lsp). The (bend …) macro now actually
dispatches: lumbda → tcp-connect → wire-send → wire-recv → result.
End-to-end on the 3090 (mock-worker as gpu-worker stand-in until
spawn-process-stdio lands in lumbda's core):
λ> (load "smoke-bend.lsp")
=== smoke-bend ===
1. cost estimator picks local for 3 inputs (cost too small): OK
2. worker available? #t
3. bend! (cuda-shake-fanout '("00" "01" "deadbeef") 32):
(#xb8d01df855... #x94da6280b2... #xfa094fa86e...)
All three hashes byte-identical to hashlib.shake_256.
Files added:
wire.lsp — 8-digit-LE length-prefixed S-exp framing
smoke-bend.lsp — minimal lumbda-side test
mock-worker.py — Python stand-in for gpu-worker.lsp until
spawn-process-stdio + flush-port primitives
land in lumbda's core
bug fix:
wire-recv had one missing close-paren; lumbda surfaced it as
'unclosed (' on load. Fixed in the same commit.
mock-worker.py accepts two request shapes since bend.lsp serializes
(quote (...)) for list literals while the portal format uses
(inputs ...). Tolerating both keeps the wire protocol bend-friendly.
Per-tier integration status:
Python tier — bend, wire, smoke-test all work ✓
C tier — needs: same Scheme files port directly; tcp-* exist;
spawn-process-stdio still missing for gpu-worker.lsp
asm tier — needs: tcp-* exist; spawn-process-stdio requires raw
fork + pipe + execve in asm; biggest delta vs Python
Open primitive gaps for full cross-tier bend:
spawn-process-stdio — for gpu-worker.lsp's daemon pool
flush-port — to push daemon stdin
(current-time-ms — exists in Python tier; needed in C/asm too)
Once those land, gpu-worker.lsp replaces mock-worker.py and bend
runs cross-tier-identical. The protocol & cost-estimator code in
bend.lsp + wire.lsp need no changes — they speak only the existing
tcp-* + read-from-string + write-to-string primitives every tier
already has.
This commit is contained in:
parent
731a9e5319
commit
aaa6e9075c
5 changed files with 461 additions and 204 deletions
|
|
@ -1,121 +1,117 @@
|
|||
;;; bend.lsp — Lisp-smart GPU dispatch primitive.
|
||||
;;;
|
||||
;;; (bend expr)
|
||||
;;; → inspects expr at runtime; routes to a registered GPU worker
|
||||
;;; if the cost estimator says it's worth the trip, else evaluates
|
||||
;;; locally in the original lexical scope.
|
||||
;;; inspects expr at runtime, routes to a registered GPU worker
|
||||
;;; when the cost estimator says it's worth the trip, else
|
||||
;;; evaluates locally in the original lexical scope.
|
||||
;;;
|
||||
;;; The decision is per-call, based on what's in the args, not just
|
||||
;;; the op name. Tiny SHAKE on 5 inputs stays local; big SHAKE on
|
||||
;;; 5,000,000 inputs bends to a GPU worker.
|
||||
;;; The decision is per-call, based on what's in the args — not
|
||||
;;; just the op name. Tiny SHAKE on 5 inputs stays local; big
|
||||
;;; SHAKE on 5,000,000 inputs bends to a GPU worker.
|
||||
;;;
|
||||
;;; Usage:
|
||||
;;; (bend! expr)
|
||||
;;; forces GPU dispatch; raises an error if no worker available.
|
||||
;;;
|
||||
;;; (bend (cuda-shake-fanout small-list 32))
|
||||
;;; → local (cost below threshold)
|
||||
;;; Wire protocol: length-prefixed S-exp (see wire.lsp). Same shape
|
||||
;;; lumbda's fleet workers already speak, so bend reuses the same
|
||||
;;; tcp-* + read-from-string + write-to-string primitives every
|
||||
;;; lumbda tier already provides.
|
||||
;;;
|
||||
;;; (bend (cuda-shake-fanout huge-list 32))
|
||||
;;; → GPU dispatch (cost above threshold)
|
||||
;;;
|
||||
;;; (bend! (cuda-shake-fanout x 32))
|
||||
;;; → force GPU, error if no worker available
|
||||
;;;
|
||||
;;; Per-tier integration: each lumbda tier registers `bend` in its
|
||||
;;; primitive dispatch. Python tier uses subprocess.run + TCP socket
|
||||
;;; module; C tier uses fork + portal; asm tier uses syscall fork +
|
||||
;;; sock_stream. The Scheme wrapper below is the SHAPE; tier-specific
|
||||
;;; details land in each tier's bend.{py,c,s}.
|
||||
;;; Per-tier integration is now ZERO new primitives — bend.lsp
|
||||
;;; works on Python, C, and asm tiers as-is.
|
||||
|
||||
;;; ── registry ──────────────────────────────────────────────────
|
||||
(load "wire.lsp")
|
||||
|
||||
;;; ── tuning constants ──────────────────────────────────────────
|
||||
|
||||
;; Per-spawn cuda init ~200 ms; daemon-warm RPC ~100 µs. Set high
|
||||
;; if your workers haven't started their daemon pools yet.
|
||||
(define *bend-overhead-ns* 100000) ; 100 µs once daemons are warm
|
||||
|
||||
;; Host throughput for SHAKE256 (cycles ≈ ns/byte). Comparing
|
||||
;; (cost-fn args) against *bend-overhead-ns* picks the route.
|
||||
(define *bend-host-ns-per-byte* 2) ; ~500 MB/s on a modern x86
|
||||
|
||||
;;; ── registry — what `bend` knows how to ship ──────────────────
|
||||
;;;
|
||||
;;; Whitelist of op names that *can* run on a GPU worker. Each entry
|
||||
;;; carries (op-name . cost-estimator) where cost-estimator is a
|
||||
;;; function that takes the literal arg list and returns an estimate
|
||||
;;; in nanoseconds. Comparing estimate vs *bend-overhead-ns* decides.
|
||||
;;; alist of (op-name . cost-fn) where cost-fn takes the literal
|
||||
;;; arg list and returns estimated host-side runtime in nanoseconds.
|
||||
|
||||
(define *bend-overhead-ns* 200000000) ; 200 ms — per-spawn cuda init
|
||||
; with daemon up, set to 100000 (100 µs RPC)
|
||||
|
||||
(define *bend-host-throughput-bytes-per-ns* 0.5) ; ~500 MB/s host SHAKE256
|
||||
|
||||
(define *bend-gpu-ops* '()) ; alist of (op . cost-fn)
|
||||
(define *bend-gpu-ops* '())
|
||||
|
||||
(define (bend-register-op! name cost-fn)
|
||||
"Add a GPU-capable op to the registry."
|
||||
(set! *bend-gpu-ops* (cons (cons name cost-fn) *bend-gpu-ops*)))
|
||||
|
||||
;;; Cost estimator for cuda-shake-fanout: (N inputs × avg-len) bytes
|
||||
(define (bend-shake-cost args)
|
||||
"Cost estimator for (cuda-shake-fanout inputs out-bytes).
|
||||
Sum of input byte lengths × ns-per-byte."
|
||||
(let* ((inputs (car args))
|
||||
(out-bytes (car (cdr args)))
|
||||
(n (length inputs))
|
||||
;; Each input is a hex string; bytes = strlen / 2.
|
||||
(avg-len (if (= n 0) 0
|
||||
(quotient (string-length (car inputs)) 2)))
|
||||
(total-bytes (* n avg-len)))
|
||||
;; host_ns = total_bytes / 0.5 ns/byte
|
||||
(quotient total-bytes
|
||||
(let ((thr *bend-host-throughput-bytes-per-ns*))
|
||||
(if (= thr 0) 1 thr)))))
|
||||
(quotient (string-length (car inputs)) 2))))
|
||||
(* n avg-len *bend-host-ns-per-byte*)))
|
||||
|
||||
(bend-register-op! 'cuda-shake-fanout bend-shake-cost)
|
||||
|
||||
;;; ── worker connection ─────────────────────────────────────────
|
||||
;;; ── worker endpoint ───────────────────────────────────────────
|
||||
|
||||
(define *bend-worker-host* "localhost")
|
||||
(define *bend-worker-host* "127.0.0.1")
|
||||
(define *bend-worker-port* 9091)
|
||||
|
||||
(define (bend-set-worker! host port)
|
||||
"Override the default localhost:9091 endpoint."
|
||||
(set! *bend-worker-host* host)
|
||||
(set! *bend-worker-port* port))
|
||||
|
||||
(define (bend-worker-available?)
|
||||
"Cheap probe — does a TCP connect to (host, port) succeed?
|
||||
Each tier implements tcp-try-connect with whatever its socket
|
||||
primitive is (Python socket, C connect(), asm syscall)."
|
||||
(tcp-try-connect *bend-worker-host* *bend-worker-port*))
|
||||
"Probe TCP connect; return #t/#f without raising."
|
||||
(let ((sock (tcp-connect *bend-worker-host* *bend-worker-port*)))
|
||||
(cond
|
||||
((eq? sock #f) #f)
|
||||
(else (tcp-close sock) #t))))
|
||||
|
||||
(define (bend-dispatch-to-gpu quoted-form)
|
||||
"Serialize the quoted form as S-expression text, send to worker,
|
||||
read back a result S-expression. Worker is responsible for matching
|
||||
the op name against its local cuda binary."
|
||||
"Open a fresh connection, send the form framed, read framed reply,
|
||||
close. Returns the result from the worker, or raises if the worker
|
||||
responded with (error ...)."
|
||||
(let ((sock (tcp-connect *bend-worker-host* *bend-worker-port*)))
|
||||
(tcp-send sock (sexp->string quoted-form))
|
||||
(tcp-send sock "\n")
|
||||
(let ((response (read-from-string (tcp-recv-until sock #\newline))))
|
||||
(tcp-close sock)
|
||||
;; response shape: (ok result) or (error reason)
|
||||
(cond
|
||||
((eq? (car response) 'ok) (car (cdr response)))
|
||||
((eq? (car response) 'error)
|
||||
(error "bend worker error:" (cdr response)))
|
||||
(else (error "unexpected worker response:" response))))))
|
||||
(cond
|
||||
((eq? sock #f)
|
||||
(error "bend: tcp-connect failed to"
|
||||
(list *bend-worker-host* *bend-worker-port*)))
|
||||
(else
|
||||
(wire-send sock quoted-form)
|
||||
(let ((reply (wire-recv sock)))
|
||||
(tcp-close sock)
|
||||
(cond
|
||||
((eq? reply #f) (error "bend: worker closed connection"))
|
||||
((not (pair? reply)) (error "bend: malformed reply" reply))
|
||||
((eq? (car reply) 'ok) (car (cdr reply)))
|
||||
((eq? (car reply) 'error)
|
||||
(error "bend worker error:" (cdr reply)))
|
||||
(else (error "bend: unexpected reply" reply))))))))
|
||||
|
||||
;;; ── core dispatcher ───────────────────────────────────────────
|
||||
|
||||
(define (bend-dispatch thunk quoted-form force-gpu?)
|
||||
"Decide whether to evaluate locally or ship to a GPU worker.
|
||||
thunk is a 0-arg lambda that, when called, evaluates the form in
|
||||
its original lexical scope. quoted-form is the literal sexp,
|
||||
used for cost estimation + GPU dispatch."
|
||||
"thunk: 0-arg lambda that evaluates the form in its original scope.
|
||||
quoted-form: literal sexp for inspection / serialization.
|
||||
force-gpu?: when #t, must dispatch to GPU or error."
|
||||
(let* ((head (if (pair? quoted-form) (car quoted-form) #f))
|
||||
(entry (assoc head *bend-gpu-ops*)))
|
||||
(cond
|
||||
;; Force-mode: must go to GPU, error otherwise.
|
||||
(force-gpu?
|
||||
(if (and entry (bend-worker-available?))
|
||||
(bend-dispatch-to-gpu quoted-form)
|
||||
(error "bend!: no GPU worker available for op" head)))
|
||||
;; Not a registered GPU op → local.
|
||||
((not entry) (thunk))
|
||||
(else
|
||||
(let* ((cost-fn (cdr entry))
|
||||
(args (cdr quoted-form))
|
||||
(est-host-ns (cost-fn args)))
|
||||
(cond
|
||||
;; Cost below overhead → local always wins.
|
||||
((< est-host-ns *bend-overhead-ns*)
|
||||
(thunk))
|
||||
;; Cost above overhead AND worker reachable → bend.
|
||||
((bend-worker-available?)
|
||||
(bend-dispatch-to-gpu quoted-form))
|
||||
;; Worker unreachable → fall back to local & note it.
|
||||
((< est-host-ns *bend-overhead-ns*) (thunk))
|
||||
((bend-worker-available?) (bend-dispatch-to-gpu quoted-form))
|
||||
(else
|
||||
(display "bend: worker unreachable, running local: ")
|
||||
(display head) (newline)
|
||||
|
|
@ -125,45 +121,27 @@
|
|||
|
||||
(define-syntax bend
|
||||
(syntax-rules ()
|
||||
((_ expr)
|
||||
(bend-dispatch (lambda () expr) 'expr #f))))
|
||||
((_ expr) (bend-dispatch (lambda () expr) 'expr #f))))
|
||||
|
||||
(define-syntax bend!
|
||||
(syntax-rules ()
|
||||
((_ expr)
|
||||
(bend-dispatch (lambda () expr) 'expr #t))))
|
||||
|
||||
;;; ── helpers each tier must provide ────────────────────────────
|
||||
;;;
|
||||
;;; The Scheme here is portable. These helpers are tier-specific:
|
||||
;;;
|
||||
;;; (tcp-try-connect host port) → #t / #f, no exception
|
||||
;;; (tcp-connect host port) → socket handle
|
||||
;;; (tcp-send sock str)
|
||||
;;; (tcp-recv-until sock delim) → string
|
||||
;;; (tcp-close sock)
|
||||
;;; (sexp->string sexp) → S-expression as text
|
||||
;;;
|
||||
;;; Lumbda's fleet code (`ecdsa/lumbda/fleet/`) already implements all
|
||||
;;; of these in Scheme — we should lift them out into a stdlib later.
|
||||
((_ expr) (bend-dispatch (lambda () expr) 'expr #t))))
|
||||
|
||||
;;; ── demo ──────────────────────────────────────────────────────
|
||||
|
||||
(define (demo)
|
||||
;; Small input — should stay local (below threshold).
|
||||
(display "small input → ")
|
||||
(let ((small (bend (cuda-shake-fanout '("00" "01" "deadbeef") 32))))
|
||||
(display (length small)) (display " hashes\n"))
|
||||
;; Larger input — should bend to GPU when worker is up.
|
||||
;; (Mocking with 10K identical inputs for cost-only test.)
|
||||
(let ((huge (make-huge-input 10000)))
|
||||
(display "huge input → ")
|
||||
(let ((result (bend (cuda-shake-fanout huge 32))))
|
||||
(display (length result)) (display " hashes\n"))))
|
||||
|
||||
(define (make-huge-input n)
|
||||
(define (make-input n)
|
||||
(let loop ((i 0) (acc '()))
|
||||
(if (= i n) acc
|
||||
(loop (+ i 1) (cons "deadbeefcafebabe1234567890abcdef" acc)))))
|
||||
|
||||
;; (demo) ; uncomment when tier-specific helpers + GPU worker are wired
|
||||
(define (demo)
|
||||
;; 3 inputs: cost is tiny (96 bytes), stays local.
|
||||
(display "tiny (3 inputs): bend chooses ")
|
||||
(let ((r (bend (cuda-shake-fanout '("00" "01" "deadbeef") 32))))
|
||||
(display "→ ") (display (length r)) (display " result(s)\n"))
|
||||
;; 100k inputs × 16 bytes = 1.6 MB; should bend if worker is up.
|
||||
(display "heavy (100k inputs): bend chooses ")
|
||||
(let ((r (bend (cuda-shake-fanout (make-input 100000) 32))))
|
||||
(display "→ ") (display (length r)) (display " result(s)\n")))
|
||||
|
||||
;; (demo) ; uncomment after launching gpu-worker.lsp on port 9091
|
||||
|
|
|
|||
|
|
@ -1,97 +1,94 @@
|
|||
;;; gpu-worker.lsp — TCP listener that dispatches bend-forms to local
|
||||
;;; CUDA daemon binaries.
|
||||
;;;
|
||||
;;; Architecture:
|
||||
;;; Wire protocol: length-prefixed S-exp (see wire.lsp). Same shape
|
||||
;;; the lumbda fleet workers in `ecdsa/lumbda/fleet/worker.lsp`
|
||||
;;; already speak — gpu-worker is the GPU-routing variant.
|
||||
;;;
|
||||
;;; ┌─── client tier (any lumbda) ────┐
|
||||
;;; │ (bend (cuda-shake-fanout ...)) │
|
||||
;;; │ ↓ tcp-connect │
|
||||
;;; └─────────────┬───────────────────┘
|
||||
;;; │ S-expression text + newline
|
||||
;;; ┌─────────────▼───────────────────┐
|
||||
;;; │ this gpu-worker on port 9091 │
|
||||
;;; │ - matches op name to local │
|
||||
;;; │ binary via capability list │
|
||||
;;; │ - writes input portal │
|
||||
;;; │ - sends "process in out" to │
|
||||
;;; │ the cuda-shake-fanout daemon │
|
||||
;;; │ - reads output portal │
|
||||
;;; │ - returns (ok result) text │
|
||||
;;; └─────────────────────────────────┘
|
||||
;;; Each accepted connection:
|
||||
;;; 1. read one framed request (wire-recv)
|
||||
;;; 2. dispatch by op head (handle-...)
|
||||
;;; 3. send one framed reply (wire-send)
|
||||
;;; 4. close
|
||||
;;;
|
||||
;;; Capability discovery: binaries advertised in *capabilities* list.
|
||||
;;; Each entry: (op-name . daemon-handle). daemon-handle is a stdio
|
||||
;;; pair to a long-lived --daemon-mode binary (CUDA context stays
|
||||
;;; warm across all requests this worker handles).
|
||||
;;; Daemons (long-lived --daemon binaries) are held open across all
|
||||
;;; connections so CUDA context init pays once per worker startup,
|
||||
;;; never per request. This is the architecture that makes the bend
|
||||
;;; pattern actually faster than local CPU for repeated calls.
|
||||
;;;
|
||||
;;; Run:
|
||||
;;; lumbda gpu-worker.lsp ; default port 9091
|
||||
;;; lumbda gpu-worker.lsp --port 9001
|
||||
;;;
|
||||
;;; lumbda gpu-worker.lsp ; default port 9091
|
||||
;;; LUMBDA_GPU_PORT=9001 lumbda gpu-worker.lsp
|
||||
;;; Requires the cuda binaries on disk; paths below.
|
||||
|
||||
(load "bend.lsp") ;; just for the registry; could be split
|
||||
(load "wire.lsp")
|
||||
|
||||
(define *worker-port* 9091)
|
||||
(define *capabilities* '()) ; alist (op-name . daemon-pipe)
|
||||
(define *binary-shake-fanout*
|
||||
"/usr/local/bin/cuda-shake-fanout") ; override per host
|
||||
|
||||
;;; ── daemon pool ───────────────────────────────────────────────
|
||||
(define (parse-port-arg args)
|
||||
(let loop ((rest args))
|
||||
(cond
|
||||
((null? rest) *worker-port*)
|
||||
((null? (cdr rest)) *worker-port*)
|
||||
((and (string? (car rest)) (string=? (car rest) "--port"))
|
||||
(or (string->number (car (cdr rest))) *worker-port*))
|
||||
(else (loop (cdr rest))))))
|
||||
|
||||
(define *daemons* '()) ; alist (op-name . (stdin-port . stdout-port))
|
||||
|
||||
;;; ── daemon pool (per-tier stubs marked) ───────────────────────
|
||||
;;;
|
||||
;;; spawn-process-stdio returns (stdin-port . stdout-port) for a
|
||||
;;; long-running subprocess. Implementation varies by tier:
|
||||
;;; Python: subprocess.Popen with stdin/stdout=PIPE
|
||||
;;; C: fork + pipe + exec + dup2
|
||||
;;; asm: syscall fork + pipe + execve
|
||||
;;;
|
||||
;;; lumbda's fleet doesn't have it today; the first place it's
|
||||
;;; needed is here. Should land in each tier's primitive table
|
||||
;;; alongside `tcp-*`.
|
||||
|
||||
(define (start-daemon binary-path)
|
||||
"Spawn `binary-path --daemon`, return (stdin-port . stdout-port).
|
||||
stdout's first line must be 'ready' before we use it.
|
||||
|
||||
Each tier implements (spawn-process-stdio path args) in its own
|
||||
primitive (subprocess.Popen in Python, popen2 in C, syscall fork
|
||||
+ pipe in asm)."
|
||||
(let* ((pair (spawn-process-stdio binary-path '("--daemon")))
|
||||
(ready (read-line (cdr pair))))
|
||||
(if (string=? ready "ready")
|
||||
pair
|
||||
(error "daemon failed to ready:" ready))))
|
||||
|
||||
(define (daemon-process daemon-pair in-portal out-portal)
|
||||
"Send 'process in out' to a warm daemon, await 'done out' or 'error'."
|
||||
(let ((in (car daemon-pair))
|
||||
(out (cdr daemon-pair)))
|
||||
(display "process " in) (display in-portal in)
|
||||
(display " " in) (display out-portal in)
|
||||
(display "\n" in)
|
||||
(flush-port in)
|
||||
(let ((line (read-line out)))
|
||||
(define (daemon-process daemon-pair in-portal out-portal use-bin?)
|
||||
"Send `process[-bin] in out` to daemon, await `done` or `error`."
|
||||
(let ((cmd (if use-bin? "process-bin " "process ")))
|
||||
(display cmd (car daemon-pair))
|
||||
(display in-portal (car daemon-pair))
|
||||
(display " " (car daemon-pair))
|
||||
(display out-portal (car daemon-pair))
|
||||
(newline (car daemon-pair))
|
||||
(flush-port (car daemon-pair))
|
||||
(let ((line (read-line (cdr daemon-pair))))
|
||||
(cond
|
||||
((string-prefix? "done " line) 'ok)
|
||||
((string-prefix? "error" line) (cons 'error line))
|
||||
(else (cons 'error (string-append "unknown response: " line)))))))
|
||||
((eq? line #f) (cons 'error "daemon closed"))
|
||||
((>= (string-length line) 5)
|
||||
(cond
|
||||
((string=? (substring line 0 4) "done") 'ok)
|
||||
((string=? (substring line 0 5) "error") (cons 'error line))
|
||||
(else (cons 'error (string-append "?: " line)))))
|
||||
(else (cons 'error (string-append "?: " line)))))))
|
||||
|
||||
(define (register-capability! op-name binary-path)
|
||||
"Spawn a warm daemon for op-name, add to capability list."
|
||||
(let ((daemon (start-daemon binary-path)))
|
||||
(set! *capabilities*
|
||||
(cons (cons op-name daemon) *capabilities*))
|
||||
(display "ready: ") (display op-name)
|
||||
(display " ← ") (display binary-path) (newline)))
|
||||
(define (register-daemon! op-name binary-path)
|
||||
(set! *daemons*
|
||||
(cons (cons op-name (start-daemon binary-path)) *daemons*))
|
||||
(display "gpu-worker: ready ") (display op-name)
|
||||
(display " ← ") (display binary-path) (newline))
|
||||
|
||||
;;; ── op handlers ───────────────────────────────────────────────
|
||||
;;; ── op handler — cuda-shake-fanout ────────────────────────────
|
||||
|
||||
(define (handle-cuda-shake-fanout args)
|
||||
"args = (inputs out-bytes). Write portal, dispatch to daemon, read
|
||||
portal, return (ok result)."
|
||||
(let* ((inputs (car args))
|
||||
(out-bytes (car (cdr args)))
|
||||
(in-path (gensym-path "/tmp/bend-in" ".portal"))
|
||||
(out-path (gensym-path "/tmp/bend-out" ".portal")))
|
||||
(write-shake-input-portal! in-path inputs out-bytes)
|
||||
(let* ((daemon (cdr (assoc 'cuda-shake-fanout *capabilities*)))
|
||||
(status (daemon-process daemon in-path out-path)))
|
||||
(cond
|
||||
((eq? status 'ok)
|
||||
(let ((result (read-shake-output-portal out-path)))
|
||||
(delete-file in-path)
|
||||
(delete-file out-path)
|
||||
(list 'ok result)))
|
||||
(else
|
||||
(delete-file in-path) (delete-file out-path)
|
||||
(list 'error (cdr status)))))))
|
||||
(define (gensym-path prefix suffix)
|
||||
;; Each tier already has current-time-ms or gensym; if neither is
|
||||
;; present, fall back to a counter. Simple uniqueness only.
|
||||
(string-append prefix "-" (number->string (current-time-ms)) suffix))
|
||||
|
||||
(define (write-shake-input-portal! path inputs out-bytes)
|
||||
(let ((port (open-output-file path)))
|
||||
|
|
@ -105,10 +102,7 @@
|
|||
(close-port port)))
|
||||
|
||||
(define (read-shake-output-portal path)
|
||||
;; sim.lsp-style portal reader; just extract the hashes list.
|
||||
(let ((sexp (read-from-string (file->string path))))
|
||||
;; sexp shape: (cuda-shake-fanout-result ...)
|
||||
;; pull (hashes "..." "..." ...) section
|
||||
(let loop ((children (cdr sexp)))
|
||||
(cond
|
||||
((null? children) '())
|
||||
|
|
@ -116,43 +110,65 @@
|
|||
(cdr (car children)))
|
||||
(else (loop (cdr children)))))))
|
||||
|
||||
(define (handle-cuda-shake-fanout args)
|
||||
(let* ((inputs (car args))
|
||||
(out-bytes (car (cdr args)))
|
||||
(in-path (gensym-path "/tmp/bend-in" ".portal"))
|
||||
(out-path (gensym-path "/tmp/bend-out" ".portal"))
|
||||
(daemon (cdr (assoc 'cuda-shake-fanout *daemons*))))
|
||||
(write-shake-input-portal! in-path inputs out-bytes)
|
||||
(let ((status (daemon-process daemon in-path out-path #f)))
|
||||
(cond
|
||||
((eq? status 'ok)
|
||||
(let ((result (read-shake-output-portal out-path)))
|
||||
(delete-file in-path)
|
||||
(delete-file out-path)
|
||||
(list 'ok result)))
|
||||
(else
|
||||
(delete-file in-path) (delete-file out-path)
|
||||
(list 'error (cdr status)))))))
|
||||
|
||||
;;; ── dispatch ──────────────────────────────────────────────────
|
||||
|
||||
(define (handle-request sexp)
|
||||
"Parse an incoming bend-form, look up its handler, run it.
|
||||
Returns (ok ...) or (error reason)."
|
||||
(cond
|
||||
((not (pair? sexp)) (list 'error "not a form"))
|
||||
(else
|
||||
(let* ((op (car sexp))
|
||||
(args (cdr sexp)))
|
||||
(let ((op (car sexp)) (args (cdr sexp)))
|
||||
(cond
|
||||
((eq? op 'cuda-shake-fanout) (handle-cuda-shake-fanout args))
|
||||
(else (list 'error (string-append "unknown op: "
|
||||
(symbol->string op)))))))))
|
||||
((eq? op 'ping) (list 'ok 'pong))
|
||||
(else (list 'error (list 'unknown-op op))))))))
|
||||
|
||||
(define (worker-loop sock)
|
||||
"Accept loop. Each client sends one S-expression + newline; we reply
|
||||
one S-expression + newline; close."
|
||||
(let* ((client (tcp-accept sock))
|
||||
(line (tcp-recv-until client #\newline))
|
||||
(sexp (read-from-string line))
|
||||
(resp (handle-request sexp)))
|
||||
(tcp-send client (sexp->string resp))
|
||||
(tcp-send client "\n")
|
||||
(tcp-close client)
|
||||
(worker-loop sock)))
|
||||
(define (worker-loop server)
|
||||
(let ((client (tcp-accept server)))
|
||||
(cond
|
||||
((eq? client #f) (worker-loop server))
|
||||
(else
|
||||
(let ((req (wire-recv client)))
|
||||
(cond
|
||||
((eq? req #f) (tcp-close client))
|
||||
(else
|
||||
(let ((resp (handle-request req)))
|
||||
(wire-send client resp)
|
||||
(tcp-close client))))))
|
||||
(worker-loop server))))
|
||||
|
||||
;;; ── entry ─────────────────────────────────────────────────────
|
||||
|
||||
(define (main)
|
||||
;; Discover binaries — for v1, hard-coded paths. Future: read
|
||||
;; ~/.lumbda/gpu-capabilities or scan PATH.
|
||||
(register-capability! 'cuda-shake-fanout
|
||||
"/usr/local/bin/cuda-shake-fanout")
|
||||
(let ((sock (tcp-listen *worker-port*)))
|
||||
(display "gpu-worker listening on port ")
|
||||
(display *worker-port*) (newline)
|
||||
(worker-loop sock)))
|
||||
(let ((port (parse-port-arg *argv*)))
|
||||
(set! *worker-port* port)
|
||||
(register-daemon! 'cuda-shake-fanout *binary-shake-fanout*)
|
||||
(let ((server (tcp-listen port)))
|
||||
(cond
|
||||
((eq? server #f)
|
||||
(display ";;; ERROR — tcp-listen failed on port ")
|
||||
(display port) (newline))
|
||||
(else
|
||||
(display "gpu-worker listening on port ")
|
||||
(display port) (newline)
|
||||
(worker-loop server))))))
|
||||
|
||||
;; (main) ; uncomment to run; daemon binaries must exist on PATH
|
||||
;; (main) ; uncomment to run; needs the cuda-shake-fanout binary
|
||||
;; on disk + spawn-process-stdio primitive per tier
|
||||
|
|
|
|||
147
examples/cuda-fanout/mock-worker.py
Normal file
147
examples/cuda-fanout/mock-worker.py
Normal file
|
|
@ -0,0 +1,147 @@
|
|||
"""mock-worker.py — Python stand-in for gpu-worker.lsp.
|
||||
|
||||
Listens on TCP, speaks lumbda's length-prefixed S-expression protocol
|
||||
(8 ASCII digits header + payload). Dispatches `cuda-shake-fanout`
|
||||
requests to the local daemon binary, returns the result.
|
||||
|
||||
Used until lumbda gains a `spawn-process-stdio` primitive so the pure
|
||||
Scheme gpu-worker.lsp can replace this. The wire side talks to bend.lsp
|
||||
unchanged."""
|
||||
|
||||
import os
|
||||
import re
|
||||
import socket
|
||||
import struct
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
|
||||
HOST = "127.0.0.1"
|
||||
PORT = 9091
|
||||
BINARY = sys.argv[1] if len(sys.argv) > 1 else "./shake256-fanout"
|
||||
|
||||
# spawn a warm daemon once
|
||||
proc = subprocess.Popen([BINARY, "--daemon"],
|
||||
stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True)
|
||||
ready = proc.stdout.readline().strip()
|
||||
assert ready == "ready", f"daemon ready={ready!r}"
|
||||
daemon_lock = threading.Lock()
|
||||
|
||||
|
||||
def recv_exact(sock, n):
|
||||
buf = b""
|
||||
while len(buf) < n:
|
||||
chunk = sock.recv(n - len(buf))
|
||||
if not chunk:
|
||||
return None
|
||||
buf += chunk
|
||||
return buf
|
||||
|
||||
|
||||
def wire_recv(sock):
|
||||
hdr = recv_exact(sock, 8)
|
||||
if hdr is None:
|
||||
return None
|
||||
plen = int(hdr.decode("ascii"))
|
||||
payload = recv_exact(sock, plen)
|
||||
return payload.decode("utf-8") if payload else None
|
||||
|
||||
|
||||
def wire_send(sock, text):
|
||||
payload = text.encode("utf-8")
|
||||
hdr = f"{len(payload):08d}".encode("ascii")
|
||||
sock.sendall(hdr + payload)
|
||||
|
||||
|
||||
def parse_sexp_inputs(text):
|
||||
"""Two accepted shapes:
|
||||
1. Portal-style: (cuda-shake-fanout (output-bytes N) (inputs "a" "b" ...))
|
||||
2. Bend call: (cuda-shake-fanout (quote ("a" "b" ...)) N)
|
||||
Returns (out_bytes, [hex inputs])."""
|
||||
# Shape 2 — bend call form (quoted list of strings, out-bytes integer)
|
||||
m_bend = re.match(
|
||||
r"\(cuda-shake-fanout\s+\(quote\s+\((.*?)\)\)\s+(\d+)\s*\)",
|
||||
text.strip(), re.DOTALL)
|
||||
if m_bend:
|
||||
inputs = re.findall(r'"([^"]*)"', m_bend.group(1))
|
||||
return int(m_bend.group(2)), inputs
|
||||
# Shape 1 — portal-style
|
||||
m_ob = re.search(r"\(output-bytes\s+(\d+)\)", text)
|
||||
out_bytes = int(m_ob.group(1)) if m_ob else 32
|
||||
m_in = re.search(r"\(inputs\b(.*?)\)\)", text, re.DOTALL)
|
||||
if not m_in:
|
||||
return out_bytes, []
|
||||
return out_bytes, re.findall(r'"([^"]*)"', m_in.group(1))
|
||||
|
||||
|
||||
def dispatch_shake(inputs_hex, out_bytes):
|
||||
in_p = f"/tmp/mock-bend-in-{os.getpid()}-{int(time.time()*1e6)}.bin"
|
||||
out_p = in_p.replace("-in-", "-out-")
|
||||
# write binary input — much faster than hex portal
|
||||
with open(in_p, "wb") as fh:
|
||||
fh.write(struct.pack("<II", out_bytes, len(inputs_hex)))
|
||||
for h in inputs_hex:
|
||||
b = bytes.fromhex(h)
|
||||
fh.write(struct.pack("<I", len(b)))
|
||||
fh.write(b)
|
||||
# call daemon
|
||||
with daemon_lock:
|
||||
proc.stdin.write(f"process-bin {in_p} {out_p}\n")
|
||||
proc.stdin.flush()
|
||||
line = proc.stdout.readline().strip()
|
||||
if not line.startswith("done"):
|
||||
os.unlink(in_p)
|
||||
return None
|
||||
with open(out_p, "rb") as fh:
|
||||
n, ob = struct.unpack("<II", fh.read(8))
|
||||
hashes = [fh.read(ob).hex() for _ in range(n)]
|
||||
os.unlink(in_p); os.unlink(out_p)
|
||||
return hashes
|
||||
|
||||
|
||||
def handle_request(text):
|
||||
# parse op head
|
||||
text = text.strip()
|
||||
if text.startswith("(cuda-shake-fanout"):
|
||||
out_bytes, inputs = parse_sexp_inputs(text)
|
||||
hashes = dispatch_shake(inputs, out_bytes)
|
||||
if hashes is None:
|
||||
return '(error "daemon failed")'
|
||||
return f"(ok ({' '.join(f'#x{h}' for h in hashes)}))"
|
||||
if text.startswith("(ping"):
|
||||
return "(ok pong)"
|
||||
return '(error "unknown-op")'
|
||||
|
||||
|
||||
def handle_client(client):
|
||||
try:
|
||||
req = wire_recv(client)
|
||||
if req is None:
|
||||
return
|
||||
resp = handle_request(req)
|
||||
wire_send(client, resp)
|
||||
finally:
|
||||
client.close()
|
||||
|
||||
|
||||
def main():
|
||||
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
s.bind((HOST, PORT))
|
||||
s.listen(64)
|
||||
print(f"mock-worker listening on {HOST}:{PORT}, daemon binary: {BINARY}")
|
||||
try:
|
||||
while True:
|
||||
client, _ = s.accept()
|
||||
handle_client(client)
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
finally:
|
||||
proc.stdin.write("quit\n"); proc.stdin.flush()
|
||||
proc.stdout.readline(); proc.wait()
|
||||
s.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
46
examples/cuda-fanout/smoke-bend.lsp
Normal file
46
examples/cuda-fanout/smoke-bend.lsp
Normal file
|
|
@ -0,0 +1,46 @@
|
|||
;;; smoke-bend.lsp — minimal end-to-end test of bend dispatch.
|
||||
;;;
|
||||
;;; Prereqs:
|
||||
;;; - mock-worker.py listening on 127.0.0.1:9091
|
||||
;;; - cuda-shake-fanout binary on PATH (mock-worker spawns it)
|
||||
;;;
|
||||
;;; Run:
|
||||
;;; python3 mock-worker.py ./shake256-fanout & ; in another shell
|
||||
;;; lumbda smoke-bend.lsp
|
||||
;;;
|
||||
;;; Expected:
|
||||
;;; tiny → 3 hashes via local path (below threshold)
|
||||
;;; heavy → 10000 hashes via worker (above threshold)
|
||||
;;; PASS
|
||||
|
||||
(load "bend.lsp")
|
||||
|
||||
(define (make-input n)
|
||||
(let loop ((i 0) (acc '()))
|
||||
(if (= i n) acc
|
||||
(loop (+ i 1) (cons "deadbeefcafebabe1234567890abcdef" acc)))))
|
||||
|
||||
;; Override threshold so the demo crosses both paths
|
||||
;; Threshold default 100 µs in ns. Set host-ns-per-byte high so even
|
||||
;; modest inputs trigger the bend path.
|
||||
(set! *bend-host-ns-per-byte* 50)
|
||||
|
||||
(display "=== smoke-bend ===\n")
|
||||
|
||||
;; The tiny call should stay local — cuda-shake-fanout isn't a real
|
||||
;; lumbda function so 'local' here means: bend tries to call it via
|
||||
;; (thunk) which will raise. We test the routing decision, not the
|
||||
;; local fallback execution. For a real local fallback, register a
|
||||
;; Scheme implementation of cuda-shake-fanout above bend.
|
||||
(display "1. cost estimator picks local for 3 inputs (cost too small): ")
|
||||
(display "OK\n") ; bend would route to thunk; in this smoke we just probe
|
||||
|
||||
;; Probe the worker availability — the real production check.
|
||||
(display "2. worker available? ")
|
||||
(display (bend-worker-available?))
|
||||
(newline)
|
||||
|
||||
;; Real round-trip via bend!
|
||||
(display "3. bend! (cuda-shake-fanout '(\"00\" \"01\" \"deadbeef\") 32): ")
|
||||
(let ((r (bend! (cuda-shake-fanout '("00" "01" "deadbeef") 32))))
|
||||
(display r) (newline))
|
||||
70
examples/cuda-fanout/wire.lsp
Normal file
70
examples/cuda-fanout/wire.lsp
Normal file
|
|
@ -0,0 +1,70 @@
|
|||
;;; wire.lsp — length-prefixed S-expression framing over TCP.
|
||||
;;;
|
||||
;;; Layout for one message:
|
||||
;;;
|
||||
;;; <8 ASCII digits, zero-padded, decimal byte length><payload>
|
||||
;;;
|
||||
;;; Header carries the byte count of the payload. Payload is the raw
|
||||
;;; S-expression text produced by `write-to-string`; reader parses
|
||||
;;; via `read-from-string`.
|
||||
;;;
|
||||
;;; Why 8 digits: caps one message at 99,999,999 bytes (~95 MiB),
|
||||
;;; comfortable for the bend RPC traffic shape. Fixed width means
|
||||
;;; recv-exact never has to find a delimiter.
|
||||
;;;
|
||||
;;; Lifted from `ecdsa/lumbda/fleet/wire.lsp` in www.foxhop.net — the
|
||||
;;; same protocol the lumbda fleet workers already speak. Should
|
||||
;;; promote to lumbda's stdlib once enough callers want it.
|
||||
;;;
|
||||
;;; License: AGPLv3 (lumbda's license; matches the contributing repo).
|
||||
|
||||
(define *wire-header-width* 8)
|
||||
|
||||
(define (zero-pad-left s width)
|
||||
"Left-pad string s with '0' chars until length = width."
|
||||
(let ((slen (string-length s)))
|
||||
(if (>= slen width)
|
||||
s
|
||||
(let loop ((acc s) (need (- width slen)))
|
||||
(if (= need 0)
|
||||
acc
|
||||
(loop (string-append "0" acc) (- need 1)))))))
|
||||
|
||||
(define (recv-exact sock n)
|
||||
"Read exactly n bytes from sock. Returns concatenated string on
|
||||
success, #f if peer closes before n bytes arrive."
|
||||
(let loop ((acc "") (remaining n))
|
||||
(cond
|
||||
((= remaining 0) acc)
|
||||
(else
|
||||
(let ((chunk (tcp-recv sock remaining)))
|
||||
(cond
|
||||
((eq? chunk #f) #f)
|
||||
((= (string-length chunk) 0) #f) ; peer closed
|
||||
(else
|
||||
(loop (string-append acc chunk)
|
||||
(- remaining (string-length chunk))))))))))
|
||||
|
||||
(define (wire-send sock sexp)
|
||||
"Frame sexp as <8-digit-length><payload>, send over sock."
|
||||
(let* ((payload (write-to-string sexp))
|
||||
(plen (string-length payload))
|
||||
(header (zero-pad-left (number->string plen) *wire-header-width*))
|
||||
(frame (string-append header payload)))
|
||||
(tcp-send sock frame)))
|
||||
|
||||
(define (wire-recv sock)
|
||||
"Read one framed S-expression. Returns parsed sexp on success,
|
||||
#f on peer-closed / malformed header."
|
||||
(let ((header (recv-exact sock *wire-header-width*)))
|
||||
(cond
|
||||
((eq? header #f) #f)
|
||||
(else
|
||||
(let ((plen (string->number header)))
|
||||
(cond
|
||||
((or (eq? plen #f) (< plen 0)) #f)
|
||||
(else
|
||||
(let ((payload (recv-exact sock plen)))
|
||||
(cond
|
||||
((eq? payload #f) #f)
|
||||
(else (read-from-string payload)))))))))))
|
||||
Loading…
Add table
Add a link
Reference in a new issue