perf(buffer): O(1) store/evict via per-priority FIFO queues (P2)
ober
5a117295fb246d40e19e960735f49abdb5b6e714
--- a/jsecmon/buffer.ss +++ b/jsecmon/buffer.ss @@ -41,11 +41,16 @@ ;; A buffered event. payload is an opaque (already-encrypted) bytevector. (defstruct stored-event (seq timestamp-ms severity payload)) - ;; The buffer: events is a FIFO list (oldest first); seq is the next number. - (defstruct ebuffer (events max-size seq)) + ;; The buffer: events live in 4 FIFO queues, one per eviction-priority 0..3, + ;; so push and priority eviction are O(1) instead of rebuilding a single list + ;; on every store (the old (append evicted (list se)) was O(n) per store and + ;; O(n^2) across a burst). `seq` is the next number and `len` the live count. + ;; Each queue is a cell holding #f (empty) or a (head . tail) pair of cons cells. + (defstruct ebuffer (queues max-size seq len)) - ;; defstruct gives the raw 3-arg make-ebuffer; the public ctor takes just the cap. - (def (make-buffer max-size) (make-ebuffer '() max-size 0)) + ;; defstruct gives the raw 4-arg make-ebuffer; the public ctor takes just the cap. + (def (make-buffer max-size) + (make-ebuffer (vector (box #f) (box #f) (box #f) (box #f)) max-size 0 0)) ;; ── priority from event type (mirrors ring.rs event_severity_u8) ───────────── ;; NB: this is the eviction priority 0..3, which is its OWN table — not the @@ -87,48 +92,87 @@ (bytevector-u8-ref data 16) pay)))))) + ;; ── per-priority FIFO queue cells (O(1) push-back / pop-front) ─────────────── + (def (qpush! qbox se) + (let* ((q (unbox qbox)) (cell (cons se '()))) + (set-box! qbox (if q + (begin (set-cdr! (cdr q) cell) (cons (car q) cell)) + (cons cell cell))))) + (def (qpop! qbox) + (let ((q (unbox qbox))) + (and q (let* ((head (car q)) (se (car head)) (rest (cdr head))) + (set-box! qbox (and (pair? rest) (cons rest (cdr q)))) + se)))) + (def (qpeek qbox) + (let ((q (unbox qbox))) (and q (caar q)))) + (def (qlist qbox) + (let ((q (unbox qbox))) (if q (car q) '()))) + + ;; Merge two seq-ordered lists into one seq-ordered list. Each priority queue + ;; is already in insertion (seq) order, so a pairwise merge yields global order. + (def (merge-2 a b) + (let loop ((a a) (b b) (acc '())) + (cond + ((null? a) (if (null? acc) b (append (reverse acc) b))) + ((null? b) (if (null? acc) a (append (reverse acc) a))) + ((< (stored-event-seq (car a)) (stored-event-seq (car b))) + (loop (cdr a) b (cons (car a) acc))) + (else (loop a (cdr b) (cons (car b) acc)))))) + (def (all-events-sorted buf) + (let ((qs (ebuffer-queues buf))) + (merge-2 (merge-2 (qlist (vector-ref qs 0)) (qlist (vector-ref qs 1))) + (merge-2 (qlist (vector-ref qs 2)) (qlist (vector-ref qs 3)))))) + ;; ── eviction ───────────────────────────────────────────────────────────────── - ;; Drop the first (oldest) event of the lowest present severity; if none below - ;; critical, drop the oldest (front). Returns the trimmed list. - (def (remove-first pred lst) - (let loop ((xs lst) (acc '())) - (cond ((null? xs) (reverse acc)) ;; not found: unchanged - ((pred (car xs)) (append (reverse acc) (cdr xs))) - (else (loop (cdr xs) (cons (car xs) acc)))))) - (def (evict-lowest-priority lst) - (let loop ((target 0)) - (cond ((>= target 3) (if (null? lst) lst (cdr lst))) ;; pop_front (oldest critical) - ((find (lambda (e) (= (stored-event-severity e) target)) lst) - (remove-first (lambda (e) (= (stored-event-severity e) target)) lst)) - (else (loop (+ target 1)))))) + ;; Drop the oldest event of the lowest present priority: pop the front of the + ;; lowest non-empty priority queue. O(1) — the front of a priority queue is the + ;; oldest event of that priority, and only priorities 0..3 exist (matching the + ;; old "oldest of lowest severity, oldest-critical last" behaviour). + (def (evict-one! buf) + (let ((qs (ebuffer-queues buf))) + (let loop ((s 0)) + (cond + ((>= s 4) (void)) + ((unbox (vector-ref qs s)) + (qpop! (vector-ref qs s)) + (ebuffer-len-set! buf (- (ebuffer-len buf) 1))) + (else (loop (+ s 1))))))) ;; ── store / poll / clear ───────────────────────────────────────────────────── (def (buffer-store! buf timestamp-ms severity payload) ;; -> assigned seq (let ((seq (ebuffer-seq buf))) (ebuffer-seq-set! buf (+ seq 1)) - (let ((se (make-stored-event seq timestamp-ms severity payload)) - (evicted (let loop ((evs (ebuffer-events buf))) - (if (>= (length evs) (ebuffer-max-size buf)) - (loop (evict-lowest-priority evs)) - evs)))) - (ebuffer-events-set! buf (append evicted (list se))) + (let ((se (make-stored-event seq timestamp-ms severity payload))) + (when (>= (ebuffer-len buf) (ebuffer-max-size buf)) + (evict-one! buf)) + (qpush! (vector-ref (ebuffer-queues buf) (max 0 (min 3 severity))) se) + (ebuffer-len-set! buf (+ (ebuffer-len buf) 1)) seq))) (def (buffer-events-after buf after-seq) - (filter (lambda (e) (> (stored-event-seq e) after-seq)) (ebuffer-events buf))) + (filter (lambda (e) (> (stored-event-seq e) after-seq)) (all-events-sorted buf))) (def (buffer-events-in-range buf start-ms end-ms) (filter (lambda (e) (and (>= (stored-event-timestamp-ms e) start-ms) (<= (stored-event-timestamp-ms e) end-ms))) - (ebuffer-events buf))) + (all-events-sorted buf))) - (def (buffer-len buf) (length (ebuffer-events buf))) - (def (buffer-empty? buf) (= (buffer-len buf) 0)) + (def (buffer-len buf) (ebuffer-len buf)) + (def (buffer-empty? buf) (= (ebuffer-len buf) 0)) (def (buffer-latest-seq buf) (ebuffer-seq buf)) - ;; Drop every front event with seq <= the delivered seq (FIFO ack). + ;; Drop every event with seq <= the delivered seq (FIFO ack). Each priority + ;; queue is in seq order, so we pop its front while it qualifies. + (def (clear-queue-before! qbox seq) + (let loop ((n 0)) + (let ((front (qpeek qbox))) + (if (and front (<= (stored-event-seq front) seq)) + (begin (qpop! qbox) (loop (+ n 1))) + n)))) (def (buffer-clear-before! buf seq) - (let loop ((evs (ebuffer-events buf))) - (if (and (pair? evs) (<= (stored-event-seq (car evs)) seq)) - (loop (cdr evs)) - (ebuffer-events-set! buf evs))))) + (let ((qs (ebuffer-queues buf))) + (ebuffer-len-set! buf (- (ebuffer-len buf) + (+ (clear-queue-before! (vector-ref qs 0) seq) + (clear-queue-before! (vector-ref qs 1) seq) + (clear-queue-before! (vector-ref qs 2) seq) + (clear-queue-before! (vector-ref qs 3) seq)))))))