File size: 3,167 Bytes
3b70664 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | (in-package #:nekod.socket)
;;; Backpressure controls — prevent unbounded memory growth from infinite streams
(defstruct backpressure-config
(maximum-read-buffer (* 1024 1024) :type integer)
(maximum-event-size (* 256 1024) :type integer)
(maximum-header-size (* 64 1024) :type integer)
(consumer-queue-depth 1024 :type integer)
(high-water-mark 768 :type integer)
(low-water-mark 256 :type integer))
(defvar *backpressure* (make-backpressure-config))
(defstruct backpressure-state
(queue-size 0 :type integer)
(paused nil :type boolean)
(events-dropped 0 :type integer)
(total-bytes-buffered 0 :type integer))
(defvar *bp-state* (make-backpressure-state))
(defun check-event-size (size)
"Signal event-too-large if event exceeds maximum."
(when (> size (backpressure-config-maximum-event-size *backpressure*))
(error 'nekod:event-too-large
:message "Event exceeds maximum size"
:size size
:limit (backpressure-config-maximum-event-size *backpressure*))))
(defun check-header-size (size)
"Signal http-header-too-large if header exceeds maximum."
(when (> size (backpressure-config-maximum-header-size *backpressure*))
(error 'nekod:http-header-too-large
:message "Header exceeds maximum size"
:size size
:limit (backpressure-config-maximum-header-size *backpressure*))))
(defun should-pause-p ()
"Return T if consumer queue is at high water mark."
(>= (backpressure-state-queue-size *bp-state*)
(backpressure-config-high-water-mark *backpressure*)))
(defun should-resume-p ()
"Return T if consumer queue has drained to low water mark."
(<= (backpressure-state-queue-size *bp-state*)
(backpressure-config-low-water-mark *backpressure*)))
(defun enqueue-event (event-string)
"Attempt to enqueue event. Returns :accepted, :paused, or :dropped."
(let ((size (length event-string)))
(check-event-size size)
(cond
((>= (backpressure-state-queue-size *bp-state*)
(backpressure-config-consumer-queue-depth *backpressure*))
(incf (backpressure-state-events-dropped *bp-state*))
:dropped)
((should-pause-p)
(setf (backpressure-state-paused *bp-state*) t)
(incf (backpressure-state-queue-size *bp-state*))
(incf (backpressure-state-total-bytes-buffered *bp-state*) size)
:paused)
(t
(incf (backpressure-state-queue-size *bp-state*))
(incf (backpressure-state-total-bytes-buffered *bp-state*) size)
(when (and (backpressure-state-paused *bp-state*) (should-resume-p))
(setf (backpressure-state-paused *bp-state*) nil))
:accepted))))
(defun dequeue-event ()
"Mark one event as consumed."
(when (> (backpressure-state-queue-size *bp-state*) 0)
(decf (backpressure-state-queue-size *bp-state*))
(when (and (backpressure-state-paused *bp-state*) (should-resume-p))
(setf (backpressure-state-paused *bp-state*) nil))
t))
(defun reset-backpressure ()
"Reset backpressure state."
(setf *bp-state* (make-backpressure-state)))
|