feat(capy): add deque and channel libraries

Author: playX18

lib/capy/channel.sls

@@ -0,0 +1,162 @@
+(library (capy channel)
+  (export
+    closed-channel-violation?
+    channel-close
+    channel-send
+    channel-recv
+    channel-closed? 
+    make-channel
+
+  )
+  (import (rnrs)
+          (capy deque)
+          (core threading))
+
+  (define-record-type slot 
+    (fields 
+      value))
+
+  (define-condition-type &closed-channel &violation 
+    make-closed-channel-violation
+    closed-channel-violation?)
+
+  (define (channel-closed-violation who msg)
+    (let ((c (condition 
+              (make-closed-channel-violation)
+              (make-message-condition msg))))
+      (raise 
+        (if who 
+          (condition c (make-who-condition who))
+          c))))
+
+  ;; A Go-like channel.
+  ;; - When `cap` is zero: unbuffered
+  ;; - When `cap` is positive: buffered
+  ;; Close semantics: 
+  ;; - close() forbids further sends (raises <closed-channel-condition> on send)
+  ;; - recv() returns value, `channel-closed` singleton when closed and drained, blocks otherwise.
+  (define-record-type channel 
+    (fields 
+      lock
+      (mutable %closed?) 
+      buf 
+      (mutable slot)
+      (mutable recv-waiting)
+      (mutable send-waiting)
+      cv-send
+      cv-recv 
+      cap)
+    (protocol 
+      (lambda (new)
+        (lambda (cap)
+          (new 
+            (make-mutex)
+            #f 
+            (make-deque cap) 
+            #f 
+            0 
+            0 
+            (make-condition) 
+            (make-condition)
+            cap)))))
+
+  (define chan (make-channel 0))
+
+  (define (channel-close chan)
+    (with-mutex (channel-lock chan)
+      (channel-%closed?-set! chan #t)
+      (condition-broadcast (channel-cv-recv chan))
+      (condition-broadcast (channel-cv-send chan))))
+  
+  (define (channel-send chan v)
+    "Send a value `v` to the channel `chan`.
+    - Blocks until it can be delivered (buffer space or rendezvous).
+    - Raises channel-closed-violation if the channel is closed."
+
+    (with-mutex (channel-lock chan)
+      (when (channel-%closed? chan)
+        (channel-closed-violation 'send "send on closed channel"))
+      (cond 
+        [(= (channel-cap chan) 0) ; unbounded send
+          (channel-send-waiting-set! chan (+ 1 (channel-send-waiting chan)))
+          ;; rendezvous: wait for a receiver, then handoff via slot.
+          (let loop () 
+            (when (and (not (channel-%closed? chan))
+                       (= (channel-recv-waiting chan) 0))
+
+              (condition-wait (channel-cv-send chan) (channel-lock chan))
+              (loop)))
+          (channel-send-waiting-set! chan 
+                                     (- (channel-send-waiting chan) 1))
+          ;; somebody might've closed the channel while we waited
+          
+          (when (channel-%closed? chan)
+            (channel-closed-violation 'send "send on closed channel"))
+          
+          (channel-slot-set! chan (make-slot v))
+          
+          (condition-signal (channel-cv-recv chan))
+          
+          
+          ;; wait until receiver takes it (slot becomes #f)
+          (let wait-loop ()
+            (when (and (not (channel-%closed? chan))
+                       (channel-slot chan))
+
+              (condition-wait (channel-cv-send chan) (channel-lock chan))
+              (wait-loop)))]
+        [else 
+          ;; buffered: wait for space
+          (let loop () 
+            (when (and (not (channel-%closed? chan))
+                       (= (deque-length (channel-buf chan))
+                          (channel-cap chan)))
+
+              (condition-wait (channel-cv-send chan) (channel-lock chan))
+              (loop)))
+          (when (channel-%closed? chan)
+            (channel-closed-violation 'send "send on closed channel"))
+          (deque-push-back! (channel-buf chan) v)
+          (condition-signal (channel-cv-recv chan))])))
+
+  (define (channel-recv chan)
+
+    (with-mutex (channel-lock chan)
+      (cond 
+        [(= (channel-cap chan) 0) ; unbuffered 
+          (channel-recv-waiting-set! chan (+ 1 (channel-recv-waiting chan)))
+          (condition-signal (channel-cv-send chan)) ; wake a sender
+          (let loop ()
+            (when (and (not (channel-%closed? chan))
+                       (not (channel-slot chan)))
+              (condition-wait (channel-cv-recv chan) (channel-lock chan))
+              (loop)))
+          (channel-recv-waiting-set! chan (- (channel-recv-waiting chan) 1))
+          
+          (cond 
+            [(channel-slot chan) => 
+              (lambda (slot) 
+                (channel-slot-set! chan #f)
+                (condition-signal (channel-cv-send chan))
+                (slot-value slot))]
+            [else 
+              #f])]
+        [else 
+          (let loop () 
+            (when (and (not (channel-%closed? chan))
+                       (= (deque-length (channel-buf chan)) 0))
+              
+              (condition-wait (channel-cv-recv chan) (channel-lock chan))
+              (loop)))
+          (if (> (deque-length (channel-buf chan)) 0)
+            (let ((v (deque-pop-front! (channel-buf chan))))
+              (condition-signal (channel-cv-send chan))
+              v)
+            #f)])))
+  
+  (define (channel-closed? chan)
+    "Returns #t if the channel is closed, #f otherwise.
+    
+    Note: a closed channel may still have buffered values to receive."
+    (with-mutex (channel-lock chan)
+      (channel-%closed? chan))))

lib/capy/deque.sls

@@ -0,0 +1,166 @@
+(library (capy deque)
+  (export
+    make-deque
+    deque
+    deque-capacity
+    deque-length
+    deque-push-back!
+    deque-push-front!
+    deque-pop-back!
+    deque-pop-front!
+    deque-ref
+    deque-set!
+    deque->vector
+    vector->deque
+    list->deque
+    deque-append!)
+  (import (rnrs)
+          (core struct)
+          (only (scheme base) vector-copy)
+          (only (scheme r5rs) modulo)
+          (only (capy) register-tuple-printer! print))
+
+  (define-struct %deque 
+    (buf head len))
+
+  (register-tuple-printer! 
+    'type:%deque
+    (lambda (obj port quote?)
+      (display "#<deque " port)
+      (let ((len (deque-length obj))
+            (last (- (deque-length obj) 1)))
+        (let loop ((i 0))
+          (when (< i len)
+            (print (deque-ref obj i) port quote?)
+            (when (< i last)
+              (display " " port))
+            (loop (+ i 1)))))
+      (display ">" port)))
+
+
+  (define (make-deque capacity)
+    (let ((initial-capacity (if (>= capacity 1)
+                                capacity
+                                8)))
+      (make-%deque (make-vector initial-capacity) 0 0)))
+
+  (define (deque . args)
+    (let ((dq (make-deque (max 8 (length args)))))
+      (for-each (lambda (v) (deque-push-back! dq v)) args)
+      dq))
+
+  (define (grow! dq)
+    (let* ((old-buf (%deque-buf dq))
+           (old-cap (vector-length old-buf))
+           (new-cap (* 2 old-cap))
+           (new-buf (make-vector new-cap)))
+      ;; Copy elements linearly to the new buffer
+      (do ((i 0 (+ i 1)))
+          ((= i (deque-len dq)))
+        (vector-set! new-buf i
+                     (vector-ref old-buf
+                                 (modulo (+ (deque-head dq) i) old-cap))))
+      (deque-buf-set! dq new-buf)
+      (deque-head-set! dq 0)))
+  (define (deque-capacity dq)
+    (vector-length (%deque-buf dq)))
+  (define (deque-length dq)
+    (%deque-len dq))
+
+  (define (deque-push-back! dq value)
+    (when (= (%deque-len dq) (deque-capacity dq))
+      (grow! dq))
+    (let* ((buf (%deque-buf dq))
+           (head (%deque-head dq))
+           (len (%deque-len dq))
+           (cap (vector-length buf))
+           (tail-index (modulo (+ head len) cap)))
+      (vector-set! buf tail-index value)
+      (%deque-len-set! dq (+ len 1))))
+  
+  (define (deque-push-front! dq value)
+    (when (= (%deque-len dq) (deque-capacity dq))
+      (grow! dq))
+    (let* ((buf (%deque-buf dq))
+           (head (%deque-head dq))
+           (len (%deque-len dq))
+           (cap (vector-length buf))
+           (new-head (modulo (+ head (- cap 1)) cap)))
+      (vector-set! buf new-head value)
+      (%deque-head-set! dq new-head)
+      (%deque-len-set! dq (+ len 1))))
+  
+  (define (deque-pop-back! dq)
+    (unless (> (%deque-len dq) 0)
+      (error 'deque-pop-back! "empty deque"))
+    
+    (let* ((buf (%deque-buf dq))
+           (head (%deque-head dq))
+           (len (%deque-len dq))
+           (cap (vector-length buf))
+           (tail-index (modulo (+ head (- len 1)) cap))
+           (value (vector-ref buf tail-index)))
+      (%deque-len-set! dq (- len 1))
+      value))
+  
+  (define (deque-pop-front! dq)
+    (unless (> (%deque-len dq) 0)
+      (error 'deque-pop-front! "empty deque"))
+    
+    (let* ((buf (%deque-buf dq))
+           (head (%deque-head dq))
+           (len (%deque-len dq))
+           (cap (vector-length buf))
+           (value (vector-ref buf head))
+           (new-head (modulo (+ head 1) cap)))
+      (%deque-head-set! dq new-head)
+      (%deque-len-set! dq (- len 1))
+      value))
+  
+  (define (deque-ref dq index)
+    (unless (and (>= index 0) (< index (%deque-len dq)))
+      (error 'deque-ref "index out of bounds"))
+    (let* ((buf (%deque-buf dq))
+           (head (%deque-head dq))
+           (cap (vector-length buf))
+           (real-index (modulo (+ head index) cap)))
+      (vector-ref buf real-index)))
+  
+  (define (deque-set! dq index value)
+    (unless (and (>= index 0) (< index (%deque-len dq)))
+      (error 'deque-set! "index out of bounds"))
+    (let* ((buf (%deque-buf dq))
+           (head (%deque-head dq))
+           (cap (vector-length buf))
+           (real-index (modulo (+ head index) cap)))
+      (vector-set! buf real-index value)))
+      
+  (define (deque->vector dq)
+    (let* ((buf (%deque-buf dq))
+           (head (%deque-head dq))
+           (len (%deque-len dq))
+           (cap (vector-length buf))
+           (result (make-vector len)))
+      (do ((i 0 (+ i 1)))
+          ((= i len) result)
+        (vector-set! result i
+                     (vector-ref buf
+                                 (modulo (+ head i) cap))))))
+
+  (define (vector->deque vec)
+    (let* ((len (vector-length vec))
+           (dq (make-deque len)))
+      (do ((i 0 (+ i 1)))
+          ((= i len) dq)
+        (deque-push-back! dq (vector-ref vec i)))))
+
+  (define (list->deque lst)
+    (let ((dq (make-deque (max 8 (length lst)))))
+      (for-each (lambda (v) (deque-push-back! dq v)) lst)
+      dq))
+
+  (define (deque-append! dq1 dq2)
+    (do ((i 0 (+ i 1)))
+        ((= i (deque-length dq2)))
+      (deque-push-back! dq1 (deque-ref dq2 i)))))
+