log+cli: serialize concurrent log + stdout writes
ober
687b05a6ace539556fa3bc2e066b49bf1db4c33f
--- a/src/jcode/core/log.ss +++ b/src/jcode/core/log.ss @@ -12,7 +12,12 @@ close-trace-log! err->string) -(import :std/misc/string) +(import :std/misc/string + ;; Use Chez's raw mutex directly so `with-mutex` works on it. + ;; The prelude shadows make-mutex with a Gerbil-wrapped variant + ;; that with-mutex (a Chez macro) cannot operate on. + (rename (only (chezscheme) make-mutex) + (make-mutex chez-make-mutex))) (def *log-level* (make-parameter 'info)) @@ -28,6 +33,14 @@ (def *trace-port* #f) +;; Global mutex serializing all log writes. Without this, concurrent +;; fprintf from multiple threads (worker tool threads + main + watchdog) +;; can interleave inside the port buffer and -- in some Chez builds -- +;; deadlock when port-internal locks are taken in different orders by +;; different threads. Serializing makes log output ordered AND avoids +;; that deadlock. +(def *log-mutex* (chez-make-mutex)) + (def (tracing?) (and *trace-port* #t)) (def (pad3 n) @@ -66,10 +79,11 @@ (def (log-at level label name msg data) (let* ((line (format "[~a] ~a: ~a" label name msg)) (suffix (if (null? data) "" (string-append " " (format-alist data))))) - (when (level-enabled? level (*log-level*)) - (fprintf (current-error-port) "~a~a~n" line suffix)) - (when *trace-port* - (fprintf *trace-port* "~a ~a~a~n" (trace-timestamp) line suffix)))) + (with-mutex *log-mutex* + (when (level-enabled? level (*log-level*)) + (fprintf (current-error-port) "~a~a~n" line suffix)) + (when *trace-port* + (fprintf *trace-port* "~a ~a~a~n" (trace-timestamp) line suffix))))) (def (level-enabled? level min-level) (case level @@ -100,4 +114,5 @@ (let* ((data (if (null? rest) '() (car rest))) (line (format "[TRACE] ~a: ~a" name msg)) (suffix (if (null? data) "" (string-append " " (format-alist data))))) - (fprintf *trace-port* "~a ~a~a~n" (trace-timestamp) line suffix)))) + (with-mutex *log-mutex* + (fprintf *trace-port* "~a ~a~a~n" (trace-timestamp) line suffix))))) --- a/src/jcode/ui/cli.ss +++ b/src/jcode/ui/cli.ss @@ -23,7 +23,12 @@ :jcode/ui/tui :jcode/ui/serve :jerboa/core - :jerboa/runtime) + :jerboa/runtime + ;; Use Chez's raw mutex (with-mutex requires it). The prelude's + ;; make-mutex returns a Gerbil-wrapped mutex that with-mutex + ;; can't operate on. + (rename (only (chezscheme) make-mutex) + (make-mutex chez-make-mutex))) (def logger (make-logger "cli")) (def *version* "0.1.0") @@ -402,29 +407,37 @@ EXAMPLES: (def *md-in-code-block* #f) (def *backtick* (integer->char 96)) +;; Serializes md-* writes (stdout printf + buffer mutation). Both the +;; main thread (streaming tokens) and worker tool threads (tool-indicator +;; cb) can hit these — without a lock, concurrent printf + set! produces +;; garbled output and can deadlock against Chez's port-internal locks. +(def *md-mutex* (chez-make-mutex)) + (def (md-reset!) (set! *md-line-buf* "") (set! *md-in-code-block* #f)) (def (md-stream-token token) - (set! *md-line-buf* (string-append *md-line-buf* token)) - (let flush-lines () - (let ((idx (string-contains *md-line-buf* "\n"))) - (when idx - (let ((line (substring *md-line-buf* 0 idx)) - (rest (substring *md-line-buf* (+ idx 1) (string-length *md-line-buf*)))) - (md-render-line line) - (newline) - (flush-output-port (current-output-port)) - (set! *md-line-buf* rest) - (flush-lines)))))) + (with-mutex *md-mutex* + (set! *md-line-buf* (string-append *md-line-buf* token)) + (let flush-lines () + (let ((idx (string-contains *md-line-buf* "\n"))) + (when idx + (let ((line (substring *md-line-buf* 0 idx)) + (rest (substring *md-line-buf* (+ idx 1) (string-length *md-line-buf*)))) + (md-render-line line) + (newline) + (flush-output-port (current-output-port)) + (set! *md-line-buf* rest) + (flush-lines))))))) (def (md-flush!) - (when (> (string-length *md-line-buf*) 0) - (md-render-line *md-line-buf*) - (newline) - (flush-output-port (current-output-port)) - (set! *md-line-buf* ""))) + (with-mutex *md-mutex* + (when (> (string-length *md-line-buf*) 0) + (md-render-line *md-line-buf*) + (newline) + (flush-output-port (current-output-port)) + (set! *md-line-buf* "")))) (def (md-render-line line) (let ((fence (string-append (string *backtick*) (string *backtick*) (string *backtick*)))) @@ -491,15 +504,20 @@ EXAMPLES: (def (tool-indicator event name args) (case event ((start) - ;; Flush any pending LLM text so it appears ABOVE the tool line + ;; Flush any pending LLM text so it appears ABOVE the tool line. + ;; md-flush! takes *md-mutex* on its own; we then re-take the same + ;; mutex to serialize the printf block below against streaming + ;; tokens from concurrent threads. Mutex acquisition order: always + ;; md-flush! -> *md-mutex*, never the reverse. (md-flush!) - (when (> (string-length *md-line-buf*) 0) - (newline)) - (printf "~n\x1b;[36m⟡ ~a\x1b;[0m" name) - (let ((summary (tool-args-summary name args))) - (when summary (printf " \x1b;[2m~a\x1b;[0m" summary))) - (newline) - (flush-output-port (current-output-port))) + (with-mutex *md-mutex* + (when (> (string-length *md-line-buf*) 0) + (newline)) + (printf "~n\x1b;[36m⟡ ~a\x1b;[0m" name) + (let ((summary (tool-args-summary name args))) + (when summary (printf " \x1b;[2m~a\x1b;[0m" summary))) + (newline) + (flush-output-port (current-output-port)))) ((end) (void)))) (def (tool-args-summary name args)