;;; -*- Mode: Scheme -*-

;;;; Rendezvous Concurrency Abstraction
;;;; Synchronous Channels

;;; This code is written by Taylor Campbell and placed in the Public
;;; Domain.  All warranties are disclaimed.

(define-synchronized-record-type :channel
  (%make-channel priority readers writers)
  (priority)       ; PRIORITY is the only synchronized field
  channel?
  (priority channel-priority set-channel-priority!)
  (readers channel-readers)
  (writers channel-writers))

(define (make-channel)
  (%make-channel 1 (make-queue) (make-queue)))

(define (channel-send channel message)
  (synchronize (channel-send-rendezvous channel message)))

(define (channel-send-rendezvous channel message)
  (polling-rendezvous
   (lambda ()
     (cond ((dequeue-reader! channel)
            => (lambda (reader)
                 (values (let ((priority (channel-priority channel)))
                           (set-channel-priority! channel
                                                  (+ priority 1))
                           priority)
                         (lambda (prepare-revival)
                           (set-channel-priority! channel 1)
                           (prepare-revival reader message)
                           (values)))))
           (else
            (values #f
                    (lambda (suspension)
                      (enqueue! (channel-writers channel)
                                (cons suspension message)))))))))

(define (dequeue-reader! channel)
  (let ((readers (channel-readers channel)))
    (let loop ()
      (if (queue-empty? readers)
          #f
          (let ((reader (dequeue! readers)))
            (if (suspension-revived? reader)
                (loop)
                reader))))))

(define (channel-receive channel)
  (synchronize (channel-receive-rendezvous channel)))

(define (channel-receive-rendezvous channel)
  (polling-rendezvous
   (lambda ()
     (cond ((dequeue-writer! channel)
            => (lambda (suspension.message)
                 (values (let ((priority (channel-priority channel)))
                           (set-channel-priority! channel
                                                  (+ priority 1))
                           priority)
                         (lambda (prepare-revival)
                           (set-channel-priority! channel 1)
                           (prepare-revival (car suspension.message))
                           (cdr suspension.message)))))
           (else
            (values #f
                    (lambda (suspension)
                      (enqueue! (channel-readers channel)
                                suspension))))))))

(define (dequeue-writer! channel)
  (let ((writers (channel-writers channel)))
    (let loop ()
      (if (queue-empty? writers)
          #f
          (let ((suspension.message (dequeue! writers)))
            (if (suspension-revived? (car suspension.message))
                (loop)
                suspension.message))))))